hyperhive/hive-c0re/src/migrate.rs

//! Startup convergence. Three phases, all idempotent and unguarded:
//! harness files, applied + proposed repos, meta repo. They re-run every
//! boot on purpose — each one is a no-op once its state is already
//! correct.
//!
//! Deliberately *not* here, and the distinction is the point:
//!
//! - **One-shot, marker-guarded migrations.** Two used to live here
//!   (repointing containers onto the meta flake, renaming `root` to
//!   `h-root`); both targeted layouts no live hive still has. Add one
//!   only if it cannot be expressed as convergence, and expect to delete
//!   it once every hive has passed it.
//! - **Create-time setup.** A thing that is true from birth is set where
//!   the agent is created, not re-asserted each morning: ruth's
//!   `manage_root_agent`, for one, is stored in the swarm's
//!   `agent-permissions` bucket when the swarm UI creates her.
//!
//! Kill-switch: `HIVE_SKIP_META_MIGRATION=1`. Full sequence and phase
//! details: `docs/agent-lifecycle/config-changes.md::Startup migrations (older hosts)`.

use std::path::Path;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result};
use tokio::process::Command;

use crate::coordinator::Coordinator;
use crate::lifecycle::{self, AGENT_PREFIX};
use crate::meta;

const KILL_SWITCH: &str = "HIVE_SKIP_META_MIGRATION";

/// Per-shellout timeout for the blocking startup convergence. `run` is
/// awaited *before* the daemon starts serving (main.rs), so any child
/// process that wedges here freezes the whole daemon — admin socket +
/// dashboard included — with no diagnostics: a git shellout was observed
/// blocked for 86min under a concurrent `nixos-rebuild`. Every shellout
/// runs under a timeout that kills the child on elapse, so a stuck phase
/// degrades to a logged warning instead of a hung boot.
const GIT_TIMEOUT: Duration = Duration::from_mins(2);

/// Substring that identifies the *current* agent flake boilerplate.
/// Bumped whenever the template changes so the startup migration
/// re-renders existing agents onto the new shape. Today the marker
/// is the `flakeInputs` module-arg forwarding line — older templates
/// (raw `import ./agent.nix`) get rewritten on next hive-c0re start.
const MODULE_FLAKE_MARKER: &str = "_module.args.flakeInputs";

pub async fn run(coord: &Arc<Coordinator>) -> Result<()> {
    if std::env::var(KILL_SWITCH).is_ok() {
        tracing::info!("migration: {KILL_SWITCH} set — skipping");
        return Ok(());
    }
    // Stale meta index lock: a previous hive-c0re crash mid-`git add`
    // can leave `.git/index.lock` behind, which blocks every
    // subsequent meta op until somebody `rm`s it manually. We just
    // booted so nothing of ours is holding it; safe to clear.
    let meta_lock = crate::paths::meta_git_index_lock();
    if meta_lock.exists() {
        match std::fs::remove_file(&meta_lock) {
            Ok(()) => tracing::warn!("cleared stale meta/.git/index.lock"),
            Err(e) => tracing::warn!(error = ?e, "clear stale meta lock failed"),
        }
    }
    let names = match enumerate_agents().await {
        Ok(names) => names,
        Err(e) => {
            tracing::warn!(error = ?e, "migration: container list failed; skipping per-agent phases");
            Vec::new()
        }
    };
    tracing::info!(count = names.len(), "migration: scanning");

    // Phase 0: move harness-owned files out of state/ into harness/.
    // Idempotent — rename is a no-op if the source doesn't exist and
    // the destination already does.
    tracing::debug!("migration: phase 0 (harness files)");
    for name in &names {
        migrate_harness_files(name);
    }

    // Phase 1 + 2: per-agent applied + proposed.
    tracing::debug!("migration: phase 1+2 (applied + proposed repos)");
    // An archetype agent has neither repo, and seeding one here would hand it
    // a config it is not built from.
    let archetypes = crate::archetypes::read_at(&crate::archetypes::path()).unwrap_or_else(|e| {
        tracing::warn!(error = ?e, "migration: stored archetypes unreadable; treating every agent as custom");
        crate::archetypes::Stored::new()
    });
    for name in &names {
        if archetypes.contains_key(name.as_str()) {
            continue;
        }
        tracing::debug!(%name, "migration: applied+proposed");
        if let Err(e) = migrate_applied_repo(name.as_str()).await {
            tracing::warn!(%name, error = ?e, "migration: applied repo rewrite failed");
        }
        let proposed_dir = Coordinator::agent_proposed_dir(name);
        let proposed = lifecycle::setup_proposed(&proposed_dir, name.as_str());
        match tokio::time::timeout(GIT_TIMEOUT, proposed).await {
            Ok(Err(e)) => tracing::warn!(%name, error = ?e, "migration: setup_proposed failed"),
            Err(_) => {
                tracing::warn!(%name, timeout = ?GIT_TIMEOUT, "migration: setup_proposed timed out — skipping");
            }
            Ok(Ok(())) => {}
        }
    }

    // Phase 3: meta repo.
    tracing::debug!("migration: phase 3 (meta sync_agents)");
    // `sync_agents` renders exactly the agents it is given, so an empty
    // stand-in for an unreadable list would drop every agent from the flake.
    let agents = match lifecycle::agents_for_meta_listing().await {
        Ok(agents) => agents,
        Err(e) => {
            tracing::warn!(error = ?e, "migration: container list failed; skipping meta sync_agents");
            return Ok(());
        }
    };
    match tokio::time::timeout(GIT_TIMEOUT, meta::sync_agents(&coord.hive_env(), &agents)).await {
        Ok(Err(e)) => tracing::warn!(error = ?e, "migration: meta sync_agents failed"),
        Err(_) => {
            tracing::warn!(timeout = ?GIT_TIMEOUT, "migration: meta sync_agents timed out — skipping");
        }
        Ok(Ok(())) => {}
    }

    Ok(())
}

/// Move harness-owned sqlite/config files out of the agent-visible state dir
/// and into the sibling harness dir. Best-effort: logs warnings but never
/// fails. Idempotent — each file is only moved if present at the old path
/// and absent at the new path.
fn migrate_harness_files(name: &hive_types::Ident) {
    const HARNESS_FILES: &[&str] = &[
        "hyperhive-events.sqlite",
        "hyperhive-turn-stats.sqlite",
        "hyperhive-model",
    ];
    let state_dir = Coordinator::agent_notes_dir(name);
    let harness_dir = Coordinator::agent_harness_dir(name);
    if let Err(e) = std::fs::create_dir_all(&harness_dir) {
        tracing::warn!(%name, error = ?e, "migration: create harness dir failed");
        return;
    }
    for file in HARNESS_FILES {
        let src = state_dir.join(file);
        let dst = harness_dir.join(file);
        if !src.exists() || dst.exists() {
            continue;
        }
        match std::fs::rename(&src, &dst) {
            Ok(()) => tracing::info!(%name, %file, "migration: moved to harness dir"),
            Err(e) => {
                tracing::warn!(%name, %file, error = ?e, "migration: move to harness dir failed");
            }
        }
    }
}

async fn enumerate_agents() -> Result<Vec<hive_types::Ident>> {
    let containers = lifecycle::list().await?;
    Ok(containers
        .into_iter()
        .filter_map(|c| {
            let name = c.strip_prefix(AGENT_PREFIX)?;
            hive_types::Ident::parse(name).ok()
        })
        .collect())
}

async fn migrate_applied_repo(name: &str) -> Result<()> {
    migrate_applied_dir(name, &crate::paths::applied_dir(name)).await
}

/// Rewrites the applied repo at `dir` to the module-only flake. A missing
/// `flake.nix` counts as "marker absent"; any other read error fails the
/// migration so the agent's file is never overwritten unread.
async fn migrate_applied_dir(name: &str, dir: &Path) -> Result<()> {
    if !dir.join(".git").exists() {
        return Ok(());
    }
    let flake_path = dir.join("flake.nix");
    let cur = match std::fs::read_to_string(&flake_path) {
        Ok(cur) => cur,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => String::new(),
        Err(e) => return Err(e).with_context(|| format!("read {}", flake_path.display())),
    };
    if cur.contains(MODULE_FLAKE_MARKER) {
        return Ok(());
    }
    let want = lifecycle::initial_flake_nix();
    std::fs::write(&flake_path, want).with_context(|| format!("write {}", flake_path.display()))?;
    raw_git(
        dir,
        &[
            "-c",
            "user.name=c0re",
            "-c",
            "user.email=c0re@hyperhive.local",
            "add",
            "flake.nix",
        ],
    )
    .await?;
    raw_git(
        dir,
        &[
            "-c",
            "user.name=c0re",
            "-c",
            "user.email=c0re@hyperhive.local",
            "commit",
            "-m",
            "migration: module-only flake",
        ],
    )
    .await?;
    // Relocate deployed/0 to the migration commit so
    // setup_applied's existence check passes.
    raw_git(dir, &["tag", "-f", "deployed/0", "HEAD"]).await?;
    tracing::info!(%name, "migration: applied repo migrated to module-only flake");
    Ok(())
}

/// Run a command to completion under a timeout, capturing its output. On
/// timeout the child is killed (`kill_on_drop`) and an error is returned,
/// so a wedged shellout can never freeze startup migration. `what` is a
/// human label surfaced in the timeout error + `with_context`.
async fn output_with_timeout(
    mut cmd: Command,
    timeout: Duration,
    what: &str,
) -> Result<std::process::Output> {
    cmd.kill_on_drop(true);
    match tokio::time::timeout(timeout, cmd.output()).await {
        Ok(r) => r.with_context(|| format!("run {what}")),
        Err(_) => anyhow::bail!("{what} timed out after {timeout:?} (child killed)"),
    }
}

async fn raw_git(dir: &Path, args: &[&str]) -> Result<()> {
    let mut cmd = lifecycle::git_command();
    cmd.current_dir(dir).args(args);
    let label = format!("git {} in {}", args.join(" "), dir.display());
    let out = output_with_timeout(cmd, GIT_TIMEOUT, &label).await?;
    if !out.status.success() {
        anyhow::bail!(
            "git {} failed: {}",
            args.join(" "),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    Ok(())
}

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

    #[tokio::test]
    async fn unreadable_flake_is_left_untouched() {
        let dir = tempfile::tempdir().expect("tempdir");
        std::fs::create_dir(dir.path().join(".git")).expect("mkdir .git");
        let flake = dir.path().join("flake.nix");
        let bytes = [0xff_u8, 0xfe, b'x'];
        std::fs::write(&flake, bytes).expect("write flake");

        let res = migrate_applied_dir("t", dir.path()).await;

        assert!(res.is_err(), "non-UTF-8 flake.nix must fail the migration");
        assert_eq!(std::fs::read(&flake).expect("read back"), bytes);
    }
}