//! Bash subprocess runner: spawns `sh -c <cmd>` tasks, writes status
//! files under `harness/bash-tasks/`, and fires hyperhive wake signals
//! on completion. Mirrors the logic previously embedded in `hive-ag3nt`.
//!
//! Files under `tasks_dir()`:
//! - `<id>.json`  — task metadata + status (pending → running → done)
//! - `<id>.out`   — captured stdout (streamed while running)
//! - `<id>.err`   — captured stderr (streamed while running)
//!
//! Tasks with status `running` on daemon boot are marked `interrupted`
//! (the process died with the previous daemon). A best-effort wake is
//! still sent so the agent is not silently blocked.
//!
//! The runner kills the child process on timeout — `tokio::process::Child::drop()`
//! does not kill children, so we explicitly call `child.kill().await`.

use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use anyhow::Result;
use tokio::io::AsyncWriteExt as _;

use crate::paths;
use crate::protocol::{TaskFile, TaskStatus};

/// Poll interval for the runner background loop.
const POLL_INTERVAL: Duration = Duration::from_millis(200);

/// Soft cap on stdout/stderr captured in the done JSON summary.
/// Full output always lives in the `.out`/`.err` files.
pub const SUMMARY_BYTES: usize = 4096;

/// Default task timeout.
pub const DEFAULT_TIMEOUT_SECS: u64 = 180;

/// Maximum inline wait (cap on `wait_seconds`).
pub const MAX_WAIT_SECS: u64 = 30;

/// Poll interval used by the inline-wait loops.
const POLL_MS: u64 = 100;

static TASK_SEQ: AtomicU64 = AtomicU64::new(0);

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

fn now_unix() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs() as i64
}

/// Generate a task ID: `<timestamp_hex><seq_hex>`.
#[must_use]
pub fn new_task_id() -> String {
    let t = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis();
    let seq = TASK_SEQ.fetch_add(1, Ordering::Relaxed);
    format!("{t:013x}{seq:04x}")
}

/// Write a task file atomically (tmp + rename).
fn write_task(task: &TaskFile) -> std::io::Result<()> {
    let json = serde_json::to_string_pretty(task)
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
    let dest = paths::task_json(&task.id);
    let tmp = dest.with_extension("json.tmp");
    std::fs::write(&tmp, &json)?;
    std::fs::rename(&tmp, &dest)
}

/// Read a task file. Returns `None` if the file doesn't exist or is
/// unparseable.
#[must_use]
pub fn read_task(id: &str) -> Option<TaskFile> {
    let s = std::fs::read_to_string(paths::task_json(id)).ok()?;
    serde_json::from_str(&s).ok()
}

// ---------------------------------------------------------------------------
// Loose-ends file (generic MCP loose-ends protocol)
// ---------------------------------------------------------------------------

/// Rewrite `mcp-loose-ends/bash.json` with a summary of all currently
/// active (Pending or Running) tasks. The harness scans this directory
/// generically in `get_loose_ends` — no bash-specific code needed there.
///
/// File format: a JSON array of plain-text summary strings, one per
/// loose-end item. The harness includes them verbatim in the output.
/// Atomic write (tmp + rename) so the harness never reads a partial file.
fn refresh_loose_ends() {
    let active = active_tasks();
    let dir = paths::mcp_loose_ends_dir();
    if let Err(e) = std::fs::create_dir_all(&dir) {
        tracing::warn!(error = ?e, "bash_runner: create mcp-loose-ends dir failed");
        return;
    }
    let items: Vec<String> = active
        .iter()
        .map(|t| {
            let age = now_unix() - t.created_at;
            format!(
                "bash task `{}` status={:?}, cmd: `{}`, age {}s",
                t.id, t.status, t.cmd, age
            )
        })
        .collect();
    let dest = dir.join("bash.json");
    let tmp = dest.with_extension("json.tmp");
    let json = serde_json::to_string(&items).unwrap_or_else(|_| "[]".to_owned());
    if let Err(e) = std::fs::write(&tmp, &json).and_then(|_| std::fs::rename(&tmp, &dest)) {
        tracing::warn!(error = ?e, "bash_runner: write mcp-loose-ends/bash.json failed");
    }
}

// ---------------------------------------------------------------------------
// Public API used by daemon dispatch
// ---------------------------------------------------------------------------

/// Submit a new pending task. Returns the task ID.
///
/// # Errors
///
/// Returns an error if the tasks directory cannot be created or the
/// task file cannot be written.
pub fn submit_task(cmd: String, timeout_secs: Option<u64>) -> Result<String> {
    std::fs::create_dir_all(paths::tasks_dir())?;
    let id = new_task_id();
    let task = TaskFile {
        id: id.clone(),
        cmd,
        timeout_secs: timeout_secs.unwrap_or(DEFAULT_TIMEOUT_SECS),
        status: TaskStatus::Pending,
        created_at: now_unix(),
        started_at: None,
        completed_at: None,
        exit_code: None,
        stdout_tail: None,
        stderr_tail: None,
    };
    write_task(&task)?;
    refresh_loose_ends();
    Ok(id)
}

/// Return all tasks currently in `Pending` or `Running` state.
#[must_use]
pub fn active_tasks() -> Vec<TaskFile> {
    let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else {
        return Vec::new();
    };
    let mut out = Vec::new();
    for entry in rd.flatten() {
        let path = entry.path();
        if path.extension().and_then(|e| e.to_str()) != Some("json") {
            continue;
        }
        let Some(id) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else {
            continue;
        };
        let Some(task) = read_task(&id) else { continue };
        if matches!(task.status, TaskStatus::Pending | TaskStatus::Running) {
            out.push(task);
        }
    }
    out
}

/// Inline wait: poll `read_task(id)` until terminal state or deadline.
/// Returns the final task on success, or `None` if it never completed.
pub async fn wait_for_task(id: &str, wait_secs: u64) -> Option<TaskFile> {
    let cap = wait_secs.min(MAX_WAIT_SECS);
    if cap == 0 {
        return read_task(id);
    }
    let deadline = tokio::time::Instant::now() + Duration::from_secs(cap);
    loop {
        match read_task(id) {
            None => break,
            Some(task) => {
                if matches!(
                    task.status,
                    TaskStatus::Done | TaskStatus::TimedOut | TaskStatus::Interrupted
                ) {
                    return Some(task);
                }
            }
        }
        if tokio::time::Instant::now() >= deadline {
            break;
        }
        tokio::time::sleep(Duration::from_millis(POLL_MS)).await;
    }
    read_task(id)
}

// ---------------------------------------------------------------------------
// Runner background loop
// ---------------------------------------------------------------------------

/// Spawn the background runner loop as a detached tokio task. `socket`
/// is the path to the per-agent broker socket used to deliver completion
/// wake signals. Call once at daemon startup.
pub fn spawn_loop(socket: PathBuf) {
    tokio::spawn(async move {
        run_loop(socket).await;
    });
}

async fn run_loop(socket: PathBuf) {
    if let Err(e) = std::fs::create_dir_all(paths::tasks_dir()) {
        tracing::warn!(error = ?e, "bash_runner: create tasks dir failed");
    }
    mark_interrupted(&socket).await;

    let claimed: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new()));

    loop {
        poll_once(&socket, &claimed).await;
        tokio::time::sleep(POLL_INTERVAL).await;
    }
}

/// On boot, flip any `running` tasks to `interrupted` and fire a wake.
async fn mark_interrupted(socket: &Path) {
    let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else {
        return;
    };
    for entry in rd.flatten() {
        let path = entry.path();
        if path.extension().and_then(|e| e.to_str()) != Some("json") {
            continue;
        }
        let Some(id) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else {
            continue;
        };
        let Some(mut task) = read_task(&id) else {
            continue;
        };
        if task.status != TaskStatus::Running {
            continue;
        }
        tracing::warn!(id = %id, "bash_runner: marking interrupted task");
        task.status = TaskStatus::Interrupted;
        task.completed_at = Some(now_unix());
        if let Err(e) = write_task(&task) {
            tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted state failed");
        }
        refresh_loose_ends();
        send_wake(socket, &id, "interrupted (daemon restarted)", None).await;
    }
}

async fn poll_once(socket: &Path, claimed: &Arc<Mutex<HashSet<String>>>) {
    let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else {
        return;
    };
    for entry in rd.flatten() {
        let path = entry.path();
        if path.extension().and_then(|e| e.to_str()) != Some("json") {
            continue;
        }
        let Some(id) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else {
            continue;
        };
        {
            let guard = claimed.lock().unwrap();
            if guard.contains(&id) {
                continue;
            }
        }
        let Some(task) = read_task(&id) else { continue };
        if task.status != TaskStatus::Pending {
            continue;
        }
        // Claim before spawning to avoid double-spawn across poll iterations.
        claimed.lock().unwrap().insert(id.clone());
        let socket = socket.to_path_buf();
        let claimed = claimed.clone();
        tokio::spawn(async move {
            run_task(task, &socket).await;
            claimed.lock().unwrap().remove(&id);
        });
    }
}

// ---------------------------------------------------------------------------
// Task execution
// ---------------------------------------------------------------------------

async fn run_task(mut task: TaskFile, socket: &Path) {
    let id = task.id.clone();
    tracing::info!(id = %id, cmd = %task.cmd, "bash_runner: starting task");

    task.status = TaskStatus::Running;
    task.started_at = Some(now_unix());
    if let Err(e) = write_task(&task) {
        tracing::warn!(id = %id, error = ?e, "bash_runner: write running state failed");
    }
    refresh_loose_ends();

    let out_path = paths::task_out(&id);
    let err_path = paths::task_err(&id);
    let timeout = Duration::from_secs(task.timeout_secs);

    let (timed_out, exit_code) = match exec_cmd(&task.cmd, &out_path, &err_path, timeout).await {
        Ok((code, false)) => (false, Some(code)),
        Ok((_, true)) => {
            tracing::warn!(id = %id, "bash_runner: task timed out");
            (true, None)
        }
        Err(e) => {
            tracing::warn!(id = %id, error = ?e, "bash_runner: exec error");
            (false, None)
        }
    };

    let stdout_tail = tail_file(&out_path, SUMMARY_BYTES);
    let stderr_tail = tail_file(&err_path, SUMMARY_BYTES);

    task.status = if timed_out {
        TaskStatus::TimedOut
    } else {
        TaskStatus::Done
    };
    task.completed_at = Some(now_unix());
    task.exit_code = exit_code;
    task.stdout_tail = stdout_tail.clone().filter(|s| !s.is_empty());
    task.stderr_tail = stderr_tail.clone().filter(|s| !s.is_empty());

    if let Err(e) = write_task(&task) {
        tracing::warn!(id = %id, error = ?e, "bash_runner: write done state failed");
    }
    refresh_loose_ends();

    let summary = if timed_out {
        format!("timed out after {}s", task.timeout_secs)
    } else {
        format!("exit={}", exit_code.unwrap_or(-1))
    };
    let out_snippet = stdout_tail.as_deref().unwrap_or("").trim();
    let err_snippet = stderr_tail.as_deref().unwrap_or("").trim();
    send_wake(socket, &id, &summary, Some((out_snippet, err_snippet))).await;
}

/// Run `sh -c cmd`, streaming output to files. Returns `(exit_code, timed_out)`.
async fn exec_cmd(
    cmd: &str,
    out_path: &Path,
    err_path: &Path,
    timeout: Duration,
) -> Result<(i32, bool)> {
    use tokio::process::Command;
    // SAFETY: `nice` is async-signal-safe and modifies only the calling
    // process's scheduling priority before exec. No allocations, no locks.
    let mut cmd_builder = Command::new("sh");
    cmd_builder
        .arg("-c")
        .arg(cmd)
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped());
    unsafe {
        cmd_builder.pre_exec(|| {
            libc::nice(10);
            Ok(())
        });
    }
    let mut child = cmd_builder.spawn()?;

    let stdout = child.stdout.take().expect("stdout piped");
    let stderr = child.stderr.take().expect("stderr piped");

    let out_path = out_path.to_path_buf();
    let err_path = err_path.to_path_buf();

    let copy_out = tokio::spawn(copy_stream_to_file(
        tokio::io::BufReader::new(stdout),
        out_path,
    ));
    let copy_err = tokio::spawn(copy_stream_to_file(
        tokio::io::BufReader::new(stderr),
        err_path,
    ));

    match tokio::time::timeout(timeout, child.wait()).await {
        Ok(Ok(status)) => {
            let _ = copy_out.await;
            let _ = copy_err.await;
            Ok((status.code().unwrap_or(-1), false))
        }
        Ok(Err(e)) => Err(e.into()),
        Err(_elapsed) => {
            let _ = child.kill().await;
            let _ = child.wait().await;
            let _ = copy_out.await;
            let _ = copy_err.await;
            Ok((-1, true))
        }
    }
}

async fn copy_stream_to_file<R>(mut reader: R, path: PathBuf)
where
    R: tokio::io::AsyncRead + Unpin,
{
    match tokio::fs::File::create(&path).await {
        Ok(mut f) => {
            let _ = tokio::io::copy(&mut reader, &mut f).await;
            let _ = f.flush().await;
        }
        Err(e) => {
            tracing::warn!(path = %path.display(), error = ?e, "bash_runner: open output file failed");
        }
    }
}

fn tail_file(path: &Path, max_bytes: usize) -> Option<String> {
    let data = std::fs::read(path).ok()?;
    let slice = if data.len() > max_bytes {
        &data[data.len() - max_bytes..]
    } else {
        &data
    };
    Some(String::from_utf8_lossy(slice).into_owned())
}

// ---------------------------------------------------------------------------
// Wake delivery
// ---------------------------------------------------------------------------

pub(crate) async fn send_wake(
    socket: &Path,
    id: &str,
    summary: &str,
    output: Option<(&str, &str)>,
) {
    let mut body = format!("bash task `{id}` finished: {summary}");
    if let Some((stdout, stderr)) = output {
        if !stdout.is_empty() {
            body.push_str("\n\nstdout:\n```\n");
            body.push_str(stdout);
            body.push_str("\n```");
        }
        if !stderr.is_empty() {
            body.push_str("\n\nstderr:\n```\n");
            body.push_str(stderr);
            body.push_str("\n```");
        }
    }
    let req = hive_sh4re::AgentRequest::Wake {
        from: format!("bash-task-{id}"),
        body,
        transient: true,
    };

    use tokio::io::{AsyncBufReadExt as _, BufReader};
    use tokio::net::UnixStream;

    match UnixStream::connect(socket).await {
        Ok(stream) => {
            let (read, mut write) = stream.into_split();
            let line = match serde_json::to_string(&req) {
                Ok(mut s) => {
                    s.push('\n');
                    s
                }
                Err(e) => {
                    tracing::warn!(id = %id, error = ?e, "bash_runner: serialise wake failed");
                    return;
                }
            };
            if write.write_all(line.as_bytes()).await.is_err() {
                tracing::warn!(id = %id, "bash_runner: write wake failed");
                return;
            }
            let _ = write.shutdown().await;
            // Drain the response so the server doesn't get ECONNRESET.
            let mut resp = String::new();
            let _ = BufReader::new(read).read_line(&mut resp).await;
            tracing::info!(id = %id, "bash_runner: wake delivered");
        }
        Err(e) => {
            tracing::warn!(id = %id, error = ?e, "bash_runner: connect wake socket failed");
        }
    }
}