unit 0.29.0

// spawn.rs — True self-replication for unit
//
// A running unit can package itself (binary + state + prelude) into a
// single blob and spawn a new independent process — locally or on a
// remote machine. The nanobot metaphor becomes literal.
//
// Package format:
//   Magic: "UREP" (4 bytes)
//   Version: u8
//   Binary size: u64
//   State size: u64
//   Prelude size: u64
//   Binary bytes
//   State bytes
//   Prelude bytes

use std::time::Instant;

// ---------------------------------------------------------------------------
// Package format
// ---------------------------------------------------------------------------

const PACKAGE_MAGIC: &[u8; 4] = b"UREP";
const PACKAGE_VERSION: u8 = 1;
const HEADER_SIZE: usize = 4 + 1 + 8 + 8 + 8; // 29 bytes

/// Build a replication package: binary + state + prelude.
pub fn build_package(state: &[u8]) -> Result<Vec<u8>, String> {
    let binary = read_own_binary()?;
    let prelude = include_bytes!("prelude.fs");

    let total = HEADER_SIZE + binary.len() + state.len() + prelude.len();
    let mut buf = Vec::with_capacity(total);

    // Header.
    buf.extend_from_slice(PACKAGE_MAGIC);
    buf.push(PACKAGE_VERSION);
    buf.extend_from_slice(&(binary.len() as u64).to_be_bytes());
    buf.extend_from_slice(&(state.len() as u64).to_be_bytes());
    buf.extend_from_slice(&(prelude.len() as u64).to_be_bytes());

    // Payloads.
    buf.extend_from_slice(&binary);
    buf.extend_from_slice(state);
    buf.extend_from_slice(prelude);

    Ok(buf)
}

/// Estimate package size without building it.
pub fn package_size_estimate(state_size: usize) -> Result<usize, String> {
    let binary_size = read_own_binary()?.len();
    let prelude_size = include_bytes!("prelude.fs").len();
    Ok(HEADER_SIZE + binary_size + state_size + prelude_size)
}

/// Unpack a replication package. Returns (binary, state, prelude).
#[allow(clippy::type_complexity)]
pub fn unpack_package(data: &[u8]) -> Result<(Vec<u8>, Vec<u8>, Vec<u8>), String> {
    if data.len() < HEADER_SIZE {
        return Err("package too small".into());
    }
    if &data[0..4] != PACKAGE_MAGIC {
        return Err("bad magic".into());
    }
    if data[4] != PACKAGE_VERSION {
        return Err(format!("unsupported version {}", data[4]));
    }

    let binary_size = u64::from_be_bytes(data[5..13].try_into().unwrap()) as usize;
    let state_size = u64::from_be_bytes(data[13..21].try_into().unwrap()) as usize;
    let prelude_size = u64::from_be_bytes(data[21..29].try_into().unwrap()) as usize;

    let expected = HEADER_SIZE + binary_size + state_size + prelude_size;
    if data.len() < expected {
        return Err(format!("truncated: have {} need {}", data.len(), expected));
    }

    let mut pos = HEADER_SIZE;
    let binary = data[pos..pos + binary_size].to_vec();
    pos += binary_size;
    let state = data[pos..pos + state_size].to_vec();
    pos += state_size;
    let prelude = data[pos..pos + prelude_size].to_vec();

    Ok((binary, state, prelude))
}

/// Read this process's own executable binary.
fn read_own_binary() -> Result<Vec<u8>, String> {
    let path = std::env::current_exe().map_err(|e| format!("current_exe: {}", e))?;
    std::fs::read(&path).map_err(|e| format!("read binary: {}", e))
}

// ---------------------------------------------------------------------------
// Child tracking
// ---------------------------------------------------------------------------

/// Information about a spawned child unit process.
#[derive(Clone, Debug)]
pub struct ChildInfo {
    pub pid: u32,
    pub port: u16,
    pub node_id: [u8; 8],
    pub spawned_at: Instant,
}

/// Tracks self-replication state including children and spawn limits.
#[derive(Clone, Debug)]
pub struct SpawnState {
    pub children: Vec<ChildInfo>,
    pub generation: u32,
    pub parent_id: Option<[u8; 8]>,
    pub max_children: usize,
    pub accept_replicate: bool,
    pub quarantine: bool,
    pub last_spawn: Option<Instant>,
    pub spawn_cooldown_secs: u64,
}

impl Default for SpawnState {
    fn default() -> Self {
        Self::new()
    }
}

impl SpawnState {
    /// Creates a new spawn state with default limits.
    pub fn new() -> Self {
        SpawnState {
            children: Vec::new(),
            generation: 0,
            parent_id: None,
            max_children: 10,
            accept_replicate: true,
            quarantine: false,
            last_spawn: None,
            spawn_cooldown_secs: 30,
        }
    }

    /// Checks whether spawning is allowed, returning an error if not.
    pub fn can_spawn(&self) -> Result<(), String> {
        if self.quarantine {
            return Err("quarantine active".into());
        }
        if self.children.len() >= self.max_children {
            return Err(format!("max children reached ({})", self.max_children));
        }
        if let Some(last) = self.last_spawn {
            let elapsed = last.elapsed().as_secs();
            if elapsed < self.spawn_cooldown_secs {
                return Err(format!(
                    "cooldown: {}s remaining",
                    self.spawn_cooldown_secs - elapsed
                ));
            }
        }
        Ok(())
    }

    /// Like [`can_spawn`](Self::can_spawn) but ALSO refuses when the host has
    /// no resource headroom — the binding-constraint ceiling.
    ///
    /// Every existing guard (quarantine, max_children, cooldown) still applies;
    /// this layers one additional refusal on top via
    /// [`HostResources::has_headroom`](crate::resources::HostResources::has_headroom).
    /// Because `has_headroom` fails closed, an unavailable reading (no `/proc`,
    /// a parse failure, or wasm32) refuses too: a coordinate that cannot
    /// measure its host must not replicate. The ceiling is a wall, not a
    /// target — see [`CEILING_UTILIZATION`](crate::resources::CEILING_UTILIZATION).
    pub fn can_spawn_within(
        &self,
        res: &crate::resources::HostResources,
    ) -> Result<(), String> {
        self.can_spawn()?;
        if !res.has_headroom() {
            return Err("ceiling: host at/over 80%".into());
        }
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Local spawn
// ---------------------------------------------------------------------------

/// Spawn a new unit process on the local machine.
/// Returns (pid, port, child_id).
pub fn spawn_local(
    package: &[u8],
    parent_port: u16,
    child_generation: u32,
) -> Result<(u32, u16, [u8; 8]), String> {
    spawn_local_with_energy(package, parent_port, child_generation, None)
}

/// Spawn a new unit process, optionally passing inherited energy via
/// the `UNIT_CHILD_ENERGY` environment variable.
pub fn spawn_local_with_energy(
    package: &[u8],
    parent_port: u16,
    child_generation: u32,
    child_energy: Option<i64>,
) -> Result<(u32, u16, [u8; 8]), String> {
    let (binary, state, _prelude) = unpack_package(package)?;

    // Generate a child ID.
    let mut child_id = [0u8; 8];
    if let Ok(mut f) = std::fs::File::open("/dev/urandom") {
        use std::io::Read;
        let _ = f.read_exact(&mut child_id);
    }
    let child_hex: String = child_id.iter().map(|b| format!("{:02x}", b)).collect();

    // Create spawn directory.
    let home = std::env::var("HOME").unwrap_or_else(|_| ".".to_string());
    let spawn_dir = format!("{}/.unit/spawn/{}", home, child_hex);
    std::fs::create_dir_all(&spawn_dir).map_err(|e| format!("mkdir: {}", e))?;

    // Write binary.
    let bin_path = format!("{}/unit", spawn_dir);
    std::fs::write(&bin_path, &binary).map_err(|e| format!("write binary: {}", e))?;

    // chmod +x
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let perms = std::fs::Permissions::from_mode(0o755);
        std::fs::set_permissions(&bin_path, perms).map_err(|e| format!("chmod: {}", e))?;
    }

    // Write state.
    let state_dir = format!("{}/.unit/{}", home, child_hex);
    std::fs::create_dir_all(&state_dir).map_err(|e| format!("mkdir state: {}", e))?;
    std::fs::write(format!("{}/state.bin", state_dir), &state)
        .map_err(|e| format!("write state: {}", e))?;

    // Write the node-id file so the child boots with this identity.
    std::fs::write(
        format!("{}/.unit/spawn/{}/node-id", home, child_hex),
        &child_hex,
    )
    .map_err(|e| format!("write node-id: {}", e))?;

    // Pick a port: 0 = OS-assigned.
    let child_port = 0u16;

    // Launch the child process.
    let mut cmd = std::process::Command::new(&bin_path);
    cmd.env("UNIT_PORT", child_port.to_string())
        .env("UNIT_PEERS", format!("127.0.0.1:{}", parent_port))
        .env("UNIT_GENERATION", child_generation.to_string())
        .env("UNIT_PARENT_ID", child_hex.clone())
        .env("UNIT_NODE_ID", child_hex.clone());
    if let Some(energy) = child_energy {
        cmd.env("UNIT_CHILD_ENERGY", energy.to_string());
    }
    let child = cmd
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::inherit())
        .stderr(std::process::Stdio::inherit())
        .spawn()
        .map_err(|e| format!("spawn: {}", e))?;

    let pid = child.id();

    Ok((pid, child_port, child_id))
}

// ---------------------------------------------------------------------------
// TCP replication (send package to remote host)
// ---------------------------------------------------------------------------

/// Send a replication package to a remote host via TCP.
pub fn send_package(addr: &str, package: &[u8]) -> Result<(), String> {
    use std::io::Write;
    use std::net::TcpStream;
    use std::time::Duration;

    let mut stream = TcpStream::connect(addr).map_err(|e| format!("connect {}: {}", addr, e))?;
    stream.set_write_timeout(Some(Duration::from_secs(30))).ok();

    // Send length prefix + package.
    let len_bytes = (package.len() as u64).to_be_bytes();
    stream
        .write_all(&len_bytes)
        .map_err(|e| format!("write len: {}", e))?;
    stream
        .write_all(package)
        .map_err(|e| format!("write pkg: {}", e))?;

    Ok(())
}

/// Listen for incoming replication packages on a TCP port.
/// Runs in a background thread. Calls `on_receive` for each package.
pub fn start_replication_listener(port: u16) -> Result<std::sync::mpsc::Receiver<Vec<u8>>, String> {
    use std::io::Read;
    use std::net::TcpListener;

    let listener = TcpListener::bind(format!("127.0.0.1:{}", port))
        .map_err(|e| format!("bind TCP {}: {}", port, e))?;

    let (tx, rx) = std::sync::mpsc::channel();

    std::thread::spawn(move || {
        for mut stream in listener.incoming().flatten() {
            stream
                .set_read_timeout(Some(std::time::Duration::from_secs(30)))
                .ok();

            // Read length prefix.
            let mut len_buf = [0u8; 8];
            if stream.read_exact(&mut len_buf).is_err() {
                continue;
            }
            let pkg_len = u64::from_be_bytes(len_buf) as usize;
            if pkg_len > 100_000_000 {
                // Sanity: reject > 100MB.
                continue;
            }

            // Read package.
            let mut pkg = vec![0u8; pkg_len];
            if stream.read_exact(&mut pkg).is_err() {
                continue;
            }

            // Validate header.
            if pkg.len() >= 4 && &pkg[0..4] == PACKAGE_MAGIC {
                let _ = tx.send(pkg);
            }
        }
    });

    Ok(rx)
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use crate::resources::HostResources;

    // A reading with ample headroom (50% binding constraint, under the 80%
    // ceiling) and one that's over the ceiling (90% memory used).
    fn under_ceiling() -> HostResources {
        HostResources::from_parts(1000, 500, 0.0, 4)
    }
    fn over_ceiling() -> HostResources {
        HostResources::from_parts(1000, 100, 0.0, 4)
    }

    #[test]
    fn can_spawn_within_allows_when_healthy_and_under_ceiling() {
        let s = SpawnState::new();
        assert!(s.can_spawn().is_ok());
        assert!(s.can_spawn_within(&under_ceiling()).is_ok());
    }

    #[test]
    fn can_spawn_within_refuses_over_ceiling() {
        let s = SpawnState::new();
        // The pre-existing guards are all clear...
        assert!(s.can_spawn().is_ok());
        // ...but the ceiling refuses on its own.
        let err = s.can_spawn_within(&over_ceiling()).unwrap_err();
        assert!(err.contains("ceiling"), "unexpected error: {err}");
    }

    #[test]
    fn can_spawn_within_fails_closed_on_unavailable() {
        let s = SpawnState::new();
        // Unmeasurable host (no /proc, wasm32) → refuse.
        let err = s.can_spawn_within(&HostResources::unavailable()).unwrap_err();
        assert!(err.contains("ceiling"), "unexpected error: {err}");
    }

    #[test]
    fn can_spawn_within_still_honors_preexisting_guards() {
        // Quarantine refuses even with ample headroom.
        let mut q = SpawnState::new();
        q.quarantine = true;
        let err = q.can_spawn_within(&under_ceiling()).unwrap_err();
        assert!(err.contains("quarantine"), "expected quarantine: {err}");

        // max_children refuses even with ample headroom.
        let mut full = SpawnState::new();
        full.max_children = 1;
        full.children.push(ChildInfo {
            pid: 1,
            port: 0,
            node_id: [0u8; 8],
            spawned_at: Instant::now(),
        });
        let err = full.can_spawn_within(&under_ceiling()).unwrap_err();
        assert!(err.contains("max children"), "expected max children: {err}");
    }
}