//! Per-container host-side config: the nspawn conf rewrite (bind mounts,
//! network isolation, forwarded credentials), the systemd resource-limits
//! drop-in, and the `write_dropins` verb that re-applies both.

use std::path::{Path, PathBuf};

use anyhow::{Context, Result};
use hive_priv_sock::{BindMount, CredentialMount};

use crate::coordinator::{AgentPaths, HiveEnv};

use super::{
    AGENT_PREFIX, CONTAINER_RUNTIME_MOUNT, CONTAINER_SHARED_MOUNT, container_claude_mount,
    container_name, validate,
};

/// Re-apply the per-container host-side config: nspawn flags (bind
/// mounts etc.), the systemd resource-limits drop-in, and a daemon
/// reload so both take effect on the next unit (re)start. Idempotent —
/// the job queue's `WriteDropin` node, also folded into every `Swap`
/// (rebuild is the reconcile verb).
pub async fn write_dropins(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
    validate(name)?;
    let container = container_name(name);
    set_nspawn_flags(&container, &paths.agent, &paths.claude, &paths.notes).await?;
    let (cpu_quota, memory_max) =
        crate::resource_limits::effective(name, &hive.agent_cpu_quota, &hive.agent_memory_max);
    set_resource_limits(
        &container,
        &cpu_quota,
        &memory_max,
        hive.agent_cpu_weight,
        hive.agent_io_weight,
    )
    .await?;
    systemd_daemon_reload().await
}

/// Write a systemd drop-in for `container@<container>.service` that applies
/// the agent's effective resource caps — its per-agent overrides from
/// `meta/resource-limits.json` where set, the hive-wide defaults
/// otherwise. Goes under `/run/systemd/system/...` so it's ephemeral
/// (regenerated on every spawn / rebuild).
///
/// The weights are hive-wide (`services.hyperhive.agentCpuWeight` /
/// `agentIoWeight`) — unlike the caps they have no per-agent override in
/// `meta/resource-limits.json`, so they come straight off [`HiveEnv`].
/// `None` (the nix option set to `null`) means the weight line is left out
/// and the container keeps the kernel default.
async fn set_resource_limits(
    container: &str,
    cpu_quota: &str,
    memory_max: &str,
    cpu_weight: Option<u32>,
    io_weight: Option<u32>,
) -> Result<()> {
    crate::priv_client::write_resource_limits(
        container, memory_max, cpu_quota, cpu_weight, io_weight,
    )
    .await
}

async fn systemd_daemon_reload() -> Result<()> {
    crate::priv_client::daemon_reload().await
}

/// Where the manager sees the applied trees of every agent, read-only.
/// Manager runs `git fetch /applied/<n>/.git refs/tags/*:refs/tags/applied/*`
/// to learn what hive-c0re deployed (or rejected, or failed to
/// build); the RO bind makes accidental writes impossible from
/// inside the container.
pub const CONTAINER_MANAGER_APPLIED_MOUNT: &str = "/applied";

/// Host path behind every `/agents/<name>/config` mount: the **applied**
/// (deployed) repo, not the working clone at `agents/<name>/config`. That
/// clone is where a config change is staged, so it can hold a proposal
/// that is still under review or was rejected outright — mounting it shows
/// an agent a config which does not govern it. Both mounts (an agent's own
/// and a parent's view of a child's) go through here so they cannot drift.
///
/// Never empty under a live container: `provision_container` runs
/// `setup_applied` before `create_only` makes the container at all.
fn config_bind_source(name: &str) -> PathBuf {
    crate::paths::applied_dir(name)
}

/// Append bind flags for `child`'s state and config dirs into `binds`.
/// See docs/persistence.md ("Parent access to child state") for what a
/// parent may touch and why. Creates missing host-side directories so
/// nspawn doesn't refuse to start; missing dirs are non-fatal.
///
/// **Three dirs, three different answers** — the uniformity of the
/// original loop is what hid that:
///
/// - `state` is **read-write**: a parent reads and writes a child's notes
///   to recover it, which is the one case that needs to work while the
///   child is down.
/// - `config` is **read-only**, and read-only for the *parent* is the
///   point: a config change is a PR against the child's repo on the
///   forge, reviewed and merged, never an edit in place. A writable mount
///   here is a second path to the same file that skips the review — the
///   boundary would then be a convention rather than a permission.
/// - `harness` is **absent entirely**. It holds the child's own runtime
///   material — `bash-tasks/`, turn-stats and event sqlite dbs — none of
///   which a parent has a stated reason to read, let alone write.
///   hive-c0re still reads it directly on the host (`stats::hive_stats`),
///   which needs no bind mount into the parent.
///
/// ⚠️ The seeding done at `InitConfig` approval is **not** affected by the
/// `config` flag and must not be read as a reason to widen it: that runs
/// as hive-c0re against the host path (see `actions.rs`, which seeds the
/// repo and wires its forge remote inline), and `read_only` on a bind
/// constrains writers *inside* the container only.
fn bind_child_agent_dirs(child: &str, binds: &mut Vec<BindMount>) {
    let Ok(child) = hive_types::Ident::parse(child) else {
        tracing::warn!(%child, "skipping child bind: invalid agent name");
        return;
    };
    let child_root = crate::paths::agent_state_dir(&child);
    for (sub, host, read_only) in [
        ("state", child_root.join("state"), false),
        ("config", config_bind_source(child.as_str()), true),
    ] {
        let _ = std::fs::create_dir_all(&host);
        binds.push(BindMount {
            host_path: host.to_string_lossy().into_owned(),
            container_path: format!("/agents/{child}/{sub}"),
            read_only,
        });
    }
}

/// Idempotently rewrite the lines in `/etc/nixos-containers/<container>.conf`
/// that hive-c0re owns: `PRIVATE_NETWORK` (forced 0 so the agent's web UI port
/// is reachable on the host) and `EXTRA_NSPAWN_FLAGS` (the runtime-dir bind).
/// The start script expands `$EXTRA_NSPAWN_FLAGS` unquoted into the
/// `systemd-nspawn` command.
#[allow(
    clippy::too_many_lines,
    reason = "one contiguous nspawn-flag assembly block; the length is the flag \
              surface itself, splitting it would just hide the shape"
)]
async fn set_nspawn_flags(
    container: &str,
    runtime_dir: &Path,
    claude_dir: &Path,
    notes_dir: &Path,
) -> Result<()> {
    // Ensure /shared directory exists before binding. systemd-nspawn requires the bind source to exist.
    let shared_root = crate::paths::shared_root();
    std::fs::create_dir_all(&shared_root)
        .with_context(|| format!("create {}", shared_root.display()))?;
    // Make /shared writable by every agent. Containers share host uids (no
    // PrivateUsers), but each agent is a distinct unix user, so a root-owned
    // 0755 dir leaves them unable to write — the documented "read/write for
    // all agents" contract was broken. A setgid group would need a
    // pinned GID declared in every container plus all agent users joined to
    // it (cross-container coordination + a rebuild cascade); instead we use
    // the /tmp model — sticky world-writable (1777). The sticky bit lets any
    // agent create files while protecting each agent's entries from deletion
    // by the others, and matches /shared's documented "free-for-all, may be
    // deleted/lost" semantics without touching any per-agent config.
    {
        use std::os::unix::fs::PermissionsExt as _;
        let perms = std::fs::Permissions::from_mode(0o1777);
        std::fs::set_permissions(&shared_root, perms)
            .with_context(|| format!("chmod 1777 {}", shared_root.display()))?;
    }
    // Ensure /knowledge dir exists. It may be empty until forge seeds it;
    // nspawn refuses to start if the bind source is missing entirely.
    std::fs::create_dir_all(crate::knowledge::LOCAL_DIR)
        .with_context(|| format!("create {}", crate::knowledge::LOCAL_DIR))?;

    // Logical agent name — strip the `h-` prefix.
    // For the manager: `h-ruth` → `ruth`. For sub-agents: `h-iris` → `iris`.
    let agent_name = container.strip_prefix(AGENT_PREFIX).unwrap_or(container);

    // Claude credentials land at `/home/<agent>/.claude` so the
    // `claude` CLI (which reads `$HOME/.claude`) finds them. The
    // harness service's environment sets `HOME` to the same path
    // (`agent.nix` / `ruth.nix` templates), so no `--setenv` plumbing
    // is needed here — the bind alone is enough.
    let claude_mount = container_claude_mount(agent_name);

    // No hive-wide secrets are forwarded into agent containers. hive-priv
    // still accepts a credential list (see `write_nspawn_flags`), but
    // nothing produces one: the only entry was the OTEL upstream token,
    // and an agent has no business holding the hive's credential for
    // anything outside it.
    let load_creds: Vec<CredentialMount> = Vec::new();

    let mut binds: Vec<BindMount> = vec![
        BindMount {
            host_path: runtime_dir.to_string_lossy().into_owned(),
            container_path: CONTAINER_RUNTIME_MOUNT.to_owned(),
            read_only: false,
        },
        BindMount {
            host_path: claude_dir.to_string_lossy().into_owned(),
            container_path: claude_mount,
            read_only: false,
        },
        BindMount {
            host_path: shared_root.to_string_lossy().into_owned(),
            container_path: CONTAINER_SHARED_MOUNT.to_owned(),
            read_only: false,
        },
        BindMount {
            host_path: crate::knowledge::LOCAL_DIR.to_owned(),
            container_path: crate::knowledge::CONTAINER_MOUNT.to_owned(),
            read_only: true,
        },
    ];

    // Own state, harness, and config dirs — same for every agent including
    // the manager. Config is RO: an agent must not edit its own config; changes
    // only ever flow through the approval queue.
    binds.push(BindMount {
        host_path: notes_dir.to_string_lossy().into_owned(),
        container_path: format!("/agents/{agent_name}/state"),
        read_only: false,
    });
    if let Some(state_parent) = notes_dir.parent() {
        let harness_dir = state_parent.join("harness");
        if !harness_dir.exists() {
            let _ = std::fs::create_dir_all(&harness_dir);
        }
        binds.push(BindMount {
            host_path: harness_dir.to_string_lossy().into_owned(),
            container_path: format!("/agents/{agent_name}/harness"),
            read_only: false,
        });
    }
    hive_types::Ident::parse(agent_name)
        .map_err(|e| anyhow::anyhow!("invalid agent name {agent_name:?}: {e}"))?;
    let own_config = config_bind_source(agent_name);
    std::fs::create_dir_all(&own_config)
        .with_context(|| format!("create {}", own_config.display()))?;
    binds.push(BindMount {
        host_path: own_config.to_string_lossy().into_owned(),
        container_path: format!("/agents/{agent_name}/config"),
        read_only: true,
    });

    // Topology-driven child mounts: every direct child of this agent gets
    // its state dir RW and its config dir RO. See `bind_child_agent_dirs`
    // for why each is what it is.
    let direct_children = crate::topology::children_of(agent_name);
    for child in &direct_children {
        bind_child_agent_dirs(child, &mut binds);
    }

    // `can_manage_top_level_agents` role: additionally mount every
    // parentless agent in the topology as a virtual child. Enables
    // recovery — a role holder can update those agents' configs even
    // when they are down. Also grants RO access to /applied and /meta.
    if crate::topology::has_role(
        agent_name,
        crate::topology::ROLE_CAN_MANAGE_TOP_LEVEL_AGENTS,
    ) {
        let top_level = crate::topology::top_level_agents();
        for tl in &top_level {
            if !direct_children.contains(tl) {
                bind_child_agent_dirs(tl, &mut binds);
            }
        }
        // systemd-nspawn refuses to start a container whose bind
        // source doesn't exist. The meta repo is created by the
        // startup migration, but make sure the directory is there
        // before the role holder comes up in case set_nspawn_flags
        // fires first (e.g. cold start with no agents).
        let meta_root = crate::paths::meta_root();
        std::fs::create_dir_all(&meta_root)
            .with_context(|| format!("create {}", meta_root.display()))?;
        binds.push(BindMount {
            host_path: crate::paths::applied_root().to_string_lossy().into_owned(),
            container_path: CONTAINER_MANAGER_APPLIED_MOUNT.to_owned(),
            read_only: true,
        });
        binds.push(BindMount {
            host_path: meta_root.to_string_lossy().into_owned(),
            container_path: crate::meta::CONTAINER_MANAGER_META_MOUNT.to_owned(),
            read_only: true,
        });
    }

    // Web-socket subdir: bind-mount `/run/hive-agent/<name>/` into the
    // container so the harness can bind `web.sock` there and the host-side
    // gateway sees it. Subdir bind (not socket file) keeps the inode
    // visible after the harness unlinks a stale socket on rebind.
    // Applies to manager and sub-agents alike.
    let socket_dir = crate::agent_sockets::agent_dir_for(agent_name);
    std::fs::create_dir_all(&socket_dir)
        .with_context(|| format!("create {}", socket_dir.display()))?;
    // Ownership is NOT repaired here. The dir's owner + mode are declared by
    // the tmpfiles.d entry (`SyncAgentTmpfiles`), which is the mechanism that
    // re-applies on every boot and every spawn — so a chown made here was
    // silently reverted the next time any agent was spawned or destroyed.
    // This `create_dir_all` only covers the window before that sync lands.
    binds.push(BindMount {
        host_path: socket_dir.to_string_lossy().into_owned(),
        container_path: socket_dir.to_string_lossy().into_owned(),
        read_only: false,
    });

    // Every container runs in a private network namespace with a veth
    // pair attached to the host bridge — including the manager (all
    // hive-c0re<->agent comms go through bind-mounted UDS, not TCP).
    // There is no non-isolated mode to fall back to, and the settings
    // are process-global, so anything wrong here was already fatal at
    // daemon startup; this call cannot newly fail.
    let isolation = super::network_isolation_from_env()?;
    tracing::info!(
        %agent_name, gateway_ip = %isolation.gateway_ip, bridge = %isolation.bridge,
        "network isolation: PRIVATE_NETWORK=1 (DHCP)"
    );

    // Delegate the actual conf-file rewrite to hive-priv (runs as root).
    crate::priv_client::write_nspawn_flags(container, &binds, isolation, &load_creds).await
}

#[cfg(test)]
mod tests {
    use super::{BindMount, bind_child_agent_dirs};

    fn child_binds() -> Vec<BindMount> {
        let mut binds = Vec::new();
        bind_child_agent_dirs("kiddo", &mut binds);
        binds
    }

    /// The boundary, asserted as a whole rather than per-dir: a parent
    /// sees a child's `state` and `config`, and nothing else.
    #[test]
    fn parent_sees_only_child_state_and_config() {
        let paths: Vec<String> = child_binds()
            .into_iter()
            .map(|b| b.container_path)
            .collect();
        assert_eq!(paths, ["/agents/kiddo/state", "/agents/kiddo/config"]);
    }

    /// The `config` mount names the **deployed** tree, not the working
    /// clone the proposal is staged in. Asserted as "outside the child's
    /// own dir" rather than by equality: the point is that the two are
    /// different objects, which is what makes the mount unable to show a
    /// config that was never approved. Equality with `applied_dir` would
    /// restate the implementation and pass under any future relocation.
    #[test]
    fn child_config_mount_is_the_deployed_tree_not_the_working_clone() {
        let working_clone =
            crate::paths::agent_state_dir(&hive_types::Ident::parse("kiddo").expect("valid ident"));
        let config = child_binds()
            .into_iter()
            .find(|b| b.container_path.ends_with("/config"))
            .expect("a config bind");
        assert!(
            !std::path::Path::new(&config.host_path).starts_with(&working_clone),
            "config mount must not come from the child's working clone: {}",
            config.host_path
        );
    }

    /// The regression this exists for. `harness` holds the child's own
    /// runtime material and was only ever mounted because one loop
    /// treated all three dirs alike — re-adding it to that loop is a
    /// one-word change that nothing else would catch.
    #[test]
    fn parent_never_sees_a_child_harness_dir() {
        for bind in child_binds() {
            assert!(
                !bind.container_path.contains("harness"),
                "child harness must not be bound into a parent: {}",
                bind.container_path
            );
            assert!(
                !bind.host_path.contains("harness"),
                "child harness must not be bound into a parent: {}",
                bind.host_path
            );
        }
    }

    /// An unparseable name yields no mounts at all — the guard must fail
    /// closed, since the alternative is a path built from unvalidated
    /// input.
    #[test]
    fn an_invalid_child_name_binds_nothing() {
        let mut binds = Vec::new();
        bind_child_agent_dirs("../escape", &mut binds);
        assert!(binds.is_empty());
    }
}