clusterflux-public/crates/clusterflux-dap/src/runtime_client.rs

use std::io::{BufRead, BufReader};
use std::path::Path;
use std::process::{Child, Stdio};
use std::time::{Duration, Instant};

use anyhow::{anyhow, Result};
use base64::{engine::general_purpose::STANDARD as BASE64_STANDARD, Engine as _};
use clusterflux_core::TaskInstanceId;
use serde_json::{json, Value};

use crate::virtual_model::{AdapterState, RuntimeLaunchRecord};

mod debug_protocol;
mod local_tools;
mod transport;
pub(crate) use debug_protocol::parse_task_restart_response;
use debug_protocol::{
    coordinator_debug_epoch_request, parse_debug_epoch_response, wait_for_debug_epoch_state,
};
use local_tools::{child_stderr_suffix, local_tool_command};
pub(crate) use transport::client_user_request;
use transport::coordinator_request;

pub(crate) struct LocalRuntimeSession {
    coordinator: Option<Child>,
    worker: Option<Child>,
}

impl Drop for LocalRuntimeSession {
    fn drop(&mut self) {
        for child in [&mut self.worker, &mut self.coordinator] {
            if let Some(mut child) = child.take() {
                let _ = child.kill();
                let _ = child.wait();
            }
        }
    }
}

struct DebugBundle {
    module_base64: String,
    digest: String,
    entry_export: String,
    entry_definition: String,
}

struct ChildGuard(Option<Child>);

impl ChildGuard {
    fn new(child: Child) -> Self {
        Self(Some(child))
    }

    fn child_mut(&mut self) -> &mut Child {
        self.0.as_mut().expect("guarded child is present")
    }

    fn take(&mut self) -> Child {
        self.0.take().expect("guarded child is present")
    }
}

impl Drop for ChildGuard {
    fn drop(&mut self) {
        if let Some(mut child) = self.0.take() {
            let _ = child.kill();
            let _ = child.wait();
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct DebugEpochRecord {
    pub(crate) epoch: u64,
    pub(crate) command: String,
    pub(crate) affected_tasks: usize,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct DebugEpochStatusRecord {
    pub(crate) epoch: u64,
    pub(crate) command: String,
    pub(crate) expected_tasks: usize,
    pub(crate) acknowledgements: Vec<Value>,
    pub(crate) fully_frozen: bool,
    pub(crate) partially_frozen: bool,
    pub(crate) fully_resumed: bool,
    pub(crate) failed: bool,
    pub(crate) failure_messages: Vec<String>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct TaskRestartRecord {
    pub(crate) accepted: bool,
    pub(crate) restarted_task_instance: Option<TaskInstanceId>,
    pub(crate) restarted_attempt_id: Option<String>,
    pub(crate) clean_boundary_available: bool,
    pub(crate) requires_whole_process_restart: bool,
    pub(crate) active_task: bool,
    pub(crate) completed_event_observed: bool,
    pub(crate) message: String,
}

pub(crate) enum RuntimeContinuationOutcome {
    Breakpoint(RuntimeLaunchRecord),
    Exception(RuntimeLaunchRecord),
    Terminal(RuntimeLaunchRecord),
}

pub(crate) fn run_local_services_runtime(
    state: &AdapterState,
) -> Result<(RuntimeLaunchRecord, LocalRuntimeSession)> {
    let repo = std::env::current_dir()?;
    let mut coordinator_command = local_tool_command(
        "CLUSTERFLUX_COORDINATOR_BIN",
        "clusterflux-coordinator",
        "clusterflux-coordinator",
        &repo,
    );
    let mut coordinator = coordinator_command
        .args(["--listen", "127.0.0.1:0", "--allow-local-trusted-loopback"])
        .current_dir(&repo)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;
    let result = (|| {
        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"))?
            .to_owned();
        coordinator_request(
            &listen,
            json!({
                "type": "create_project",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
                "name": "DAP local services project",
            }),
        )?;
        let enrollment = coordinator_request(
            &listen,
            json!({
                "type": "create_node_enrollment_grant",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
                "ttl_seconds": 60,
            }),
        )?;
        let enrollment_grant = enrollment
            .get("grant")
            .and_then(Value::as_str)
            .ok_or_else(|| anyhow!("local coordinator omitted the node enrollment grant"))?
            .to_owned();

        let mut worker_command = local_tool_command(
            "CLUSTERFLUX_NODE_BIN",
            "clusterflux-node",
            "clusterflux-node",
            &repo,
        );
        let mut worker = ChildGuard::new(
            worker_command
                .args([
                    "--coordinator",
                    &listen,
                    "--tenant",
                    state.tenant.as_str(),
                    "--project-id",
                    state.project_id.as_str(),
                    "--node",
                    "dap-node",
                    "--enrollment-grant",
                    &enrollment_grant,
                    "--worker",
                    "--project-root",
                    &state.project,
                    "--assignment-poll-ms",
                    "20",
                ])
                .current_dir(&repo)
                .stdout(Stdio::null())
                .stderr(Stdio::piped())
                .spawn()?,
        );
        wait_for_local_node(&listen, state, "dap-node", worker.child_mut())?;

        let record = launch_services_debug_entrypoint(&listen, state, &repo)?;
        Ok((record, worker.take()))
    })();
    match result {
        Ok((record, worker)) => Ok((
            record,
            LocalRuntimeSession {
                coordinator: Some(coordinator),
                worker: Some(worker),
            },
        )),
        Err(error) => {
            let _ = coordinator.kill();
            let _ = coordinator.wait();
            Err(error)
        }
    }
}

fn build_debug_bundle(state: &AdapterState, repo: &Path) -> Result<DebugBundle> {
    let mut command = local_tool_command(
        "CLUSTERFLUX_CLI_BIN",
        "clusterflux",
        "clusterflux-cli",
        repo,
    );
    let output = command
        .args(["build", "--project", &state.project, "--json"])
        .current_dir(repo)
        .output()?;
    if !output.status.success() {
        return Err(anyhow!(
            "Clusterflux bundle build failed before debug launch: {}",
            String::from_utf8_lossy(&output.stderr).trim()
        ));
    }
    let report: Value = serde_json::from_slice(&output.stdout)?;
    let module_path = report
        .pointer("/bundle_artifact/module")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow!("bundle build omitted module path"))?;
    let module_path = if Path::new(module_path).is_absolute() {
        Path::new(module_path).to_path_buf()
    } else {
        repo.join(module_path)
    };
    let module = std::fs::read(&module_path)?;
    let entrypoints: Value = serde_json::from_slice(&std::fs::read(
        module_path
            .parent()
            .ok_or_else(|| anyhow!("bundle module path has no parent"))?
            .join("entrypoints.json"),
    )?)?;
    let descriptor = entrypoints
        .as_array()
        .and_then(|descriptors| {
            descriptors.iter().find(|descriptor| {
                descriptor.get("name").and_then(Value::as_str) == Some(state.entry.as_str())
            })
        })
        .ok_or_else(|| anyhow!("bundle has no entrypoint `{}`", state.entry))?;
    let entry_export = descriptor
        .get("export")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow!("entrypoint `{}` omitted its Wasm export", state.entry))?
        .to_owned();
    let entry_definition = descriptor
        .get("stable_id")
        .and_then(Value::as_str)
        .ok_or_else(|| {
            anyhow!(
                "entrypoint `{}` omitted its stable definition ID",
                state.entry
            )
        })?
        .to_owned();
    let digest = report
        .pointer("/bundle_artifact/bundle_digest")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow!("bundle build omitted bundle digest"))?
        .to_owned();
    Ok(DebugBundle {
        module_base64: BASE64_STANDARD.encode(module),
        digest,
        entry_export,
        entry_definition,
    })
}

fn launch_services_debug_entrypoint(
    coordinator: &str,
    state: &AdapterState,
    repo: &Path,
) -> Result<RuntimeLaunchRecord> {
    let bundle = build_debug_bundle(state, repo)?;
    let started = coordinator_request(
        coordinator,
        client_user_request(
            state,
            json!({
                "type": "start_process",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
                "process": state.process.as_str(),
                "restart": state.restart_existing,
            }),
        ),
    )?;
    let epoch = started
        .get("epoch")
        .and_then(Value::as_u64)
        .ok_or_else(|| anyhow!("coordinator did not report a process epoch"))?;
    let probe_symbols = state.requested_probe_symbols();
    if probe_symbols.is_empty() {
        return Err(anyhow!(
            "no executable Clusterflux probe corresponds to the configured source breakpoints"
        ));
    }
    coordinator_request(
        coordinator,
        client_user_request(
            state,
            json!({
                "type": "set_debug_breakpoints",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
                "process": state.process.as_str(),
                "probe_symbols": probe_symbols,
            }),
        ),
    )?;
    let launch = coordinator_request(
        coordinator,
        client_user_request(
            state,
            json!({
                "type": "launch_task",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
                "task_spec": {
                    "tenant": state.tenant,
                    "project": state.project_id,
                    "process": state.process.as_str(),
                    "task_definition": bundle.entry_definition,
                    "task_instance": format!("ti:{}:main", state.process),
                    "dispatch": {
                        "kind": "coordinator_node_wasm",
                        "export": bundle.entry_export,
                        "abi": "entrypoint_v1",
                    },
                    "environment_id": null,
                    "environment": null,
                    "environment_digest": null,
                    "required_capabilities": [],
                    "dependency_cache": null,
                    "source_snapshot": null,
                    "required_artifacts": [],
                    "args": [],
                    "vfs_epoch": epoch,
                    "bundle_digest": bundle.digest,
                },
                "wait_for_node": true,
                "artifact_path": "/vfs/artifacts/dap-output.txt",
                "wasm_module_base64": bundle.module_base64,
            }),
        ),
    )?;
    if launch.get("type").and_then(Value::as_str) != Some("main_launched") {
        return Err(anyhow!(
            "coordinator did not start the capless main runtime: {}",
            serde_json::to_string(&launch)?
        ));
    }
    let breakpoint = wait_for_breakpoint_hit(coordinator, state)?;
    let debug_epoch = breakpoint
        .get("hit_epoch")
        .and_then(Value::as_u64)
        .ok_or_else(|| anyhow!("breakpoint status omitted the hit debug epoch"))?;
    let frozen = wait_for_debug_epoch_state_at(coordinator, state, debug_epoch, true)?;
    let fully_frozen = frozen
        .get("fully_frozen")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let partially_frozen = frozen
        .get("partially_frozen")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let expected = frozen
        .get("expected_tasks")
        .and_then(Value::as_array)
        .map(Vec::len)
        .unwrap_or(0);
    let acknowledged = frozen
        .get("acknowledgements")
        .and_then(Value::as_array)
        .map(Vec::len)
        .unwrap_or(0);
    if expected == 0 || (!fully_frozen && !partially_frozen) {
        return Err(anyhow!(
            "debug epoch {debug_epoch} settled without a frozen participant"
        ));
    }
    let task_snapshots = fetch_task_snapshots(coordinator, state)?;
    let process_statuses = coordinator_request(
        coordinator,
        client_user_request(
            state,
            json!({
                "type": "list_processes",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
            }),
        ),
    )?;
    let process_status = process_statuses
        .get("processes")
        .and_then(Value::as_array)
        .and_then(|processes| {
            processes.iter().find(|process| {
                process.get("process").and_then(Value::as_str) == Some(state.process.as_str())
            })
        })
        .cloned();
    let node = frozen
        .get("acknowledgements")
        .and_then(Value::as_array)
        .and_then(|items| items.first())
        .and_then(|item| item.get("node"))
        .and_then(Value::as_str)
        .unwrap_or("coordinator-main")
        .to_owned();
    Ok(RuntimeLaunchRecord {
        coordinator: coordinator.to_owned(),
        node,
        node_report: json!({
            "process": started,
            "task_launch": launch,
            "breakpoint": breakpoint,
            "debug_epoch": frozen,
            "process_status": process_status,
            "process_statuses": process_statuses,
            "task_snapshots": task_snapshots,
        }),
        task_events: json!({ "events": [] }),
        placed_task_launched: true,
        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,
        debug_epoch: Some(debug_epoch),
        stopped_task: breakpoint
            .get("hit_task")
            .and_then(Value::as_str)
            .map(str::to_owned),
        stopped_probe_symbol: breakpoint
            .get("hit_probe_symbol")
            .and_then(Value::as_str)
            .map(str::to_owned),
        all_participants_frozen: fully_frozen && acknowledged == expected,
    })
}

fn wait_for_local_node(
    coordinator: &str,
    state: &AdapterState,
    node: &str,
    worker: &mut Child,
) -> Result<()> {
    let deadline = Instant::now() + Duration::from_secs(30);
    loop {
        if let Some(status) = worker.try_wait()? {
            return Err(anyhow!(
                "local Clusterflux worker `{node}` exited before attaching ({status}){}",
                child_stderr_suffix(worker)
            ));
        }
        let response = coordinator_request(
            coordinator,
            json!({
                "type": "list_node_descriptors",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
            }),
        )?;
        if response
            .get("descriptors")
            .and_then(Value::as_array)
            .is_some_and(|descriptors| {
                descriptors
                    .iter()
                    .any(|descriptor| descriptor.get("id").and_then(Value::as_str) == Some(node))
            })
        {
            return Ok(());
        }
        if Instant::now() >= deadline {
            let _ = worker.kill();
            let _ = worker.wait();
            return Err(anyhow!(
                "local Clusterflux worker `{node}` did not attach within 30 seconds{}",
                child_stderr_suffix(worker)
            ));
        }
        std::thread::sleep(Duration::from_millis(50));
    }
}

fn wait_for_breakpoint_hit(coordinator: &str, state: &AdapterState) -> Result<Value> {
    let deadline = Instant::now() + Duration::from_secs(30);
    loop {
        let response = coordinator_request(
            coordinator,
            client_user_request(
                state,
                json!({
                    "type": "inspect_debug_breakpoints",
                    "tenant": state.tenant,
                    "project": state.project_id,
                    "actor_user": state.actor_user,
                    "process": state.process.as_str(),
                }),
            ),
        )?;
        if response.get("hit_epoch").and_then(Value::as_u64).is_some() {
            return Ok(response);
        }
        if Instant::now() >= deadline {
            return Err(anyhow!(
                "executing Wasm did not reach any configured source breakpoint within 30 seconds"
            ));
        }
        std::thread::sleep(Duration::from_millis(50));
    }
}

fn wait_for_debug_epoch_state_at(
    coordinator: &str,
    state: &AdapterState,
    epoch: u64,
    frozen: bool,
) -> Result<Value> {
    let deadline = Instant::now() + Duration::from_secs(60);
    loop {
        let response = coordinator_request(
            coordinator,
            client_user_request(
                state,
                json!({
                    "type": "inspect_debug_epoch",
                    "tenant": state.tenant,
                    "project": state.project_id,
                    "actor_user": state.actor_user,
                    "process": state.process.as_str(),
                    "epoch": epoch,
                }),
            ),
        )?;
        let partially_frozen = frozen
            && response
                .get("partially_frozen")
                .and_then(Value::as_bool)
                .unwrap_or(false);
        if response.get("fully_frozen").and_then(Value::as_bool) == Some(true)
            || partially_frozen
            || (!frozen && response.get("fully_resumed").and_then(Value::as_bool) == Some(true))
        {
            return Ok(response);
        }
        if response.get("failed").and_then(Value::as_bool) == Some(true) {
            return Err(anyhow!(
                "debug epoch {epoch} participant failed: {}",
                response
                    .get("failure_messages")
                    .and_then(Value::as_array)
                    .into_iter()
                    .flatten()
                    .filter_map(Value::as_str)
                    .collect::<Vec<_>>()
                    .join("; ")
            ));
        }
        let ready_field = if frozen {
            "fully_frozen"
        } else {
            "fully_resumed"
        };
        if Instant::now() >= deadline {
            return Err(anyhow!(
                "debug epoch {epoch} did not reach {ready_field} within 60 seconds"
            ));
        }
        std::thread::sleep(Duration::from_millis(50));
    }
}

pub(crate) fn run_live_services_runtime(state: &AdapterState) -> Result<RuntimeLaunchRecord> {
    let coordinator =
        crate::view_state::normalize_coordinator_endpoint(&state.coordinator_endpoint);
    let repo = std::env::current_dir()?;
    launch_services_debug_entrypoint(&coordinator, state, &repo)
}

fn fetch_task_snapshots(coordinator: &str, state: &AdapterState) -> Result<Value> {
    coordinator_request(
        coordinator,
        client_user_request(
            state,
            json!({
                "type": "list_task_snapshots",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
                "process": state.process.as_str(),
            }),
        ),
    )
}

fn fetch_current_process_status(
    coordinator: &str,
    state: &AdapterState,
) -> Result<(Value, Option<Value>)> {
    let statuses = coordinator_request(
        coordinator,
        client_user_request(
            state,
            json!({
                "type": "list_processes",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
            }),
        ),
    )?;
    let current = statuses
        .get("processes")
        .and_then(Value::as_array)
        .and_then(|processes| {
            processes.iter().find(|process| {
                process.get("process").and_then(Value::as_str) == Some(state.process.as_str())
            })
        })
        .cloned();
    Ok((statuses, current))
}

pub(crate) fn relaunch_services_main_runtime(state: &AdapterState) -> Result<RuntimeLaunchRecord> {
    let coordinator =
        crate::view_state::normalize_coordinator_endpoint(&state.coordinator_endpoint);
    let repo = std::env::current_dir()?;
    let mut restart_state = state.clone();
    restart_state.restart_existing = true;
    launch_services_debug_entrypoint(&coordinator, &restart_state, &repo)
}

pub(crate) fn attach_services_runtime(state: &AdapterState) -> Result<RuntimeLaunchRecord> {
    let coordinator =
        crate::view_state::normalize_coordinator_endpoint(&state.coordinator_endpoint);
    let debug_attach = coordinator_request(
        &coordinator,
        client_user_request(
            state,
            json!({
                "type": "debug_attach",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
                "process": state.process.as_str(),
            }),
        ),
    )?;
    let allowed = debug_attach
        .get("authorization")
        .and_then(|authorization| authorization.get("allowed"))
        .and_then(Value::as_bool)
        .unwrap_or(false);
    if !allowed {
        let reason = debug_attach
            .get("authorization")
            .and_then(|authorization| authorization.get("reason"))
            .and_then(Value::as_str)
            .unwrap_or("debug attach was denied by coordinator authorization");
        return Err(anyhow!("debug attach denied: {reason}"));
    }
    let events = coordinator_request(
        &coordinator,
        client_user_request(
            state,
            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);
    let task_snapshots = fetch_task_snapshots(&coordinator, state)?;
    let active_snapshot_count = task_snapshots
        .get("snapshots")
        .and_then(Value::as_array)
        .map(|snapshots| {
            snapshots
                .iter()
                .filter(|snapshot| {
                    snapshot.get("current").and_then(Value::as_bool) == Some(true)
                        && matches!(
                            snapshot.get("state").and_then(Value::as_str),
                            Some("queued" | "running" | "failed_awaiting_action")
                        )
                })
                .count()
        })
        .unwrap_or(0);
    let process_statuses = coordinator_request(
        &coordinator,
        client_user_request(
            state,
            json!({
                "type": "list_processes",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
            }),
        ),
    )?;
    let process_status = process_statuses
        .get("processes")
        .and_then(Value::as_array)
        .and_then(|processes| {
            processes.iter().find(|process| {
                process.get("process").and_then(Value::as_str) == Some(state.process.as_str())
            })
        })
        .cloned();
    let node = process_status
        .as_ref()
        .and_then(|status| status.get("connected_nodes"))
        .and_then(Value::as_array)
        .and_then(|nodes| nodes.first())
        .and_then(Value::as_str)
        .unwrap_or("coordinator-main")
        .to_owned();
    let active_main = process_status
        .as_ref()
        .and_then(|status| status.get("main_task_instance"))
        .and_then(Value::as_str)
        .is_some();

    Ok(RuntimeLaunchRecord {
        coordinator,
        node,
        node_report: json!({
            "debug_attach": debug_attach,
            "process_status": process_status,
            "process_statuses": process_statuses,
            "task_snapshots": task_snapshots,
        }),
        task_events: events,
        placed_task_launched: active_main || active_snapshot_count > 0,
        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,
        debug_epoch: None,
        stopped_task: None,
        stopped_probe_symbol: None,
        all_participants_frozen: false,
    })
}

pub(crate) fn create_debug_epoch(
    state: &AdapterState,
    stopped_task: &TaskInstanceId,
    reason: &str,
) -> Result<DebugEpochRecord> {
    let response = coordinator_debug_epoch_request(
        state,
        client_user_request(
            state,
            json!({
                "type": "create_debug_epoch",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
                "process": state.process.as_str(),
                "stopped_task": stopped_task.as_str(),
                "reason": reason,
            }),
        ),
    )?;
    parse_debug_epoch_response(response)
}

pub(crate) fn resume_debug_epoch(state: &AdapterState, epoch: u64) -> Result<DebugEpochRecord> {
    let response = coordinator_debug_epoch_request(
        state,
        client_user_request(
            state,
            json!({
                "type": "resume_debug_epoch",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
                "process": state.process.as_str(),
                "epoch": epoch,
            }),
        ),
    )?;
    parse_debug_epoch_response(response)
}

pub(crate) fn wait_for_debug_epoch_frozen(
    state: &AdapterState,
    epoch: u64,
) -> Result<DebugEpochStatusRecord> {
    wait_for_debug_epoch_state(state, epoch, true)
}

pub(crate) fn wait_for_debug_epoch_resumed(
    state: &AdapterState,
    epoch: u64,
) -> Result<DebugEpochStatusRecord> {
    wait_for_debug_epoch_state(state, epoch, false)
}

pub(crate) fn wait_for_services_runtime_outcome(
    state: &AdapterState,
    previous_debug_epoch: u64,
) -> Result<RuntimeContinuationOutcome> {
    let coordinator =
        crate::view_state::normalize_coordinator_endpoint(&state.coordinator_endpoint);
    let join_timeout = clusterflux_core::limits::task_join_timeout();
    let deadline = Instant::now() + join_timeout;
    loop {
        // Terminal task events remain inspectable after the active process slot is
        // released. Read them before active-process debug state so a fast main exit
        // cannot turn a successful continuation into an authorization error.
        let events = coordinator_request(
            &coordinator,
            client_user_request(
                state,
                json!({
                    "type": "list_task_events",
                    "tenant": state.tenant,
                    "project": state.project_id,
                    "actor_user": state.actor_user,
                    "process": state.process.as_str(),
                }),
            ),
        )?;
        if let Some(outcome) = terminal_runtime_outcome(&coordinator, state, &events) {
            return Ok(outcome);
        }
        let task_snapshots = fetch_task_snapshots(&coordinator, state)?;
        if let Some((failed_task, _attempt_id)) = failed_awaiting_action_snapshot(&task_snapshots) {
            let failed_task = failed_task.to_owned();
            let failed_event = events
                .get("events")
                .and_then(Value::as_array)
                .into_iter()
                .flatten()
                .rev()
                .find(|event| {
                    event.get("task").and_then(Value::as_str) == Some(failed_task.as_str())
                        && event.get("terminal_state").and_then(Value::as_str) == Some("failed")
                })
                .cloned()
                .unwrap_or_else(|| json!({}));
            let event_count = events
                .get("events")
                .and_then(Value::as_array)
                .map(Vec::len)
                .unwrap_or(state.runtime_event_count);
            let (process_statuses, process_status) =
                fetch_current_process_status(&coordinator, state)?;
            return Ok(RuntimeContinuationOutcome::Exception(RuntimeLaunchRecord {
                coordinator,
                node: failed_event
                    .get("node")
                    .and_then(Value::as_str)
                    .unwrap_or("unknown")
                    .to_owned(),
                node_report: json!({
                    "task_snapshots": task_snapshots,
                    "process_status": process_status,
                    "process_statuses": process_statuses,
                    "failed_awaiting_action": failed_task,
                }),
                task_events: events,
                placed_task_launched: true,
                status_code: failed_event
                    .get("status_code")
                    .and_then(Value::as_i64)
                    .map(|status| status as i32)
                    .or(Some(1)),
                stdout_bytes: failed_event
                    .get("stdout_bytes")
                    .and_then(Value::as_u64)
                    .unwrap_or(0),
                stderr_bytes: failed_event
                    .get("stderr_bytes")
                    .and_then(Value::as_u64)
                    .unwrap_or(0),
                stdout_tail: failed_event
                    .get("stdout_tail")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_owned(),
                stderr_tail: failed_event
                    .get("stderr_tail")
                    .and_then(Value::as_str)
                    .unwrap_or("task failed awaiting operator action")
                    .to_owned(),
                stdout_truncated: failed_event
                    .get("stdout_truncated")
                    .and_then(Value::as_bool)
                    .unwrap_or(false),
                stderr_truncated: failed_event
                    .get("stderr_truncated")
                    .and_then(Value::as_bool)
                    .unwrap_or(false),
                artifact_path: failed_event
                    .get("artifact_path")
                    .and_then(Value::as_str)
                    .map(str::to_owned),
                event_count,
                debug_epoch: None,
                stopped_task: Some(failed_task),
                stopped_probe_symbol: None,
                all_participants_frozen: false,
            }));
        }

        let breakpoint = match coordinator_request(
            &coordinator,
            client_user_request(
                state,
                json!({
                    "type": "inspect_debug_breakpoints",
                    "tenant": state.tenant,
                    "project": state.project_id,
                    "actor_user": state.actor_user,
                    "process": state.process.as_str(),
                }),
            ),
        ) {
            Ok(breakpoint) => breakpoint,
            Err(inspect_error) => {
                // Main completion records its terminal event and clears ephemeral
                // debug state in one coordinator pump. That pump can occur between
                // the event read above and this inspection request. Re-read the
                // durable event stream before treating the missing debug state as
                // an adapter failure.
                let events = coordinator_request(
                    &coordinator,
                    client_user_request(
                        state,
                        json!({
                            "type": "list_task_events",
                            "tenant": state.tenant,
                            "project": state.project_id,
                            "actor_user": state.actor_user,
                            "process": state.process.as_str(),
                        }),
                    ),
                )?;
                if let Some(outcome) = terminal_runtime_outcome(&coordinator, state, &events) {
                    return Ok(outcome);
                }
                return Err(inspect_error);
            }
        };
        if let Some(epoch) = breakpoint.get("hit_epoch").and_then(Value::as_u64) {
            if epoch > previous_debug_epoch {
                let frozen = wait_for_debug_epoch_state_at(&coordinator, state, epoch, true)?;
                let node = frozen
                    .get("acknowledgements")
                    .and_then(Value::as_array)
                    .and_then(|items| items.first())
                    .and_then(|item| item.get("node"))
                    .and_then(Value::as_str)
                    .unwrap_or("unknown")
                    .to_owned();
                let task_snapshots = fetch_task_snapshots(&coordinator, state)?;
                let (process_statuses, process_status) =
                    fetch_current_process_status(&coordinator, state)?;
                let fully_frozen = frozen
                    .get("fully_frozen")
                    .and_then(Value::as_bool)
                    .unwrap_or(false);
                return Ok(RuntimeContinuationOutcome::Breakpoint(
                    RuntimeLaunchRecord {
                        coordinator,
                        node,
                        node_report: json!({
                            "breakpoint": breakpoint,
                            "debug_epoch": frozen,
                            "task_snapshots": task_snapshots,
                            "process_status": process_status,
                            "process_statuses": process_statuses,
                        }),
                        task_events: json!({ "events": [] }),
                        placed_task_launched: true,
                        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: state.runtime_event_count,
                        debug_epoch: Some(epoch),
                        stopped_task: breakpoint
                            .get("hit_task")
                            .and_then(Value::as_str)
                            .map(str::to_owned),
                        stopped_probe_symbol: breakpoint
                            .get("hit_probe_symbol")
                            .and_then(Value::as_str)
                            .map(str::to_owned),
                        all_participants_frozen: fully_frozen,
                    },
                ));
            }
        }

        if Instant::now() >= deadline {
            return Err(anyhow::Error::new(clusterflux_core::limits::TaskJoinError::timeout(
                clusterflux_core::TaskInstanceId::from("coordinator-main"),
                join_timeout,
            ))
            .context(format!(
                "continued virtual process {} neither reached another executing Wasm breakpoint nor recorded terminal entrypoint state",
                state.process
            )));
        }
        std::thread::sleep(Duration::from_millis(100));
    }
}

pub(crate) fn failed_awaiting_action_snapshot(task_snapshots: &Value) -> Option<(&str, &str)> {
    task_snapshots
        .get("snapshots")?
        .as_array()?
        .iter()
        .find(|snapshot| {
            snapshot.get("current").and_then(Value::as_bool) == Some(true)
                && snapshot.get("state").and_then(Value::as_str) == Some("failed_awaiting_action")
        })
        .and_then(|snapshot| {
            Some((
                snapshot.get("task")?.as_str()?,
                snapshot.get("attempt_id")?.as_str()?,
            ))
        })
}

fn terminal_runtime_outcome(
    coordinator: &str,
    state: &AdapterState,
    events: &Value,
) -> Option<RuntimeContinuationOutcome> {
    let event = events
        .get("events")
        .and_then(Value::as_array)?
        .get(state.runtime_event_count..)?
        .iter()
        .rev()
        .find(|event| event.get("executor").and_then(Value::as_str) == Some("coordinator_main"))?;
    let status_code = event
        .get("status_code")
        .and_then(Value::as_i64)
        .map(|status| status as i32)
        .or_else(
            || match event.get("terminal_state").and_then(Value::as_str) {
                Some("completed") => Some(0),
                Some("failed" | "cancelled") => Some(1),
                _ => None,
            },
        );
    Some(RuntimeContinuationOutcome::Terminal(RuntimeLaunchRecord {
        coordinator: coordinator.to_owned(),
        node: event
            .get("node")
            .and_then(Value::as_str)
            .unwrap_or("coordinator-main")
            .to_owned(),
        node_report: json!({ "terminal_event": event }),
        task_events: events.clone(),
        placed_task_launched: true,
        status_code,
        stdout_bytes: event
            .get("stdout_bytes")
            .and_then(Value::as_u64)
            .unwrap_or(0),
        stderr_bytes: event
            .get("stderr_bytes")
            .and_then(Value::as_u64)
            .unwrap_or(0),
        stdout_tail: event
            .get("stdout_tail")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_owned(),
        stderr_tail: event
            .get("stderr_tail")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_owned(),
        stdout_truncated: event
            .get("stdout_truncated")
            .and_then(Value::as_bool)
            .unwrap_or(false),
        stderr_truncated: event
            .get("stderr_truncated")
            .and_then(Value::as_bool)
            .unwrap_or(false),
        artifact_path: event
            .get("artifact_path")
            .and_then(Value::as_str)
            .map(str::to_owned),
        event_count: events
            .get("events")
            .and_then(Value::as_array)
            .map(Vec::len)
            .unwrap_or(0),
        debug_epoch: None,
        stopped_task: None,
        stopped_probe_symbol: None,
        all_participants_frozen: false,
    }))
}

pub(crate) fn restart_task(
    state: &AdapterState,
    task: &TaskInstanceId,
) -> Result<TaskRestartRecord> {
    let repo = std::env::current_dir()?;
    let replacement = build_debug_bundle(state, &repo)?;
    let response = coordinator_debug_epoch_request(
        state,
        client_user_request(
            state,
            json!({
                "type": "restart_task",
                "tenant": state.tenant,
                "project": state.project_id,
                "actor_user": state.actor_user,
                "process": state.process.as_str(),
                "task": task.as_str(),
                "replacement_bundle": {
                    "bundle_digest": replacement.digest,
                    "wasm_module_base64": replacement.module_base64,
                },
            }),
        ),
    )?;
    parse_task_restart_response(response)
}