netsky 0.1.6

//! `netsky restart [N] [--handoff <file>]` — atomic constellation respawn.
//!
//! Kills `^agent[0-9]+$` tmux sessions (never agentinfinity), respawns via
//! `netsky up N`, waits for agent0 readiness, delivers the handoff to
//! agent0's inbox plus a durable archive copy, then waits for each clone.

use std::fs;
use std::path::{Path, PathBuf};
use std::thread;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use netsky_core::consts::{
    AGENT0_NAME, CLONE_PREFIX, ENV_HANDOFF_KEEP, HANDOFF_FROM, HANDOFF_KEEP_DEFAULT,
    RESTART_AGENT0_TOS_WAIT_S, RESTART_AGENT0_UP_WAIT_S, RESTART_STATUS_KEEP,
    RESTART_TEARDOWN_SETTLE_MS,
};
use netsky_core::paths::{agent0_inbox_dir, handoff_archive_dir, restart_status_dir};
use netsky_sh::tmux;

pub fn run(n: u32, handoff_path: Option<&str>) -> netsky_core::Result<()> {
    let status = StatusWriter::new();
    let result = run_impl(n, handoff_path.map(Path::new), &status);
    if let Err(e) = &result {
        status.write_errored(&e.to_string());
    }
    result
}

fn run_impl(n: u32, handoff: Option<&Path>, status: &StatusWriter) -> netsky_core::Result<()> {
    // Pre-flight the spawn dependencies BEFORE teardown. If claude / tmux /
    // netsky is missing, bailing here leaves the running constellation
    // intact; bailing after teardown leaves the system with no agents and
    // nothing able to spawn them. The watchdog would then crash-recover
    // in a loop, never making progress.
    netsky_core::spawn::require_deps()?;

    teardown_agents();
    thread::sleep(Duration::from_millis(RESTART_TEARDOWN_SETTLE_MS));

    println!("[netsky-restart] spawning constellation (N={n})");
    // Restart preserves the legacy claude-default constellation. A
    // `netsky restart --type` flag is a follow-up: the watchdog's
    // crash-recovery path needs to know which runtime the dying agent0
    // was running before it can respawn into the same flavor, and that
    // requires a state-file the watchdog reads at tick time. Until that
    // lands, restart brings the constellation up with codex clones and
    // claude agent0 — matching the runtime allocation rule.
    crate::cmd::up::run(n, crate::cli::AgentType::Codex)?;
    status.write_spawned();

    // agent0 MUST reach its `/up` session-N marker before the rest of
    // the restart is meaningful — clone waits, handoff delivery, and
    // the "complete" log all depend on agent0 being alive and
    // processing. Returning Ok here on failure would let the detached
    // restart log a success the watchdog's crashloop counter (P0-1)
    // interprets as a healthy recovery, which is the exact phantom
    // the blocker review caught. Bail instead; the status writer
    // records the error, and the outer `run` already re-writes
    // errored on any propagated error so the last-known-error is
    // durable for the next tick's escalation page.
    if !wait_session_up(AGENT0_NAME) {
        let err = agent0_up_failure_error(RESTART_AGENT0_UP_WAIT_S);
        status.write_errored(&err.to_string());
        return Err(err);
    }
    status.write_up_detected();

    match handoff {
        Some(p) if p != Path::new("/dev/null") => {
            deliver_handoff(p)?;
        }
        _ => println!("[netsky-restart] no handoff (file missing or empty)"),
    }

    for i in 1..=n {
        let _ = wait_session_up(&format!("{CLONE_PREFIX}{i}"));
    }

    println!("[netsky-restart] complete");
    Ok(())
}

/// Build the error returned when agent0's tmux session spawned but
/// never reached its `/up` `session <N>` marker within
/// [`RESTART_AGENT0_UP_WAIT_S`]. Exposed as a pure helper so tests can
/// pin the error message shape without spinning up tmux.
fn agent0_up_failure_error(wait_s: u64) -> netsky_core::Error {
    netsky_core::Error::Message(format!(
        "restart failed at agent0 step: agent0 did not finish /up within {wait_s}s"
    ))
}

fn teardown_agents() {
    for s in tmux::list_sessions()
        .into_iter()
        .filter(|s| is_agent_session(s))
    {
        println!("[netsky-restart] killing session {s}");
        let _ = tmux::kill_session(&s);
    }
}

fn is_agent_session(name: &str) -> bool {
    let rest = match name.strip_prefix(CLONE_PREFIX) {
        Some(r) => r,
        None => return false,
    };
    !rest.is_empty() && rest.chars().all(|c| c.is_ascii_digit())
}

/// Dismiss the dev-channels TOS (up to 30s) then poll for `/up` completion
/// (up to 90s). Returns true iff the session reached the `session <N>`
/// marker before the timeout. Callers that depend on /up having run
/// (e.g. handoff delivery) must check the return value.
fn wait_session_up(sess: &str) -> bool {
    if !tmux::session_is_alive(sess) {
        println!("[netsky-restart] warn: session {sess} did not spawn; skipping");
        return false;
    }

    if netsky_core::spawn::dismiss_tos(sess, Duration::from_secs(RESTART_AGENT0_TOS_WAIT_S)) {
        println!("[netsky-restart] {sess}: TOS dismissed");
    }

    let mut found = false;
    for _ in 0..RESTART_AGENT0_UP_WAIT_S {
        if let Ok(out) = tmux::capture_pane(sess, None)
            && contains_session_marker(&out)
        {
            found = true;
            break;
        }
        thread::sleep(Duration::from_secs(1));
    }
    if found {
        println!("[netsky-restart] {sess}: /up complete");
    } else {
        println!(
            "[netsky-restart] warn: {sess} did not finish /up within {RESTART_AGENT0_UP_WAIT_S}s"
        );
    }
    found
}

fn contains_session_marker(pane: &str) -> bool {
    for line in pane.lines() {
        if let Some((_, rest)) = line.split_once("session ")
            && rest.chars().next().is_some_and(|c| c.is_ascii_digit())
        {
            return true;
        }
    }
    false
}

fn deliver_handoff(file: &Path) -> netsky_core::Result<()> {
    deliver_handoff_core(file, &agent0_inbox_dir(), &handoff_archive_dir())
}

/// Pure-ish core of handoff delivery: read `source`, write the JSON
/// envelope atomically into `inbox`, mirror it into `archive`, then
/// remove `source` so the draft doesn't leak. Parameterized for tests.
///
/// The source rm is the reap-on-delivery contract: the only caller that
/// writes to `source` is `write_crash_handoff` (watchdog), and once the
/// envelope has landed in both inbox and archive the draft is durable
/// in its new home. Leaving the draft behind is how $TMPDIR accumulated
/// 13 stale files over the first two weeks of the rewrite.
fn deliver_handoff_core(source: &Path, inbox: &Path, archive: &Path) -> netsky_core::Result<()> {
    let text = match fs::read_to_string(source) {
        Ok(t) => t,
        Err(e) => {
            println!(
                "[netsky-restart] warn: handoff read failed for {}: {e}",
                source.display()
            );
            return Ok(());
        }
    };
    if text.is_empty() {
        println!("[netsky-restart] no handoff (file missing or empty)");
        return Ok(());
    }

    fs::create_dir_all(inbox)?;
    fs::create_dir_all(archive)?;

    let ts_iso = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
    let ts_compact = chrono::Utc::now().format("%Y%m%dT%H%M%SZ");
    let pid = std::process::id();
    let name = format!("{ts_compact}-{pid}-from-{HANDOFF_FROM}.json");

    let envelope = serde_json::json!({
        "from": HANDOFF_FROM,
        "text": text,
        "ts": ts_iso,
    });
    let body = serde_json::to_string(&envelope)?;

    // Delivery order: inbox is the authoritative delivery; archive is
    // a record OF delivery. Committing to inbox first (atomic rename)
    // then copying to archive means a crash between steps 2 and 3
    // leaves inbox=truth + archive=empty, which is correct (agent0
    // got the handoff, archive just lacks a record). Codex-review
    // concern #3: the pre-fix order (write tmp -> copy tmp to archive
    // -> rename tmp to inbox) let a crash between copy and rename
    // leave a phantom archive entry for a handoff agent0 never
    // received.
    let tmp = inbox.join(format!(".{name}.tmp"));
    let final_ = inbox.join(&name);
    let archive_path = archive.join(&name);
    fs::write(&tmp, &body)?;
    fs::rename(&tmp, &final_)?;
    fs::copy(&final_, &archive_path)?;

    println!(
        "[netsky-restart] handoff delivered to {name} (archive: {})",
        archive_path.display()
    );

    // Reap the source draft now that both inbox + archive copies are durable.
    match fs::remove_file(source) {
        Ok(()) => println!(
            "[netsky-restart] handoff source removed: {}",
            source.display()
        ),
        Err(e) => println!(
            "[netsky-restart] warn: handoff source rm failed for {}: {e}",
            source.display()
        ),
    }

    prune_archive(archive)?;
    Ok(())
}

fn prune_archive(dir: &Path) -> netsky_core::Result<()> {
    let keep = std::env::var(ENV_HANDOFF_KEEP)
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .unwrap_or(HANDOFF_KEEP_DEFAULT);

    let mut entries: Vec<(SystemTime, PathBuf)> = Vec::new();
    for e in fs::read_dir(dir)? {
        let e = e?;
        let p = e.path();
        if p.extension().and_then(|s| s.to_str()) != Some("json") {
            continue;
        }
        let mtime = e.metadata()?.modified().unwrap_or(UNIX_EPOCH);
        entries.push((mtime, p));
    }
    entries.sort_by(|a, b| b.0.cmp(&a.0));
    for (_, p) in entries.into_iter().skip(keep) {
        let _ = fs::remove_file(p);
    }
    Ok(())
}

// ---- P0-2: restart-child status writer -----------------------------------

/// Writes a single status json per restart invocation at phase
/// transitions so the next watchdog tick can tell "restart succeeded"
/// from "restart died silently". Without this, the detached restart is
/// fire-and-forget: the tick sees RESTART_INFLIGHT_FILE disappear and
/// guesses. That guess is the 2026-04-15 15:51Z crashloop.
///
/// Filename = `<ts>-<pid>.json`. `<ts>` is UTC compact (YYYYMMDDTHHMMSSZ)
/// from the first write; the same file is overwritten on subsequent
/// writes within one invocation so the watchdog only ever reads the
/// latest phase. Older files from prior restarts are pruned by
/// mtime-sort down to `RESTART_STATUS_KEEP`.
///
/// Errors are swallowed: a status-write failure must never crash the
/// restart itself. Best-effort. The watchdog handles missing files.
struct StatusWriter {
    pid: u32,
    started_at: String,
    path: PathBuf,
}

impl StatusWriter {
    fn new() -> Self {
        let pid = std::process::id();
        let ts_compact = chrono::Utc::now().format("%Y%m%dT%H%M%SZ").to_string();
        let ts_iso = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
        let dir = restart_status_dir();
        let _ = fs::create_dir_all(&dir);
        let path = dir.join(format!("{ts_compact}-{pid}.json"));
        Self {
            pid,
            started_at: ts_iso,
            path,
        }
    }

    fn write_spawned(&self) {
        self.write_phase("spawned", None, None);
    }

    fn write_up_detected(&self) {
        self.write_phase("up-detected", Some(0), None);
    }

    fn write_errored(&self, error: &str) {
        self.write_phase("errored", Some(1), Some(error));
    }

    fn write_phase(&self, phase: &str, exit_code: Option<i32>, error: Option<&str>) {
        let updated_at = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
        let rendered = match render_status_body(
            self.pid,
            &self.started_at,
            phase,
            exit_code,
            error,
            &updated_at,
        ) {
            Some(s) => s,
            None => return,
        };
        // Atomic write: tmp + rename so a reader never catches a half
        // write. The watchdog mtime-sorts to pick the latest; an atomic
        // rename keeps mtime consistent with content.
        let tmp = self.path.with_extension("json.tmp");
        if fs::write(&tmp, &rendered).is_err() {
            return;
        }
        let _ = fs::rename(&tmp, &self.path);
        let _ = prune_status_dir(&restart_status_dir());
    }
}

/// Pure-ish body renderer. Exposed so tests can pin the shape without
/// touching disk.
fn render_status_body(
    pid: u32,
    started_at: &str,
    phase: &str,
    exit_code: Option<i32>,
    error: Option<&str>,
    updated_at: &str,
) -> Option<String> {
    let body = serde_json::json!({
        "pid": pid,
        "started_at": started_at,
        "phase": phase,
        "exit_code": exit_code,
        "error": error,
        "updated_at": updated_at,
    });
    serde_json::to_string(&body).ok()
}

/// Prune `dir` to the N most-recent `.json` files by mtime. Mirrors
/// `prune_archive` above. Best-effort: errors silently tolerated.
fn prune_status_dir(dir: &Path) -> netsky_core::Result<()> {
    let mut entries: Vec<(SystemTime, PathBuf)> = Vec::new();
    for e in fs::read_dir(dir)? {
        let e = e?;
        let p = e.path();
        if p.extension().and_then(|s| s.to_str()) != Some("json") {
            continue;
        }
        let mtime = e.metadata()?.modified().unwrap_or(UNIX_EPOCH);
        entries.push((mtime, p));
    }
    entries.sort_by(|a, b| b.0.cmp(&a.0));
    for (_, p) in entries.into_iter().skip(RESTART_STATUS_KEEP) {
        let _ = fs::remove_file(p);
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::tempdir;

    #[test]
    fn agent_session_filter() {
        assert!(is_agent_session("agent0"));
        assert!(is_agent_session("agent42"));
        assert!(!is_agent_session("agentinfinity"));
        assert!(!is_agent_session("netsky-ticker"));
    }

    #[test]
    fn up_completion_probe() {
        assert!(contains_session_marker("agent0 session 3 starting at ..."));
        assert!(!contains_session_marker("no marker here"));
    }

    // ---- P0-2: restart-status writer ------------------------------------

    #[test]
    fn status_body_shape_spawned() {
        let rendered = render_status_body(
            1234,
            "2026-04-15T16:00:00Z",
            "spawned",
            None,
            None,
            "2026-04-15T16:00:05Z",
        )
        .expect("serialize");
        let v: serde_json::Value = serde_json::from_str(&rendered).expect("parse");
        assert_eq!(v["pid"], 1234);
        assert_eq!(v["phase"], "spawned");
        assert!(v["exit_code"].is_null());
        assert!(v["error"].is_null());
    }

    #[test]
    fn status_body_shape_errored_carries_error_text() {
        let rendered = render_status_body(
            9,
            "2026-04-15T16:00:00Z",
            "errored",
            Some(1),
            Some("tmux refused: command too long"),
            "2026-04-15T16:00:05Z",
        )
        .expect("serialize");
        let v: serde_json::Value = serde_json::from_str(&rendered).expect("parse");
        assert_eq!(v["phase"], "errored");
        assert_eq!(v["exit_code"], 1);
        assert_eq!(v["error"], "tmux refused: command too long");
    }

    #[test]
    fn prune_status_dir_keeps_only_most_recent() {
        let dir = tempdir().unwrap();
        // Write RESTART_STATUS_KEEP + 5 files, each older than the next.
        let total = RESTART_STATUS_KEEP + 5;
        for i in 0..total {
            let p = dir.path().join(format!("{i:04}.json"));
            fs::write(&p, "{}").unwrap();
            // Backdate mtime so sort is deterministic — earliest files
            // get the oldest mtimes, should be pruned.
            let age_mins = total - i;
            let stamp = format!("2025010100{age_mins:02}"); // YYYYMMDDhhmm
            std::process::Command::new("touch")
                .args(["-t", &stamp, p.to_str().unwrap()])
                .status()
                .unwrap();
        }
        prune_status_dir(dir.path()).unwrap();
        let remaining: Vec<_> = fs::read_dir(dir.path())
            .unwrap()
            .filter_map(|e| e.ok())
            .filter(|e| e.path().extension().and_then(|s| s.to_str()) == Some("json"))
            .collect();
        assert_eq!(remaining.len(), RESTART_STATUS_KEEP);
    }

    #[test]
    fn prune_status_dir_ignores_non_json() {
        let dir = tempdir().unwrap();
        fs::write(dir.path().join("a.json"), "{}").unwrap();
        fs::write(dir.path().join("a.json.tmp"), "{}").unwrap();
        fs::write(dir.path().join("random.txt"), "x").unwrap();
        prune_status_dir(dir.path()).unwrap();
        // All three survive — only one .json, under the keep limit.
        assert!(dir.path().join("a.json").exists());
        assert!(dir.path().join("a.json.tmp").exists());
        assert!(dir.path().join("random.txt").exists());
    }

    // ---- P0-4: crash-handoff deliver-and-reap ---------------------------

    #[test]
    fn deliver_handoff_removes_source_after_copy() {
        let dir = tempdir().unwrap();
        let source = dir.path().join("src.txt");
        let inbox = dir.path().join("inbox");
        let archive = dir.path().join("archive");
        std::fs::write(&source, "hello handoff").unwrap();

        deliver_handoff_core(&source, &inbox, &archive).unwrap();

        assert!(!source.exists(), "source draft must be reaped");
        let inbox_entries: Vec<_> = std::fs::read_dir(&inbox)
            .unwrap()
            .filter_map(|e| e.ok())
            .map(|e| e.file_name().into_string().unwrap())
            .filter(|n| !n.starts_with('.'))
            .collect();
        assert_eq!(inbox_entries.len(), 1, "exactly one envelope in inbox");
        let archive_entries: Vec<_> = std::fs::read_dir(&archive)
            .unwrap()
            .filter_map(|e| e.ok())
            .map(|e| e.file_name().into_string().unwrap())
            .collect();
        assert_eq!(archive_entries.len(), 1, "exactly one envelope in archive");
        assert_eq!(inbox_entries, archive_entries);

        let envelope_path = inbox.join(&inbox_entries[0]);
        let body = std::fs::read_to_string(&envelope_path).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert_eq!(parsed["text"].as_str().unwrap(), "hello handoff");
        assert_eq!(parsed["from"].as_str().unwrap(), HANDOFF_FROM);
    }

    #[test]
    fn deliver_handoff_no_op_on_empty_source() {
        let dir = tempdir().unwrap();
        let source = dir.path().join("src.txt");
        let inbox = dir.path().join("inbox");
        let archive = dir.path().join("archive");
        std::fs::write(&source, "").unwrap();

        deliver_handoff_core(&source, &inbox, &archive).unwrap();

        // Empty input: nothing delivered, source left alone for forensic
        // inspection. We only reap on a successful delivery.
        assert!(source.exists(), "empty-source draft must not be reaped");
        assert!(!inbox.exists() || std::fs::read_dir(&inbox).unwrap().count() == 0);
        assert!(!archive.exists() || std::fs::read_dir(&archive).unwrap().count() == 0);
    }

    // ---- codex-review wave-1A fix 1 (BLOCKER) ---------------------------
    // agent0 /up failure must return Err so the detached restart exits
    // non-zero + the crashloop counter (P0-1) sees the error. Pre-fix,
    // run_impl printed "complete" and returned Ok on a failed /up — the
    // phantom-success codex's CODEX_FEEDBACK.md blocker described.

    #[test]
    fn agent0_up_failure_error_names_the_timeout() {
        let err = agent0_up_failure_error(90);
        let msg = err.to_string();
        assert!(
            msg.contains("restart failed at agent0 step"),
            "error must name the failing step for the crashloop page: {msg}"
        );
        assert!(
            msg.contains("within 90s"),
            "error must include the actual timeout: {msg}"
        );
    }

    #[test]
    fn agent0_up_failure_error_is_message_variant() {
        // Pins the error variant so call-sites that match on Error::*
        // don't drift silently. Message is the current contract — when
        // we promote to a typed variant, this test updates with intent.
        let err = agent0_up_failure_error(5);
        assert!(
            matches!(err, netsky_core::Error::Message(_)),
            "agent0_up failure should surface as Error::Message until promoted"
        );
    }

    // ---- codex-review wave-1A fix 4 -------------------------------------
    // Delivery order: inbox first (atomic rename = commit of delivery),
    // archive second (from the committed inbox file). Codex flagged the
    // pre-fix order (write tmp -> copy tmp to archive -> rename tmp to
    // inbox) as a phantom-delivery hazard: a crash between copy + rename
    // left an archive entry for a handoff agent0 never received.

    #[test]
    fn deliver_handoff_archive_mirrors_inbox_content() {
        // Post-fix: archive is copied FROM the committed inbox file,
        // not the tmp. This test pins the consistency invariant both
        // files end up with identical content. The stronger invariant —
        // archive never exists without inbox — is structural in the op
        // order and not easily testable without process control.
        let dir = tempdir().unwrap();
        let source = dir.path().join("src.txt");
        let inbox = dir.path().join("inbox");
        let archive = dir.path().join("archive");
        std::fs::write(&source, "payload-AAA").unwrap();

        deliver_handoff_core(&source, &inbox, &archive).unwrap();

        let inbox_files: Vec<_> = fs::read_dir(&inbox).unwrap().flatten().collect();
        let archive_files: Vec<_> = fs::read_dir(&archive).unwrap().flatten().collect();
        assert_eq!(inbox_files.len(), 1, "exactly one envelope in inbox");
        assert_eq!(archive_files.len(), 1, "exactly one record in archive");

        let i = std::fs::read_to_string(inbox_files[0].path()).unwrap();
        let a = std::fs::read_to_string(archive_files[0].path()).unwrap();
        assert_eq!(i, a, "archive must mirror inbox content byte-for-byte");
        assert!(
            i.contains("payload-AAA"),
            "envelope must include the source payload"
        );
    }

    #[test]
    fn deliver_handoff_leaves_no_tmp_in_inbox_after_success() {
        // The pre-fix tmp lifecycle (write tmp, copy tmp, rename tmp)
        // left nothing in the inbox except the final + an atomic-rename
        // consumed tmp. Post-fix behavior is identical on the inbox
        // side; assert no `.tmp` debris lingers.
        let dir = tempdir().unwrap();
        let source = dir.path().join("src.txt");
        let inbox = dir.path().join("inbox");
        let archive = dir.path().join("archive");
        std::fs::write(&source, "payload").unwrap();

        deliver_handoff_core(&source, &inbox, &archive).unwrap();

        let stray_tmps: Vec<_> = fs::read_dir(&inbox)
            .unwrap()
            .flatten()
            .filter(|e| {
                e.path()
                    .file_name()
                    .and_then(|n| n.to_str())
                    .map(|s| s.ends_with(".tmp") || s.contains(".tmp"))
                    .unwrap_or(false)
            })
            .collect();
        assert!(
            stray_tmps.is_empty(),
            "inbox must not contain .tmp debris after successful delivery: {stray_tmps:?}"
        );
    }
}