//! OpenShell gRPC client — mTLS connection, sandbox readiness polling.
//! Also provides CLI wrappers for sandbox lifecycle, SSH config, and remote exec.

use std::net::IpAddr;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::time::Duration;

use tokio::process::Command;
use tonic::transport::{Certificate, Channel, ClientTlsConfig, Identity};

use crate::openshell_proto::openshell::datamodel::v1::{
    Sandbox, SandboxCondition, SandboxPhase, SandboxStatus,
};
use crate::openshell_proto::openshell::v1::open_shell_client::OpenShellClient;
use crate::openshell_proto::openshell::v1::{ExecSandboxRequest, GetSandboxRequest};

/// SANDBOX_PHASE_READY value from openshell.datamodel.v1.SandboxPhase.
const SANDBOX_PHASE_READY: i32 = SandboxPhase::Ready as i32;

/// SANDBOX_PHASE_ERROR value from openshell.datamodel.v1.SandboxPhase.
const SANDBOX_PHASE_ERROR: i32 = SandboxPhase::Error as i32;

/// Timeout for SSH ControlMaster probe and control operations (`-O check`, `-O exit`).
///
/// These operations target a local Unix-domain socket — they complete in milliseconds
/// under normal conditions. 5 seconds is generous enough to absorb a loaded system
/// while still surfacing a dead or unresponsive master quickly.
const SSH_CONTROL_OP_TIMEOUT_SECS: u64 = 5;

/// Path to `mcp.json` inside an OpenShell sandbox.
pub const SANDBOX_MCP_JSON_PATH: &str = "/sandbox/mcp.json";

const DEFAULT_GATEWAY_ENDPOINT: &str = "https://127.0.0.1:8080";
pub const DEFAULT_REBUILDABLE_BACKUP_EXCLUDES: &[&str] = &[".cache", ".venv", ".npm", ".uv"];

/// Generate deterministic fallback sandbox name from agent name.
///
/// This deliberately still returns `rightclaw-{agent_name}` after the
/// `right-agent` rename to keep existing pre-rename agents working —
/// their `agent.yaml` has no explicit `sandbox.name`, so this fallback
/// is what their bot uses to find their existing sandbox.
///
/// New agents (created via `right agent init`) get an explicit
/// `sandbox.name: right-{agent_name}` written into `agent.yaml` and
/// never hit this fallback.
pub fn sandbox_name(agent_name: &str) -> String {
    format!("rightclaw-{agent_name}")
}

/// SSH host alias for an agent's sandbox (used in SSH config).
pub fn ssh_host(agent_name: &str) -> String {
    format!("openshell-rightclaw-{agent_name}")
}

/// Resolve sandbox name: explicit from config, or deterministic fallback.
pub fn resolve_sandbox_name(agent_name: &str, explicit_name: Option<&str>) -> String {
    explicit_name
        .map(str::to_owned)
        .unwrap_or_else(|| sandbox_name(agent_name))
}

/// SSH host alias from sandbox name (not agent name).
pub fn ssh_host_for_sandbox(sandbox_name: &str) -> String {
    format!("openshell-{sandbox_name}")
}

/// Filesystem path to the OpenSSH ControlMaster socket for a sandbox.
///
/// Keyed on sandbox name so each sandbox gets its own master. Sandbox
/// migration produces a new sandbox name → new ControlPath → no stale-master
/// reuse risk. See spec for rationale (OpenSSH matches multiplex candidates
/// on `(user, host, port)` and the host alias is per-sandbox).
pub fn control_master_socket_path(ssh_config_dir: &Path, sandbox_name: &str) -> PathBuf {
    ssh_config_dir.join(format!("{sandbox_name}.cm"))
}

/// Build the ControlMaster directive block to append to an SSH config file.
///
/// Uses the resolved absolute socket path (no `~` expansion ambiguity) so the
/// directive behaves identically regardless of the bot's `HOME` env var.
pub fn control_master_directives(ssh_config_dir: &Path, sandbox_name: &str) -> String {
    let socket = control_master_socket_path(ssh_config_dir, sandbox_name);
    format!(
        "\n\
        # Connection multiplexing — see docs/superpowers/specs/2026-05-05-ssh-controlmaster-design.md\n\
        ControlMaster auto\n\
        ControlPath {}\n\
        ControlPersist yes\n",
        socket.display()
    )
}

/// Resolve the default mTLS directory for the OpenShell gateway.
///
/// Checks `OPENSHELL_MTLS_DIR` env var first, then falls back to the
/// platform config directory: `<config_dir>/openshell/gateways/openshell/mtls`.
pub fn default_mtls_dir() -> PathBuf {
    std::env::var("OPENSHELL_MTLS_DIR")
        .map(PathBuf::from)
        .unwrap_or_else(|_| {
            // OpenShell uses XDG_CONFIG_HOME (~/.config) on all platforms,
            // not the macOS-native ~/Library/Application Support.
            let config_base = std::env::var("XDG_CONFIG_HOME")
                .map(PathBuf::from)
                .unwrap_or_else(|_| {
                    dirs::home_dir()
                        .unwrap_or_else(|| PathBuf::from("/"))
                        .join(".config")
                });
            config_base.join("openshell/gateways/openshell/mtls")
        })
}

/// Result of the OpenShell pre-flight check.
#[derive(Debug)]
pub enum OpenShellStatus {
    /// mTLS certs are present — sandbox mode can proceed.
    Ready(PathBuf),
    /// `openshell` binary is not in PATH.
    NotInstalled,
    /// Binary is installed but no gateway has been started (mTLS dir missing).
    NoGateway(PathBuf),
    /// Gateway metadata exists but mTLS certs are missing/corrupt.
    BrokenGateway(PathBuf),
}

/// Check whether the OpenShell environment is ready for sandbox mode.
///
/// Returns a diagnostic enum that callers can use to give targeted
/// guidance (install, gateway start, or destroy+recreate).
pub fn preflight_check() -> OpenShellStatus {
    let mtls_dir = default_mtls_dir();

    // Check if all three mTLS files are present.
    let has_certs = mtls_dir.join("ca.crt").exists()
        && mtls_dir.join("tls.crt").exists()
        && mtls_dir.join("tls.key").exists();

    if has_certs {
        return OpenShellStatus::Ready(mtls_dir);
    }

    // No certs — figure out why.
    if which::which("openshell").is_err() {
        return OpenShellStatus::NotInstalled;
    }

    // Binary exists — check if a gateway has been configured.
    // `openshell gateway info` exits non-zero when no gateway metadata exists.
    let gateway_exists = std::process::Command::new("openshell")
        .args(["gateway", "info"])
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status()
        .map(|s| s.success())
        .unwrap_or(false);

    if gateway_exists {
        OpenShellStatus::BrokenGateway(mtls_dir)
    } else {
        OpenShellStatus::NoGateway(mtls_dir)
    }
}

fn gateway_endpoint() -> String {
    std::env::var("OPENSHELL_GATEWAY_ENDPOINT")
        .ok()
        .map(|s| s.trim().to_owned())
        .filter(|s| !s.is_empty())
        .or_else(gateway_endpoint_from_status)
        .unwrap_or_else(|| DEFAULT_GATEWAY_ENDPOINT.to_owned())
}

fn gateway_endpoint_from_status() -> Option<String> {
    let output = std::process::Command::new("openshell")
        .arg("status")
        .env("NO_COLOR", "1")
        .output()
        .ok()?;

    if !output.status.success() {
        return None;
    }

    gateway_endpoint_from_status_output(&String::from_utf8_lossy(&output.stdout))
        .or_else(|| gateway_endpoint_from_status_output(&String::from_utf8_lossy(&output.stderr)))
}

fn gateway_endpoint_from_status_output(output: &str) -> Option<String> {
    output
        .find("Server:")
        .or_else(|| output.find("Endpoint:"))
        .and_then(|idx| extract_first_url(&output[idx..]))
}

fn extract_first_url(text: &str) -> Option<String> {
    let https = text.find("https://");
    let http = text.find("http://");
    let start = match (https, http) {
        (Some(https), Some(http)) => https.min(http),
        (Some(https), None) => https,
        (None, Some(http)) => http,
        (None, None) => return None,
    };
    let endpoint: String = text[start..]
        .chars()
        .take_while(|ch| !ch.is_whitespace() && !ch.is_control())
        .collect();

    (!endpoint.is_empty()).then_some(endpoint)
}

/// Connect to the OpenShell gRPC server with mTLS.
///
/// Reads CA cert, client cert, and client key from `mtls_dir`:
/// - `ca.crt`  — CA certificate
/// - `tls.crt` — client certificate
/// - `tls.key` — client private key
pub async fn connect_grpc(mtls_dir: &Path) -> miette::Result<OpenShellClient<Channel>> {
    let ca_pem = tokio::fs::read(mtls_dir.join("ca.crt"))
        .await
        .map_err(|e| miette::miette!("failed to read ca.crt from {}: {e:#}", mtls_dir.display()))?;

    let client_cert = tokio::fs::read(mtls_dir.join("tls.crt"))
        .await
        .map_err(|e| {
            miette::miette!("failed to read tls.crt from {}: {e:#}", mtls_dir.display())
        })?;

    let client_key = tokio::fs::read(mtls_dir.join("tls.key"))
        .await
        .map_err(|e| {
            miette::miette!("failed to read tls.key from {}: {e:#}", mtls_dir.display())
        })?;

    let tls = ClientTlsConfig::new()
        .ca_certificate(Certificate::from_pem(ca_pem))
        .identity(Identity::from_pem(client_cert, client_key))
        .domain_name("localhost");

    let endpoint = gateway_endpoint();
    let channel = Channel::from_shared(endpoint.clone())
        .map_err(|e| miette::miette!("invalid OpenShell gRPC endpoint '{endpoint}': {e:#}"))?
        .tls_config(tls)
        .map_err(|e| miette::miette!("TLS config error: {e:#}"))?
        .connect()
        .await
        .map_err(|e| miette::miette!("gRPC connect to {endpoint} failed: {e:#}"))?;

    Ok(OpenShellClient::new(channel))
}

/// Check if a sandbox exists (any phase). Returns `false` only on `NotFound`.
pub async fn sandbox_exists(
    client: &mut OpenShellClient<Channel>,
    name: &str,
) -> miette::Result<bool> {
    match client
        .get_sandbox(GetSandboxRequest {
            name: name.to_owned(),
        })
        .await
    {
        Ok(_) => Ok(true),
        Err(status) if status.code() == tonic::Code::NotFound => Ok(false),
        Err(e) => Err(miette::miette!("GetSandbox RPC failed for '{name}': {e:#}")),
    }
}

/// Poll until a sandbox is fully deleted (gRPC returns `NotFound`).
pub async fn wait_for_deleted(
    client: &mut OpenShellClient<Channel>,
    name: &str,
    timeout_secs: u64,
    poll_interval_secs: u64,
) -> miette::Result<()> {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);
    let interval = Duration::from_secs(poll_interval_secs);

    loop {
        if !sandbox_exists(client, name).await? {
            tracing::info!(sandbox = name, "sandbox fully deleted");
            return Ok(());
        }

        if tokio::time::Instant::now() + interval > deadline {
            return Err(miette::miette!(
                "sandbox '{name}' was not deleted within {timeout_secs}s"
            ));
        }

        tracing::debug!(
            sandbox = name,
            "sandbox still exists, polling again in {poll_interval_secs}s"
        );
        tokio::time::sleep(interval).await;
    }
}

/// Check whether a sandbox has reached READY phase.
///
/// Returns `Ok(false)` when the sandbox does not exist yet (gRPC `NotFound`),
/// which is the normal state right after `spawn_sandbox` — the create process
/// hasn't registered the sandbox in OpenShell's registry yet.
pub async fn is_sandbox_ready(
    client: &mut OpenShellClient<Channel>,
    name: &str,
) -> miette::Result<bool> {
    let Some(readiness) = get_sandbox_readiness(client, name).await? else {
        return Ok(false);
    };
    Ok(readiness.is_ready())
}

/// Poll until a sandbox reaches READY phase, or timeout.
pub async fn wait_for_ready(
    client: &mut OpenShellClient<Channel>,
    name: &str,
    timeout_secs: u64,
    poll_interval_secs: u64,
) -> miette::Result<()> {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);
    let interval = Duration::from_secs(poll_interval_secs);
    let mut last_readiness = None;

    loop {
        match get_sandbox_readiness(client, name).await? {
            Some(readiness) if readiness.is_ready() => {
                tracing::info!(sandbox = name, "sandbox is READY");
                return Ok(());
            }
            Some(readiness) if readiness.is_error() => {
                return Err(miette::miette!(
                    "sandbox '{name}' entered ERROR while waiting for READY: {}",
                    readiness.describe()
                ));
            }
            Some(readiness) => {
                tracing::debug!(
                    sandbox = name,
                    phase = sandbox_phase_name(readiness.phase),
                    status = %readiness.status_summary,
                    "sandbox not ready"
                );
                last_readiness = Some(readiness);
            }
            None => {
                tracing::debug!(
                    sandbox = name,
                    "sandbox not found yet (expected during creation)"
                );
            }
        }

        if tokio::time::Instant::now() + interval > deadline {
            let last_status = last_readiness
                .as_ref()
                .map(SandboxReadiness::describe)
                .unwrap_or_else(|| "sandbox was not found".to_owned());
            return Err(miette::miette!(
                "sandbox '{name}' did not become READY within {timeout_secs}s ({last_status})"
            ));
        }

        tracing::debug!(
            sandbox = name,
            "sandbox not ready, polling again in {poll_interval_secs}s"
        );
        tokio::time::sleep(interval).await;
    }
}

#[derive(Debug, Clone, Eq, PartialEq)]
struct SandboxReadiness {
    phase: i32,
    status_summary: String,
}

impl SandboxReadiness {
    fn from_sandbox(sandbox: Sandbox) -> Self {
        Self {
            phase: sandbox.phase,
            status_summary: summarize_sandbox_status(sandbox.status.as_ref()),
        }
    }

    fn is_ready(&self) -> bool {
        self.phase == SANDBOX_PHASE_READY
    }

    fn is_error(&self) -> bool {
        self.phase == SANDBOX_PHASE_ERROR
    }

    fn describe(&self) -> String {
        format!(
            "phase={} status={}",
            sandbox_phase_name(self.phase),
            self.status_summary
        )
    }
}

async fn get_sandbox_readiness(
    client: &mut OpenShellClient<Channel>,
    name: &str,
) -> miette::Result<Option<SandboxReadiness>> {
    let resp = match client
        .get_sandbox(GetSandboxRequest {
            name: name.to_owned(),
        })
        .await
    {
        Ok(r) => r,
        Err(status) if status.code() == tonic::Code::NotFound => return Ok(None),
        Err(e) => {
            return Err(miette::miette!("GetSandbox RPC failed for '{name}': {e:#}"));
        }
    };

    let sandbox = resp
        .into_inner()
        .sandbox
        .ok_or_else(|| miette::miette!("GetSandbox returned empty response for '{name}'"))?;

    Ok(Some(SandboxReadiness::from_sandbox(sandbox)))
}

fn sandbox_phase_name(phase: i32) -> &'static str {
    match SandboxPhase::try_from(phase) {
        Ok(SandboxPhase::Unspecified) => "UNSPECIFIED",
        Ok(SandboxPhase::Provisioning) => "PROVISIONING",
        Ok(SandboxPhase::Ready) => "READY",
        Ok(SandboxPhase::Error) => "ERROR",
        Ok(SandboxPhase::Deleting) => "DELETING",
        Ok(SandboxPhase::Unknown) => "UNKNOWN",
        Err(_) => "UNRECOGNIZED",
    }
}

fn summarize_sandbox_status(status: Option<&SandboxStatus>) -> String {
    let Some(status) = status else {
        return "unavailable".to_owned();
    };

    let mut parts = Vec::new();
    push_status_field(&mut parts, "sandbox_name", &status.sandbox_name);
    push_status_field(&mut parts, "agent_pod", &status.agent_pod);
    push_status_field(&mut parts, "agent_fd", &status.agent_fd);
    push_status_field(&mut parts, "sandbox_fd", &status.sandbox_fd);

    if status.conditions.is_empty() {
        parts.push("conditions=[]".to_owned());
    } else {
        let mut conditions = status
            .conditions
            .iter()
            .take(5)
            .map(summarize_sandbox_condition)
            .collect::<Vec<_>>();
        if status.conditions.len() > 5 {
            conditions.push(format!("... {} more", status.conditions.len() - 5));
        }
        parts.push(format!("conditions=[{}]", conditions.join("; ")));
    }

    if parts.is_empty() {
        "empty".to_owned()
    } else {
        parts.join(" ")
    }
}

fn summarize_sandbox_condition(condition: &SandboxCondition) -> String {
    let mut parts = Vec::new();
    match (
        condition.r#type.trim().is_empty(),
        condition.status.trim().is_empty(),
    ) {
        (false, false) => parts.push(format!(
            "{}={}",
            compact_status_field(&condition.r#type),
            compact_status_field(&condition.status)
        )),
        (false, true) => parts.push(compact_status_field(&condition.r#type)),
        (true, false) => parts.push(format!(
            "status={}",
            compact_status_field(&condition.status)
        )),
        (true, true) => {}
    }
    push_status_field(&mut parts, "reason", &condition.reason);
    push_status_field(&mut parts, "message", &condition.message);
    push_status_field(
        &mut parts,
        "last_transition_time",
        &condition.last_transition_time,
    );

    if parts.is_empty() {
        "empty".to_owned()
    } else {
        parts.join(" ")
    }
}

fn push_status_field(parts: &mut Vec<String>, name: &str, value: &str) {
    if value.trim().is_empty() {
        return;
    }
    parts.push(format!("{name}={}", compact_status_field(value)));
}

fn compact_status_field(value: &str) -> String {
    let compact = value.split_whitespace().collect::<Vec<_>>().join(" ");
    const MAX_CHARS: usize = 180;
    if compact.chars().count() <= MAX_CHARS {
        return compact;
    }

    let mut truncated = compact.chars().take(MAX_CHARS).collect::<String>();
    truncated.push_str("...");
    truncated
}

/// Spawn an OpenShell sandbox. Returns a [`ProcessGroupChild`] handle.
///
/// On Drop, the openshell CLI process and all its descendants (ssh,
/// ssh-proxy, internal k3s spawns) are killed via `killpg(SIGKILL)`.
/// Callers that need the sandbox to outlive the Rust process must
/// `std::mem::forget` the returned handle (we currently have no such
/// callers — the CLI process does nothing useful after READY, and the
/// sandbox itself lives in k3s state, not in the CLI process).
pub fn spawn_sandbox(
    name: &str,
    policy_path: &Path,
    upload_dir: Option<&Path>,
) -> miette::Result<right_process::ProcessGroupChild> {
    let mut cmd = Command::new("openshell");
    cmd.args(["sandbox", "create", "--name", name, "--policy"]);
    cmd.arg(policy_path);
    cmd.arg("--no-tty");

    if let Some(dir) = upload_dir {
        cmd.arg("--upload");
        cmd.arg(dir);
    }

    // `openshell sandbox create` is long-running; callers wait for READY via
    // gRPC and kill this process afterwards. Do not pipe output unless it is
    // drained concurrently, or fresh CI runs can block before the sandbox is
    // registered.
    cmd.stdout(Stdio::inherit());
    cmd.stderr(Stdio::inherit());

    let child = right_process::ProcessGroupChild::spawn(cmd)
        .map_err(|e| miette::miette!("failed to spawn openshell sandbox create: {e:#}"))?;

    tracing::info!(sandbox = name, "spawned sandbox create process");
    Ok(child)
}

/// Test-only concurrency limiter for live OpenShell sandbox tests.
///
/// Holds one of [`max_concurrent_sandbox_tests`] exclusive file locks so the
/// total number of concurrently running sandbox-creating tests across the
/// entire workspace — including tests in separate test binaries — is capped.
///
/// Drop releases the lock. Acquire via [`acquire_sandbox_slot`].
pub struct SandboxTestSlot {
    _file: std::fs::File,
}

/// Default maximum number of live OpenShell sandbox tests allowed in parallel.
/// See [`acquire_sandbox_slot`].
pub const DEFAULT_MAX_CONCURRENT_SANDBOX_TESTS: u8 = 30;

const SANDBOX_SLOT_LIMIT_ENV: &str = "RIGHT_MAX_CONCURRENT_SANDBOX_TESTS";

fn sandbox_slot_limit_from_env_value(value: Option<&str>) -> u8 {
    value
        .and_then(|value| value.parse::<u8>().ok())
        .filter(|value| *value > 0)
        .unwrap_or(DEFAULT_MAX_CONCURRENT_SANDBOX_TESTS)
}

/// Maximum number of live OpenShell sandbox tests allowed to run in parallel.
///
/// Defaults to [`DEFAULT_MAX_CONCURRENT_SANDBOX_TESTS`]. CI jobs can set
/// `RIGHT_MAX_CONCURRENT_SANDBOX_TESTS=1` to serialize only live sandbox
/// creation while preserving normal Cargo test parallelism for the rest of the
/// workspace.
pub fn max_concurrent_sandbox_tests() -> u8 {
    sandbox_slot_limit_from_env_value(std::env::var(SANDBOX_SLOT_LIMIT_ENV).ok().as_deref())
}

/// Acquire one of [`max_concurrent_sandbox_tests`] parallel-sandbox-test slots.
///
/// Blocks (polling every 200ms) until a slot is free. Cross-process via
/// `fs4` advisory file locks on `$TMPDIR/right-sandbox-slot-{N}.lock`.
/// Drop the returned [`SandboxTestSlot`] to release the slot.
///
/// This function is intended for use in test code that calls [`spawn_sandbox`]
/// — it prevents `cargo test --workspace` from fork-bombing the host when
/// multiple test binaries run in parallel, each starting Docker/k3s containers.
pub fn acquire_sandbox_slot() -> SandboxTestSlot {
    loop {
        for slot in 1..=max_concurrent_sandbox_tests() {
            let path = std::env::temp_dir().join(format!("right-sandbox-slot-{slot}.lock"));
            let file = std::fs::OpenOptions::new()
                .create(true)
                .truncate(false)
                .write(true)
                .open(&path)
                .expect("open sandbox-slot lock file");
            match file.try_lock() {
                Ok(()) => return SandboxTestSlot { _file: file },
                Err(std::fs::TryLockError::WouldBlock) => continue, // another holder
                Err(std::fs::TryLockError::Error(e)) => {
                    panic!("sandbox-slot lock {}: {e:#}", path.display())
                }
            }
        }
        std::thread::sleep(std::time::Duration::from_millis(200));
    }
}

/// Test-only mutex for any test resource keyed by a logical name.
///
/// Held across process boundaries (including different git worktrees and
/// different `cargo test` binaries) via an advisory file lock at
/// `$TMPDIR/right-test-name-<sanitized>.lock`. Drop releases the lock; the
/// kernel also releases on process death.
///
/// Acquire via [`acquire_test_name_lock`].
pub struct TestNameLock {
    _file: std::fs::File,
}

/// Acquire an exclusive cross-process lock on `name`.
///
/// Two callers — in the same process or different processes, including
/// different worktrees running the same test — that pass the same `name`
/// will serialize. Different names do not contend with each other.
///
/// Blocks (polling every 100ms) until the lock is free. Drop the returned
/// [`TestNameLock`] to release.
///
/// Use this to wrap any test that owns a non-discriminated shared resource:
/// a sandbox name, a fixed TCP port, etc. The 30-slot cap from
/// [`acquire_sandbox_slot`] is orthogonal and still applies for sandbox
/// tests.
pub fn acquire_test_name_lock(name: &str) -> TestNameLock {
    let safe: String = name
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
                c
            } else {
                '_'
            }
        })
        .collect();
    let path = std::env::temp_dir().join(format!("right-test-name-{safe}.lock"));
    let file = std::fs::OpenOptions::new()
        .create(true)
        .truncate(false)
        .write(true)
        .open(&path)
        .unwrap_or_else(|e| panic!("open test-name lock {}: {e:#}", path.display()));
    loop {
        match file.try_lock() {
            Ok(()) => return TestNameLock { _file: file },
            Err(std::fs::TryLockError::WouldBlock) => {
                std::thread::sleep(std::time::Duration::from_millis(100));
            }
            Err(std::fs::TryLockError::Error(e)) => {
                panic!("lock test-name file {}: {e:#}", path.display())
            }
        }
    }
}

/// Run `openshell sandbox ssh-config NAME` and write the output to
/// `config_dir/{name}.ssh-config`. Returns the path of the written file.
pub async fn generate_ssh_config(name: &str, config_dir: &Path) -> miette::Result<PathBuf> {
    let mut cmd = Command::new("openshell");
    cmd.args(["sandbox", "ssh-config", name]);
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());

    let mut child = right_process::ProcessGroupChild::spawn(cmd)
        .map_err(|e| miette::miette!("failed to spawn openshell sandbox ssh-config: {e:#}"))?;

    let output = child
        .wait_with_output()
        .await
        .map_err(|e| miette::miette!("openshell sandbox ssh-config wait failed: {e:#}"))?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        return Err(miette::miette!(
            "openshell sandbox ssh-config failed (exit {}): {stderr}",
            output.status
        ));
    }

    let dest = config_dir.join(format!("{name}.ssh-config"));
    let mut content = output.stdout;
    let directives = control_master_directives(config_dir, name);
    content.extend_from_slice(directives.as_bytes());
    tokio::fs::write(&dest, &content)
        .await
        .map_err(|e| miette::miette!("failed to write ssh-config to {}: {e:#}", dest.display()))?;

    tracing::info!(sandbox = name, path = %dest.display(), "wrote ssh-config");
    Ok(dest)
}

/// Apply a policy to a running sandbox via `openshell policy set`.
///
/// Uses `--wait` to block until the sandbox confirms it loaded the new policy.
pub async fn apply_policy(name: &str, policy_path: &Path) -> miette::Result<()> {
    let mut cmd = Command::new("openshell");
    cmd.args(["policy", "set", name, "--policy"])
        .arg(policy_path)
        .args(["--wait", "--timeout", "30"]);
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());

    let mut child = right_process::ProcessGroupChild::spawn(cmd)
        .map_err(|e| miette::miette!("failed to spawn openshell policy set: {e:#}"))?;

    let output = child
        .wait_with_output()
        .await
        .map_err(|e| miette::miette!("openshell policy set wait failed: {e:#}"))?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        return Err(miette::miette!(
            "openshell policy set failed (exit {}): {stderr}",
            output.status
        ));
    }

    tracing::info!(sandbox = name, "policy applied");
    Ok(())
}

/// Execute a command inside a sandbox over SSH.
///
/// Uses `-F config_path` to pick up the sandbox SSH config, and
/// `host` as the SSH host alias (see [`ssh_host`]).
/// Returns stdout on success.
///
/// # ControlMaster compatibility
///
/// This function intentionally does NOT use [`ProcessGroupChild`]. The SSH
/// config appended by [`generate_ssh_config`] includes `ControlMaster auto /
/// ControlPersist yes`, so the first `ssh_exec` call spawns a background
/// master process and a ProxyCommand that serves as the master's transport.
/// Both are in the *same OS process group* as the initial `ssh` client.
///
/// Using `ProcessGroupChild` (which calls `killpg(SIGKILL)` on drop) would
/// kill the ProxyCommand and the master with it — defeating multiplexing.
/// Instead, we use a plain `tokio::process::Child` with `kill_on_drop(false)`.
///
/// On timeout or `tokio::select!` cancellation the direct `ssh` client process
/// is killed individually via an RAII [`SshClientKillGuard`]; the background
/// master and its ProxyCommand are left intact so subsequent calls can still
/// reuse them. Cleanup happens via [`tear_down_control_master`] at bot
/// shutdown or sandbox migration, and via [`clean_stale_control_master`] at
/// bot startup.
pub async fn ssh_exec(
    config_path: &Path,
    host: &str,
    cmd: &[&str],
    timeout_secs: u64,
) -> miette::Result<String> {
    let mut command = Command::new("ssh");
    command.arg("-F").arg(config_path);
    command.arg(host);
    command.arg("--");
    command.arg(quote_ssh_remote_args(cmd.iter().copied())?);
    command.stdout(Stdio::piped());
    command.stderr(Stdio::piped());
    command.kill_on_drop(false);

    let child = command
        .spawn()
        .map_err(|e| miette::miette!("failed to spawn ssh: {e:#}"))?;

    // Arm the kill guard immediately after spawn.  The guard SIGKILLs the
    // direct ssh client PID on drop unless explicitly disarmed — covering
    // timeout, `tokio::select!` cancellation, task abort, and panics alike.
    let mut kill_guard = SshClientKillGuard::new(child.id());

    let timeout_dur = Duration::from_secs(timeout_secs);
    match tokio::time::timeout(timeout_dur, child.wait_with_output()).await {
        Ok(Ok(output)) => {
            // Process exited cleanly — no kill needed.
            kill_guard.disarm();
            if !output.status.success() {
                let stderr = String::from_utf8_lossy(&output.stderr);
                tracing::error!(host, ?cmd, %stderr, "ssh exec failed");
                return Err(miette::miette!(
                    "ssh exec on '{host}' failed (exit {}): {stderr}",
                    output.status
                ));
            }
            Ok(String::from_utf8_lossy(&output.stdout).into_owned())
        }
        Ok(Err(e)) => Err(miette::miette!("ssh exec failed: {e:#}")),
        Err(_) => {
            // Timeout: kill_guard drops at the end of this arm and SIGKILLs
            // the ssh client while leaving the ControlMaster intact.
            Err(miette::miette!("ssh exec timed out after {timeout_secs}s"))
        }
    }
    // kill_guard drops here on every non-Ok(Ok) path; SIGKILL fires unless
    // disarmed above.
}

/// Drop guard that SIGKILLs a captured ssh PID unless explicitly disarmed.
/// Used by [`ssh_exec`] to guarantee the ssh client doesn't outlive the future
/// even on `tokio::select!` cancellation or task abort.
struct SshClientKillGuard(Option<u32>);

impl SshClientKillGuard {
    fn new(pid: Option<u32>) -> Self {
        Self(pid)
    }

    fn disarm(&mut self) {
        self.0 = None;
    }
}

impl Drop for SshClientKillGuard {
    fn drop(&mut self) {
        if let Some(pid) = self.0.take() {
            let _ = nix::sys::signal::kill(
                nix::unistd::Pid::from_raw(pid as i32),
                nix::sys::signal::Signal::SIGKILL,
            );
        }
    }
}

/// Run a single OpenSSH ControlMaster control operation
/// (`ssh -F <config> -O <op> <host>`) and report whether it exited 0.
///
/// Used by the three control-op callers (`check`, `clean_stale`, `tear_down`).
/// Failures and timeouts return `false` and log at `debug` — the caller
/// decides what to do with the answer.
///
/// Uses plain `tokio::process::Child` with `kill_on_drop(false)` for the same
/// reason as [`ssh_exec`]: `ProcessGroupChild`'s `killpg(SIGKILL)` would also
/// kill the master's ProxyCommand if it happened to share the spawning ssh's
/// process group. Control ops do not establish a connection, so they don't
/// fork a ProxyCommand themselves — but using the consistent pattern keeps
/// the rationale on one shape and avoids future "let's unify these" mistakes.
async fn run_ssh_control_op(config_path: &Path, host: &str, op: &str) -> bool {
    let mut command = Command::new("ssh");
    command.arg("-F").arg(config_path);
    command.args(["-O", op]);
    command.arg(host);
    command.stdout(Stdio::null());
    command.stderr(Stdio::null());
    command.kill_on_drop(false);

    let child = match command.spawn() {
        Ok(c) => c,
        Err(e) => {
            tracing::debug!(host, op, error = %e, "ssh -O spawn failed");
            return false;
        }
    };

    match tokio::time::timeout(
        Duration::from_secs(SSH_CONTROL_OP_TIMEOUT_SECS),
        child.wait_with_output(),
    )
    .await
    {
        Ok(Ok(o)) => o.status.success(),
        Ok(Err(e)) => {
            tracing::debug!(host, op, error = %e, "ssh -O wait failed");
            false
        }
        Err(_) => {
            tracing::debug!(
                host,
                op,
                "ssh -O timed out after {}s",
                SSH_CONTROL_OP_TIMEOUT_SECS,
            );
            false
        }
    }
}

/// Check whether an OpenSSH ControlMaster is alive at the ControlPath
/// specified in the given SSH config.
///
/// Returns `true` iff `ssh -F <config> -O check <host>` exits 0. Returns
/// `false` for any other exit, spawn failure, or timeout. This is a probe,
/// not an error path — the caller decides what to do with the answer.
pub async fn check_control_master(config_path: &Path, host: &str) -> bool {
    run_ssh_control_op(config_path, host, "check").await
}

/// At bot startup, clean up a stale ControlMaster socket left behind by
/// a previous SIGKILL'd bot process.
///
/// If the socket file does not exist: no-op.
/// If the socket exists and a master is alive (`ssh -O check` succeeds):
///   leave it alone — the new bot will reuse the existing master.
/// If the socket exists but no master is alive:
///   best-effort `ssh -O exit` (in case the master is reachable from
///   somewhere we don't see), then `rm -f` the socket so the next ssh
///   call can establish a fresh master without `bind: Address already in use`.
pub async fn clean_stale_control_master(
    config_path: &Path,
    host: &str,
    socket_path: &Path,
) -> miette::Result<()> {
    if !socket_path.exists() {
        tracing::debug!(socket = %socket_path.display(), "no stale control-master socket");
        return Ok(());
    }

    if check_control_master(config_path, host).await {
        tracing::info!(
            socket = %socket_path.display(),
            "control-master from previous bot is alive, will reuse",
        );
        return Ok(());
    }

    tracing::info!(
        socket = %socket_path.display(),
        "stale control-master socket found, cleaning up",
    );

    // Best-effort exit; ignore the result (master is dead anyway).
    let _ = run_ssh_control_op(config_path, host, "exit").await;

    if let Err(e) = std::fs::remove_file(socket_path)
        && e.kind() != std::io::ErrorKind::NotFound
    {
        return Err(miette::miette!(
            "failed to remove stale control-master socket {}: {e:#}",
            socket_path.display(),
        ));
    }

    Ok(())
}

/// Tear down a ControlMaster: send `ssh -O exit` and remove the socket file.
///
/// Best-effort and non-fatal — used during graceful shutdown and sandbox
/// migration. Never returns an error; outcomes are logged at info, debug,
/// or warn depending on what happened.
pub async fn tear_down_control_master(config_path: &Path, host: &str, socket_path: &Path) {
    let _ = run_ssh_control_op(config_path, host, "exit").await;

    match std::fs::remove_file(socket_path) {
        Ok(_) => {
            tracing::info!(socket = %socket_path.display(), "control-master socket removed");
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            tracing::debug!(socket = %socket_path.display(), "control-master socket already gone");
        }
        Err(e) => {
            tracing::warn!(
                socket = %socket_path.display(),
                error = %e,
                "failed to remove control-master socket",
            );
        }
    }
}

/// Stream a sandbox tarball to a local file via SSH.
///
/// Runs tar from inside the sandbox root and rewrites archive entries under
/// `sandbox_path`, piping stdout directly to `dest_path` without buffering in
/// memory.
pub async fn ssh_tar_download(
    config_path: &Path,
    ssh_host: &str,
    sandbox_path: &str,
    dest_path: &Path,
    include_rebuildable: bool,
    timeout_secs: u64,
) -> miette::Result<()> {
    let mut command = Command::new("ssh");
    command.arg("-F").arg(config_path);
    command.arg(ssh_host);
    command.arg("--");
    let tar_args = sandbox_tar_download_args(sandbox_path, include_rebuildable)?;
    command.arg(quote_ssh_remote_args(tar_args.iter().map(String::as_str))?);
    command.stdout(Stdio::piped());
    command.stderr(Stdio::piped());

    let mut child = right_process::ProcessGroupChild::spawn(command)
        .map_err(|e| miette::miette!("failed to spawn ssh for tar download: {e:#}"))?;

    let mut stdout = child
        .stdout()
        .ok_or_else(|| miette::miette!("no stdout handle from ssh tar download"))?;
    let mut stderr = child
        .stderr()
        .ok_or_else(|| miette::miette!("no stderr handle from ssh tar download"))?;

    let mut file = tokio::fs::File::create(dest_path).await.map_err(|e| {
        miette::miette!(
            "failed to create backup file {}: {e:#}",
            dest_path.display()
        )
    })?;

    // Run stdout→file copy concurrently with child wait via try_join! — no orphan
    // spawn to leak on timeout. Dropping the outer future cancels both branches.
    let copy_fut = async {
        tokio::io::copy(&mut stdout, &mut file)
            .await
            .map_err(|e| miette::miette!("I/O error during tar download: {e:#}"))?;
        Ok::<_, miette::Error>(())
    };

    let stderr_fut = async {
        let mut stderr_output = Vec::new();
        tokio::io::AsyncReadExt::read_to_end(&mut stderr, &mut stderr_output)
            .await
            .map_err(|e| miette::miette!("I/O error reading tar stderr: {e:#}"))?;
        Ok::<_, miette::Error>(String::from_utf8_lossy(&stderr_output).into_owned())
    };

    let wait_fut = async {
        child
            .wait()
            .await
            .map_err(|e| miette::miette!("ssh wait failed: {e:#}"))
    };

    let timeout_dur = Duration::from_secs(timeout_secs);
    let ((), stderr, child_result) = tokio::time::timeout(timeout_dur, async {
        tokio::try_join!(copy_fut, stderr_fut, wait_fut)
    })
    .await
    .map_err(|_| miette::miette!("ssh tar download timed out after {timeout_secs}s"))??;

    if !child_result.success() {
        let stderr = stderr.trim();
        if !stderr.is_empty() {
            return Err(miette::miette!(
                "ssh tar download failed (exit {child_result}): {stderr}"
            ));
        }
        return Err(miette::miette!(
            "ssh tar download failed (exit {child_result})"
        ));
    }
    Ok(())
}

/// Quote argv for OpenSSH remote command execution.
///
/// OpenSSH does not preserve remote argv. It sends one command string to the
/// remote login shell, so callers must pass the returned string as exactly one
/// argument after the SSH host or `--`.
///
/// Returns `Err` only if an argument contains an interior NUL byte (the sole
/// `shlex::try_join` failure mode).
pub fn quote_ssh_remote_args<'a>(
    args: impl IntoIterator<Item = &'a str>,
) -> miette::Result<String> {
    shlex::try_join(args)
        .map_err(|e| miette::miette!("failed to quote SSH remote command args: {e}"))
}

fn sandbox_tar_download_args(
    sandbox_path: &str,
    include_rebuildable: bool,
) -> miette::Result<Vec<String>> {
    let archive_root = sandbox_path.trim_matches('/');
    if archive_root.is_empty() {
        miette::bail!("sandbox path must not be empty");
    }

    // `flags=rh` rewrites archive member names and hardlink targets while
    // preserving symlink targets such as `./target`.
    let mut args = vec![
        "tar".to_string(),
        "czpf".to_string(),
        "-".to_string(),
        "-C".to_string(),
        format!("/{archive_root}"),
        format!("--transform=flags=rh;s,^\\.$,{archive_root},"),
        format!("--transform=flags=rh;s,^\\./,{archive_root}/,"),
    ];

    if !include_rebuildable {
        for path in DEFAULT_REBUILDABLE_BACKUP_EXCLUDES {
            // GNU tar evaluates excludes before transforms, so match names as
            // seen under `-C /sandbox .`, not final `sandbox/...` archive names.
            args.push(format!("--exclude=./{path}"));
            args.push(format!("--exclude=./{path}/*"));
        }
    }

    args.push(".".to_string());
    Ok(args)
}

/// Stream a local tar.gz file into the sandbox via SSH `tar xzpf -`.
///
/// Pipes `src_path` to SSH stdin, extracting the archived `sandbox/...` root
/// into the writable `/sandbox` directory.
pub async fn ssh_tar_upload(
    config_path: &Path,
    ssh_host: &str,
    src_path: &Path,
    timeout_secs: u64,
) -> miette::Result<()> {
    let mut command = Command::new("ssh");
    command.arg("-F").arg(config_path);
    command.arg(ssh_host);
    command.arg("--");
    command.arg(quote_ssh_remote_args([
        "tar",
        "xzpf",
        "-",
        "-C",
        "/sandbox",
        "--strip-components=1",
        "sandbox",
    ])?);
    command.stdin(Stdio::piped());
    command.stdout(Stdio::piped());
    command.stderr(Stdio::piped());

    let mut child = right_process::ProcessGroupChild::spawn(command)
        .map_err(|e| miette::miette!("failed to spawn ssh for tar upload: {e:#}"))?;

    let mut stdin = child
        .stdin()
        .ok_or_else(|| miette::miette!("no stdin handle from ssh tar upload"))?;

    let src_path = src_path.to_owned();
    // Run the write task concurrently with child wait via try_join! — no orphan
    // spawn to leak on timeout. Dropping the outer future cancels both branches.
    let write_fut = async {
        let mut file = tokio::fs::File::open(&src_path).await.map_err(|e| {
            miette::miette!("failed to open backup file {}: {e:#}", src_path.display())
        })?;
        tokio::io::copy(&mut file, &mut stdin)
            .await
            .map_err(|e| miette::miette!("I/O error during tar upload: {e:#}"))?;
        // Drop stdin to signal EOF to the remote tar process.
        drop(stdin);
        Ok::<_, miette::Error>(())
    };

    let wait_fut = async {
        child
            .wait_with_output()
            .await
            .map_err(|e| miette::miette!("ssh wait failed: {e:#}"))
    };

    let timeout_dur = Duration::from_secs(timeout_secs);
    let (write_result, output_result) =
        tokio::time::timeout(timeout_dur, async { tokio::join!(write_fut, wait_fut) })
            .await
            .map_err(|_| miette::miette!("ssh tar upload timed out after {timeout_secs}s"))?;

    let output = output_result?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        let local_error = write_result
            .as_ref()
            .err()
            .map(|e| format!("; local upload error: {e:#}"))
            .unwrap_or_default();
        return Err(miette::miette!(
            "ssh tar upload failed (exit {}): {stderr}{local_error}",
            output.status,
        ));
    }

    write_result?;
    Ok(())
}

/// Upload a file from host into a running sandbox.
///
/// `sandbox_dir` must be a directory path ending with `/`.
/// The file lands in `sandbox_dir` with its original name from `host_path`.
pub async fn upload_file(sandbox: &str, host_path: &Path, sandbox_dir: &str) -> miette::Result<()> {
    if !sandbox_dir.ends_with('/') {
        miette::bail!(
            "upload destination must be a directory path ending with '/', got: {sandbox_dir}"
        );
    }

    if host_path.is_dir() {
        return upload_directory(sandbox, host_path, sandbox_dir).await;
    }

    upload_single_file(sandbox, host_path, sandbox_dir).await
}

/// Upload a single file to a sandbox directory.
async fn upload_single_file(
    sandbox: &str,
    host_path: &Path,
    sandbox_dir: &str,
) -> miette::Result<()> {
    let mut cmd = Command::new("openshell");
    cmd.args(["sandbox", "upload", sandbox])
        .arg(host_path)
        .arg(sandbox_dir);
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());

    let mut child = right_process::ProcessGroupChild::spawn(cmd)
        .map_err(|e| miette::miette!("failed to spawn openshell upload: {e:#}"))?;

    let output = child
        .wait_with_output()
        .await
        .map_err(|e| miette::miette!("openshell upload wait failed: {e:#}"))?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        return Err(miette::miette!("openshell upload failed: {stderr}"));
    }
    Ok(())
}

/// Upload a directory by individually uploading each file in parallel.
///
/// OpenShell CLI has a known bug where directory uploads silently drop files.
/// Workaround: walk the directory tree and upload each file individually.
async fn upload_directory(sandbox: &str, host_dir: &Path, sandbox_dir: &str) -> miette::Result<()> {
    use futures::stream::{self, StreamExt};

    let dir_name = host_dir
        .file_name()
        .ok_or_else(|| miette::miette!("directory has no name: {}", host_dir.display()))?
        .to_string_lossy();

    let mut uploads = Vec::new();
    for entry in walkdir::WalkDir::new(host_dir) {
        let entry = entry.map_err(|e| miette::miette!("walkdir error: {e:#}"))?;
        if !entry.file_type().is_file() {
            continue;
        }
        let rel = entry
            .path()
            .strip_prefix(host_dir)
            .map_err(|e| miette::miette!("strip_prefix failed: {e:#}"))?;
        // Destination: sandbox_dir + dir_name + relative path's parent
        let dest_dir = std::path::Path::new(sandbox_dir)
            .join(&*dir_name)
            .join(rel.parent().unwrap_or(std::path::Path::new("")));
        let dest = format!("{}/", dest_dir.display());
        uploads.push((entry.path().to_path_buf(), dest));
    }

    if uploads.is_empty() {
        return Ok(());
    }

    let results: Vec<miette::Result<()>> = stream::iter(uploads.into_iter().map(|(path, dest)| {
        let sandbox = sandbox.to_owned();
        async move { upload_single_file(&sandbox, &path, &dest).await }
    }))
    .buffer_unordered(10)
    .collect()
    .await;

    for result in results {
        result?;
    }
    Ok(())
}

fn ensure_download_parent(host_dest: &Path) -> miette::Result<()> {
    let parent = host_dest.parent().unwrap_or_else(|| Path::new("."));
    std::fs::create_dir_all(parent).map_err(|e| {
        miette::miette!(
            "failed to create parent directory {}: {e:#}",
            parent.display()
        )
    })
}

fn remove_stale_directory_at_dest(host_dest: &Path) -> miette::Result<()> {
    if let Ok(meta) = std::fs::symlink_metadata(host_dest)
        && meta.is_dir()
    {
        std::fs::remove_dir_all(host_dest).map_err(|e| {
            miette::miette!(
                "failed to remove stale directory at {}: {e:#}",
                host_dest.display()
            )
        })?;
    }
    Ok(())
}

fn move_downloaded_file_into_place(staged: &Path, host_dest: &Path) -> miette::Result<()> {
    remove_stale_directory_at_dest(host_dest)?;
    std::fs::rename(staged, host_dest).map_err(|e| {
        miette::miette!(
            "failed to move downloaded file to {}: {e:#}",
            host_dest.display()
        )
    })
}

/// Download a file or directory from a sandbox to the host.
/// Download a single file from the sandbox, placing it at exactly `host_dest`.
///
/// The `openshell sandbox download` CLI always treats its DEST argument as a
/// directory and writes the file inside it using the source basename. This
/// wrapper hides that quirk: it downloads into a private temp directory in
/// `host_dest`'s parent, then renames the file into place. Missing parent
/// directories are created. Any pre-existing file at `host_dest` is replaced.
pub async fn download_file(
    sandbox: &str,
    sandbox_path: &str,
    host_dest: &Path,
) -> miette::Result<()> {
    let basename = Path::new(sandbox_path)
        .file_name()
        .ok_or_else(|| miette::miette!("sandbox_path has no file name: {sandbox_path}"))?;

    ensure_download_parent(host_dest)?;
    let parent = host_dest.parent().unwrap_or_else(|| Path::new("."));

    // Stage the download in a temp dir alongside host_dest so the final
    // rename stays on the same filesystem (atomic rename, no cross-device copy).
    let staging = tempfile::tempdir_in(parent).map_err(|e| {
        miette::miette!(
            "failed to create staging dir in {}: {e:#}",
            parent.display()
        )
    })?;

    let mut cmd = Command::new("openshell");
    cmd.args(["sandbox", "download", sandbox, sandbox_path])
        .arg(staging.path());
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());

    let mut child = right_process::ProcessGroupChild::spawn(cmd)
        .map_err(|e| miette::miette!("failed to spawn openshell download: {e:#}"))?;

    let output = child
        .wait_with_output()
        .await
        .map_err(|e| miette::miette!("openshell download wait failed: {e:#}"))?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        return Err(miette::miette!("openshell download failed: {stderr}"));
    }

    let staged = staging.path().join(basename);
    if !staged.exists() {
        return Err(miette::miette!(
            "openshell download reported success but expected file not found at {}",
            staged.display()
        ));
    }

    move_downloaded_file_into_place(&staged, host_dest)?;

    Ok(())
}

/// Delete a sandbox. Best-effort — logs a warning on failure but does not
/// propagate the error (stale sandboxes that don't exist shouldn't block callers).
pub async fn delete_sandbox(name: &str) {
    let mut cmd = Command::new("openshell");
    cmd.args(["sandbox", "delete", name]);
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());

    let mut child = match right_process::ProcessGroupChild::spawn(cmd) {
        Ok(c) => c,
        Err(e) => {
            tracing::warn!(sandbox = name, "failed to spawn openshell delete: {e:#}");
            return;
        }
    };

    let result = child.wait_with_output().await;

    match result {
        Ok(output) if output.status.success() => {
            tracing::info!(sandbox = name, "deleted sandbox");
        }
        Ok(output) => {
            let stderr = String::from_utf8_lossy(&output.stderr);
            tracing::warn!(
                sandbox = name,
                exit = %output.status,
                %stderr,
                "failed to delete sandbox (best-effort)"
            );
        }
        Err(e) => {
            tracing::warn!(sandbox = name, "failed to wait for openshell delete: {e:#}");
        }
    }
}

/// Prepare the staging directory for sandbox upload.
///
/// Copies curated files from the agent directory into staging/,
/// excluding credentials (sandbox gets its own via login flow) and plugins (not used).
pub fn prepare_staging_dir(agent_dir: &Path, upload_dir: &Path) -> miette::Result<()> {
    let staging_claude = upload_dir.join(".claude");
    if staging_claude.exists() {
        std::fs::remove_dir_all(&staging_claude)
            .map_err(|e| miette::miette!("failed to clean staging/.claude: {e:#}"))?;
    }
    std::fs::create_dir_all(&staging_claude)
        .map_err(|e| miette::miette!("failed to create staging/.claude: {e:#}"))?;

    let src_claude = agent_dir.join(".claude");

    // Minimal CC bootstrap files only — agent defs, skills, and schemas
    // are deployed via /platform/ store during initial_sync.
    let copy_items: &[&str] = &["settings.json", "reply-schema.json"];

    for item in copy_items {
        let src = src_claude.join(item);
        let dst = staging_claude.join(item);
        if !src.exists() {
            continue;
        }
        std::fs::copy(&src, &dst)
            .map_err(|e| miette::miette!("failed to copy {} to staging: {e:#}", item))?;
    }

    // Copy .claude.json (trust/onboarding — at agent root, not inside .claude/)
    let claude_json_src = agent_dir.join(".claude.json");
    let claude_json_dst = upload_dir.join(".claude.json");
    if claude_json_src.exists() {
        std::fs::copy(&claude_json_src, &claude_json_dst)
            .map_err(|e| miette::miette!("failed to copy .claude.json to staging: {e:#}"))?;
    }

    // Copy mcp.json
    let mcp_json_src = agent_dir.join("mcp.json");
    let mcp_json_dst = upload_dir.join("mcp.json");
    if mcp_json_src.exists() {
        std::fs::copy(&mcp_json_src, &mcp_json_dst)
            .map_err(|e| miette::miette!("failed to copy mcp.json to staging: {e:#}"))?;
    }

    // Seed agent-editable files that the agent needs from turn 1.
    // Staging only runs when creating a new sandbox, so this never
    // overwrites agent edits on an existing sandbox. After initial upload,
    // the sandbox copy is authoritative — reverse_sync pulls agent edits
    // back to host; there is no host→sandbox sync after creation.
    let seed_files: &[&str] = &["TOOLS.md"];
    for name in seed_files {
        let src = agent_dir.join(name);
        let dst = upload_dir.join(name);
        if src.exists() {
            std::fs::copy(&src, &dst)
                .map_err(|e| miette::miette!("failed to copy {} to staging: {e:#}", name))?;
        }
    }

    tracing::info!("prepared staging dir for sandbox upload");
    Ok(())
}

/// Recursively copy a directory, resolving symlinks to regular files.
/// Skips entries that fail to read (e.g. broken symlinks).
pub fn copy_dir_resolve_symlinks(src: &Path, dst: &Path) -> std::io::Result<()> {
    std::fs::create_dir_all(dst)?;
    for entry in std::fs::read_dir(src)? {
        let entry = entry?;
        let src_path = entry.path();
        let dst_path = dst.join(entry.file_name());

        // Resolve symlinks via entry.metadata() (follows symlinks, avoids extra syscall).
        let meta = match entry.metadata() {
            Ok(m) => m,
            Err(e) => {
                tracing::warn!(path = %src_path.display(), "skipping unresolvable entry: {e}");
                continue;
            }
        };

        if meta.is_dir() {
            copy_dir_resolve_symlinks(&src_path, &dst_path)?;
        } else if meta.is_file() {
            std::fs::copy(&src_path, &dst_path)?;
        }
    }
    Ok(())
}

/// Outcome of `ensure_sandbox` — tells the caller what happened.
#[derive(Debug, PartialEq)]
pub enum SandboxOutcome {
    Created,
    Recreated,
}

/// Create a sandbox, handling the case where one already exists.
///
/// - If no sandbox exists: create it, wait for READY, return `Created`.
/// - If sandbox exists and `force_recreate`: delete + create, return `Recreated`.
/// - If sandbox exists and not force: return error.
///
/// `sandbox` is the literal OpenShell sandbox name (e.g. `right-foo`). Callers
/// must pass the same name they will write into `agent.yaml::sandbox.name`,
/// otherwise the bot won't be able to find the sandbox at startup.
pub async fn ensure_sandbox(
    sandbox: &str,
    policy_path: &Path,
    staging_dir: Option<&Path>,
    force_recreate: bool,
) -> miette::Result<SandboxOutcome> {
    let mtls_dir = match preflight_check() {
        OpenShellStatus::Ready(dir) => dir,
        OpenShellStatus::NotInstalled => {
            return Err(miette::miette!(
                help = "Install OpenShell and run `openshell auth login`,\n  \
                        or use `--sandbox-mode none` to run without a sandbox",
                "OpenShell is required for sandbox mode 'openshell'"
            ));
        }
        OpenShellStatus::NoGateway(_) => {
            return Err(miette::miette!(
                help = "Run `openshell gateway start`,\n  \
                        or use `--sandbox-mode none`",
                "OpenShell gateway is not running"
            ));
        }
        OpenShellStatus::BrokenGateway(dir) => {
            return Err(miette::miette!(
                help = "Try `openshell gateway destroy && openshell gateway start`,\n  \
                        or use `--sandbox-mode none`",
                "OpenShell gateway exists but mTLS certificates are missing at {}",
                dir.display()
            ));
        }
    };

    let mut grpc_client = connect_grpc(&mtls_dir).await?;
    // Use sandbox_exists (not is_sandbox_ready) to detect sandboxes in ANY phase,
    // including DELETING — otherwise we'd skip wait_for_deleted and create would
    // conflict with a sandbox still being torn down.
    let exists = sandbox_exists(&mut grpc_client, sandbox).await?;

    if exists && !force_recreate {
        return Err(miette::miette!(
            help = "Use --force to recreate the sandbox,\n  \
                    or `right agent config` to update an existing agent",
            "Sandbox '{sandbox}' already exists"
        ));
    }

    if exists {
        tracing::info!(sandbox = %sandbox, "deleting existing sandbox for recreate");
        delete_sandbox(sandbox).await;
        wait_for_deleted(&mut grpc_client, sandbox, 60, 2).await?;
    }

    tracing::info!(sandbox = %sandbox, "creating sandbox");
    let mut child = spawn_sandbox(sandbox, policy_path, staging_dir)?;

    tokio::select! {
        result = wait_for_ready(&mut grpc_client, sandbox, 120, 2) => {
            result?;
            drop(child);
        }
        status = child.wait() => {
            let status = status.map_err(|e| miette::miette!("sandbox create child wait failed: {e:#}"))?;
            if !status.success() {
                return Err(miette::miette!(
                    "openshell sandbox create for '{sandbox}' exited with {status} before reaching READY"
                ));
            }
        }
    }

    // SSH readiness probe — gRPC READY doesn't guarantee SSH transport is up.
    // 60s timeout: sandboxes with TLS termination + proxy can take 30-50s after READY.
    let sandbox_id = resolve_sandbox_id(&mut grpc_client, sandbox).await?;
    wait_for_ssh(&mut grpc_client, &sandbox_id, 60, 2).await?;

    let outcome = if exists {
        SandboxOutcome::Recreated
    } else {
        SandboxOutcome::Created
    };
    tracing::info!(sandbox = %sandbox, ?outcome, "sandbox ready");
    Ok(outcome)
}

/// Resolve sandbox ID from sandbox name via gRPC `GetSandbox`.
pub async fn resolve_sandbox_id(
    client: &mut OpenShellClient<Channel>,
    name: &str,
) -> miette::Result<String> {
    let resp = client
        .get_sandbox(GetSandboxRequest {
            name: name.to_owned(),
        })
        .await
        .map_err(|e| miette::miette!("GetSandbox failed for '{name}': {e:#}"))?;

    let sandbox = resp
        .into_inner()
        .sandbox
        .ok_or_else(|| miette::miette!("GetSandbox returned empty response for '{name}'"))?;

    let metadata = sandbox.metadata.ok_or_else(|| {
        miette::miette!("GetSandbox returned sandbox without metadata for '{name}'")
    })?;
    if metadata.id.is_empty() {
        return Err(miette::miette!(
            "GetSandbox returned sandbox metadata without id for '{name}'"
        ));
    }

    Ok(metadata.id)
}

/// Execute a command inside a sandbox via gRPC `ExecSandbox` (single attempt).
///
/// Returns `Ok((stdout, exit_code))` on success, or an error if the RPC or
/// stream fails (e.g. SSH transport not ready yet).
async fn exec_in_sandbox_once(
    client: &mut OpenShellClient<Channel>,
    sandbox_id: &str,
    command: &[&str],
    timeout_seconds: u32,
) -> miette::Result<(String, i32)> {
    use crate::openshell_proto::openshell::v1::exec_sandbox_event::Payload;

    let req = ExecSandboxRequest {
        sandbox_id: sandbox_id.to_owned(),
        command: command.iter().map(|s| s.to_string()).collect(),
        timeout_seconds,
        ..Default::default()
    };

    let mut stream = client
        .exec_sandbox(req)
        .await
        .map_err(|e| miette::miette!("ExecSandbox RPC failed: {e:#}"))?
        .into_inner();

    let mut stdout = Vec::new();
    let mut exit_code = -1;

    loop {
        match stream.message().await {
            Ok(Some(event)) => match event.payload {
                Some(Payload::Stdout(chunk)) => stdout.extend_from_slice(&chunk.data),
                Some(Payload::Exit(exit)) => exit_code = exit.exit_code,
                _ => {}
            },
            Ok(None) => break,
            Err(e) => {
                return Err(miette::miette!("ExecSandbox stream error: {e:#}"));
            }
        }
    }

    let stdout_str = String::from_utf8_lossy(&stdout).to_string();
    Ok((stdout_str, exit_code))
}

/// Default timeout for OpenShell `ExecSandbox` commands (seconds). Long enough
/// for small shell operations (`test -f`, `getent`, etc.) but short enough to
/// surface hung commands quickly. Callers that need more (e.g. `claude upgrade`)
/// should pass an explicit timeout.
pub const DEFAULT_EXEC_TIMEOUT_SECS: u32 = 10;

/// Execute a command inside a sandbox via gRPC `ExecSandbox` and return (stdout, exit_code).
///
/// `timeout_seconds` is the server-side timer: OpenShell kills the process and
/// returns exit 124 after this many seconds. Use `DEFAULT_EXEC_TIMEOUT_SECS`
/// for cheap shell probes; bump it for commands that fetch over the network
/// (`claude upgrade`, npm installs, etc.).
///
/// Retries up to 5 times with 1s backoff if the SSH transport isn't ready yet
/// (common immediately after sandbox creation — gRPC reports READY before SSH is up).
/// Retries cover both RPC-level and stream-level errors.
pub async fn exec_in_sandbox(
    client: &mut OpenShellClient<Channel>,
    sandbox_id: &str,
    command: &[&str],
    timeout_seconds: u32,
) -> miette::Result<(String, i32)> {
    let mut last_err = String::new();

    for attempt in 0..5u32 {
        if attempt > 0 {
            tracing::debug!(
                sandbox_id,
                attempt,
                "exec: retrying after SSH transport delay"
            );
            tokio::time::sleep(Duration::from_secs(1)).await;
        }

        match exec_in_sandbox_once(client, sandbox_id, command, timeout_seconds).await {
            Ok(result) => return Ok(result),
            Err(e) => {
                last_err = format!("{e:#}");
                tracing::debug!(sandbox_id, attempt, error = %last_err, "exec: attempt failed");
            }
        }
    }

    Err(miette::miette!(
        "ExecSandbox failed after 5 attempts: {last_err}"
    ))
}

/// Poll until SSH transport inside the sandbox is ready.
///
/// gRPC READY doesn't guarantee SSH is accepting connections — there's a gap
/// where `ExecSandbox` fails with "Connection reset by peer". This probe
/// runs a trivial `echo` command until it succeeds.
pub async fn wait_for_ssh(
    client: &mut OpenShellClient<Channel>,
    sandbox_id: &str,
    timeout_secs: u64,
    poll_interval_secs: u64,
) -> miette::Result<()> {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_secs);
    let interval = Duration::from_secs(poll_interval_secs);

    loop {
        match exec_in_sandbox_once(
            client,
            sandbox_id,
            &["echo", "ready"],
            DEFAULT_EXEC_TIMEOUT_SECS,
        )
        .await
        {
            Ok(_) => {
                tracing::info!(sandbox_id, "SSH transport is ready");
                return Ok(());
            }
            Err(e) => {
                if tokio::time::Instant::now() + interval > deadline {
                    return Err(miette::miette!(
                        "SSH transport not ready after {timeout_secs}s: {e:#}"
                    ));
                }
                tracing::debug!(sandbox_id, error = %e, "SSH not ready, retrying");
                tokio::time::sleep(interval).await;
            }
        }
    }
}

pub(crate) fn parse_getent_ahosts_ips(stdout: &str) -> Vec<IpAddr> {
    let mut ips = Vec::new();
    for token in stdout
        .lines()
        .filter_map(|line| line.split_whitespace().next())
    {
        let Ok(ip) = token.parse::<IpAddr>() else {
            continue;
        };
        if !ips.contains(&ip) {
            ips.push(ip);
        }
    }
    ips
}

/// Resolve host alias IPs as seen from inside a sandbox.
///
/// Runs `getent ahosts host.openshell.internal` inside the sandbox via gRPC exec
/// and parses every unique IPv4/IPv6 address. OpenShell's SSRF proxy resolves
/// through the sandbox view as well, so host-side guesses are not authoritative.
pub async fn resolve_host_ips(
    client: &mut OpenShellClient<Channel>,
    sandbox_id: &str,
) -> miette::Result<Vec<IpAddr>> {
    let (stdout, exit_code) = exec_in_sandbox(
        client,
        sandbox_id,
        &["getent", "ahosts", "host.openshell.internal"],
        DEFAULT_EXEC_TIMEOUT_SECS,
    )
    .await?;

    if exit_code != 0 || stdout.trim().is_empty() {
        return Err(miette::miette!(
            "host.openshell.internal not resolvable in sandbox '{sandbox_id}' (getent exit {exit_code})"
        ));
    }

    let ips = parse_getent_ahosts_ips(&stdout);
    if ips.is_empty() {
        return Err(miette::miette!(
            "host.openshell.internal resolved without valid IPs in sandbox '{sandbox_id}': {stdout}"
        ));
    }

    tracing::info!(sandbox_id, ?ips, "resolved host.openshell.internal");
    Ok(ips)
}

/// Verify that critical files exist in the sandbox after creation/upload.
///
/// Uses gRPC `ExecSandbox` to check file existence (fast, no download).
/// If any are missing, attempts individual re-upload from `host_source_dir`
/// and re-checks. Fails with a detailed error listing all missing files.
pub async fn verify_sandbox_files(
    sandbox_name: &str,
    host_source_dir: &Path,
    sandbox_base_path: &str,
    expected_files: &[&str],
) -> miette::Result<()> {
    let mtls_dir = match preflight_check() {
        OpenShellStatus::Ready(dir) => dir,
        _ => return Ok(()), // Can't verify without OpenShell
    };
    let mut client = connect_grpc(&mtls_dir).await?;
    let sandbox_id = resolve_sandbox_id(&mut client, sandbox_name).await?;

    tracing::info!(
        sandbox = sandbox_name,
        count = expected_files.len(),
        "verify: checking files"
    );

    // Build a single shell command to check all files at once.
    let checks: Vec<String> = expected_files
        .iter()
        .map(|f| format!("test -f {sandbox_base_path}{f} && echo 'OK:{f}' || echo 'MISSING:{f}'"))
        .collect();
    let check_cmd = checks.join("; ");

    let (output, exit_code) = exec_in_sandbox(
        &mut client,
        &sandbox_id,
        &["sh", "-c", &check_cmd],
        DEFAULT_EXEC_TIMEOUT_SECS,
    )
    .await?;

    tracing::debug!(sandbox = sandbox_name, exit_code, output = %output.trim(), "verify: check result");

    // Parse which files are missing.
    let mut missing_files: Vec<&str> = Vec::new();
    for &filename in expected_files {
        let ok_marker = format!("OK:{filename}");
        if !output.contains(&ok_marker) {
            missing_files.push(filename);
        }
    }

    if missing_files.is_empty() {
        tracing::info!(sandbox = sandbox_name, "verify: all expected files present");
        return Ok(());
    }

    // Re-upload missing files individually.
    tracing::warn!(
        sandbox = sandbox_name,
        missing = ?missing_files,
        "verify: some files missing, attempting individual re-upload"
    );

    for &filename in &missing_files {
        let host_path = host_source_dir.join(filename);
        if host_path.exists() {
            upload_file(sandbox_name, &host_path, sandbox_base_path)
                .await
                .map_err(|e| miette::miette!("verify: re-upload {filename} failed: {e:#}"))?;
            tracing::info!(file = filename, "verify: re-uploaded");
        } else {
            tracing::debug!(file = filename, "verify: not on host, skipping");
        }
    }

    // Re-check after re-upload.
    let (output2, _) = exec_in_sandbox(
        &mut client,
        &sandbox_id,
        &["sh", "-c", &check_cmd],
        DEFAULT_EXEC_TIMEOUT_SECS,
    )
    .await?;

    let mut still_missing: Vec<String> = Vec::new();
    for &filename in &missing_files {
        let ok_marker = format!("OK:{filename}");
        if !output2.contains(&ok_marker) {
            still_missing.push(filename.to_string());
        }
    }

    if still_missing.is_empty() {
        tracing::info!(
            sandbox = sandbox_name,
            "verify: all files present after re-upload"
        );
        Ok(())
    } else {
        Err(miette::miette!(
            help = "This is likely an OpenShell upload bug.\n  \
                    Try running `right init` again.",
            "Sandbox '{}' is missing critical files after re-upload: {}",
            sandbox_name,
            still_missing.join(", ")
        ))
    }
}

/// Compare filesystem and landlock sections of two `SandboxPolicy` values.
///
/// Returns `true` if they differ — indicating a sandbox migration is required
/// (filesystem policy changes cannot be hot-reloaded; only network policy can).
pub fn filesystem_policy_changed(
    old: &crate::openshell_proto::openshell::sandbox::v1::SandboxPolicy,
    new: &crate::openshell_proto::openshell::sandbox::v1::SandboxPolicy,
) -> bool {
    let fs_changed = match (&old.filesystem, &new.filesystem) {
        (Some(a), Some(b)) => {
            a.include_workdir != b.include_workdir
                || a.read_only != b.read_only
                || a.read_write != b.read_write
        }
        (None, None) => false,
        _ => true,
    };

    let ll_changed = match (&old.landlock, &new.landlock) {
        (Some(a), Some(b)) => a.compatibility != b.compatibility,
        (None, None) => false,
        _ => true,
    };

    fs_changed || ll_changed
}

/// Fetch the currently active policy from a sandbox via gRPC.
///
/// Returns `None` if the revision exists but the policy payload is not populated.
pub async fn get_active_policy(
    client: &mut OpenShellClient<Channel>,
    name: &str,
) -> miette::Result<Option<crate::openshell_proto::openshell::sandbox::v1::SandboxPolicy>> {
    use crate::openshell_proto::openshell::v1::GetSandboxPolicyStatusRequest;

    let resp = client
        .get_sandbox_policy_status(GetSandboxPolicyStatusRequest {
            name: name.to_owned(),
            version: 0, // latest
            global: false,
        })
        .await
        .map_err(|e| miette::miette!("GetSandboxPolicyStatus RPC failed: {e:#}"))?;

    let revision = resp
        .into_inner()
        .revision
        .ok_or_else(|| miette::miette!("GetSandboxPolicyStatus returned no revision"))?;

    Ok(revision.policy)
}

/// Parse a policy YAML string and extract the filesystem-relevant fields into a `SandboxPolicy`.
///
/// Only populates `filesystem` and `landlock` — enough for comparison via
/// [`filesystem_policy_changed`]. Network policies are ignored.
pub fn parse_policy_yaml_filesystem(
    yaml: &str,
) -> miette::Result<crate::openshell_proto::openshell::sandbox::v1::SandboxPolicy> {
    use crate::openshell_proto::openshell::sandbox::v1::{FilesystemPolicy, LandlockPolicy};

    let doc: serde_json::Value = serde_saphyr::from_str(yaml)
        .map_err(|e| miette::miette!("failed to parse policy YAML: {e:#}"))?;

    let fs_policy = doc.get("filesystem_policy").map(|fs| FilesystemPolicy {
        include_workdir: fs
            .get("include_workdir")
            .and_then(|v| v.as_bool())
            .unwrap_or(false),
        read_only: fs
            .get("read_only")
            .and_then(|v| v.as_array())
            .map(|seq| {
                seq.iter()
                    .filter_map(|v| v.as_str().map(String::from))
                    .collect()
            })
            .unwrap_or_default(),
        read_write: fs
            .get("read_write")
            .and_then(|v| v.as_array())
            .map(|seq| {
                seq.iter()
                    .filter_map(|v| v.as_str().map(String::from))
                    .collect()
            })
            .unwrap_or_default(),
    });

    let landlock = doc.get("landlock").map(|ll| LandlockPolicy {
        compatibility: ll
            .get("compatibility")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_owned(),
    });

    Ok(
        crate::openshell_proto::openshell::sandbox::v1::SandboxPolicy {
            filesystem: fs_policy,
            landlock,
            ..Default::default()
        },
    )
}

#[cfg(test)]
#[path = "openshell_tests.rs"]
mod tests;
