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

use std::collections::BTreeMap;
use std::fs;
use std::io::{self, BufRead, BufReader, Write};
use std::net::TcpStream;
use std::path::Path;
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};

use anyhow::{anyhow, Result};
use disasmer_core::{DebugRuntimeState, Digest, ProcessId, TaskId};
use disasmer_node::WasmtimeTaskRuntime;
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,
    locals_ref: i64,
    wasm_locals_ref: i64,
    args_ref: i64,
    runtime_ref: i64,
    output_ref: i64,
    target_ref: i64,
    vfs_ref: i64,
    command_ref: i64,
    task: TaskId,
    name: String,
    line: i64,
    state: DebugRuntimeState,
    freeze_supported: bool,
    recent_output: Vec<String>,
    stdout_bytes: u64,
    stderr_bytes: u64,
    stdout_tail: String,
    stderr_tail: String,
    stdout_truncated: bool,
    stderr_truncated: bool,
}

#[derive(Clone)]
struct AdapterState {
    process: ProcessId,
    epoch: u64,
    entry: String,
    project: String,
    source_path: String,
    runtime_backend: RuntimeBackend,
    operator_endpoint: String,
    tenant: String,
    project_id: String,
    actor_user: String,
    session_mode: DapSessionMode,
    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,
    LiveServices,
}

#[derive(Clone, Debug, PartialEq, Eq)]
enum DapSessionMode {
    Launch,
    Attach,
}

impl RuntimeBackend {
    fn description(&self) -> &'static str {
        match self {
            RuntimeBackend::Simulated => "simulated debug adapter state",
            RuntimeBackend::LocalServices => "local services",
            RuntimeBackend::LiveServices => "live services",
        }
    }
}

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,
            source_path: "src/build.rs".to_owned(),
            runtime_backend: RuntimeBackend::Simulated,
            operator_endpoint: "disasmer.michelpaulissen.com:9443".to_owned(),
            tenant: "tenant".to_owned(),
            project_id: "project".to_owned(),
            actor_user: "dap".to_owned(),
            session_mode: DapSessionMode::Launch,
            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,
        operator_endpoint: String,
        tenant: String,
        project_id: String,
        actor_user: String,
        source_path: Option<String>,
    ) {
        self.process = process_id(&project, &entry);
        self.entry = entry;
        self.source_path = source_path.unwrap_or_else(|| infer_source_path(&project));
        self.project = project;
        self.runtime_backend = runtime_backend;
        self.operator_endpoint = normalize_operator_endpoint(&operator_endpoint);
        self.tenant = tenant;
        self.project_id = project_id;
        self.actor_user = actor_user;
        self.session_mode = DapSessionMode::Launch;
        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) {
        if !record.placed_task_launched {
            self.command_status = format!(
                "coordinator-side virtual process started through {}; no capability task launched yet",
                self.runtime_backend.description()
            );
            self.last_task_failed = false;
            self.runtime_artifact_path = None;
            self.runtime_event_count = record.event_count;
            if let Some(thread) = self.threads.get_mut(&MAIN_THREAD) {
                thread.recent_output.push(format!(
                    "coordinator-side process started through {}",
                    self.runtime_backend.description()
                ));
                thread
                    .recent_output
                    .push("a Command-capability virtual task will need a capable node when the process reaches it".to_owned());
            }
            let _ = write_view_state(self, &record);
            return;
        }

        let task_failed = record.status_code != Some(0);
        self.command_status = format!(
            "{} through {} with status {:?}",
            if task_failed { "failed" } else { "completed" },
            self.runtime_backend.description(),
            record.status_code
        );
        self.last_task_failed = task_failed;
        self.runtime_artifact_path = record.artifact_path.clone();
        self.runtime_event_count = record.event_count;
        if let Some(thread) = self.threads.get_mut(&LINUX_THREAD) {
            thread.stdout_bytes = record.stdout_bytes;
            thread.stderr_bytes = record.stderr_bytes;
            thread.stdout_tail = record.stdout_tail.clone();
            thread.stderr_tail = record.stderr_tail.clone();
            thread.stdout_truncated = record.stdout_truncated;
            thread.stderr_truncated = record.stderr_truncated;
            thread.recent_output.push(format!(
                "attached 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
                ));
            }
        }
        let _ = write_view_state(self, &record);
    }
}

#[derive(Clone, Debug)]
struct RuntimeLaunchRecord {
    coordinator: String,
    node: String,
    node_report: Value,
    task_events: Value,
    placed_task_launched: bool,
    status_code: Option<i32>,
    stdout_bytes: u64,
    stderr_bytes: u64,
    stdout_tail: String,
    stderr_tail: String,
    stdout_truncated: bool,
    stderr_truncated: bool,
    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,
        locals_ref: 5000 + id,
        wasm_locals_ref: 9000 + id,
        args_ref: 2000 + id,
        runtime_ref: 3000 + id,
        output_ref: 4000 + id,
        target_ref: 6000 + id,
        vfs_ref: 7000 + id,
        command_ref: 8000 + id,
        task: TaskId::from(task),
        name: name.to_owned(),
        line,
        state: DebugRuntimeState::Running,
        freeze_supported: true,
        recent_output: vec![format!("{name}: waiting")],
        stdout_bytes: 0,
        stderr_bytes: 0,
        stdout_tail: String::new(),
        stderr_tail: String::new(),
        stdout_truncated: false,
        stderr_truncated: false,
    }
}

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" | "attach" => {
                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,
                    Some("live-services") => RuntimeBackend::LiveServices,
                    _ => RuntimeBackend::Simulated,
                };
                let operator_endpoint = args
                    .get("operatorEndpoint")
                    .and_then(Value::as_str)
                    .unwrap_or("disasmer.michelpaulissen.com:9443")
                    .to_owned();
                let tenant = args
                    .get("tenant")
                    .and_then(Value::as_str)
                    .unwrap_or("tenant")
                    .to_owned();
                let project_id = args
                    .get("projectId")
                    .or_else(|| args.get("project_id"))
                    .and_then(Value::as_str)
                    .unwrap_or("project")
                    .to_owned();
                let actor_user = args
                    .get("actorUser")
                    .or_else(|| args.get("actor_user"))
                    .and_then(Value::as_str)
                    .unwrap_or("dap")
                    .to_owned();
                let source_path = args
                    .get("sourcePath")
                    .or_else(|| args.get("source_path"))
                    .and_then(Value::as_str)
                    .map(str::to_owned);
                state.configure_launch(
                    project,
                    state.entry.clone(),
                    runtime_backend,
                    operator_endpoint,
                    tenant,
                    project_id,
                    actor_user,
                    source_path,
                );
                if command == "attach" {
                    state.session_mode = DapSessionMode::Attach;
                    state.command_status =
                        format!("attached to existing virtual process {}", state.process);
                }
                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.session_mode == DapSessionMode::Attach {
                    state.command_status =
                        format!("attached to existing virtual process {}", state.process);
                } else {
                    match 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;
                            }
                        },
                        RuntimeBackend::LiveServices => match run_live_services_runtime(&state) {
                            Ok(record) => {
                                state.apply_runtime_record(record);
                            }
                            Err(err) => {
                                writer.error_response(
                                    &request,
                                    format!("live services runtime launch failed: {err:#}"),
                                )?;
                                continue;
                            }
                        },
                        RuntimeBackend::Simulated => {
                            state.command_status =
                                "running under simulated debug adapter state".to_owned();
                        }
                    }
                }
                writer.response(&request, true, json!({}))?;
                let session_message = if state.session_mode == DapSessionMode::Attach {
                    format!(
                        "Attached to Disasmer virtual process {} with {:?} runtime\n",
                        state.process, state.runtime_backend
                    )
                } else {
                    format!(
                        "Disasmer bundle refreshed for entry `{}`; starting virtual process {} with {:?} runtime\n",
                        state.entry, state.process, state.runtime_backend
                    )
                };
                writer.output("console", session_message)?;
                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);
                let source_path = stack_source_path(&state);
                writer.response(
                    &request,
                    true,
                    json!({
                        "stackFrames": [{
                            "id": thread.frame_id,
                            "name": format!("{}::run", thread.name),
                                "line": thread.line,
                                "column": 1,
                                "source": {
                                    "name": source_name(&source_path),
                                    "path": source_path,
                                    "presentationHint": "normal"
                                }
                            }],
                            "totalFrames": 1
                    }),
                )?;
            }
            "source" => match source_response(&state, &request) {
                Ok(body) => writer.response(&request, true, body)?,
                Err(err) => writer.error_response(&request, err.to_string())?,
            },
            "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": "Source Locals",
                                "variablesReference": thread.locals_ref,
                                "expensive": false,
                                "presentationHint": "locals"
                            },
                            {
                                "name": "Wasm Frame Locals",
                                "variablesReference": thread.wasm_locals_ref,
                                "expensive": false,
                                "presentationHint": "locals"
                            },
                            {
                                "name": "Task Args and Handles",
                                "variablesReference": thread.args_ref,
                                "expensive": false,
                                "presentationHint": "arguments"
                            },
                            {
                                "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" => {
                let thread_id = request_thread_id(&request).unwrap_or(LINUX_THREAD);
                let next_breakpoint = next_breakpoint_after(&state, thread_id);
                for thread in state.threads.values_mut() {
                    if thread.state == DebugRuntimeState::Frozen {
                        thread.state = DebugRuntimeState::Running;
                    }
                }
                writer.response(&request, true, json!({ "allThreadsContinued": true }))?;
                writer.event(
                    "continued",
                    json!({
                        "threadId": thread_id,
                        "allThreadsContinued": true,
                    }),
                )?;
                if let Some((stopped_thread, line)) = next_breakpoint {
                    if state.runtime_backend == RuntimeBackend::LiveServices
                        && stopped_thread == LINUX_THREAD
                        && state.runtime_event_count == 0
                    {
                        match launch_live_placed_task(&state) {
                            Ok(record) => state.apply_runtime_record(record),
                            Err(err) => {
                                let message = format!(
                                    "Command-capability virtual task placement failed after coordinator-side process start: {err:#}"
                                );
                                state.command_status = message.clone();
                                state.last_task_failed = true;
                                if let Some(thread) = state.threads.get_mut(&LINUX_THREAD) {
                                    thread.state = DebugRuntimeState::Failed(message.clone());
                                    thread.recent_output.push(message.clone());
                                }
                                writer.output("stderr", format!("{message}\n"))?;
                                writer.event(
                                    "stopped",
                                    json!({
                                        "reason": "exception",
                                        "description": "Disasmer capability task unavailable",
                                        "threadId": LINUX_THREAD,
                                        "allThreadsStopped": true,
                                    }),
                                )?;
                                continue;
                            }
                        }
                    }
                    match freeze_all_at_line(&mut state, stopped_thread, line, 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)?;
                        }
                    }
                }
            }
            "next" | "stepIn" | "stepOut" => {
                let thread_id = request_thread_id(&request).unwrap_or(LINUX_THREAD);
                let description = match command {
                    "next" => "step over",
                    "stepIn" => "step in",
                    "stepOut" => "step out",
                    _ => "step",
                };
                step_thread(&mut state, thread_id, description);
                writer.response(&request, true, json!({}))?;
                writer.event(
                    "stopped",
                    json!({
                        "reason": "step",
                        "description": format!("Disasmer Debug Epoch {description}"),
                        "threadId": thread_id,
                        "allThreadsStopped": 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> {
    let line = state
        .breakpoints
        .iter()
        .copied()
        .find(|line| thread_for_source_line(state, *line) == stopped_thread)
        .or_else(|| state.breakpoints.first().copied())
        .unwrap_or_else(|| state.threads[&stopped_thread].line);
    freeze_all_at_line(state, stopped_thread, line, forced_failure_thread)
}

fn freeze_all_at_line(
    state: &mut AdapterState,
    stopped_thread: i64,
    line: 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) {
        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()
        .map(|line| thread_for_source_line(state, *line))
        .unwrap_or(LINUX_THREAD)
}

fn next_breakpoint_after(state: &AdapterState, thread_id: i64) -> Option<(i64, i64)> {
    let current_line = state.threads.get(&thread_id)?.line;
    state
        .breakpoints
        .iter()
        .copied()
        .filter(|line| *line > current_line)
        .min()
        .map(|line| (thread_for_source_line(state, line), line))
}

fn thread_for_source_line(state: &AdapterState, line: i64) -> i64 {
    if let Some(function_name) = source_function_name_at_line(state, line) {
        let lower = function_name.to_ascii_lowercase();
        if lower.contains("linux") {
            return LINUX_THREAD;
        }
        if lower.contains("windows") {
            return WINDOWS_THREAD;
        }
        if lower.contains("package") {
            return PACKAGE_THREAD;
        }
        return MAIN_THREAD;
    }

    state
        .threads
        .values()
        .find(|thread| thread.line == line)
        .map(|thread| thread.id)
        .unwrap_or(MAIN_THREAD)
}

fn source_function_name_at_line(state: &AdapterState, line: i64) -> Option<String> {
    let source =
        fs::read_to_string(resolve_source_path(&state.project, &state.source_path)).ok()?;
    let lines = source.lines().collect::<Vec<_>>();
    let mut index = usize::try_from(line).ok()?.saturating_sub(1);
    if index >= lines.len() {
        index = lines.len().saturating_sub(1);
    }
    lines[..=index]
        .iter()
        .rev()
        .find_map(|line| parse_rust_function_name(line))
}

fn parse_rust_function_name(line: &str) -> Option<String> {
    let start = line.find("fn ")? + 3;
    let rest = &line[start..];
    let name = rest
        .chars()
        .take_while(|ch| ch.is_ascii_alphanumeric() || *ch == '_')
        .collect::<String>();
    (!name.is_empty()).then_some(name)
}

fn step_thread(state: &mut AdapterState, thread_id: i64, description: &str) {
    state.epoch += 1;
    let resolved_thread = if state.threads.contains_key(&thread_id) {
        thread_id
    } else {
        LINUX_THREAD
    };
    if let Some(thread) = state.threads.get_mut(&resolved_thread) {
        thread.state = DebugRuntimeState::Frozen;
        thread.line += 1;
        thread
            .recent_output
            .push(format!("debug epoch {} {description}", state.epoch));
    }
}

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

    if thread.locals_ref == reference {
        return json!({ "variables": source_locals_variables(state, thread) });
    }

    if thread.wasm_locals_ref == reference {
        return json!({ "variables": wasm_frame_local_variables(state, thread) });
    }

    if thread.args_ref == reference {
        return json!({
            "variables": [
                {
                    "name": "target",
                    "value": format!("Target {{ task = \"{}\", env = \"{}\" }}", thread.task, task_environment(thread)),
                    "variablesReference": thread.target_ref
                },
                {
                    "name": "task_arguments",
                    "value": task_arguments_value(thread),
                    "variablesReference": 0
                },
                {
                    "name": "return_value",
                    "value": return_value_for_thread(state, thread),
                    "variablesReference": 0
                },
                {
                    "name": "artifact",
                    "value": format!("Artifact {{ id = \"{}\" }}", artifact_handle_value(state, thread)),
                    "variablesReference": 0
                },
                {
                    "name": "source_snapshot",
                    "value": format!("SourceSnapshot {{ digest = \"source://local-checkout/{}\" }}", project_snapshot_suffix(&state.project)),
                    "variablesReference": 0
                },
                {
                    "name": "blob",
                    "value": format!("Blob {{ digest = \"{}\" }}", blob_digest(state, thread)),
                    "variablesReference": 0
                },
                {
                    "name": "vfs_mounts",
                    "value": "{ /vfs/artifacts, /vfs/sources, /vfs/blobs }",
                    "variablesReference": thread.vfs_ref
                },
                {
                    "name": "unavailable-local-diagnostic",
                    "value": "source-level locals that cannot be inspected keep task args and handles visible",
                    "variablesReference": 0
                }
            ]
        });
    }

    if thread.target_ref == reference {
        return json!({
            "variables": [
                {
                    "name": "task",
                    "value": thread.task.to_string(),
                    "variablesReference": 0
                },
                {
                    "name": "environment",
                    "value": task_environment(thread),
                    "variablesReference": 0
                },
                {
                    "name": "kind",
                    "value": if thread.id == MAIN_THREAD { "coordinator-side wasm entrypoint" } else { "capability task" },
                    "variablesReference": 0
                },
                {
                    "name": "required_capability",
                    "value": if thread.id == MAIN_THREAD { "none" } else { "Command" },
                    "variablesReference": 0
                }
            ]
        });
    }

    if thread.vfs_ref == reference {
        return json!({
            "variables": [
                {
                    "name": "/vfs/artifacts",
                    "value": artifact_handle_value(state, thread),
                    "variablesReference": 0
                },
                {
                    "name": "/vfs/sources",
                    "value": format!("source://local-checkout/{}", project_snapshot_suffix(&state.project)),
                    "variablesReference": 0
                },
                {
                    "name": "/vfs/blobs",
                    "value": blob_digest(state, thread),
                    "variablesReference": 0
                },
                {
                    "name": "durability",
                    "value": "best-effort until explicitly exported or stored",
                    "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
                },
                {
                    "name": "command_spec",
                    "value": command_spec_summary(state, thread),
                    "variablesReference": thread.command_ref
                },
                {
                    "name": "stdout_tail",
                    "value": output_tail_display(&thread.stdout_tail, thread.stdout_bytes, thread.stdout_truncated, "stdout"),
                    "variablesReference": 0
                },
                {
                    "name": "stderr_tail",
                    "value": output_tail_display(&thread.stderr_tail, thread.stderr_bytes, thread.stderr_truncated, "stderr"),
                    "variablesReference": 0
                }
            ]
        });
    }

    if thread.command_ref == reference {
        return json!({
            "variables": [
                {
                    "name": "program",
                    "value": command_program(thread),
                    "variablesReference": 0
                },
                {
                    "name": "args",
                    "value": command_args_summary(state, thread),
                    "variablesReference": 0
                },
                {
                    "name": "required_capability",
                    "value": if thread.id == MAIN_THREAD { "none" } else { "Command" },
                    "variablesReference": 0
                },
                {
                    "name": "status",
                    "value": state.command_status,
                    "variablesReference": 0
                }
            ]
        });
    }

    let mut variables = vec![
        json!({
            "name": "stdout_tail",
            "value": output_tail_display(&thread.stdout_tail, thread.stdout_bytes, thread.stdout_truncated, "stdout"),
            "variablesReference": 0
        }),
        json!({
            "name": "stderr_tail",
            "value": output_tail_display(&thread.stderr_tail, thread.stderr_bytes, thread.stderr_truncated, "stderr"),
            "variablesReference": 0
        }),
    ];
    variables.extend(
        thread
            .recent_output
            .iter()
            .enumerate()
            .map(|(index, output)| {
                json!({
                    "name": format!("line_{index}"),
                    "value": output,
                    "variablesReference": 0
                })
            }),
    );

    json!({ "variables": variables })
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct SourceLocal {
    name: String,
    line: i64,
    value: Option<String>,
}

fn source_locals_variables(state: &AdapterState, thread: &VirtualThread) -> Vec<Value> {
    let (locals, diagnostic) = source_locals_for_thread(state, thread);
    let mut variables = locals
        .into_iter()
        .map(|local| {
            let has_value = local.value.is_some();
            let value = local.value.unwrap_or_else(|| {
                format!(
                    "cannot be inspected: source-level local from line {} is known, but DWARF/runtime value snapshot is unavailable",
                    local.line
                )
            });
            json!({
                "name": local.name,
                "value": value,
                "type": if has_value { "disasmer-source-local" } else { "unavailable-local" },
                "variablesReference": 0
            })
        })
        .collect::<Vec<_>>();
    variables.push(json!({
        "name": "unavailable-local-diagnostic",
        "value": diagnostic,
        "type": "unavailable-local",
        "variablesReference": 0
    }));
    variables
}

const TASK_ADD_ONE_WAT: &str = r#"
(module
  (func (export "task_add_one") (param i32) (result i32)
    local.get 0
    i32.const 1
    i32.add))
"#;

fn wasm_frame_local_variables(state: &AdapterState, thread: &VirtualThread) -> Vec<Value> {
    let function_name = source_function_name_at_line(state, thread.line);
    if function_name.as_deref() != Some("task_add_one") {
        return vec![json!({
            "name": "wasm-local-diagnostic",
            "value": "no Wasm frame-local snapshot is available at this source location",
            "type": "unavailable-local",
            "variablesReference": 0
        })];
    }

    let probe = WasmtimeTaskRuntime::new().and_then(|runtime| {
        runtime.freeze_resume_i32_export_probe(TASK_ADD_ONE_WAT, "task_add_one", 41)
    });
    match probe {
        Ok(probe) => {
            let mut variables = probe
                .local_values
                .into_iter()
                .map(|(name, value)| {
                    json!({
                        "name": name,
                        "value": value,
                        "type": "wasm-frame-local",
                        "variablesReference": 0
                    })
                })
                .collect::<Vec<_>>();
            variables.push(json!({
                "name": "wasm-local-diagnostic",
                "value": "captured from Wasmtime guest-debug frame state in the product node runtime",
                "type": "diagnostic",
                "variablesReference": 0
            }));
            variables
        }
        Err(err) => vec![json!({
            "name": "wasm-local-diagnostic",
            "value": format!("cannot inspect Wasm frame locals: {err:#}"),
            "type": "unavailable-local",
            "variablesReference": 0
        })],
    }
}

fn source_locals_for_thread(
    state: &AdapterState,
    thread: &VirtualThread,
) -> (Vec<SourceLocal>, String) {
    let source_path = resolve_source_path(&state.project, &state.source_path);
    let Ok(source) = fs::read_to_string(&source_path) else {
        return (
            Vec::new(),
            format!(
                "cannot be inspected: source file `{}` is unavailable; task args and handles remain visible",
                source_path.display()
            ),
        );
    };
    let lines = source.lines().collect::<Vec<_>>();
    if lines.is_empty() {
        return (
            Vec::new(),
            "cannot be inspected: source file is empty; task args and handles remain visible"
                .to_owned(),
        );
    }
    let current = usize::try_from(thread.line)
        .ok()
        .and_then(|line| line.checked_sub(1))
        .unwrap_or(0)
        .min(lines.len() - 1);
    let function_start = lines[..=current]
        .iter()
        .rposition(|line| parse_rust_function_name(line).is_some())
        .unwrap_or(0);
    let mut locals = Vec::new();
    let mut runtime_values = BTreeMap::new();
    let mut index = function_start;
    while index <= current {
        let Some(name) = parse_let_binding_name(lines[index]) else {
            index += 1;
            continue;
        };

        let line = (index + 1) as i64;
        let mut statement = lines[index].to_owned();
        while !statement.contains(';') && index < current {
            index += 1;
            statement.push('\n');
            statement.push_str(lines[index]);
        }

        let runtime_value = infer_disasmer_source_local_value(state, &statement, &runtime_values);
        if let Some(value) = runtime_value.clone() {
            runtime_values.insert(name.clone(), value);
        }
        locals.push(SourceLocal {
            name,
            line,
            value: runtime_value.map(|value| value.display(state)),
        });
        index += 1;
    }
    let diagnostic = if locals.is_empty() {
        "cannot be inspected: no source-level `let` locals are in scope for this frame; task args and handles remain visible"
            .to_owned()
    } else if locals.iter().any(|local| local.value.is_some()) {
        "selected source-level Disasmer API locals are populated from virtual-thread runtime state; non-Disasmer locals remain best-effort until DWARF value snapshots are available"
            .to_owned()
    } else {
        "cannot be inspected: selected source-level local names are best-effort until DWARF/runtime value snapshots are available"
            .to_owned()
    };
    (locals, diagnostic)
}

fn parse_let_binding_name(line: &str) -> Option<String> {
    let rest = line.trim_start().strip_prefix("let ")?;
    let rest = rest
        .trim_start()
        .strip_prefix("mut ")
        .unwrap_or(rest)
        .trim_start();
    let name = rest
        .chars()
        .take_while(|ch| ch.is_ascii_alphanumeric() || *ch == '_')
        .collect::<String>();
    (!name.is_empty()).then_some(name)
}

#[derive(Clone, Debug, PartialEq, Eq)]
enum SourceLocalRuntimeValue {
    TaskHandle {
        task: TaskId,
        thread_id: i64,
        env: String,
        state: String,
    },
    ThreadId(i64),
    Artifact(String),
}

impl SourceLocalRuntimeValue {
    fn display(&self, _state: &AdapterState) -> String {
        match self {
            SourceLocalRuntimeValue::TaskHandle {
                task,
                thread_id,
                env,
                state,
            } => format!(
                "TaskHandle {{ task = \"{task}\", virtual_thread_id = {thread_id}, env = \"{env}\", state = {state} }}"
            ),
            SourceLocalRuntimeValue::ThreadId(thread_id) => thread_id.to_string(),
            SourceLocalRuntimeValue::Artifact(artifact) => {
                format!("Artifact {{ id = \"{artifact}\" }}")
            }
        }
    }
}

fn infer_disasmer_source_local_value(
    state: &AdapterState,
    statement: &str,
    runtime_values: &BTreeMap<String, SourceLocalRuntimeValue>,
) -> Option<SourceLocalRuntimeValue> {
    if let Some(task_function) = extract_call_argument(statement, "disasmer::spawn::task(") {
        let task_name = extract_string_argument(statement, ".name(");
        let spawned_thread =
            thread_for_spawn_statement(state, &task_function, task_name.as_deref())?;
        return Some(SourceLocalRuntimeValue::TaskHandle {
            task: spawned_thread.task.clone(),
            thread_id: spawned_thread.id,
            env: extract_env_name(statement)
                .unwrap_or_else(|| task_environment(spawned_thread).to_owned()),
            state: format!("{:?}", spawned_thread.state),
        });
    }

    if let Some(receiver) = receiver_before_method(statement, ".virtual_thread_id()") {
        if let Some(SourceLocalRuntimeValue::TaskHandle { thread_id, .. }) =
            runtime_values.get(&receiver)
        {
            return Some(SourceLocalRuntimeValue::ThreadId(*thread_id));
        }
    }

    if let Some(receiver) = receiver_before_method(statement, ".join()") {
        if let Some(SourceLocalRuntimeValue::TaskHandle { thread_id, .. }) =
            runtime_values.get(&receiver)
        {
            if let Some(thread) = state.threads.get(thread_id) {
                return Some(SourceLocalRuntimeValue::Artifact(artifact_handle_value(
                    state, thread,
                )));
            }
        }
    }

    None
}

fn extract_call_argument(statement: &str, marker: &str) -> Option<String> {
    let start = statement.find(marker)? + marker.len();
    let rest = &statement[start..];
    let end = rest.find(')')?;
    let value = rest[..end].trim();
    (!value.is_empty()).then_some(value.to_owned())
}

fn extract_string_argument(statement: &str, marker: &str) -> Option<String> {
    let start = statement.find(marker)? + marker.len();
    let rest = &statement[start..];
    let quoted = rest.strip_prefix('"')?;
    let end = quoted.find('"')?;
    Some(quoted[..end].to_owned())
}

fn extract_env_name(statement: &str) -> Option<String> {
    if statement.contains("windows_env") || statement.contains("env!(\"windows\")") {
        return Some("windows".to_owned());
    }
    if statement.contains("linux_env") || statement.contains("env!(\"linux\")") {
        return Some("linux".to_owned());
    }
    None
}

fn receiver_before_method(statement: &str, method: &str) -> Option<String> {
    let before = statement.split(method).next()?.trim_end();
    let receiver = before
        .chars()
        .rev()
        .take_while(|ch| ch.is_ascii_alphanumeric() || *ch == '_')
        .collect::<String>()
        .chars()
        .rev()
        .collect::<String>();
    (!receiver.is_empty()).then_some(receiver)
}

fn thread_for_spawn_statement<'a>(
    state: &'a AdapterState,
    task_function: &str,
    task_name: Option<&str>,
) -> Option<&'a VirtualThread> {
    if let Some(task_name) = task_name {
        if let Some(thread) = state
            .threads
            .values()
            .find(|thread| thread.name == task_name)
        {
            return Some(thread);
        }
    }
    let normalized = task_function.replace('_', "-");
    state
        .threads
        .values()
        .find(|thread| thread.task.as_str() == normalized)
}

fn task_environment(thread: &VirtualThread) -> &'static str {
    match thread.id {
        WINDOWS_THREAD => "windows",
        MAIN_THREAD => "coordinator",
        _ => "linux",
    }
}

fn task_arguments_value(thread: &VirtualThread) -> String {
    if thread.task.as_str() == "task_add_one" {
        "[input: i32]".to_owned()
    } else {
        "[]".to_owned()
    }
}

fn return_value_for_thread(state: &AdapterState, thread: &VirtualThread) -> String {
    match thread.id {
        LINUX_THREAD | PACKAGE_THREAD => {
            format!(
                "Artifact {{ id = \"{}\" }}",
                artifact_handle_value(state, thread)
            )
        }
        MAIN_THREAD => "not returned yet: coordinator-side workflow still running".to_owned(),
        WINDOWS_THREAD => {
            "not returned yet: Windows capability task waits for a capable node".to_owned()
        }
        _ => "not returned yet".to_owned(),
    }
}

fn artifact_handle_value(state: &AdapterState, thread: &VirtualThread) -> String {
    if thread.id == LINUX_THREAD {
        return state.runtime_artifact_path.clone().unwrap_or_else(|| {
            format!("artifact://{}/{}/app.tar.zst", state.process, thread.task)
        });
    }
    if thread.id == PACKAGE_THREAD {
        return "artifact://package/release.tar.zst".to_owned();
    }
    format!("artifact://{}/{}/app.tar.zst", state.process, thread.task)
}

fn blob_digest(state: &AdapterState, thread: &VirtualThread) -> String {
    Digest::from_parts([
        b"blob:v1".as_slice(),
        state.project.as_bytes(),
        thread.task.as_str().as_bytes(),
    ])
    .as_str()
    .to_owned()
}

fn command_spec_summary(state: &AdapterState, thread: &VirtualThread) -> String {
    format!(
        "{} {} ({})",
        command_program(thread),
        command_args_summary(state, thread),
        state.command_status
    )
}

fn command_program(thread: &VirtualThread) -> &'static str {
    if thread.id == MAIN_THREAD {
        "coordinator-side wasm"
    } else {
        "cargo"
    }
}

fn command_args_summary(state: &AdapterState, thread: &VirtualThread) -> String {
    if thread.id == MAIN_THREAD {
        return format!("entry={}", state.entry);
    }
    format!("test --quiet --manifest-path {}/Cargo.toml", state.project)
}

fn output_tail_display(tail: &str, bytes: u64, truncated: bool, stream: &str) -> String {
    if tail.is_empty() {
        return format!("no {stream} captured ({bytes} bytes)");
    }
    if truncated {
        format!("{tail}\n... truncated")
    } else {
        tail.to_owned()
    }
}

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_live_services_runtime(state: &AdapterState) -> Result<RuntimeLaunchRecord> {
    let coordinator = normalize_operator_endpoint(&state.operator_endpoint);
    let started = coordinator_request(
        &coordinator,
        json!({
            "type": "start_process",
            "tenant": state.tenant,
            "project": state.project_id,
            "process": state.process.as_str(),
        }),
    )?;
    let epoch = started
        .get("epoch")
        .and_then(Value::as_u64)
        .ok_or_else(|| anyhow!("coordinator did not report a process epoch"))?;

    Ok(RuntimeLaunchRecord {
        coordinator,
        node: "coordinator-side-process".to_owned(),
        node_report: json!({
            "coordinator_process": started,
            "coordinator_epoch": epoch,
            "capability_task": "not_launched",
        }),
        task_events: json!({ "events": [] }),
        placed_task_launched: false,
        status_code: None,
        stdout_bytes: 0,
        stderr_bytes: 0,
        stdout_tail: String::new(),
        stderr_tail: String::new(),
        stdout_truncated: false,
        stderr_truncated: false,
        artifact_path: None,
        event_count: 0,
    })
}

fn launch_live_placed_task(state: &AdapterState) -> Result<RuntimeLaunchRecord> {
    let coordinator = normalize_operator_endpoint(&state.operator_endpoint);
    let launch = coordinator_request(
        &coordinator,
        json!({
            "type": "launch_task",
            "tenant": state.tenant,
            "project": state.project_id,
            "actor_user": state.actor_user,
            "process": state.process.as_str(),
            "task": "compile-linux",
            "environment": null,
            "environment_digest": null,
            "required_capabilities": ["Command"],
            "dependency_cache": null,
            "source_snapshot": null,
            "required_artifacts": [],
            "quota_available": true,
            "policy_allowed": true,
            "command": "cargo",
            "command_args": [
                "test",
                "--quiet",
                "--manifest-path",
                Path::new(&state.project)
                    .join("Cargo.toml")
                    .to_string_lossy()
                    .into_owned()
            ],
            "artifact_path": "/vfs/artifacts/dap-output.txt",
        }),
    )
    .map_err(|err| {
        anyhow!(
            "coordinator-side process could not place virtual task with Command capability: {err:#}. Attach a capable node, for example `disasmer-node --worker --coordinator {} --tenant {} --project-id {} --node <node-id>`.",
            coordinator,
            state.tenant,
            state.project_id,
        )
    })?;
    let node = launch
        .get("placement")
        .and_then(|placement| placement.get("node"))
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow!("task launch did not report a worker node placement"))?
        .to_owned();
    let events = wait_for_placed_task_event(&coordinator, state, "compile-linux", &node)?;
    let event_count = events
        .get("events")
        .and_then(Value::as_array)
        .map(Vec::len)
        .unwrap_or(0);
    let event = events
        .get("events")
        .and_then(Value::as_array)
        .and_then(|items| {
            items.iter().find(|event| {
                event.get("task").and_then(Value::as_str) == Some("compile-linux")
                    && event.get("node").and_then(Value::as_str) == Some(node.as_str())
            })
        })
        .ok_or_else(|| anyhow!("worker node {node} did not record compile-linux completion"))?;
    let status_code = event
        .get("status_code")
        .and_then(Value::as_i64)
        .map(|value| value as i32);
    let stdout_bytes = event
        .get("stdout_bytes")
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let stderr_bytes = event
        .get("stderr_bytes")
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let artifact_path = event
        .get("artifact_path")
        .and_then(Value::as_str)
        .map(str::to_owned);
    let stdout_tail = event
        .get("stdout_tail")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned();
    let stderr_tail = event
        .get("stderr_tail")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned();
    let stdout_truncated = event
        .get("stdout_truncated")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let stderr_truncated = event
        .get("stderr_truncated")
        .and_then(Value::as_bool)
        .unwrap_or(false);

    Ok(RuntimeLaunchRecord {
        coordinator,
        node,
        node_report: json!({
            "task_launch": launch,
        }),
        task_events: events,
        placed_task_launched: true,
        status_code,
        stdout_bytes,
        stderr_bytes,
        stdout_tail,
        stderr_tail,
        stdout_truncated,
        stderr_truncated,
        artifact_path,
        event_count,
    })
}

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"))?;

    run_node_against_coordinator(
        state,
        repo,
        "cargo",
        &[
            "run",
            "-q",
            "-p",
            "disasmer-node",
            "--bin",
            "disasmer-node",
            "--",
        ],
        listen,
        "dap-node",
    )
}

fn run_node_against_coordinator(
    state: &AdapterState,
    cwd: &Path,
    command: &str,
    prefix_args: &[&str],
    coordinator: &str,
    node: &str,
) -> Result<RuntimeLaunchRecord> {
    let mut args = prefix_args
        .iter()
        .map(|arg| (*arg).to_owned())
        .collect::<Vec<_>>();
    args.extend([
        "--coordinator".to_owned(),
        coordinator.to_owned(),
        "--tenant".to_owned(),
        state.tenant.clone(),
        "--project-id".to_owned(),
        state.project_id.clone(),
        "--node".to_owned(),
        node.to_owned(),
        "--process".to_owned(),
        state.process.as_str().to_owned(),
        "--task".to_owned(),
        "compile-linux".to_owned(),
        "--project".to_owned(),
        state.project.clone(),
        "--artifact".to_owned(),
        "/vfs/artifacts/dap-output.txt".to_owned(),
    ]);

    let node_output = Command::new(command)
        .args(&args)
        .current_dir(cwd)
        .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(
        coordinator,
        json!({
            "type": "list_task_events",
            "tenant": state.tenant,
            "project": state.project_id,
            "actor_user": state.actor_user,
            "process": state.process.as_str(),
        }),
    )?;
    let event_count = events
        .get("events")
        .and_then(Value::as_array)
        .map(Vec::len)
        .unwrap_or(0);

    Ok(RuntimeLaunchRecord {
        coordinator: coordinator.to_owned(),
        node: node.to_owned(),
        node_report: report.clone(),
        task_events: events,
        placed_task_launched: true,
        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),
        stdout_tail: report
            .get("stdout_tail")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_owned(),
        stderr_tail: report
            .get("stderr_tail")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_owned(),
        stdout_truncated: report
            .get("stdout_truncated")
            .and_then(Value::as_bool)
            .unwrap_or(false),
        stderr_truncated: report
            .get("stderr_truncated")
            .and_then(Value::as_bool)
            .unwrap_or(false),
        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)?;
    let response: Value = serde_json::from_str(&line)?;
    if response.get("type").and_then(Value::as_str) == Some("error") {
        return Err(anyhow!(
            "{}",
            response
                .get("message")
                .and_then(Value::as_str)
                .unwrap_or("coordinator returned an error")
        ));
    }
    Ok(response)
}

fn wait_for_placed_task_event(
    coordinator: &str,
    state: &AdapterState,
    task: &str,
    node: &str,
) -> Result<Value> {
    let deadline = Instant::now() + Duration::from_secs(120);
    loop {
        let events = coordinator_request(
            coordinator,
            json!({
                "type": "list_task_events",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
                "process": state.process.as_str(),
            }),
        )?;
        let completed = events
            .get("events")
            .and_then(Value::as_array)
            .is_some_and(|items| {
                items.iter().any(|event| {
                    event.get("task").and_then(Value::as_str) == Some(task)
                        && event.get("node").and_then(Value::as_str) == Some(node)
                })
            });
        if completed {
            return Ok(events);
        }
        if Instant::now() >= deadline {
            return Err(anyhow!(
                "node {node} accepted virtual task `{task}` but did not report completion; make sure the explicit `disasmer-node --worker` process is still running"
            ));
        }
        std::thread::sleep(Duration::from_millis(250));
    }
}

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 infer_source_path(project: &str) -> String {
    if Path::new(project).join("src/build.rs").is_file() {
        "src/build.rs".to_owned()
    } else {
        "src/main.rs".to_owned()
    }
}

fn source_name(source_path: &str) -> &str {
    Path::new(source_path)
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or(source_path)
}

fn stack_source_path(state: &AdapterState) -> String {
    let resolved = resolve_source_path(&state.project, &state.source_path);
    let absolute = if resolved.is_absolute() {
        resolved
    } else {
        std::env::current_dir()
            .unwrap_or_else(|_| Path::new(".").to_path_buf())
            .join(resolved)
    };
    absolute
        .canonicalize()
        .unwrap_or(absolute)
        .to_string_lossy()
        .into_owned()
}

fn source_response(state: &AdapterState, request: &Value) -> Result<Value> {
    let source_path = request
        .get("arguments")
        .and_then(|arguments| arguments.get("source"))
        .and_then(|source| source.get("path"))
        .and_then(Value::as_str)
        .unwrap_or(&state.source_path);
    let content = fs::read_to_string(resolve_source_path(&state.project, source_path))?;
    Ok(json!({
        "content": content,
        "mimeType": "text/x-rustsrc",
    }))
}

fn resolve_source_path(project: &str, source_path: &str) -> std::path::PathBuf {
    let path = Path::new(source_path);
    if path.is_absolute() {
        path.to_path_buf()
    } else {
        Path::new(project).join(path)
    }
}

fn normalize_operator_endpoint(endpoint: &str) -> String {
    let trimmed = endpoint.trim();
    let without_scheme = trimmed
        .strip_prefix("https://")
        .or_else(|| trimmed.strip_prefix("http://"))
        .unwrap_or(trimmed);
    without_scheme
        .split('/')
        .next()
        .unwrap_or(without_scheme)
        .to_owned()
}

fn write_view_state(state: &AdapterState, record: &RuntimeLaunchRecord) -> Result<()> {
    let dir = Path::new(&state.project).join(".disasmer");
    fs::create_dir_all(&dir)?;
    let node_status = if record.placed_task_launched {
        if record.status_code == Some(0) {
            "completed"
        } else {
            "failed"
        }
    } else {
        "not-required-yet"
    };
    let process_status = if record.placed_task_launched {
        if record.status_code == Some(0) {
            "completed"
        } else {
            "failed"
        }
    } else {
        "running"
    };
    let log_message = if record.placed_task_launched {
        "node task output was staged as a VFS artifact"
    } else {
        "coordinator-side virtual process started; capability task not launched"
    };
    let artifact_status = if record.placed_task_launched {
        "best-effort retained on the attached node"
    } else {
        "not produced until a capability task is placed"
    };
    let view_state = json!({
        "nodes": [{
            "id": record.node,
            "status": node_status,
            "capabilities": format!("connected to {}", record.coordinator),
        }],
        "processes": [{
            "id": state.process.as_str(),
            "status": process_status,
            "entry": state.entry,
        }],
        "logs": [{
            "task": "compile-linux",
            "message": log_message,
            "bytes": format!("stdout={} stderr={}", record.stdout_bytes, record.stderr_bytes),
        }],
        "artifacts": [{
            "path": record.artifact_path.clone().unwrap_or_else(|| "/vfs/artifacts/dap-output.txt".to_owned()),
            "status": artifact_status,
            "size": "reported by node",
        }],
        "inspector": [
            {
                "label": "Runtime backend",
                "value": state.runtime_backend.description(),
            },
            {
                "label": "Coordinator",
                "value": record.coordinator,
            },
            {
                "label": "Node report",
                "value": record.node_report.to_string(),
            },
            {
                "label": "Task events",
                "value": record.task_events.to_string(),
            }
        ]
    });
    fs::write(
        dir.join("views.json"),
        serde_json::to_string_pretty(&view_state)?,
    )?;
    Ok(())
}

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.runtime_backend = RuntimeBackend::LocalServices;

        state.apply_runtime_record(RuntimeLaunchRecord {
            coordinator: "127.0.0.1:7999".to_owned(),
            node: "node".to_owned(),
            node_report: json!({}),
            task_events: json!({ "events": [] }),
            placed_task_launched: true,
            status_code: Some(1),
            stdout_bytes: 0,
            stderr_bytes: 12,
            stdout_tail: String::new(),
            stderr_tail: "failed".to_owned(),
            stdout_truncated: false,
            stderr_truncated: false,
            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("attached 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"
        }));
    }

    #[test]
    fn variables_expose_mvp_runtime_handles_and_output_tails() {
        let mut state = AdapterState::default();
        state.runtime_backend = RuntimeBackend::LocalServices;
        state.apply_runtime_record(RuntimeLaunchRecord {
            coordinator: "127.0.0.1:7999".to_owned(),
            node: "node".to_owned(),
            node_report: json!({}),
            task_events: json!({ "events": [] }),
            placed_task_launched: true,
            status_code: Some(0),
            stdout_bytes: 11,
            stderr_bytes: 7,
            stdout_tail: "hello tail\n".to_owned(),
            stderr_tail: "warn\n".to_owned(),
            stdout_truncated: false,
            stderr_truncated: false,
            artifact_path: Some("/vfs/artifacts/dap-output.txt".to_owned()),
            event_count: 1,
        });
        let linux = state
            .threads
            .get(&LINUX_THREAD)
            .expect("linux task should exist");

        let args = variables_response(&state, linux.args_ref);
        let args = args["variables"]
            .as_array()
            .expect("variables response should be an array");
        assert!(args.iter().any(|variable| {
            variable["name"] == "return_value"
                && variable["value"]
                    .as_str()
                    .is_some_and(|value| value.contains("Artifact"))
        }));
        assert!(args.iter().any(|variable| variable["name"] == "blob"));
        let target_ref = args
            .iter()
            .find(|variable| variable["name"] == "target")
            .and_then(|variable| variable["variablesReference"].as_i64())
            .expect("target should have child variables");
        let vfs_ref = args
            .iter()
            .find(|variable| variable["name"] == "vfs_mounts")
            .and_then(|variable| variable["variablesReference"].as_i64())
            .expect("vfs mounts should have child variables");

        let target = variables_response(&state, target_ref);
        assert!(target["variables"]
            .as_array()
            .unwrap()
            .iter()
            .any(|variable| variable["name"] == "environment" && variable["value"] == "linux"));

        let vfs = variables_response(&state, vfs_ref);
        assert!(vfs["variables"]
            .as_array()
            .unwrap()
            .iter()
            .any(|variable| variable["name"] == "/vfs/artifacts"));

        let runtime = variables_response(&state, linux.runtime_ref);
        let runtime = runtime["variables"].as_array().unwrap();
        let command_ref = runtime
            .iter()
            .find(|variable| variable["name"] == "command_spec")
            .and_then(|variable| variable["variablesReference"].as_i64())
            .expect("command spec should have child variables");
        assert!(runtime.iter().any(
            |variable| variable["name"] == "stdout_tail" && variable["value"] == "hello tail\n"
        ));
        assert!(runtime
            .iter()
            .any(|variable| variable["name"] == "stderr_tail" && variable["value"] == "warn\n"));

        let command = variables_response(&state, command_ref);
        assert!(command["variables"]
            .as_array()
            .unwrap()
            .iter()
            .any(|variable| variable["name"] == "required_capability"
                && variable["value"] == "Command"));
    }

    #[test]
    fn source_locals_infer_disasmer_api_values_from_runtime_state() {
        let mut state = AdapterState::default();
        state.project = Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("../../examples/launch-build-demo")
            .to_string_lossy()
            .into_owned();
        state.source_path = "src/build.rs".to_owned();
        state.threads.get_mut(&MAIN_THREAD).unwrap().line = 60;
        let thread = state.threads[&MAIN_THREAD].clone();

        let locals = variables_response(&state, thread.locals_ref);
        let locals = locals["variables"].as_array().unwrap();

        assert!(locals.iter().any(|variable| {
            variable["name"] == "linux"
                && variable["value"].as_str().is_some_and(|value| {
                    value.contains("TaskHandle")
                        && value.contains("compile-linux")
                        && value.contains("virtual_thread_id = 2")
                })
        }));
        assert!(locals
            .iter()
            .any(|variable| { variable["name"] == "linux_thread" && variable["value"] == "2" }));
        assert!(locals.iter().any(|variable| {
            variable["name"] == "linux_artifact"
                && variable["value"]
                    .as_str()
                    .is_some_and(|value| value.contains("Artifact"))
        }));
        assert!(locals.iter().any(|variable| {
            variable["name"] == "unavailable-local-diagnostic"
                && variable["type"] == "unavailable-local"
        }));
    }

    #[test]
    fn wasm_frame_locals_expose_wasmtime_runtime_values() {
        let mut state = AdapterState::default();
        state.project = Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("../../examples/launch-build-demo")
            .to_string_lossy()
            .into_owned();
        state.source_path = "src/build.rs".to_owned();
        state.threads.get_mut(&MAIN_THREAD).unwrap().line = 31;
        let thread = state.threads[&MAIN_THREAD].clone();

        let locals = variables_response(&state, thread.wasm_locals_ref);
        let locals = locals["variables"].as_array().unwrap();

        assert!(locals.iter().any(|variable| {
            variable["name"] == "wasm_local_0"
                && variable["type"] == "wasm-frame-local"
                && variable["value"]
                    .as_str()
                    .is_some_and(|value| value.contains("41"))
        }));
        assert!(locals.iter().any(|variable| {
            variable["name"] == "wasm-local-diagnostic"
                && variable["value"]
                    .as_str()
                    .is_some_and(|value| value.contains("product node runtime"))
        }));
    }
}