diff --git a/Cargo.toml b/Cargo.toml index dec6e1c0..f18cf761 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,14 @@ members = [ edition = "2024" version = "0.1.0" +[workspace.lints.clippy] +pedantic = { level = "warn", priority = -1 } +# Tolerated stylistic pedantic lints (noisy, not actionable). +missing_errors_doc = "allow" +missing_panics_doc = "allow" +module_name_repetitions = "allow" +must_use_candidate = "allow" + [workspace.dependencies] anyhow = "1" clap = { version = "4", features = ["derive"] } diff --git a/flake.nix b/flake.nix index 4d845932..9b480bbc 100644 --- a/flake.nix +++ b/flake.nix @@ -79,20 +79,30 @@ nixosModules = { agent-base = ./nix/templates/agent-base.nix; hive-c0re = ./nix/modules/hive-c0re.nix; + manager = ./nix/templates/manager.nix; }; - nixosConfigurations.agent-base = nixpkgs.lib.nixosSystem { - system = "x86_64-linux"; - modules = [ - self.nixosModules.agent-base - { - nixpkgs.overlays = [ - self.overlays.default - self.overlays.claude-unstable - ]; - } - ]; - }; + nixosConfigurations = + let + mkContainer = + module: + nixpkgs.lib.nixosSystem { + system = "x86_64-linux"; + modules = [ + module + { + nixpkgs.overlays = [ + self.overlays.default + self.overlays.claude-unstable + ]; + } + ]; + }; + in + { + agent-base = mkContainer self.nixosModules.agent-base; + manager = mkContainer self.nixosModules.manager; + }; devShells = forAllSystems ( { pkgs, ... }: @@ -105,6 +115,7 @@ rust-analyzer rustc rustfmt + sqlite ]; }; } @@ -113,9 +124,18 @@ formatter = forAllSystems ({ treefmt-eval, ... }: treefmt-eval.config.build.wrapper); checks = forAllSystems ( - { treefmt-eval, ... }: + { + treefmt-eval, + naersk-lib, + ... + }: { formatting = treefmt-eval.config.build.check self; + clippy = naersk-lib.buildPackage { + src = ./.; + mode = "clippy"; + cargoClippyOptions = orig: orig ++ [ "--all-targets" "--" "-D" "warnings" ]; + }; } ); }; diff --git a/hive-ag3nt/Cargo.toml b/hive-ag3nt/Cargo.toml index f5bc59d7..8154257c 100644 --- a/hive-ag3nt/Cargo.toml +++ b/hive-ag3nt/Cargo.toml @@ -3,6 +3,9 @@ name = "hive-ag3nt" edition.workspace = true version.workspace = true +[lints] +workspace = true + [dependencies] anyhow.workspace = true clap.workspace = true diff --git a/hive-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs index 33d2ecb3..f8a66f91 100644 --- a/hive-ag3nt/src/bin/hive-ag3nt.rs +++ b/hive-ag3nt/src/bin/hive-ag3nt.rs @@ -46,12 +46,13 @@ async fn main() -> Result<()> { match cli.cmd { Cmd::Serve { poll_ms } => serve(&cli.socket, Duration::from_millis(poll_ms)).await, Cmd::Send { to, body } => { - let resp = client::request(&cli.socket, AgentRequest::Send { to, body }).await?; + let resp: AgentResponse = + client::request(&cli.socket, &AgentRequest::Send { to, body }).await?; render(&resp)?; check(&resp) } Cmd::Recv => { - let resp = client::request(&cli.socket, AgentRequest::Recv).await?; + let resp: AgentResponse = client::request(&cli.socket, &AgentRequest::Recv).await?; render(&resp)?; check(&resp) } @@ -61,7 +62,8 @@ async fn main() -> Result<()> { async fn serve(socket: &Path, interval: Duration) -> Result<()> { tracing::info!(socket = %socket.display(), "hive-ag3nt serve"); loop { - match client::request(socket, AgentRequest::Recv).await { + let recv: Result = client::request(socket, &AgentRequest::Recv).await; + match recv { Ok(AgentResponse::Message { from, body }) => { tracing::info!(%from, %body, "inbox"); // Don't auto-reply to echoes — prevents infinite ping-pong when @@ -69,15 +71,15 @@ async fn serve(socket: &Path, interval: Duration) -> Result<()> { // manager's job (Phase 4+). if !body.starts_with("echo: ") { let reply = compute_reply(&body).await; - if let Err(e) = client::request( + let send: Result = client::request( socket, - AgentRequest::Send { + &AgentRequest::Send { to: from, body: reply, }, ) - .await - { + .await; + if let Err(e) = send { tracing::warn!(error = ?e, "send reply failed"); } } diff --git a/hive-ag3nt/src/bin/hive-m1nd.rs b/hive-ag3nt/src/bin/hive-m1nd.rs index a63ce9d6..89fb8214 100644 --- a/hive-ag3nt/src/bin/hive-m1nd.rs +++ b/hive-ag3nt/src/bin/hive-m1nd.rs @@ -1,5 +1,100 @@ -fn main() { - // Phase 4 — manager tool surface. For now, a placeholder so the binary - // exists and can be referenced from the manager nixos-container template. - println!("hive-m1nd placeholder"); +//! Manager harness. Talks to the manager socket (bind-mounted from the host +//! at `/run/hive/mcp.sock` inside the `hm1nd` container) using the privileged +//! tool surface. Phase 4 minimum: a CLI to exercise the verbs from a shell, +//! plus a `serve` loop that logs the manager's inbox. + +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use anyhow::{Result, bail}; +use clap::{Parser, Subcommand}; +use hive_ag3nt::{DEFAULT_SOCKET, client}; +use hive_sh4re::{ManagerRequest, ManagerResponse}; + +#[derive(Parser)] +#[command(name = "hive-m1nd", about = "hyperhive manager harness")] +struct Cli { + /// Path to the manager MCP socket (bind-mounted from the host). + #[arg(long, global = true, default_value = DEFAULT_SOCKET)] + socket: PathBuf, + + #[command(subcommand)] + cmd: Cmd, +} + +#[derive(Subcommand)] +enum Cmd { + /// Long-lived loop polling the manager inbox. + Serve { + #[arg(long, default_value_t = 1000)] + poll_ms: u64, + }, + /// Send a message to a sub-agent (or anywhere — the broker doesn't validate). + Send { to: String, body: String }, + /// Pop one message from the manager's inbox. + Recv, + /// Spawn a sub-agent. + Spawn { name: String }, + /// Kill a sub-agent. + Kill { name: String }, + /// Submit a config commit on the agent's config repo for user approval. + RequestApplyCommit { agent: String, commit_ref: String }, +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .init(); + + let cli = Cli::parse(); + match cli.cmd { + Cmd::Serve { poll_ms } => serve(&cli.socket, Duration::from_millis(poll_ms)).await, + Cmd::Send { to, body } => one_shot(&cli.socket, ManagerRequest::Send { to, body }).await, + Cmd::Recv => one_shot(&cli.socket, ManagerRequest::Recv).await, + Cmd::Spawn { name } => one_shot(&cli.socket, ManagerRequest::Spawn { name }).await, + Cmd::Kill { name } => one_shot(&cli.socket, ManagerRequest::Kill { name }).await, + Cmd::RequestApplyCommit { agent, commit_ref } => { + one_shot( + &cli.socket, + ManagerRequest::RequestApplyCommit { agent, commit_ref }, + ) + .await + } + } +} + +async fn one_shot(socket: &Path, req: ManagerRequest) -> Result<()> { + let resp: ManagerResponse = client::request(socket, &req).await?; + println!("{}", serde_json::to_string_pretty(&resp)?); + if let ManagerResponse::Err { message } = resp { + bail!("{message}"); + } + Ok(()) +} + +async fn serve(socket: &Path, interval: Duration) -> Result<()> { + tracing::info!(socket = %socket.display(), "hive-m1nd serve"); + loop { + let recv: Result = client::request(socket, &ManagerRequest::Recv).await; + match recv { + Ok(ManagerResponse::Message { from, body }) => { + tracing::info!(%from, %body, "manager inbox"); + } + Ok(ManagerResponse::Empty) => {} + Ok(ManagerResponse::Ok) => { + tracing::warn!("recv produced Ok (unexpected)"); + } + Ok(ManagerResponse::Err { message }) => { + tracing::warn!(%message, "recv error"); + } + Err(e) => { + tracing::warn!(error = ?e, "recv failed; retrying"); + } + } + tokio::time::sleep(interval).await; + } } diff --git a/hive-ag3nt/src/client.rs b/hive-ag3nt/src/client.rs index 8deb726c..5a314bd4 100644 --- a/hive-ag3nt/src/client.rs +++ b/hive-ag3nt/src/client.rs @@ -1,17 +1,24 @@ use std::path::Path; use anyhow::{Context, Result, bail}; -use hive_sh4re::{AgentRequest, AgentResponse}; +use serde::Serialize; +use serde::de::DeserializeOwned; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; -pub async fn request(socket: &Path, req: AgentRequest) -> Result { +/// Generic JSON-line request/response over a unix socket. One request, one +/// response, then drop. Used by both the agent and manager harnesses. +pub async fn request(socket: &Path, req: &Req) -> Result +where + Req: Serialize + ?Sized, + Resp: DeserializeOwned, +{ let stream = UnixStream::connect(socket) .await .with_context(|| format!("connect to {}", socket.display()))?; let (read, mut write) = stream.into_split(); - let mut payload = serde_json::to_string(&req)?; + let mut payload = serde_json::to_string(req)?; payload.push('\n'); write.write_all(payload.as_bytes()).await?; write.flush().await?; @@ -22,6 +29,5 @@ pub async fn request(socket: &Path, req: AgentRequest) -> Result if line.is_empty() { bail!("server closed connection without responding"); } - let resp: AgentResponse = serde_json::from_str(line.trim())?; - Ok(resp) + Ok(serde_json::from_str(line.trim())?) } diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 2d84474b..8eafdfa9 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -3,6 +3,9 @@ name = "hive-c0re" edition.workspace = true version.workspace = true +[lints] +workspace = true + [dependencies] anyhow.workspace = true clap.workspace = true diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index 05aae47e..66d484d5 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -2,7 +2,7 @@ //! authenticates the caller: connecting to `<.../agents/foo/mcp.sock>` means //! you are `foo`. -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Arc; use anyhow::{Context, Result}; @@ -18,23 +18,24 @@ pub struct AgentSocket { pub handle: JoinHandle<()>, } -pub async fn start( - agent: String, - socket_path: PathBuf, +pub fn start( + agent: &str, + socket_path: &Path, broker: Arc, ) -> Result { + let agent = agent.to_owned(); if let Some(parent) = socket_path.parent() { std::fs::create_dir_all(parent) .with_context(|| format!("create agent socket dir {}", parent.display()))?; } if socket_path.exists() { - std::fs::remove_file(&socket_path).context("remove stale agent socket")?; + std::fs::remove_file(socket_path).context("remove stale agent socket")?; } - let listener = UnixListener::bind(&socket_path) + let listener = UnixListener::bind(socket_path) .with_context(|| format!("bind agent socket {}", socket_path.display()))?; tracing::info!(%agent, socket = %socket_path.display(), "agent socket listening"); - let path = socket_path.clone(); + let path = socket_path.to_path_buf(); let handle = tokio::spawn(async move { loop { match listener.accept().await { @@ -83,7 +84,7 @@ async fn serve(stream: UnixStream, agent: String, broker: Arc) -> Result fn dispatch(req: &AgentRequest, agent: &str, broker: &Broker) -> AgentResponse { match req { AgentRequest::Send { to, body } => { - match broker.send(Message { + match broker.send(&Message { from: agent.to_owned(), to: to.clone(), body: body.clone(), diff --git a/hive-c0re/src/approvals.rs b/hive-c0re/src/approvals.rs new file mode 100644 index 00000000..0dbeac17 --- /dev/null +++ b/hive-c0re/src/approvals.rs @@ -0,0 +1,169 @@ +//! Approval queue. Manager submits via `RequestApplyCommit`; the user +//! approves/denies via the host admin CLI; on approval the host runs the +//! corresponding action (Phase 5a: `lifecycle::rebuild(agent)`). + +use std::path::Path; +use std::sync::Mutex; +use std::time::{SystemTime, UNIX_EPOCH}; + +use anyhow::{Context, Result, bail}; +use hive_sh4re::{Approval, ApprovalStatus}; +use rusqlite::{Connection, OptionalExtension, params}; + +const SCHEMA: &str = r" +CREATE TABLE IF NOT EXISTS approvals ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + agent TEXT NOT NULL, + commit_ref TEXT NOT NULL, + requested_at INTEGER NOT NULL, + status TEXT NOT NULL, + resolved_at INTEGER, + note TEXT +); +CREATE INDEX IF NOT EXISTS idx_approvals_pending + ON approvals (id) WHERE status = 'pending'; +"; + +pub struct Approvals { + conn: Mutex, +} + +impl Approvals { + pub fn open(path: &Path) -> Result { + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("create approvals db parent {}", parent.display()))?; + } + let conn = Connection::open(path) + .with_context(|| format!("open approvals db {}", path.display()))?; + conn.execute_batch(SCHEMA).context("apply approvals schema")?; + Ok(Self { + conn: Mutex::new(conn), + }) + } + + pub fn submit(&self, agent: &str, commit_ref: &str) -> Result { + let conn = self.conn.lock().unwrap(); + conn.execute( + "INSERT INTO approvals (agent, commit_ref, requested_at, status) + VALUES (?1, ?2, ?3, 'pending')", + params![agent, commit_ref, now_unix()], + )?; + Ok(conn.last_insert_rowid()) + } + + pub fn pending(&self) -> Result> { + let conn = self.conn.lock().unwrap(); + let mut stmt = conn.prepare( + "SELECT id, agent, commit_ref, requested_at, status, resolved_at, note + FROM approvals + WHERE status = 'pending' + ORDER BY id ASC", + )?; + let rows = stmt.query_map([], row_to_approval)?; + rows.collect::>>() + .map_err(Into::into) + } + + #[allow(dead_code)] // used by Phase 5b commit verification + pub fn get(&self, id: i64) -> Result> { + let conn = self.conn.lock().unwrap(); + conn.query_row( + "SELECT id, agent, commit_ref, requested_at, status, resolved_at, note + FROM approvals WHERE id = ?1", + params![id], + row_to_approval, + ) + .optional() + .map_err(Into::into) + } + + /// Mark pending -> approved (or fail if not pending). Returns the (now-updated) + /// approval so the caller can run the action and pass the agent name. + pub fn mark_approved(&self, id: i64) -> Result { + let conn = self.conn.lock().unwrap(); + let current: Option<(String, String, i64, String)> = conn + .query_row( + "SELECT agent, commit_ref, requested_at, status FROM approvals WHERE id = ?1", + params![id], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .optional()?; + let Some((agent, commit_ref, requested_at, status)) = current else { + bail!("approval {id} not found"); + }; + if status != "pending" { + bail!("approval {id} is {status}, not pending"); + } + let resolved_at = now_unix(); + conn.execute( + "UPDATE approvals SET status = 'approved', resolved_at = ?1 WHERE id = ?2", + params![resolved_at, id], + )?; + Ok(Approval { + id, + agent, + commit_ref, + requested_at, + status: ApprovalStatus::Approved, + resolved_at: Some(resolved_at), + note: None, + }) + } + + pub fn mark_denied(&self, id: i64) -> Result<()> { + let conn = self.conn.lock().unwrap(); + let affected = conn.execute( + "UPDATE approvals SET status = 'denied', resolved_at = ?1 + WHERE id = ?2 AND status = 'pending'", + params![now_unix(), id], + )?; + if affected == 0 { + bail!("approval {id} not pending"); + } + Ok(()) + } + + pub fn mark_failed(&self, id: i64, note: &str) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "UPDATE approvals SET status = 'failed', resolved_at = ?1, note = ?2 WHERE id = ?3", + params![now_unix(), note, id], + )?; + Ok(()) + } +} + +fn row_to_approval(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let status: String = row.get(4)?; + let status = match status.as_str() { + "pending" => ApprovalStatus::Pending, + "approved" => ApprovalStatus::Approved, + "denied" => ApprovalStatus::Denied, + "failed" => ApprovalStatus::Failed, + other => { + return Err(rusqlite::Error::FromSqlConversionFailure( + 4, + rusqlite::types::Type::Text, + format!("unknown approval status '{other}'").into(), + )); + } + }; + Ok(Approval { + id: row.get(0)?, + agent: row.get(1)?, + commit_ref: row.get(2)?, + requested_at: row.get(3)?, + status, + resolved_at: row.get(5)?, + note: row.get(6)?, + }) +} + +fn now_unix() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .ok() + .and_then(|d| i64::try_from(d.as_secs()).ok()) + .unwrap_or(0) +} diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index a88bbd8e..98719eea 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -8,7 +8,7 @@ use anyhow::{Context, Result}; use hive_sh4re::Message; use rusqlite::{Connection, OptionalExtension, params}; -const SCHEMA: &str = r#" +const SCHEMA: &str = r" CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, sender TEXT NOT NULL, @@ -19,7 +19,7 @@ CREATE TABLE IF NOT EXISTS messages ( ); CREATE INDEX IF NOT EXISTS idx_messages_undelivered ON messages (recipient, id) WHERE delivered_at IS NULL; -"#; +"; pub struct Broker { conn: Mutex, @@ -39,7 +39,7 @@ impl Broker { }) } - pub fn send(&self, message: Message) -> Result<()> { + pub fn send(&self, message: &Message) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "INSERT INTO messages (sender, recipient, body, sent_at) VALUES (?1, ?2, ?3, ?4)", @@ -75,6 +75,7 @@ impl Broker { fn now_unix() -> i64 { SystemTime::now() .duration_since(UNIX_EPOCH) - .map(|d| d.as_secs() as i64) + .ok() + .and_then(|d| i64::try_from(d.as_secs()).ok()) .unwrap_or(0) } diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index b2ba2f90..ce2032e6 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -1,5 +1,6 @@ -//! Runtime state shared between the host admin socket and the per-agent -//! sockets: the broker plus a map of `name -> AgentSocket`. +//! Runtime state + config shared between the host admin socket, the manager +//! socket, and the per-agent sockets: the broker, configured `agent_flake`, +//! and the map of registered agent sockets. use std::collections::HashMap; use std::path::{Path, PathBuf}; @@ -8,25 +9,32 @@ use std::sync::{Arc, Mutex}; use anyhow::{Context, Result}; use crate::agent_server::{self, AgentSocket}; +use crate::approvals::Approvals; use crate::broker::Broker; const AGENT_RUNTIME_ROOT: &str = "/run/hyperhive/agents"; +const MANAGER_RUNTIME_ROOT: &str = "/run/hyperhive/manager"; pub struct Coordinator { pub broker: Arc, + pub approvals: Arc, + pub agent_flake: String, agents: Mutex>, } impl Coordinator { - pub fn open(db_path: &Path) -> Result { + pub fn open(db_path: &Path, agent_flake: String) -> Result { let broker = Broker::open(db_path).context("open broker")?; + let approvals = Approvals::open(db_path).context("open approvals")?; Ok(Self { broker: Arc::new(broker), + approvals: Arc::new(approvals), + agent_flake, agents: Mutex::new(HashMap::new()), }) } - pub async fn register_agent(&self, name: &str) -> Result { + pub fn register_agent(&self, name: &str) -> Result { // Idempotent: drop any existing listener so re-registration (e.g. on rebuild, // or after a hive-c0re restart cleared /run/hyperhive) gets a fresh socket. self.unregister_agent(name); @@ -34,7 +42,7 @@ impl Coordinator { std::fs::create_dir_all(&agent_dir) .with_context(|| format!("create agent dir {}", agent_dir.display()))?; let socket_path = Self::socket_path(name); - let socket = agent_server::start(name.to_owned(), socket_path, self.broker.clone()).await?; + let socket = agent_server::start(name, &socket_path, self.broker.clone())?; self.agents.lock().unwrap().insert(name.to_owned(), socket); Ok(agent_dir) } @@ -53,4 +61,12 @@ impl Coordinator { pub fn socket_path(name: &str) -> PathBuf { Self::agent_dir(name).join("mcp.sock") } + + pub fn manager_dir() -> PathBuf { + PathBuf::from(MANAGER_RUNTIME_ROOT) + } + + pub fn manager_socket_path() -> PathBuf { + Self::manager_dir().join("mcp.sock") + } } diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 6346cf9e..f31f2c1a 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -6,10 +6,12 @@ use clap::{Parser, Subcommand}; use hive_sh4re::{HostRequest, HostResponse}; mod agent_server; +mod approvals; mod broker; mod client; mod coordinator; mod lifecycle; +mod manager_server; mod server; use coordinator::Coordinator; @@ -44,6 +46,12 @@ enum Cmd { Rebuild { name: String }, /// List managed containers. List, + /// List pending approval requests submitted by the manager. + Pending, + /// Approve a pending request by id; the action runs immediately. + Approve { id: i64 }, + /// Deny a pending request by id. + Deny { id: i64 }, } #[tokio::main] @@ -58,8 +66,9 @@ async fn main() -> Result<()> { let cli = Cli::parse(); match cli.cmd { Cmd::Serve { agent_flake, db } => { - let coord = Arc::new(Coordinator::open(&db)?); - server::serve(&cli.socket, &agent_flake, coord).await + let coord = Arc::new(Coordinator::open(&db, agent_flake)?); + manager_server::start(coord.clone())?; + server::serve(&cli.socket, coord).await } Cmd::Spawn { name } => { render(client::request(&cli.socket, HostRequest::Spawn { name }).await?) @@ -71,6 +80,9 @@ async fn main() -> Result<()> { render(client::request(&cli.socket, HostRequest::Rebuild { name }).await?) } Cmd::List => render(client::request(&cli.socket, HostRequest::List).await?), + Cmd::Pending => render(client::request(&cli.socket, HostRequest::Pending).await?), + Cmd::Approve { id } => render(client::request(&cli.socket, HostRequest::Approve { id }).await?), + Cmd::Deny { id } => render(client::request(&cli.socket, HostRequest::Deny { id }).await?), } } diff --git a/hive-c0re/src/manager_server.rs b/hive-c0re/src/manager_server.rs new file mode 100644 index 00000000..086c9ba3 --- /dev/null +++ b/hive-c0re/src/manager_server.rs @@ -0,0 +1,140 @@ +//! Manager socket listener. Privileged tool surface: agent-style send/recv +//! plus lifecycle verbs (Phase 4). Phase 5 will gate Spawn/Kill behind the +//! commit-approval flow; for now they hit the same code path the host admin +//! socket uses. + +use std::sync::Arc; + +use anyhow::{Context, Result}; +use hive_sh4re::{MANAGER_AGENT, ManagerRequest, ManagerResponse, Message}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{UnixListener, UnixStream}; + +use crate::coordinator::Coordinator; +use crate::lifecycle; + +pub fn start(coord: Arc) -> Result<()> { + let dir = Coordinator::manager_dir(); + std::fs::create_dir_all(&dir) + .with_context(|| format!("create manager dir {}", dir.display()))?; + let socket = Coordinator::manager_socket_path(); + if socket.exists() { + std::fs::remove_file(&socket).context("remove stale manager socket")?; + } + let listener = UnixListener::bind(&socket) + .with_context(|| format!("bind manager socket {}", socket.display()))?; + tracing::info!(socket = %socket.display(), "manager socket listening"); + + tokio::spawn(async move { + loop { + match listener.accept().await { + Ok((stream, _)) => { + let coord = coord.clone(); + tokio::spawn(async move { + if let Err(e) = serve(stream, coord).await { + tracing::warn!(error = ?e, "manager connection failed"); + } + }); + } + Err(e) => { + tracing::warn!(error = ?e, "manager listener accept failed"); + return; + } + } + } + }); + Ok(()) +} + +async fn serve(stream: UnixStream, coord: Arc) -> Result<()> { + let (read, mut write) = stream.into_split(); + let mut reader = BufReader::new(read); + let mut line = String::new(); + loop { + line.clear(); + let n = reader.read_line(&mut line).await?; + if n == 0 { + return Ok(()); + } + let resp = match serde_json::from_str::(line.trim()) { + Ok(req) => dispatch(&req, &coord).await, + Err(e) => ManagerResponse::Err { + message: format!("parse error: {e}"), + }, + }; + let mut payload = serde_json::to_string(&resp)?; + payload.push('\n'); + write.write_all(payload.as_bytes()).await?; + write.flush().await?; + } +} + +async fn dispatch(req: &ManagerRequest, coord: &Coordinator) -> ManagerResponse { + match req { + ManagerRequest::Send { to, body } => match coord.broker.send(&Message { + from: MANAGER_AGENT.to_owned(), + to: to.clone(), + body: body.clone(), + }) { + Ok(()) => ManagerResponse::Ok, + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, + }, + ManagerRequest::Recv => match coord.broker.recv(MANAGER_AGENT) { + Ok(Some(msg)) => ManagerResponse::Message { + from: msg.from, + body: msg.body, + }, + Ok(None) => ManagerResponse::Empty, + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, + }, + ManagerRequest::Spawn { name } => { + tracing::info!(%name, "manager: spawn"); + let result: Result<()> = async { + let agent_dir = coord.register_agent(name)?; + if let Err(e) = lifecycle::spawn(name, &coord.agent_flake, &agent_dir).await { + coord.unregister_agent(name); + return Err(e); + } + Ok(()) + } + .await; + match result { + Ok(()) => ManagerResponse::Ok, + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, + } + } + ManagerRequest::Kill { name } => { + tracing::info!(%name, "manager: kill"); + let result: Result<()> = async { + lifecycle::kill(name).await?; + coord.unregister_agent(name); + Ok(()) + } + .await; + match result { + Ok(()) => ManagerResponse::Ok, + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, + } + } + ManagerRequest::RequestApplyCommit { agent, commit_ref } => { + tracing::info!(%agent, %commit_ref, "manager: request_apply_commit"); + match coord.approvals.submit(agent, commit_ref) { + Ok(id) => { + tracing::info!(%id, %agent, %commit_ref, "approval queued"); + ManagerResponse::Ok + } + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, + } + } + } +} diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 2e34a9b0..8ed0c15c 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -9,7 +9,7 @@ use tokio::net::{UnixListener, UnixStream}; use crate::coordinator::Coordinator; use crate::lifecycle; -pub async fn serve(socket: &Path, agent_flake: &str, coord: Arc) -> Result<()> { +pub async fn serve(socket: &Path, coord: Arc) -> Result<()> { if let Some(parent) = socket.parent() { std::fs::create_dir_all(parent) .with_context(|| format!("create socket parent {}", parent.display()))?; @@ -20,21 +20,20 @@ pub async fn serve(socket: &Path, agent_flake: &str, coord: Arc) -> let listener = UnixListener::bind(socket) .with_context(|| format!("bind admin socket {}", socket.display()))?; - tracing::info!(socket = %socket.display(), %agent_flake, "hive-c0re listening"); + tracing::info!(socket = %socket.display(), agent_flake = %coord.agent_flake, "hive-c0re admin listening"); loop { let (stream, _) = listener.accept().await.context("accept connection")?; - let agent_flake = agent_flake.to_owned(); let coord = coord.clone(); tokio::spawn(async move { - if let Err(e) = handle(stream, &agent_flake, coord).await { + if let Err(e) = handle(stream, coord).await { tracing::warn!(error = ?e, "connection failed"); } }); } } -async fn handle(stream: UnixStream, agent_flake: &str, coord: Arc) -> Result<()> { +async fn handle(stream: UnixStream, coord: Arc) -> Result<()> { let (read, mut write) = stream.into_split(); let mut reader = BufReader::new(read); let mut line = String::new(); @@ -46,7 +45,7 @@ async fn handle(stream: UnixStream, agent_flake: &str, coord: Arc) return Ok(()); } let resp = match serde_json::from_str::(line.trim()) { - Ok(req) => dispatch(&req, agent_flake, &coord).await, + Ok(req) => dispatch(&req, &coord).await, Err(e) => HostResponse::error(format!("parse error: {e}")), }; let mut payload = serde_json::to_string(&resp)?; @@ -56,13 +55,13 @@ async fn handle(stream: UnixStream, agent_flake: &str, coord: Arc) } } -async fn dispatch(req: &HostRequest, agent_flake: &str, coord: &Coordinator) -> HostResponse { +async fn dispatch(req: &HostRequest, coord: &Coordinator) -> HostResponse { let result: anyhow::Result = async { Ok(match req { HostRequest::Spawn { name } => { tracing::info!(%name, "spawn"); - let agent_dir = coord.register_agent(name).await?; - if let Err(e) = lifecycle::spawn(name, agent_flake, &agent_dir).await { + let agent_dir = coord.register_agent(name)?; + if let Err(e) = lifecycle::spawn(name, &coord.agent_flake, &agent_dir).await { // Roll back socket registration if container creation failed. coord.unregister_agent(name); return Err(e); @@ -77,11 +76,30 @@ async fn dispatch(req: &HostRequest, agent_flake: &str, coord: &Coordinator) -> } HostRequest::Rebuild { name } => { tracing::info!(%name, "rebuild"); - let agent_dir = coord.register_agent(name).await?; - lifecycle::rebuild(name, agent_flake, &agent_dir).await?; + let agent_dir = coord.register_agent(name)?; + lifecycle::rebuild(name, &coord.agent_flake, &agent_dir).await?; HostResponse::success() } HostRequest::List => HostResponse::list(lifecycle::list().await?), + HostRequest::Pending => HostResponse::pending(coord.approvals.pending()?), + HostRequest::Approve { id } => { + let approval = coord.approvals.mark_approved(*id)?; + tracing::info!(%approval.id, %approval.agent, %approval.commit_ref, "approval applied: rebuilding agent"); + let agent_dir = coord.register_agent(&approval.agent)?; + if let Err(e) = + lifecycle::rebuild(&approval.agent, &coord.agent_flake, &agent_dir).await + { + let note = format!("{e:#}"); + let _ = coord.approvals.mark_failed(approval.id, ¬e); + return Err(e); + } + HostResponse::success() + } + HostRequest::Deny { id } => { + coord.approvals.mark_denied(*id)?; + tracing::info!(%id, "approval denied"); + HostResponse::success() + } }) } .await; diff --git a/hive-sh4re/Cargo.toml b/hive-sh4re/Cargo.toml index 45f43cfb..e4f7600c 100644 --- a/hive-sh4re/Cargo.toml +++ b/hive-sh4re/Cargo.toml @@ -3,5 +3,8 @@ name = "hive-sh4re" edition.workspace = true version.workspace = true +[lints] +workspace = true + [dependencies] serde.workspace = true diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 06683116..5f609639 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -20,6 +20,12 @@ pub enum HostRequest { Rebuild { name: String }, /// List managed containers. List, + /// List pending approval requests. + Pending, + /// Approve a pending request by id; the action runs immediately. + Approve { id: i64 }, + /// Deny a pending request by id. + Deny { id: i64 }, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -29,6 +35,30 @@ pub struct HostResponse { pub error: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub agents: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub approvals: Option>, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Approval { + pub id: i64, + pub agent: String, + pub commit_ref: String, + pub requested_at: i64, + pub status: ApprovalStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub resolved_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub note: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ApprovalStatus { + Pending, + Approved, + Denied, + Failed, } impl HostResponse { @@ -37,6 +67,7 @@ impl HostResponse { ok: true, error: None, agents: None, + approvals: None, } } @@ -45,6 +76,7 @@ impl HostResponse { ok: false, error: Some(message.into()), agents: None, + approvals: None, } } @@ -53,6 +85,16 @@ impl HostResponse { ok: true, error: None, agents: Some(agents), + approvals: None, + } + } + + pub fn pending(approvals: Vec) -> Self { + Self { + ok: true, + error: None, + agents: None, + approvals: Some(approvals), } } } @@ -94,3 +136,47 @@ pub enum AgentResponse { /// `Recv` found nothing pending. Empty, } + +// ----------------------------------------------------------------------------- +// Manager socket — /run/hyperhive/manager/mcp.sock on the host, bind-mounted +// into the manager container at /run/hive/mcp.sock. +// ----------------------------------------------------------------------------- + +/// Logical name the broker uses for the manager. +pub const MANAGER_AGENT: &str = "manager"; + +/// Requests on the manager socket. Manager has the agent surface (send/recv) +/// plus privileged lifecycle verbs. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "cmd", rename_all = "snake_case")] +pub enum ManagerRequest { + Send { + to: String, + body: String, + }, + Recv, + /// Spawn a sub-agent. Phase 5 will gate this on user approval. + Spawn { + name: String, + }, + /// Stop a sub-agent (graceful). + Kill { + name: String, + }, + /// Submit a config commit for the user to approve. `commit_ref` is opaque + /// to the host (typically a git sha pointing into the agent's config repo). + /// On approval the host applies the change via `nixos-container update`. + RequestApplyCommit { + agent: String, + commit_ref: String, + }, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ManagerResponse { + Ok, + Err { message: String }, + Message { from: String, body: String }, + Empty, +} diff --git a/nix/templates/manager.nix b/nix/templates/manager.nix new file mode 100644 index 00000000..b1912955 --- /dev/null +++ b/nix/templates/manager.nix @@ -0,0 +1,26 @@ +{ pkgs, ... }: +{ + boot.isNspawnContainer = true; + + nixpkgs.config.allowUnfreePredicate = pkg: builtins.elem (pkgs.lib.getName pkg) [ "claude-code" ]; + + environment.systemPackages = with pkgs; [ + hyperhive + claude-code + git + coreutils-full + ]; + + systemd.services.hive-m1nd = { + description = "hive-m1nd manager harness"; + wantedBy = [ "multi-user.target" ]; + after = [ "network.target" ]; + serviceConfig = { + ExecStart = "${pkgs.hyperhive}/bin/hive-m1nd serve"; + Restart = "on-failure"; + RestartSec = 2; + }; + }; + + system.stateVersion = "25.11"; +}