diff --git a/Cargo.toml b/Cargo.toml index f18cf761..dec6e1c0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,14 +10,6 @@ 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 9b480bbc..4d845932 100644 --- a/flake.nix +++ b/flake.nix @@ -79,30 +79,20 @@ nixosModules = { agent-base = ./nix/templates/agent-base.nix; hive-c0re = ./nix/modules/hive-c0re.nix; - manager = ./nix/templates/manager.nix; }; - 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; - }; + nixosConfigurations.agent-base = nixpkgs.lib.nixosSystem { + system = "x86_64-linux"; + modules = [ + self.nixosModules.agent-base + { + nixpkgs.overlays = [ + self.overlays.default + self.overlays.claude-unstable + ]; + } + ]; + }; devShells = forAllSystems ( { pkgs, ... }: @@ -115,7 +105,6 @@ rust-analyzer rustc rustfmt - sqlite ]; }; } @@ -124,18 +113,9 @@ formatter = forAllSystems ({ treefmt-eval, ... }: treefmt-eval.config.build.wrapper); checks = forAllSystems ( - { - treefmt-eval, - naersk-lib, - ... - }: + { treefmt-eval, ... }: { 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 8154257c..f5bc59d7 100644 --- a/hive-ag3nt/Cargo.toml +++ b/hive-ag3nt/Cargo.toml @@ -3,9 +3,6 @@ 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 f8a66f91..33d2ecb3 100644 --- a/hive-ag3nt/src/bin/hive-ag3nt.rs +++ b/hive-ag3nt/src/bin/hive-ag3nt.rs @@ -46,13 +46,12 @@ 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: AgentResponse = - client::request(&cli.socket, &AgentRequest::Send { to, body }).await?; + let resp = client::request(&cli.socket, AgentRequest::Send { to, body }).await?; render(&resp)?; check(&resp) } Cmd::Recv => { - let resp: AgentResponse = client::request(&cli.socket, &AgentRequest::Recv).await?; + let resp = client::request(&cli.socket, AgentRequest::Recv).await?; render(&resp)?; check(&resp) } @@ -62,8 +61,7 @@ async fn main() -> Result<()> { async fn serve(socket: &Path, interval: Duration) -> Result<()> { tracing::info!(socket = %socket.display(), "hive-ag3nt serve"); loop { - let recv: Result = client::request(socket, &AgentRequest::Recv).await; - match recv { + match client::request(socket, AgentRequest::Recv).await { Ok(AgentResponse::Message { from, body }) => { tracing::info!(%from, %body, "inbox"); // Don't auto-reply to echoes — prevents infinite ping-pong when @@ -71,15 +69,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; - let send: Result = client::request( + if let Err(e) = client::request( socket, - &AgentRequest::Send { + AgentRequest::Send { to: from, body: reply, }, ) - .await; - if let Err(e) = send { + .await + { 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 89fb8214..a63ce9d6 100644 --- a/hive-ag3nt/src/bin/hive-m1nd.rs +++ b/hive-ag3nt/src/bin/hive-m1nd.rs @@ -1,100 +1,5 @@ -//! 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; - } +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"); } diff --git a/hive-ag3nt/src/client.rs b/hive-ag3nt/src/client.rs index 5a314bd4..8deb726c 100644 --- a/hive-ag3nt/src/client.rs +++ b/hive-ag3nt/src/client.rs @@ -1,24 +1,17 @@ use std::path::Path; use anyhow::{Context, Result, bail}; -use serde::Serialize; -use serde::de::DeserializeOwned; +use hive_sh4re::{AgentRequest, AgentResponse}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; -/// 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, -{ +pub async fn request(socket: &Path, req: AgentRequest) -> Result { 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?; @@ -29,5 +22,6 @@ where if line.is_empty() { bail!("server closed connection without responding"); } - Ok(serde_json::from_str(line.trim())?) + let resp: AgentResponse = serde_json::from_str(line.trim())?; + Ok(resp) } diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 8eafdfa9..2d84474b 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -3,9 +3,6 @@ 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 66d484d5..05aae47e 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::{Path, PathBuf}; +use std::path::PathBuf; use std::sync::Arc; use anyhow::{Context, Result}; @@ -18,24 +18,23 @@ pub struct AgentSocket { pub handle: JoinHandle<()>, } -pub fn start( - agent: &str, - socket_path: &Path, +pub async fn start( + agent: String, + socket_path: PathBuf, 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.to_path_buf(); + let path = socket_path.clone(); let handle = tokio::spawn(async move { loop { match listener.accept().await { @@ -84,7 +83,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 deleted file mode 100644 index 0dbeac17..00000000 --- a/hive-c0re/src/approvals.rs +++ /dev/null @@ -1,169 +0,0 @@ -//! 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 98719eea..a88bbd8e 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,7 +75,6 @@ impl Broker { fn now_unix() -> i64 { SystemTime::now() .duration_since(UNIX_EPOCH) - .ok() - .and_then(|d| i64::try_from(d.as_secs()).ok()) + .map(|d| d.as_secs() as i64) .unwrap_or(0) } diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index ce2032e6..b2ba2f90 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -1,6 +1,5 @@ -//! 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. +//! Runtime state shared between the host admin socket and the per-agent +//! sockets: the broker plus a map of `name -> AgentSocket`. use std::collections::HashMap; use std::path::{Path, PathBuf}; @@ -9,32 +8,25 @@ 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, agent_flake: String) -> Result { + pub fn open(db_path: &Path) -> 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 fn register_agent(&self, name: &str) -> Result { + pub async 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); @@ -42,7 +34,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, &socket_path, self.broker.clone())?; + let socket = agent_server::start(name.to_owned(), socket_path, self.broker.clone()).await?; self.agents.lock().unwrap().insert(name.to_owned(), socket); Ok(agent_dir) } @@ -61,12 +53,4 @@ 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 f31f2c1a..6346cf9e 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -6,12 +6,10 @@ 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; @@ -46,12 +44,6 @@ 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] @@ -66,9 +58,8 @@ async fn main() -> Result<()> { let cli = Cli::parse(); match cli.cmd { Cmd::Serve { agent_flake, db } => { - let coord = Arc::new(Coordinator::open(&db, agent_flake)?); - manager_server::start(coord.clone())?; - server::serve(&cli.socket, coord).await + let coord = Arc::new(Coordinator::open(&db)?); + server::serve(&cli.socket, &agent_flake, coord).await } Cmd::Spawn { name } => { render(client::request(&cli.socket, HostRequest::Spawn { name }).await?) @@ -80,9 +71,6 @@ 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 deleted file mode 100644 index 086c9ba3..00000000 --- a/hive-c0re/src/manager_server.rs +++ /dev/null @@ -1,140 +0,0 @@ -//! 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 8ed0c15c..2e34a9b0 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, coord: Arc) -> Result<()> { +pub async fn serve(socket: &Path, agent_flake: &str, coord: Arc) -> Result<()> { if let Some(parent) = socket.parent() { std::fs::create_dir_all(parent) .with_context(|| format!("create socket parent {}", parent.display()))?; @@ -20,20 +20,21 @@ pub async fn serve(socket: &Path, coord: Arc) -> Result<()> { let listener = UnixListener::bind(socket) .with_context(|| format!("bind admin socket {}", socket.display()))?; - tracing::info!(socket = %socket.display(), agent_flake = %coord.agent_flake, "hive-c0re admin listening"); + tracing::info!(socket = %socket.display(), %agent_flake, "hive-c0re 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, coord).await { + if let Err(e) = handle(stream, &agent_flake, coord).await { tracing::warn!(error = ?e, "connection failed"); } }); } } -async fn handle(stream: UnixStream, coord: Arc) -> Result<()> { +async fn handle(stream: UnixStream, agent_flake: &str, coord: Arc) -> Result<()> { let (read, mut write) = stream.into_split(); let mut reader = BufReader::new(read); let mut line = String::new(); @@ -45,7 +46,7 @@ async fn handle(stream: UnixStream, coord: Arc) -> Result<()> { return Ok(()); } let resp = match serde_json::from_str::(line.trim()) { - Ok(req) => dispatch(&req, &coord).await, + Ok(req) => dispatch(&req, agent_flake, &coord).await, Err(e) => HostResponse::error(format!("parse error: {e}")), }; let mut payload = serde_json::to_string(&resp)?; @@ -55,13 +56,13 @@ async fn handle(stream: UnixStream, coord: Arc) -> Result<()> { } } -async fn dispatch(req: &HostRequest, coord: &Coordinator) -> HostResponse { +async fn dispatch(req: &HostRequest, agent_flake: &str, 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)?; - if let Err(e) = lifecycle::spawn(name, &coord.agent_flake, &agent_dir).await { + let agent_dir = coord.register_agent(name).await?; + if let Err(e) = lifecycle::spawn(name, agent_flake, &agent_dir).await { // Roll back socket registration if container creation failed. coord.unregister_agent(name); return Err(e); @@ -76,30 +77,11 @@ async fn dispatch(req: &HostRequest, coord: &Coordinator) -> HostResponse { } HostRequest::Rebuild { name } => { tracing::info!(%name, "rebuild"); - let agent_dir = coord.register_agent(name)?; - lifecycle::rebuild(name, &coord.agent_flake, &agent_dir).await?; + let agent_dir = coord.register_agent(name).await?; + lifecycle::rebuild(name, 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 e4f7600c..45f43cfb 100644 --- a/hive-sh4re/Cargo.toml +++ b/hive-sh4re/Cargo.toml @@ -3,8 +3,5 @@ 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 5f609639..06683116 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -20,12 +20,6 @@ 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)] @@ -35,30 +29,6 @@ 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 { @@ -67,7 +37,6 @@ impl HostResponse { ok: true, error: None, agents: None, - approvals: None, } } @@ -76,7 +45,6 @@ impl HostResponse { ok: false, error: Some(message.into()), agents: None, - approvals: None, } } @@ -85,16 +53,6 @@ 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), } } } @@ -136,47 +94,3 @@ 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 deleted file mode 100644 index b1912955..00000000 --- a/nix/templates/manager.nix +++ /dev/null @@ -1,26 +0,0 @@ -{ 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"; -}