clusterflux-public/crates/clusterflux-cli/src/run.rs

use std::path::{Path, PathBuf};
use std::thread;
use std::time::{Duration, Instant};

use anyhow::{Context, Result};
use base64::{engine::general_purpose::STANDARD as BASE64_STANDARD, Engine as _};
use clusterflux_control::MAX_CONTROL_FRAME_BYTES;
use clusterflux_core::{
    agent_ed25519_public_key_from_private_key, agent_workflow_request_scope_from_payload,
    sign_agent_workflow_request, signed_request_payload_digest, Digest,
};
use serde::Serialize;
use serde_json::{json, Value};

use crate::client::{stored_session_for_coordinator, JsonLineSession};
use crate::config::{
    default_hosted_coordinator_endpoint, read_cli_session, read_project_config, StoredCliSession,
};
use crate::errors::{cli_error_summary, cli_error_summary_for_category};
use crate::{BuildArgs, RunArgs};

mod local_services;
use local_services::{LocalCoordinator, LocalNodeWorker};

struct RunBundle {
    build_report: Value,
    digest: Digest,
    module_base64: String,
    module_size_bytes: usize,
    entry_export: String,
    entry_stable_id: String,
}

// The control request contains base64 (4/3 expansion), the authenticated
// envelope, TaskSpec metadata, and user/project identifiers. Reserve a fixed
// worst-case metadata budget so every module at or below this limit fits the
// existing 1 MiB control frame without creating a process first.
const INLINE_BUNDLE_REQUEST_OVERHEAD_BYTES: usize = 96 * 1024;
pub(crate) const MAX_INLINE_WASM_MODULE_BYTES: usize =
    ((MAX_CONTROL_FRAME_BYTES - INLINE_BUNDLE_REQUEST_OVERHEAD_BYTES) / 4) * 3;

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub(crate) struct RunPlan {
    pub(crate) project: PathBuf,
    pub(crate) entry: String,
    pub(crate) coordinator: CoordinatorSelection,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) hosted_coordinator_endpoint: Option<String>,
    pub(crate) session: CliSession,
}

#[derive(Clone, Debug, PartialEq, Serialize)]
pub(crate) struct RunExecutionReport {
    pub(crate) plan: RunPlan,
    pub(crate) boundary: RunBoundaryEvidence,
    pub(crate) node_report: serde_json::Value,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub(crate) struct RunBoundaryEvidence {
    pub(crate) cli_process_started_node_process: bool,
    pub(crate) cli_process_started_coordinator_process: bool,
    pub(crate) coordinator_address: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) coordinator_process_id: Option<u32>,
    pub(crate) spawned_node_process_id: u32,
    pub(crate) node_session_requests: u64,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub(crate) enum CoordinatorSelection {
    Hosted,
    LocalOverride(String),
    LocalOnly,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub(crate) enum CliSession {
    Anonymous,
    HumanSession,
    AgentPublicKey {
        agent: String,
        public_key: String,
        public_key_fingerprint: Digest,
        #[serde(skip)]
        private_key: Option<String>,
        browser_interaction_required: bool,
    },
}

impl CliSession {
    pub(crate) fn is_authenticated(&self) -> bool {
        !matches!(self, Self::Anonymous)
    }
}

pub(crate) fn run_plan(args: RunArgs, cwd: PathBuf, session: CliSession) -> Result<RunPlan> {
    let (coordinator, hosted_coordinator_endpoint) = if let Some(url) = args.coordinator {
        (CoordinatorSelection::LocalOverride(url), None)
    } else if args.local {
        (CoordinatorSelection::LocalOnly, None)
    } else if session.is_authenticated() {
        (
            CoordinatorSelection::Hosted,
            Some(default_hosted_coordinator_endpoint()),
        )
    } else {
        (CoordinatorSelection::LocalOnly, None)
    };

    Ok(RunPlan {
        project: args.project.unwrap_or(cwd),
        entry: args.entry.unwrap_or_else(|| "build".to_owned()),
        coordinator,
        hosted_coordinator_endpoint,
        session,
    })
}

fn non_interactive_run_requires_auth_report(args: RunArgs, cwd: PathBuf) -> Value {
    let project = args.project.unwrap_or(cwd);
    let entry = args.entry.unwrap_or_else(|| "build".to_owned());
    let message = "non-interactive run requires an authenticated human or agent session unless --local or --coordinator is explicit";
    let next_actions = vec![
        "clusterflux login --browser",
        "set CLUSTERFLUX_AGENT_PRIVATE_KEY for automation",
        "pass --local to run against local services",
        "pass --coordinator for an explicit self-hosted coordinator",
    ];
    let mut machine_error = cli_error_summary_for_category("authentication", message);
    if let Some(object) = machine_error.as_object_mut() {
        object.insert("next_actions".to_owned(), json!(next_actions.clone()));
        object.insert("browser_opened".to_owned(), json!(false));
    }
    json!({
        "command": "run",
        "status": "authentication_required",
        "project_root": project,
        "entry": entry,
        "non_interactive": true,
        "browser_opened": false,
        "safe_failure": true,
        "message": message,
        "next_actions": next_actions,
        "private_website_required": false,
        "machine_error": machine_error,
    })
}

pub(crate) fn run_report(args: RunArgs, cwd: PathBuf, session: CliSession) -> Result<Value> {
    if args.non_interactive
        && !args.local
        && args.coordinator.is_none()
        && !session.is_authenticated()
    {
        return Ok(non_interactive_run_requires_auth_report(args, cwd));
    }
    let plan = run_plan(args, cwd, session)?;
    if should_execute_local_node(&plan) {
        return Ok(serde_json::to_value(execute_local_node_run(plan)?)?);
    }
    coordinator_run_report(plan)
}

fn coordinator_run_report(plan: RunPlan) -> Result<Value> {
    // Build, resolve the requested entrypoint, verify the module digest, and
    // enforce the accepted inline transport boundary before mutating the
    // coordinator. A failure here therefore cannot leave an active process.
    let bundle = build_bundle_for_run(&plan.project, &plan.entry)?;
    validate_inline_bundle_size(bundle.module_size_bytes)?;
    let config = read_project_config(&plan.project)?;
    let stored_session = read_cli_session(&plan.project)?;
    let coordinator = run_coordinator_endpoint(&plan, stored_session.as_ref())?;
    let bound_session = stored_session_for_coordinator(&coordinator, stored_session.as_ref());
    let tenant = bound_session
        .map(|session| session.tenant.clone())
        .or_else(|| config.as_ref().map(|config| config.tenant.clone()))
        .unwrap_or_else(|| "tenant".to_owned());
    let project = bound_session
        .map(|session| session.project.clone())
        .or_else(|| config.as_ref().map(|config| config.project.clone()))
        .unwrap_or_else(|| "project".to_owned());
    let user = bound_session
        .map(|session| session.user.clone())
        .or_else(|| config.as_ref().map(|config| config.user.clone()))
        .unwrap_or_else(|| "user".to_owned());
    let human_session_secret =
        human_run_session_secret(&plan, &coordinator, stored_session.as_ref())?;
    if matches!(plan.session, CliSession::HumanSession)
        && human_session_secret.is_none()
        && !crate::client::is_loopback_coordinator(&coordinator)
    {
        anyhow::bail!(
            "no authenticated CLI session matches coordinator {coordinator}; run `clusterflux login --browser` from the current project"
        );
    }
    let process = "vp-current".to_owned();
    let mut session = JsonLineSession::connect(&coordinator)?;
    let mut request = json!({
        "type": "start_process",
        "tenant": tenant.clone(),
        "project": project.clone(),
        "process": process.clone(),
        "restart": false,
    });
    request = authenticated_human_or_local_trusted_workflow(
        request,
        &plan.session,
        &user,
        human_session_secret.as_deref(),
    );
    let response = request_process_start_with_rollback(
        &mut session,
        request,
        &coordinator,
        &plan.session,
        &user,
        human_session_secret.as_deref(),
        &tenant,
        &project,
        &process,
    )?;
    let run_start = run_start_summary(&response);
    let status = run_start
        .get("status")
        .and_then(Value::as_str)
        .unwrap_or("coordinator_response");
    let task_definition = bundle.entry_stable_id.clone();
    let task_instance = format!("ti:{process}:main");
    let artifact_path = format!("/vfs/artifacts/{task_instance}-output.txt");
    let launch_task_response = if status == "started" {
        let task_spec = json!({
            "tenant": tenant,
            "project": project,
            "process": process,
            "task_definition": task_definition,
            "task_instance": task_instance,
            "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": response.get("epoch").and_then(Value::as_u64).unwrap_or_default(),
            "failure_policy": "fail_fast",
            "bundle_digest": bundle.digest,
        });
        let mut launch_task_request = json!({
            "type": "launch_task",
            "tenant": tenant.clone(),
            "project": project.clone(),
            "task_spec": task_spec,
            "wait_for_node": true,
            "artifact_path": artifact_path.clone(),
            "wasm_module_base64": bundle.module_base64,
        });
        launch_task_request = authenticated_human_or_local_trusted_workflow(
            launch_task_request,
            &plan.session,
            &user,
            human_session_secret.as_deref(),
        );
        let launch = session
            .request_allow_error(launch_task_request)
            .and_then(|response| {
                if response.get("type").and_then(Value::as_str) == Some("main_launched") {
                    Ok(response)
                } else {
                    anyhow::bail!("coordinator main launch was not acknowledged: {response}")
                }
            });
        match launch {
            Ok(launch) => launch,
            Err(launch_error) => {
                let rollback = ProcessLaunchRollback {
                    cli_session: &plan.session,
                    fallback_user: &user,
                    human_session_secret: human_session_secret.as_deref(),
                    tenant: &tenant,
                    project: &project,
                    process: &process,
                };
                let cleanup = rollback_failed_process_launch_reconnecting(
                    &coordinator,
                    &rollback,
                    &json!({ "error": launch_error.to_string() }),
                );
                if let Err(cleanup_error) = cleanup {
                    anyhow::bail!("{launch_error}; process cleanup also failed: {cleanup_error}");
                }
                return Err(launch_error);
            }
        }
    } else {
        Value::Null
    };
    Ok(json!({
        "command": "run",
        "status": if launch_task_response.get("type").and_then(Value::as_str) == Some("main_launched") { "main_launched" } else { status },
        "project_root": plan.project,
        "entry": plan.entry,
        "tenant": tenant,
        "project": project,
        "user": user,
        "workflow_actor": response.get("actor").cloned().unwrap_or(Value::Null),
        "coordinator": coordinator,
        "process": process,
        "run_start": run_start,
        "task_definition": task_definition,
        "task_instance": task_instance,
        "bundle_build": bundle.build_report,
        "bundle_digest": bundle.digest,
        "bundle_module_size_bytes": bundle.module_size_bytes,
        "entry_export": bundle.entry_export,
        "entry_stable_id": bundle.entry_stable_id,
        "entry_abi_version": 1,
        "worker_placement_requested": launch_task_response.is_object(),
        "task_launch": launch_task_response,
        "coordinator_response": response,
        "coordinator_session_requests": session.requests(),
        "private_website_required": false,
    }))
}

#[allow(clippy::too_many_arguments)]
fn request_process_start_with_rollback(
    session: &mut JsonLineSession,
    request: Value,
    coordinator: &str,
    cli_session: &CliSession,
    fallback_user: &str,
    human_session_secret: Option<&str>,
    tenant: &str,
    project: &str,
    process: &str,
) -> Result<Value> {
    let rollback = ProcessLaunchRollback {
        cli_session,
        fallback_user,
        human_session_secret,
        tenant,
        project,
        process,
    };
    match session.request_allow_error(request) {
        Ok(response) => Ok(response),
        Err(start_error) => {
            let cleanup = rollback_failed_process_launch_reconnecting(
                coordinator,
                &rollback,
                &json!({ "error": start_error.to_string(), "phase": "start_process" }),
            );
            if let Err(cleanup_error) = cleanup {
                let cleanup_message = cleanup_error.to_string();
                if !cleanup_message.contains("process abort requires an active virtual process") {
                    anyhow::bail!(
                        "{start_error}; ambiguous process-start cleanup also failed: {cleanup_error}"
                    );
                }
            }
            Err(start_error)
        }
    }
}

struct ProcessLaunchRollback<'a> {
    cli_session: &'a CliSession,
    fallback_user: &'a str,
    human_session_secret: Option<&'a str>,
    tenant: &'a str,
    project: &'a str,
    process: &'a str,
}

fn validate_inline_bundle_size(module_size_bytes: usize) -> Result<()> {
    if module_size_bytes <= MAX_INLINE_WASM_MODULE_BYTES {
        return Ok(());
    }
    anyhow::bail!(
        "built Wasm module is {module_size_bytes} bytes, but the current {}-byte inline control frame supports at most about {} KiB ({} raw bytes); reduce the bundle dependency or optimization footprint. Larger bundles require an out-of-band transport. No virtual process was created",
        MAX_CONTROL_FRAME_BYTES,
        MAX_INLINE_WASM_MODULE_BYTES / 1024,
        MAX_INLINE_WASM_MODULE_BYTES,
    )
}

fn rollback_failed_process_launch(
    session: &mut JsonLineSession,
    context: &ProcessLaunchRollback<'_>,
    launch_response: &Value,
) -> Result<()> {
    let rollback = authenticated_human_or_local_trusted_workflow(
        json!({
            "type": "abort_process",
            "tenant": context.tenant,
            "project": context.project,
            "process": context.process,
        }),
        context.cli_session,
        context.fallback_user,
        context.human_session_secret,
    );
    let rollback_response = session.request_allow_error(rollback)?;
    if rollback_response.get("type").and_then(Value::as_str) != Some("process_aborted") {
        anyhow::bail!(
            "coordinator main launch failed ({launch_response}) and rollback was not acknowledged ({rollback_response}); inspect virtual process {} before retrying",
            context.process
        );
    }
    Ok(())
}

fn rollback_failed_process_launch_reconnecting(
    coordinator: &str,
    context: &ProcessLaunchRollback<'_>,
    launch_response: &Value,
) -> Result<()> {
    let mut cleanup_session = JsonLineSession::connect(coordinator)
        .context("open a fresh coordinator connection for failed-launch cleanup")?;
    rollback_failed_process_launch(&mut cleanup_session, context, launch_response)
}

fn build_bundle_for_run(project: &Path, entry: &str) -> Result<RunBundle> {
    let build_report = crate::build::build_report(
        BuildArgs {
            project: Some(project.to_path_buf()),
            source_provider: None,
            disabled_source_providers: Vec::new(),
            output: None,
            json: true,
        },
        project.to_path_buf(),
    )?;
    if build_report.get("status").and_then(Value::as_str) != Some("built") {
        let diagnostics = build_report
            .get("diagnostics")
            .cloned()
            .unwrap_or(Value::Null);
        anyhow::bail!("Clusterflux bundle build was blocked before run: {diagnostics}");
    }
    let artifact = build_report
        .get("bundle_artifact")
        .context("bundle build response omitted bundle_artifact")?;
    let module_path = artifact
        .get("module")
        .and_then(Value::as_str)
        .context("bundle build response omitted module path")?;
    let directory = artifact
        .get("directory")
        .and_then(Value::as_str)
        .context("bundle build response omitted bundle directory")?;
    let digest: Digest = serde_json::from_value(
        artifact
            .get("bundle_digest")
            .cloned()
            .context("bundle build response omitted bundle digest")?,
    )?;
    let module = std::fs::read(module_path)
        .with_context(|| format!("failed to read built Wasm module {module_path}"))?;
    let actual_digest = Digest::sha256(&module);
    if actual_digest != digest {
        anyhow::bail!(
            "built Wasm module digest changed before run: expected {digest}, actual {actual_digest}"
        );
    }
    let descriptors_path = Path::new(directory).join("entrypoints.json");
    let descriptors: Vec<Value> = serde_json::from_slice(
        &std::fs::read(&descriptors_path)
            .with_context(|| format!("failed to read {}", descriptors_path.display()))?,
    )?;
    let descriptor = descriptors
        .iter()
        .find(|descriptor| descriptor.get("name").and_then(Value::as_str) == Some(entry))
        .with_context(|| {
            let available = descriptors
                .iter()
                .filter_map(|descriptor| descriptor.get("name").and_then(Value::as_str))
                .collect::<Vec<_>>();
            format!(
                "bundle has no Clusterflux entrypoint `{entry}`; available entrypoints: {available:?}"
            )
        })?;
    if descriptor.get("abi_version").and_then(Value::as_u64) != Some(1) {
        anyhow::bail!("entrypoint `{entry}` does not use supported Clusterflux ABI version 1");
    }
    let entry_export = descriptor
        .get("export")
        .and_then(Value::as_str)
        .filter(|value| !value.trim().is_empty())
        .with_context(|| format!("entrypoint `{entry}` descriptor omitted its Wasm export"))?
        .to_owned();
    let entry_stable_id = descriptor
        .get("stable_id")
        .and_then(Value::as_str)
        .filter(|value| !value.trim().is_empty())
        .with_context(|| format!("entrypoint `{entry}` descriptor omitted its stable id"))?
        .to_owned();
    Ok(RunBundle {
        build_report,
        digest,
        module_size_bytes: module.len(),
        module_base64: BASE64_STANDARD.encode(module),
        entry_export,
        entry_stable_id,
    })
}

fn human_run_session_secret(
    plan: &RunPlan,
    coordinator: &str,
    stored_session: Option<&StoredCliSession>,
) -> Result<Option<String>> {
    if !matches!(plan.session, CliSession::HumanSession) {
        return Ok(None);
    }
    if let Ok(token) = std::env::var("CLUSTERFLUX_TOKEN") {
        if !token.trim().is_empty() {
            return Ok(Some(token));
        }
    }
    Ok(stored_session_for_coordinator(coordinator, stored_session)
        .and_then(|session| session.session_secret.clone()))
}

fn authenticated_human_or_local_trusted_workflow(
    mut request: Value,
    session: &CliSession,
    fallback_user: &str,
    human_session_secret: Option<&str>,
) -> Value {
    if matches!(session, CliSession::HumanSession) {
        if let Some(session_secret) = human_session_secret {
            if let Value::Object(request) = &mut request {
                request.remove("tenant");
                request.remove("project");
                request.remove("actor_user");
                request.remove("actor_agent");
                request.remove("agent_public_key_fingerprint");
                request.remove("agent_signature");
            }
            return json!({
                "type": "authenticated",
                "session_secret": session_secret,
                "request": request,
            });
        }
    }
    add_workflow_actor_fields(&mut request, session, fallback_user);
    request
}

fn run_coordinator_endpoint(
    plan: &RunPlan,
    stored_session: Option<&StoredCliSession>,
) -> Result<String> {
    match &plan.coordinator {
        CoordinatorSelection::Hosted => Ok(stored_session
            .filter(|session| {
                matches!(plan.session, CliSession::HumanSession) && session.session_secret.is_some()
            })
            .map(|session| session.coordinator.clone())
            .or_else(|| plan.hosted_coordinator_endpoint.clone())
            .unwrap_or_else(default_hosted_coordinator_endpoint)),
        CoordinatorSelection::LocalOverride(coordinator) => Ok(coordinator.clone()),
        CoordinatorSelection::LocalOnly => {
            anyhow::bail!("local-only run should execute through local services")
        }
    }
}

pub(crate) fn run_start_summary(response: &Value) -> Value {
    if response.get("type").and_then(Value::as_str) == Some("process_started") {
        return json!({
            "status": "started",
            "accepted": true,
            "process": response.get("process").cloned().unwrap_or(Value::Null),
            "coordinator_epoch": response.get("epoch").cloned().unwrap_or(Value::Null),
            "actor": response.get("actor").cloned().unwrap_or(Value::Null),
            "restart": false,
            "single_active_process_boundary": true,
            "next_actions": [
                "clusterflux process status",
                "clusterflux logs",
                "clusterflux process cancel"
            ],
        });
    }

    let message = response
        .get("message")
        .and_then(Value::as_str)
        .unwrap_or("coordinator rejected run");
    let active_conflict = message.contains("already has active virtual process");
    let machine_error = if active_conflict {
        cli_error_summary_for_category("active_process", message)
    } else {
        cli_error_summary(message)
    };
    let error_category = machine_error
        .get("category")
        .cloned()
        .unwrap_or_else(|| json!("unknown"));
    let stable_exit_code = machine_error
        .get("stable_exit_code")
        .cloned()
        .unwrap_or_else(|| json!(1));
    json!({
        "status": if active_conflict { "blocked_active_process" } else { "coordinator_rejected" },
        "accepted": false,
        "category": if active_conflict { "active_process_already_running" } else { "coordinator" },
        "error_category": error_category,
        "stable_exit_code": stable_exit_code,
        "machine_error": machine_error,
        "message": message,
        "restart": false,
        "single_active_process_boundary": true,
        "safe_failure": true,
        "next_actions": if active_conflict {
            json!([
                "clusterflux process list",
                "clusterflux process status",
                "clusterflux debug attach",
                "clusterflux process restart --yes",
                "clusterflux process cancel --yes",
                "clusterflux process abort --yes",
                "use another Coordinator Project"
            ])
        } else {
            json!(["clusterflux doctor", "check coordinator status"])
        },
    })
}

pub(crate) fn should_execute_local_node(plan: &RunPlan) -> bool {
    match &plan.coordinator {
        CoordinatorSelection::LocalOnly => true,
        CoordinatorSelection::LocalOverride(coordinator) => !coordinator.contains("://"),
        CoordinatorSelection::Hosted => false,
    }
}

fn execute_local_node_run(plan: RunPlan) -> Result<RunExecutionReport> {
    let environments = clusterflux_core::discover_environments(&plan.project)?;
    let detected = clusterflux_core::NodeCapabilities::detect_current();
    if environments.iter().any(|environment| {
        environment
            .requirements
            .capabilities
            .contains(&clusterflux_core::Capability::RootlessPodman)
    }) && !detected
        .capabilities
        .contains(&clusterflux_core::Capability::RootlessPodman)
    {
        anyhow::bail!(
            "local project declares a Linux container environment, but rootless Podman is not available; configure rootless Podman or attach a capable user node"
        );
    }
    let local_coordinator = match &plan.coordinator {
        CoordinatorSelection::LocalOverride(coordinator) => LocalCoordinator::external(coordinator),
        CoordinatorSelection::LocalOnly => LocalCoordinator::start_ephemeral()?,
        CoordinatorSelection::Hosted => anyhow::bail!("local node execution requires local mode"),
    };
    let coordinator_address = local_coordinator.address.clone();
    let mut coordinator_plan = plan.clone();
    coordinator_plan.coordinator =
        CoordinatorSelection::LocalOverride(format!("clusterflux+tcp://{coordinator_address}"));
    let run = coordinator_run_report(coordinator_plan)?;
    let launch_type = run.pointer("/task_launch/type").and_then(Value::as_str);
    if launch_type != Some("main_launched") {
        anyhow::bail!("local coordinator refused the Wasm entrypoint launch: {run}");
    }
    let process = run
        .get("process")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow::anyhow!("coordinator run report omitted virtual process id"))?;
    let task = run
        .pointer("/task_launch/task_instance")
        .or_else(|| run.get("task_instance"))
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow::anyhow!("coordinator run report omitted entrypoint task id"))?;
    let pre_node_process_status = wait_for_local_main_placement(&coordinator_address, process)?;
    let enrollment_grant = create_local_node_enrollment_grant(&coordinator_address)?;
    let mut worker =
        LocalNodeWorker::start(&coordinator_address, &plan.project, &enrollment_grant)?;
    let spawned_node_process_id = worker.process_id;
    let join = wait_for_local_main_completion(&coordinator_address, process, task)?;
    worker.stop();
    let node_report = json!({
        "node_status": "completed",
        "execution_substrate": "wasm",
        "task_spawn_host_import": true,
        "pre_node_process_status": pre_node_process_status,
        "run": run,
        "join": join,
    });

    Ok(RunExecutionReport {
        plan,
        boundary: RunBoundaryEvidence {
            cli_process_started_node_process: true,
            cli_process_started_coordinator_process: local_coordinator.process_id.is_some(),
            coordinator_address,
            coordinator_process_id: local_coordinator.process_id,
            spawned_node_process_id,
            node_session_requests: 0,
        },
        node_report,
    })
}

fn local_process_status(coordinator: &str) -> Result<Value> {
    let mut session = JsonLineSession::connect(coordinator)?;
    session.request_allow_error(json!({
        "type": "list_processes",
        "tenant": "tenant",
        "project": "project",
        "actor_user": "user",
    }))
}

fn create_local_node_enrollment_grant(coordinator: &str) -> Result<String> {
    let mut session = JsonLineSession::connect(coordinator)?;
    let response = session.request_allow_error(json!({
        "type": "create_node_enrollment_grant",
        "tenant": "tenant",
        "project": "project",
        "actor_user": "user",
        "ttl_seconds": 60,
    }))?;
    if response.get("type").and_then(Value::as_str) != Some("node_enrollment_grant_created") {
        anyhow::bail!("local coordinator refused node enrollment: {response}");
    }
    response
        .get("grant")
        .and_then(Value::as_str)
        .map(str::to_owned)
        .context("local coordinator omitted the node enrollment grant")
}

fn wait_for_local_main_placement(coordinator: &str, process: &str) -> Result<Value> {
    let started = Instant::now();
    loop {
        let status = local_process_status(coordinator)?;
        let process_status =
            status
                .get("processes")
                .and_then(Value::as_array)
                .and_then(|processes| {
                    processes.iter().find(|candidate| {
                        candidate.get("process").and_then(Value::as_str) == Some(process)
                    })
                });
        match process_status.and_then(|status| status.get("main_wait_state")) {
            Some(Value::String(wait_state)) if wait_state == "waiting_for_node" => {
                return Ok(status);
            }
            _ if started.elapsed() > Duration::from_secs(30) => anyhow::bail!(
                "local coordinator main `{process}` did not become observably parked on node placement before the local node launch: {status}"
            ),
            _ => thread::sleep(Duration::from_millis(10)),
        }
    }
}

fn wait_for_local_main_completion(coordinator: &str, process: &str, task: &str) -> Result<Value> {
    let started = Instant::now();
    loop {
        let mut session = JsonLineSession::connect(coordinator)?;
        let response = session.request_allow_error(json!({
            "type": "join_task",
            "tenant": "tenant",
            "project": "project",
            "actor_user": "user",
            "process": process,
            "task": task,
        }))?;
        match response.pointer("/join/state").and_then(Value::as_str) {
            Some("Completed") | Some("completed") => return Ok(response),
            Some("Failed") | Some("failed") | Some("Cancelled") | Some("cancelled") => {
                let events = session.request_allow_error(json!({
                    "type": "list_task_events",
                    "tenant": "tenant",
                    "project": "project",
                    "actor_user": "user",
                    "process": process,
                }))?;
                anyhow::bail!("local Wasm entrypoint failed: {response}; task events: {events}")
            }
            _ if started.elapsed() > clusterflux_core::limits::task_join_timeout() => {
                return Err(anyhow::Error::new(
                    clusterflux_core::limits::TaskJoinError::timeout(
                        clusterflux_core::TaskInstanceId::from(task),
                        clusterflux_core::limits::task_join_timeout(),
                    ),
                ));
            }
            _ => thread::sleep(Duration::from_millis(20)),
        }
    }
}

pub(crate) fn session_from_env() -> Result<CliSession> {
    if let Some(session) = agent_session_from_keys(
        std::env::var("CLUSTERFLUX_AGENT_ID").unwrap_or_else(|_| "agent".to_owned()),
        std::env::var("CLUSTERFLUX_AGENT_PUBLIC_KEY").ok(),
        std::env::var("CLUSTERFLUX_AGENT_PRIVATE_KEY").ok(),
    )? {
        return Ok(session);
    }
    if std::env::var_os("CLUSTERFLUX_TOKEN").is_some() {
        return Ok(CliSession::HumanSession);
    }
    Ok(CliSession::Anonymous)
}

pub(crate) fn agent_session_from_keys(
    agent: String,
    configured_public_key: Option<String>,
    private_key: Option<String>,
) -> Result<Option<CliSession>> {
    if let Some(private_key) = private_key {
        let derived_public_key = agent_ed25519_public_key_from_private_key(&private_key)
            .map_err(anyhow::Error::msg)
            .context("CLUSTERFLUX_AGENT_PRIVATE_KEY is not a valid Ed25519 private key")?;
        if let Some(configured_public_key) = configured_public_key.as_deref() {
            if configured_public_key != derived_public_key {
                anyhow::bail!(
                    "CLUSTERFLUX_AGENT_PUBLIC_KEY does not match CLUSTERFLUX_AGENT_PRIVATE_KEY"
                );
            }
        }
        return Ok(Some(CliSession::AgentPublicKey {
            agent,
            public_key_fingerprint: Digest::sha256(derived_public_key.as_bytes()),
            public_key: derived_public_key,
            private_key: Some(private_key),
            browser_interaction_required: false,
        }));
    }
    if configured_public_key.is_some() {
        anyhow::bail!(
            "CLUSTERFLUX_AGENT_PUBLIC_KEY identifies a registered key but cannot authenticate; set CLUSTERFLUX_AGENT_PRIVATE_KEY to prove key possession"
        );
    }
    Ok(None)
}

pub(crate) fn session_from_sources(project: &Path) -> Result<CliSession> {
    let session = session_from_env()?;
    if session.is_authenticated() {
        return Ok(session);
    }
    if read_cli_session(project)?.is_some() {
        return Ok(CliSession::HumanSession);
    }
    Ok(CliSession::Anonymous)
}

fn add_workflow_actor_fields(request: &mut Value, session: &CliSession, fallback_user: &str) {
    let Value::Object(map) = request else {
        return;
    };
    match session {
        CliSession::AgentPublicKey {
            agent,
            public_key: _,
            public_key_fingerprint,
            private_key,
            ..
        } => {
            map.insert("actor_agent".to_owned(), json!(agent));
            map.insert(
                "agent_public_key_fingerprint".to_owned(),
                json!(public_key_fingerprint),
            );
            if let Some(signature) = agent_signature_for_request(map, agent, private_key.as_deref())
            {
                map.insert("agent_signature".to_owned(), json!(signature));
            }
        }
        CliSession::HumanSession | CliSession::Anonymous => {
            map.insert("actor_user".to_owned(), json!(fallback_user));
        }
    }
}

fn agent_signature_for_request(
    request: &serde_json::Map<String, Value>,
    agent: &str,
    private_key: Option<&str>,
) -> Option<clusterflux_core::AgentSignedRequest> {
    let private_key = private_key?;
    let payload = Value::Object(request.clone());
    let scope = agent_workflow_request_scope_from_payload(&payload).ok()?;
    let agent = clusterflux_core::AgentId::from(agent);
    let payload_digest = signed_request_payload_digest(&payload);
    sign_agent_workflow_request(
        private_key,
        scope.for_agent(&agent),
        &payload_digest,
        crate::tools::command_nonce("agent-signature"),
        crate::tools::unix_timestamp_seconds(),
    )
    .ok()
}

#[cfg(test)]
mod transactional_launch_tests {
    use std::io::{BufRead as _, ErrorKind, Write as _};
    use std::net::TcpListener;

    use super::*;

    #[test]
    fn inline_module_limit_is_derived_from_the_control_frame() {
        assert_eq!(MAX_CONTROL_FRAME_BYTES, 1024 * 1024);
        assert_eq!(MAX_INLINE_WASM_MODULE_BYTES % 3, 0);
        let inline_limit = std::hint::black_box(MAX_INLINE_WASM_MODULE_BYTES);
        assert!((690 * 1024..=700 * 1024).contains(&inline_limit));
        validate_inline_bundle_size(MAX_INLINE_WASM_MODULE_BYTES).unwrap();
        let error = validate_inline_bundle_size(MAX_INLINE_WASM_MODULE_BYTES + 1)
            .unwrap_err()
            .to_string();
        assert!(error.contains("No virtual process was created"));
        assert!(error.contains("out-of-band transport"));
    }

    #[test]
    fn build_failure_occurs_before_any_coordinator_connection() {
        let project = tempfile::tempdir().unwrap();
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        listener.set_nonblocking(true).unwrap();
        let plan = RunPlan {
            project: project.path().to_path_buf(),
            entry: "build".to_owned(),
            coordinator: CoordinatorSelection::LocalOverride(
                listener.local_addr().unwrap().to_string(),
            ),
            hosted_coordinator_endpoint: None,
            session: CliSession::Anonymous,
        };

        let error = coordinator_run_report(plan).unwrap_err().to_string();
        assert!(
            error.contains("Cargo.toml") || error.contains("project"),
            "unexpected build error: {error}"
        );
        assert_eq!(listener.accept().unwrap_err().kind(), ErrorKind::WouldBlock);
    }

    #[test]
    fn failed_main_launch_uses_explicit_abort_rollback() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let address = listener.local_addr().unwrap();
        let server = std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            let mut line = String::new();
            std::io::BufReader::new(stream.try_clone().unwrap())
                .read_line(&mut line)
                .unwrap();
            let wire: Value = serde_json::from_str(&line).unwrap();
            let payload = wire.get("payload").unwrap();
            assert_eq!(
                payload.get("type").and_then(Value::as_str),
                Some("abort_process")
            );
            assert_eq!(
                payload.get("tenant").and_then(Value::as_str),
                Some("tenant-a")
            );
            assert_eq!(
                payload.get("project").and_then(Value::as_str),
                Some("project-a")
            );
            assert_eq!(
                payload.get("process").and_then(Value::as_str),
                Some("vp-current")
            );
            writeln!(
                stream,
                "{}",
                json!({
                    "type": "process_aborted",
                    "process": "vp-current",
                    "aborted_tasks": [],
                    "affected_nodes": []
                })
            )
            .unwrap();
        });
        let mut session = JsonLineSession::connect(&address.to_string()).unwrap();
        let rollback = ProcessLaunchRollback {
            cli_session: &CliSession::Anonymous,
            fallback_user: "user-a",
            human_session_secret: None,
            tenant: "tenant-a",
            project: "project-a",
            process: "vp-current",
        };
        rollback_failed_process_launch(
            &mut session,
            &rollback,
            &json!({"type": "error", "message": "main launch failed"}),
        )
        .unwrap();
        server.join().unwrap();
    }

    #[test]
    fn dropped_start_response_reconnects_and_aborts_ambiguous_process() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let address = listener.local_addr().unwrap().to_string();
        let server = std::thread::spawn(move || {
            let (stream, _) = listener.accept().unwrap();
            let mut start_line = String::new();
            std::io::BufReader::new(stream.try_clone().unwrap())
                .read_line(&mut start_line)
                .unwrap();
            assert!(start_line.contains("\"type\":\"start_process\""));
            drop(stream);

            let (mut cleanup_stream, _) = listener.accept().unwrap();
            let mut cleanup_line = String::new();
            std::io::BufReader::new(cleanup_stream.try_clone().unwrap())
                .read_line(&mut cleanup_line)
                .unwrap();
            assert!(cleanup_line.contains("\"type\":\"abort_process\""));
            writeln!(
                cleanup_stream,
                "{}",
                json!({
                    "type": "process_aborted",
                    "process": "vp-current",
                    "aborted_tasks": [],
                    "affected_nodes": []
                })
            )
            .unwrap();
        });
        let mut session = JsonLineSession::connect(&address).unwrap();
        let error = request_process_start_with_rollback(
            &mut session,
            json!({
                "type": "start_process",
                "tenant": "tenant-a",
                "project": "project-a",
                "actor_user": "user-a",
                "process": "vp-current",
                "restart": false
            }),
            &address,
            &CliSession::Anonymous,
            "user-a",
            None,
            "tenant-a",
            "project-a",
            "vp-current",
        )
        .unwrap_err()
        .to_string();
        assert!(error.contains("closed") || error.contains("response"));
        server.join().unwrap();
    }
}