diff --git a/Cargo.lock b/Cargo.lock index 42d70f8e..164bfbc7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14,7 +14,7 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0" dependencies = [ - "crypto-common 0.1.7", + "crypto-common", "generic-array", ] @@ -25,7 +25,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" dependencies = [ "cfg-if", - "cipher 0.4.4", + "cipher", "cpufeatures 0.2.17", ] @@ -303,19 +303,6 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" -[[package]] -name = "bcrypt" -version = "0.19.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24ae5479c93d3720e4c1dbd6b945b97457c50cb672781104768190371df1a905" -dependencies = [ - "base64", - "blowfish", - "getrandom 0.4.2", - "subtle", - "zeroize", -] - [[package]] name = "bitflags" version = "2.11.1" @@ -363,28 +350,12 @@ dependencies = [ "generic-array", ] -[[package]] -name = "blowfish" -version = "0.10.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "62ce3946557b35e71d1bbe07ec385073ce9eda05043f95de134eb578fcf1a298" -dependencies = [ - "byteorder", - "cipher 0.5.2", -] - [[package]] name = "bumpalo" version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" -[[package]] -name = "byteorder" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" - [[package]] name = "bytes" version = "1.11.1" @@ -403,7 +374,7 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "26b52a9543ae338f279b96b0b9fed9c8093744685043739079ce85cd58f289a6" dependencies = [ - "cipher 0.4.4", + "cipher", ] [[package]] @@ -435,7 +406,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3613f74bd2eac03dad61bd53dbe620703d4371614fe0bc3b9f04dd36fe4e818" dependencies = [ "cfg-if", - "cipher 0.4.4", + "cipher", "cpufeatures 0.2.17", ] @@ -447,7 +418,7 @@ checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35" dependencies = [ "aead", "chacha20", - "cipher 0.4.4", + "cipher", "poly1305", "zeroize", ] @@ -472,21 +443,11 @@ version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ - "crypto-common 0.1.7", - "inout 0.1.4", + "crypto-common", + "inout", "zeroize", ] -[[package]] -name = "cipher" -version = "0.5.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e8cf2a2c93cd704877c0858356ed03480ff301ee950b43f1cbe4573b088bfa6c" -dependencies = [ - "crypto-common 0.2.2", - "inout 0.2.2", -] - [[package]] name = "clap" version = "4.6.1" @@ -630,15 +591,6 @@ dependencies = [ "typenum", ] -[[package]] -name = "crypto-common" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" -dependencies = [ - "hybrid-array", -] - [[package]] name = "curve25519-dalek" version = "4.1.3" @@ -790,7 +742,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer", - "crypto-common 0.1.7", + "crypto-common", "subtle", ] @@ -1248,9 +1200,9 @@ dependencies = [ "anyhow", "axum", "base64", - "bcrypt", "clap", "hive-sh4re", + "libc", "listenfd", "reqwest", "rusqlite", @@ -1402,15 +1354,6 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" -[[package]] -name = "hybrid-array" -version = "0.4.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9155a582abd142abc056962c29e3ce5ff2ad5469f4246b537ed42c5deba857da" -dependencies = [ - "typenum", -] - [[package]] name = "hyper" version = "1.10.1" @@ -1676,15 +1619,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "inout" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4250ce6452e92010fdf7268ccc5d14faa80bb12fc741938534c58f16804e03c7" -dependencies = [ - "hybrid-array", -] - [[package]] name = "ipnet" version = "2.12.0" @@ -3807,7 +3741,7 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" dependencies = [ - "crypto-common 0.1.7", + "crypto-common", "subtle", ] diff --git a/Cargo.toml b/Cargo.toml index cfcfd132..b8ac38ca 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,13 +1,6 @@ [workspace] resolver = "3" -members = [ - "hive-ag3nt", - "hive-c0re", - "hive-forge", - "hive-matrix-mcp", - "hive-priv", - "hive-sh4re", -] +members = ["hive-ag3nt", "hive-c0re", "hive-forge", "hive-matrix-mcp", "hive-priv", "hive-sh4re"] [workspace.package] edition = "2024" @@ -25,7 +18,6 @@ must_use_candidate = "allow" anyhow = "1" axum = { version = "0.8", features = ["ws"] } base64 = "0.22" -bcrypt = "0.19" clap = { version = "4", features = ["derive"] } hive-sh4re = { path = "hive-sh4re" } tower-http = { version = "0.6", features = ["fs"] } @@ -53,13 +45,6 @@ tokio = { version = "1", features = [ tokio-stream = { version = "0.1", features = ["sync"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } -reqwest = { version = "0.12", default-features = false, features = [ - "json", - "rustls-tls", -] } -matrix-sdk = { version = "0.14", default-features = false, features = [ - "rustls-tls", - "sqlite", - "markdown", -] } +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } +matrix-sdk = { version = "0.14", default-features = false, features = ["rustls-tls", "sqlite", "markdown"] } futures-util = "0.3" diff --git a/docs/gateway.md b/docs/gateway.md index ebb0cf2f..d8336fce 100644 --- a/docs/gateway.md +++ b/docs/gateway.md @@ -339,55 +339,3 @@ the nginx config get custom error pages. Other gateway routes (forge / matrix / fluffychat) get nginx defaults — extending the custom-error pattern there is a separate follow-up. -## HTTP Basic auth - -`services.hyperhive.gateway.auth.enable = true` gates every request to -the main vhost (`_`) behind HTTP Basic auth. nginx's built-in `auth_basic` -module validates credentials; no extra service or host-side daemon is -required. - -**Setup:** - -```nix -services.hyperhive.gateway.auth = { - enable = true; - htpasswdFile = "/etc/hyperhive/gateway.htpasswd"; - # realm = "hyperhive"; # optional, default shown -}; -``` - -Manage users with `hivectl gateway`: - -```sh -# Add or update a user (prompted for password): -hivectl gateway create-user --file /etc/hyperhive/gateway.htpasswd alice --password-stdin - -# Add with inline password (visible in shell history — avoid for sensitive creds): -hivectl gateway create-user --file /etc/hyperhive/gateway.htpasswd bob --password hunter2 - -# Remove a user: -hivectl gateway delete-user --file /etc/hyperhive/gateway.htpasswd bob - -# List current usernames: -hivectl gateway list-users --file /etc/hyperhive/gateway.htpasswd -``` - -`hivectl gateway create-user` hashes passwords with BCrypt (cost 12) and -writes `$2y$`-prefixed hashes that nginx accepts natively. No external -`htpasswd` binary is required. The file is created on first add if absent; -its parent directory must already exist. - -The file must be readable by the `nginx` user inside the container -(`chmod 0644`). The module bind-mounts the file's parent directory -read-only into the container at `/run/gateway-auth/`; nginx reads -`/run/gateway-auth/`. - -**What is not gated:** per-agent UI routes emitted into `agents.conf` -(served under `/agent//`) inherit no auth from `/` — nginx -applies `auth_basic` per-location. Full per-agent coverage is a -follow-up. - -**Realm:** the `WWW-Authenticate: Basic realm="..."` string browsers -display in the credential dialog. Defaults to `"hyperhive"`. Must not -contain `"` or `$`. - diff --git a/hive-ag3nt/src/bash_runner.rs b/hive-ag3nt/src/bash_runner.rs index b048a705..60fd0179 100644 --- a/hive-ag3nt/src/bash_runner.rs +++ b/hive-ag3nt/src/bash_runner.rs @@ -60,7 +60,10 @@ fn tasks_dir() -> PathBuf { } else { // Pre-split fallback: derive harness/ as a sibling of state/. let state = crate::paths::state_dir(); - state.parent().map(|p| p.join("harness")).unwrap_or(state) + state + .parent() + .map(|p| p.join("harness")) + .unwrap_or(state) }; base.join("bash-tasks") } @@ -133,8 +136,9 @@ impl TaskFile { /// Write a task file atomically (tmp + rename). fn write_task(task: &TaskFile) -> std::io::Result<()> { - let json = serde_json::to_string_pretty(task) - .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + let json = serde_json::to_string_pretty(task).map_err(|e| { + std::io::Error::new(std::io::ErrorKind::InvalidData, e) + })?; let dest = task_json(&task.id); let tmp = dest.with_extension("json.tmp"); std::fs::write(&tmp, json)?; @@ -225,7 +229,7 @@ async fn run_loop(socket: PathBuf) { let claimed: Arc>> = Arc::new(Mutex::new(HashSet::new())); loop { - poll_once(&socket, &claimed).await; + poll_once(&socket, &claimed); tokio::time::sleep(POLL_INTERVAL).await; } } @@ -233,20 +237,14 @@ async fn run_loop(socket: PathBuf) { /// On boot, find any task files in `running` state and flip them to /// `interrupted`, then fire a wake so the agent unblocks. async fn mark_interrupted(socket: &Path) { - let Ok(rd) = std::fs::read_dir(tasks_dir()) else { - return; - }; + let Ok(rd) = std::fs::read_dir(tasks_dir()) else { return }; for entry in rd.flatten() { let path = entry.path(); if path.extension().and_then(|e| e.to_str()) != Some("json") { continue; } - let Some(id) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else { - continue; - }; - let Some(mut task) = read_task(&id) else { - continue; - }; + let Some(id) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else { continue }; + let Some(mut task) = read_task(&id) else { continue }; if task.status != TaskStatus::Running { continue; } @@ -260,18 +258,14 @@ async fn mark_interrupted(socket: &Path) { } } -async fn poll_once(socket: &Path, claimed: &Arc>>) { - let Ok(rd) = std::fs::read_dir(tasks_dir()) else { - return; - }; +fn poll_once(socket: &Path, claimed: &Arc>>) { + let Ok(rd) = std::fs::read_dir(tasks_dir()) else { return }; for entry in rd.flatten() { let path = entry.path(); if path.extension().and_then(|e| e.to_str()) != Some("json") { continue; } - let Some(id) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else { - continue; - }; + let Some(id) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else { continue }; { let guard = claimed.lock().unwrap(); if guard.contains(&id) { @@ -328,11 +322,7 @@ async fn run_task(mut task: TaskFile, socket: &Path) { let stdout_tail = tail_file(&out_path, SUMMARY_BYTES); let stderr_tail = tail_file(&err_path, SUMMARY_BYTES); - task.status = if timed_out { - TaskStatus::TimedOut - } else { - TaskStatus::Done - }; + task.status = if timed_out { TaskStatus::TimedOut } else { TaskStatus::Done }; task.completed_at = Some(crate::serve_common::now_unix()); task.exit_code = exit_code; task.stdout_tail = stdout_tail.clone().filter(|s| !s.is_empty()); @@ -354,12 +344,7 @@ async fn run_task(mut task: TaskFile, socket: &Path) { /// Run `sh -c cmd`, streaming output to files. Returns `(exit_code, timed_out)`. /// On timeout the child process is explicitly killed before returning. -async fn exec_cmd( - cmd: &str, - out_path: &Path, - err_path: &Path, - timeout: Duration, -) -> Result<(i32, bool)> { +async fn exec_cmd(cmd: &str, out_path: &Path, err_path: &Path, timeout: Duration) -> Result<(i32, bool)> { use tokio::process::Command; let mut child = Command::new("sh") .arg("-c") @@ -411,9 +396,7 @@ where let _ = tokio::io::copy(&mut reader, &mut f).await; let _ = f.flush().await; } - Err(e) => { - tracing::warn!(path = %path.display(), error = ?e, "bash_runner: open output file failed") - } + Err(e) => tracing::warn!(path = %path.display(), error = ?e, "bash_runner: open output file failed"), } } @@ -432,7 +415,12 @@ fn tail_file(path: &Path, max_bytes: usize) -> Option { // Wake delivery // --------------------------------------------------------------------------- -async fn send_wake(socket: &Path, id: &str, summary: &str, output: Option<(&str, &str)>) { +async fn send_wake( + socket: &Path, + id: &str, + summary: &str, + output: Option<(&str, &str)>, +) { let mut body = format!("bash task `{id}` finished: {summary}"); if let Some((stdout, stderr)) = output { if !stdout.is_empty() { diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index cf2d3c50..d2ad6ce1 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -18,9 +18,7 @@ use clap::{Parser, Subcommand}; use hive_ag3nt::events::{Bus, LiveEvent, TurnState}; use hive_ag3nt::login::{self, LoginState}; use hive_ag3nt::turn_stats::TurnStats; -use hive_ag3nt::{ - DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, plugins, serve_common, turn, web_ui, -}; +use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, plugins, serve_common, turn, web_ui}; use hive_sh4re::{ AgentRequest, AgentResponse, HelperEvent, ManagerRequest, ManagerResponse, SYSTEM_SENDER, }; @@ -131,9 +129,7 @@ fn log_system_event(bus: &Bus, from: &str, body: &str) { } else { tracing::info!(%from, %body, "system message"); } - bus.emit(LiveEvent::Note { - text: format!("[system] {body}"), - }); + bus.emit(LiveEvent::Note { text: format!("[system] {body}") }); } /// Body string for the turn-failure notification we route to @@ -144,11 +140,7 @@ fn log_system_event(bus: &Bus, from: &str, body: &str) { /// misconfigured harness still produces a parseable line. fn format_turn_failure(err: &anyhow::Error) -> String { let who = hive_ag3nt::identity::qualified_label(); - let who = if who.is_empty() { - "".to_owned() - } else { - who - }; + let who = if who.is_empty() { "".to_owned() } else { who }; format!("[system] `{who}` claude turn failed:\n{err:#}") } @@ -210,7 +202,9 @@ trait Surface { /// `(open_threads, open_reminders)` for the post-turn stats row. /// Either field is `None` when the underlying request errors. - fn post_turn_counts(socket: &Path) -> impl Future, Option)>; + fn post_turn_counts( + socket: &Path, + ) -> impl Future, Option)>; /// Send a message addressed to `` (broker resolves the /// sentinel via `topology::parent_of` at delivery time; root @@ -231,8 +225,11 @@ trait Surface { /// by co-process daemons like matrix to push events into the /// harness inbox). Errors out via `anyhow::bail!` so the calling /// binary surfaces them on stderr. - fn wake_external(socket: &Path, from: String, body: String) - -> impl Future>; + fn wake_external( + socket: &Path, + from: String, + body: String, + ) -> impl Future>; } // ---------- AgentSurface ---------- @@ -274,15 +271,13 @@ impl Surface for AgentSurface { } async fn post_turn_counts(socket: &Path) -> (Option, Option) { - let threads = match client::request::<_, AgentResponse>( - socket, - &AgentRequest::GetLooseEnds { agent: None }, - ) - .await - { - Ok(AgentResponse::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(), - _ => None, - }; + let threads = + match client::request::<_, AgentResponse>(socket, &AgentRequest::GetLooseEnds { agent: None }).await { + Ok(AgentResponse::LooseEnds { loose_ends }) => { + u64::try_from(loose_ends.len()).ok() + } + _ => None, + }; let reminders = match client::request::<_, AgentResponse>( socket, &AgentRequest::CountPendingReminders { agent: None }, @@ -420,7 +415,9 @@ impl Surface for ManagerSurface { ) .await { - Ok(ManagerResponse::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(), + Ok(ManagerResponse::LooseEnds { loose_ends }) => { + u64::try_from(loose_ends.len()).ok() + } _ => None, }; let reminders = match client::request::<_, ManagerResponse>( @@ -572,7 +569,8 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { ); tokio::spawn(async move { let (label, port, login_state, bus, socket, files, turn_lock) = web_ui_args; - if let Err(e) = web_ui::serve(label, port, login_state, bus, socket, files, turn_lock).await + if let Err(e) = + web_ui::serve(label, port, login_state, bus, socket, files, turn_lock).await { tracing::error!(error = %e, "web_ui::serve exited with error"); } @@ -660,11 +658,7 @@ async fn handle_turn( log_system_event(bus, &from, &body); tracing::info!(%from, %body, %redelivered, "inbox"); let unread = S::inbox_unread(socket).await; - bus.emit(LiveEvent::TurnStart { - from: from.clone(), - body: body.clone(), - unread, - }); + bus.emit(LiveEvent::TurnStart { from: from.clone(), body: body.clone(), unread }); bus.set_state(TurnState::Thinking); let started_at = serve_common::now_unix(); let started_instant = std::time::Instant::now(); @@ -676,10 +670,7 @@ async fn handle_turn( }; turn::emit_turn_end(bus, &outcome); bus.set_state(TurnState::Idle); - if matches!( - outcome, - turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted - ) { + if matches!(outcome, turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted) { S::ack_turn(socket).await; } if matches!(outcome, turn::TurnOutcome::RateLimited) { @@ -706,7 +697,8 @@ async fn handle_turn( } if let Some(stats) = stats { let ended_at = serve_common::now_unix(); - let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX); + let duration_ms = + i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX); let (open_threads, open_reminders) = S::post_turn_counts(socket).await; let row = serve_common::build_row( started_at, diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index 700c7fcc..d511ab28 100644 --- a/hive-ag3nt/src/events.rs +++ b/hive-ag3nt/src/events.rs @@ -80,18 +80,12 @@ fn harness_json_path() -> PathBuf { fn read_harness_state() -> (bool, bool) { // Try the new consolidated file first. - if let Ok(raw) = std::fs::read_to_string(harness_json_path()) { - if let Ok(v) = serde_json::from_str::(&raw) { - let rate_limited = v - .get("rate_limited") - .and_then(|x| x.as_bool()) - .unwrap_or(false); - let needs_login = v - .get("needs_login") - .and_then(|x| x.as_bool()) - .unwrap_or(false); - return (rate_limited, needs_login); - } + if let Ok(raw) = std::fs::read_to_string(harness_json_path()) + && let Ok(v) = serde_json::from_str::(&raw) + { + let rate_limited = v.get("rate_limited").and_then(serde_json::Value::as_bool).unwrap_or(false); + let needs_login = v.get("needs_login").and_then(serde_json::Value::as_bool).unwrap_or(false); + return (rate_limited, needs_login); } // Fall back to legacy sentinel files written by older harness builds. let state_dir = crate::paths::state_dir(); diff --git a/hive-ag3nt/src/forge_notify.rs b/hive-ag3nt/src/forge_notify.rs index 829e3584..0784aaea 100644 --- a/hive-ag3nt/src/forge_notify.rs +++ b/hive-ag3nt/src/forge_notify.rs @@ -244,7 +244,10 @@ fn is_username_byte(b: u8) -> bool { /// window so addressed agents never silently miss a mention on a long /// body. See `docs/forge.md::Body excerpt + truncation + heading /// escape` for the truncate-before-escape ordering rule. -fn extract_truncated_mention_lines<'a>(full_body: &'a str, included_excerpt: &str) -> Vec<&'a str> { +fn extract_truncated_mention_lines<'a>( + full_body: &'a str, + included_excerpt: &str, +) -> Vec<&'a str> { full_body .lines() .filter(|line| { @@ -911,10 +914,7 @@ mod tests { let full = "# @argus check this\nmore body\n"; let raw_excerpt = full; // fits entirely let lines = extract_truncated_mention_lines(full, raw_excerpt); - assert!( - lines.is_empty(), - "heading+mention inside window must not be re-surfaced, got {lines:?}" - ); + assert!(lines.is_empty(), "heading+mention inside window must not be re-surfaced, got {lines:?}"); } #[test] diff --git a/hive-ag3nt/src/identity.rs b/hive-ag3nt/src/identity.rs index 56759aee..14acb7f2 100644 --- a/hive-ag3nt/src/identity.rs +++ b/hive-ag3nt/src/identity.rs @@ -122,9 +122,7 @@ mod tests { swarm_name: Option<&str>, f: F, ) { - let _guard = ENV_LOCK - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let _guard = ENV_LOCK.lock().unwrap_or_else(std::sync::PoisonError::into_inner); let prev_label = env::var("HIVE_LABEL").ok(); let prev_domain = env::var("HYPERHIVE_HIVE_DOMAIN").ok(); let prev_hive_name = env::var("HYPERHIVE_HIVE_NAME").ok(); diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 4d79930a..f9b26d58 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -437,30 +437,31 @@ fn format_bash_status(id: &str) -> String { let Some(task) = crate::bash_runner::read_task(id) else { return format!("bash_status: unknown task id `{id}`"); }; - let mut out = format!("task `{id}`: status={status:?}", status = task.status); + let mut out = format!( + "task `{id}`: status={status:?}", + status = task.status + ); if let Some(code) = task.exit_code { let _ = write!(out, ", exit={code}"); } - if let Some(t) = task.started_at - && task.completed_at.is_none() - { + if let Some(t) = task.started_at && task.completed_at.is_none() { let age = crate::serve_common::now_unix() - t; let _ = write!(out, ", running for {age}s"); } - if let Some(t) = task.completed_at { - if let Some(s) = task.started_at { - let _ = write!(out, ", took {}s", t - s); - } + if let Some(t) = task.completed_at + && let Some(s) = task.started_at + { + let _ = write!(out, ", took {}s", t - s); } - if let Some(ref stdout) = task.stdout_tail { - if !stdout.trim().is_empty() { - let _ = write!(out, "\n\nstdout:\n```\n{}\n```", stdout.trim()); - } + if let Some(ref stdout) = task.stdout_tail + && !stdout.trim().is_empty() + { + let _ = write!(out, "\n\nstdout:\n```\n{}\n```", stdout.trim()); } - if let Some(ref stderr) = task.stderr_tail { - if !stderr.trim().is_empty() { - let _ = write!(out, "\n\nstderr:\n```\n{}\n```", stderr.trim()); - } + if let Some(ref stderr) = task.stderr_tail + && !stderr.trim().is_empty() + { + let _ = write!(out, "\n\nstderr:\n```\n{}\n```", stderr.trim()); } out } @@ -667,9 +668,7 @@ impl AgentServer { )] async fn get_loose_ends(&self, Parameters(args): Parameters) -> String { run_tool_envelope("get_loose_ends", String::new(), async move { - let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::GetLooseEnds { agent: args.agent }) - .await; + let (resp, retries) = self.dispatch(hive_sh4re::AgentRequest::GetLooseEnds { agent: args.agent }).await; let mut out = annotate_retries(format_loose_ends(resp), retries); // Append any local bash tasks still in pending/running state so // the agent sees all outstanding work in one call. @@ -679,11 +678,8 @@ impl AgentServer { let _ = write!(out, "\n\n{} active bash task(s):", active.len()); for task in &active { let age = crate::serve_common::now_unix() - task.created_at; - let _ = write!( - out, - "\n- `{}` status={:?}, cmd: `{}`, age {}s", - task.id, task.status, task.cmd, age - ); + let _ = write!(out, "\n- `{}` status={:?}, cmd: `{}`, age {}s", + task.id, task.status, task.cmd, age); } } out @@ -834,11 +830,9 @@ impl AgentServer { )] async fn bash_status(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); - run_tool_envelope( - "bash_status", - log, - async move { format_bash_status(&args.id) }, - ) + run_tool_envelope("bash_status", log, async move { + format_bash_status(&args.id) + }) .await } @@ -881,7 +875,10 @@ impl AgentServer { `since`: show entries on or newer than this (e.g. `-1h`, `2024-01-01 12:00:00`). \ `until`: show entries on or older than this." )] - async fn get_host_journal(&self, Parameters(args): Parameters) -> String { + async fn get_host_journal( + &self, + Parameters(args): Parameters, + ) -> String { let log = format!("{args:?}"); run_tool_envelope("get_host_journal", log, async move { let (resp, retries) = self @@ -1916,14 +1913,14 @@ pub enum Flavor { } /// Env var written by the meta renderer with a comma-separated list of -/// `hive_sh4re::ToolGroup` snake_case names (e.g. `"messaging,inbox,meta"`). +/// `hive_sh4re::ToolGroup` `snake_case` names (e.g. `"messaging,inbox,meta"`). /// When present, the harness expands the groups into per-tool allow entries /// instead of using the hardcoded flavor default. See `docs/conventions.md::Tool groups`. const TOOL_GROUPS_ENV: &str = "HIVE_TOOL_GROUPS"; /// `HIVE_CAPABILITIES` env var injected by `meta::render_flake` when the /// operator grants capabilities to this agent. Comma-separated -/// `hive_sh4re::Capability` snake_case names. Absent = no extra capabilities. +/// `hive_sh4re::Capability` `snake_case` names. Absent = no extra capabilities. const CAPABILITIES_ENV: &str = "HIVE_CAPABILITIES"; /// Returns the MCP tool names (without `mcp__hyperhive__` prefix) that are @@ -1979,13 +1976,12 @@ fn effective_tool_groups(flavor: Flavor) -> Vec { for token in raw.split(',') { let t = token.trim().to_ascii_lowercase(); // Parse via serde_json (the canonical deserialization path). - match serde_json::from_value::(serde_json::Value::String(t.clone())) - { - Ok(g) => groups.push(g), - Err(_) => tracing::warn!( - token = %t, - "{TOOL_GROUPS_ENV}: unknown tool group, skipping" - ), + if let Ok(g) = serde_json::from_value::( + serde_json::Value::String(t.clone()), + ) { + groups.push(g); + } else { + tracing::warn!(token = %t, "{TOOL_GROUPS_ENV}: unknown tool group, skipping"); } } if groups.is_empty() { diff --git a/hive-ag3nt/src/stats.rs b/hive-ag3nt/src/stats.rs index 301d1c50..ae98ca7e 100644 --- a/hive-ag3nt/src/stats.rs +++ b/hive-ag3nt/src/stats.rs @@ -391,11 +391,7 @@ fn summarize_durations(all: &mut [i64]) -> DurationSummary { } } -#[allow( - clippy::cast_precision_loss, - clippy::cast_possible_truncation, - clippy::cast_sign_loss -)] +#[allow(clippy::cast_precision_loss, clippy::cast_possible_truncation, clippy::cast_sign_loss)] fn percentile(sorted: &[i64], pct: u8) -> f64 { if sorted.is_empty() { return 0.0; @@ -475,15 +471,7 @@ mod tests { (started_at, ended_at, duration_ms, model, wake_from, last_input_tokens, tool_call_breakdown_json, result_kind) VALUES (?1, ?2, ?3, ?4, ?5, 1000, ?6, ?7)", - params![ - started, - started + dur / 1000, - dur, - model, - wake, - tools_json, - result - ], + params![started, started + dur / 1000, dur, model, wake, tools_json, result], ) .unwrap(); } @@ -497,14 +485,7 @@ mod tests { seed_db( &db, &[ - ( - now - 600, - 5_000, - "opus", - "recv", - "ok", - r#"{"Read":2,"Bash":1}"#, - ), + (now - 600, 5_000, "opus", "recv", "ok", r#"{"Read":2,"Bash":1}"#), (now - 300, 10_000, "opus", "recv", "ok", r#"{"Read":3}"#), (now - 100, 20_000, "sonnet", "operator", "failed", "{}"), ], diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs index 6e588f29..a513ab5a 100644 --- a/hive-ag3nt/src/web_ui.rs +++ b/hive-ag3nt/src/web_ui.rs @@ -561,9 +561,7 @@ async fn api_state(State(state): State) -> axum::Json { ctx_usage, cost_usage, links: agent_links(&state.label, state.gui_vnc_port.is_some()), - forge_public_url: std::env::var("HIVE_FORGE_PUBLIC_URL") - .ok() - .filter(|s| !s.is_empty()), + forge_public_url: std::env::var("HIVE_FORGE_PUBLIC_URL").ok().filter(|s| !s.is_empty()), hive_name: crate::identity::hive_name(), swarm_name: crate::identity::swarm_name(), }) @@ -615,7 +613,8 @@ fn agent_links(label: &str, gui_enabled: bool) -> Vec { // `{state_dir}/hyperhive-dashboard-links.json`). Shape on disk // is `{label, icon, url}` with absolute URLs — those become // `kind = External` links, passed through verbatim. - let extras_path = crate::paths::state_dir().join("hyperhive-dashboard-links.json"); + let extras_path = + crate::paths::state_dir().join("hyperhive-dashboard-links.json"); if let Ok(text) = std::fs::read_to_string(&extras_path) && !text.trim().is_empty() && let Ok(extras) = serde_json::from_str::>(&text) @@ -663,10 +662,7 @@ async fn recent_inbox(socket: &std::path::Path) -> Vec { /// Fetch reminder activity stats from the broker via the per-agent / /// manager socket. Returns None on any transport / decode failure — the /// stats are decorative, not authoritative. -async fn fetch_reminder_stats( - socket: &std::path::Path, - window_secs: u64, -) -> Option { +async fn fetch_reminder_stats(socket: &std::path::Path, window_secs: u64) -> Option { match client::request::<_, hive_sh4re::Response>( socket, &hive_sh4re::Request::ReminderRollup { @@ -960,9 +956,7 @@ async fn post_cancel_turn(State(state): State) -> Response { ), Err(e) => format!("operator: /cancel — pkill failed: {e}"), }; - state - .bus - .emit(crate::events::LiveEvent::Note { text: note }); + state.bus.emit(crate::events::LiveEvent::Note { text: note }); (axum::http::StatusCode::OK, "ok").into_response() } diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 3446e8f2..ac70638e 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -9,10 +9,10 @@ workspace = true anyhow.workspace = true axum.workspace = true base64.workspace = true -bcrypt.workspace = true reqwest.workspace = true clap.workspace = true hive-sh4re.workspace = true +libc = "0.2" listenfd = "1" rusqlite.workspace = true serde.workspace = true diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 1905ac0b..dfdb298c 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -76,7 +76,8 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { // Pre-enqueue cascade rebuilds in topological order so // agents depending on updated inputs are rebuilt after the // lock bump, matching the dashboard post_meta_update path. - let cascade_agents = crate::rebuild_queue::meta_update_cascade_agents(&inputs).await; + let cascade_agents = + crate::rebuild_queue::meta_update_cascade_agents(&inputs).await; let cascade_reason = format!("approval #{id} meta input cascade"); for name in cascade_agents { coord.rebuild_queue.enqueue( diff --git a/hive-c0re/src/agent_ports.rs b/hive-c0re/src/agent_ports.rs index de1032c4..f86ae25f 100644 --- a/hive-c0re/src/agent_ports.rs +++ b/hive-c0re/src/agent_ports.rs @@ -40,7 +40,8 @@ fn render(map: &BTreeMap) -> String { // BTreeMap → serde_json::to_string_pretty preserves key order, // so the output is deterministic across calls with the same // agent set. - serde_json::to_string_pretty(map).expect("BTreeMap is always serialisable") + serde_json::to_string_pretty(map) + .expect("BTreeMap is always serialisable") } /// Atomically write the JSON for `names` to @@ -60,10 +61,12 @@ pub fn write(names: &[String]) -> Result<()> { return Ok(()); } if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?; + std::fs::create_dir_all(parent) + .with_context(|| format!("create {}", parent.display()))?; } let tmp = path.with_extension("json.tmp"); - std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?; + std::fs::write(&tmp, &body) + .with_context(|| format!("write {}", tmp.display()))?; std::fs::rename(&tmp, &path).with_context(|| { format!( "rename {} -> {} (atomic publish)", @@ -129,6 +132,9 @@ mod tests { // BTreeMap sorts → alpha before zeta in output. let alpha_pos = body.find("alpha").expect("alpha in output"); let zeta_pos = body.find("zeta").expect("zeta in output"); - assert!(alpha_pos < zeta_pos, "sorted order broken:\n{body}"); + assert!( + alpha_pos < zeta_pos, + "sorted order broken:\n{body}" + ); } } diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index 92b7f6cf..c220f020 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -129,11 +129,9 @@ pub(crate) async fn dispatch_shared( ) -> Option { let broker = &coord.broker; Some(match req { - hive_sh4re::Request::Send { - to, - body, - in_reply_to, - } => handle_send(coord, agent, to, body, *in_reply_to), + hive_sh4re::Request::Send { to, body, in_reply_to } => { + handle_send(coord, agent, to, body, *in_reply_to) + } hive_sh4re::Request::Recv { wait_seconds, max } => { let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize; match broker @@ -225,8 +223,8 @@ pub(crate) async fn dispatch_shared( if let Err(message) = crate::limits::check_status_text(text) { return Some(hive_sh4re::Response::Err { message }); } - let path = - crate::coordinator::Coordinator::agent_notes_dir(agent).join("hyperhive-status"); + let path = crate::coordinator::Coordinator::agent_notes_dir(agent) + .join("hyperhive-status"); let result = if text.trim().is_empty() { std::fs::remove_file(&path).or_else(|e| { if e.kind() == std::io::ErrorKind::NotFound { @@ -244,9 +242,11 @@ pub(crate) async fn dispatch_shared( tokio::spawn(async move { coord2.rescan_containers_and_emit().await }); hive_sh4re::Response::Ok } - Err(e) => hive_sh4re::Response::Err { - message: format!("set_status write failed: {e}"), - }, + Err(e) => { + hive_sh4re::Response::Err { + message: format!("set_status write failed: {e}"), + } + } } } hive_sh4re::Request::GetAgentMeta { name } => { @@ -294,15 +294,7 @@ pub(crate) async fn dispatch_shared( message: format!("{e:#}"), }, }, - hive_sh4re::Request::GetHostJournal { - unit, - container, - lines, - priority, - grep, - since, - until, - } => { + hive_sh4re::Request::GetHostJournal { unit, container, lines, priority, grep, since, until } => { dispatch_host_journal(agent, unit, container, lines, priority, grep, since, until).await } // Not a shared variant. @@ -339,10 +331,7 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> Err(message) => AgentResponse::Err { message }, } } - AgentRequest::ReminderRollup { - since_secs, - agent: target, - } => { + AgentRequest::ReminderRollup { since_secs, agent: target } => { let name = resolve_agent_state_target(agent, target.as_deref()); match name { Ok(name) => match coord.broker.reminder_rollup_for(name, *since_secs) { @@ -558,7 +547,10 @@ pub(crate) fn store_remind( ) -> Result<(), String> { let max = remind_max_pending(); if max > 0 { - let pending = coord.broker.count_pending_reminders_for(agent).unwrap_or(0); + let pending = coord + .broker + .count_pending_reminders_for(agent) + .unwrap_or(0); if pending >= max { return Err(format!( "reminder rejected: agent `{agent}` already has {pending} pending \ @@ -612,9 +604,8 @@ fn prepare_remind_storage( }; let host_path = crate::reminder_scheduler::resolve_host_path(agent, &req_path) .map_err(|reason| format!("auto-save path `{req_path}` rejected: {reason}"))?; - crate::reminder_scheduler::write_payload(agent, &host_path, message).map_err(|reason| { - format!("auto-save of large reminder body to `{req_path}` failed: {reason}") - })?; + crate::reminder_scheduler::write_payload(agent, &host_path, message) + .map_err(|reason| format!("auto-save of large reminder body to `{req_path}` failed: {reason}"))?; let hint = format!( "[reminder body of {} bytes auto-saved to `{req_path}`; read with your filesystem tools]", message.len() @@ -643,26 +634,19 @@ fn auto_reminder_path(agent: &str) -> String { /// - `Some("")` where other is not a child → requires the /// `query_agent_state` capability; returns an error otherwise. /// - `Some("*")` → always rejected (hive-wide scans are manager-only). -fn resolve_agent_state_target<'a>( - caller: &'a str, - target: Option<&'a str>, -) -> Result<&'a str, String> { +fn resolve_agent_state_target<'a>(caller: &'a str, target: Option<&'a str>) -> Result<&'a str, String> { match target { None => Ok(caller), Some("*") => Err( "hive-wide query (agent=\"*\") is not available on the agent socket; \ - use the manager socket for swarm-wide scans" - .to_owned(), + use the manager socket for swarm-wide scans".to_owned() ), Some(name) => { if name == caller { return Ok(caller); } // Direct children are visible to their parent without extra capability. - if crate::topology::children_of(caller) - .iter() - .any(|c| c == name) - { + if crate::topology::children_of(caller).iter().any(|c| c == name) { return Ok(name); } if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) { diff --git a/hive-c0re/src/agent_sockets.rs b/hive-c0re/src/agent_sockets.rs index a18a1bbd..d7edbfd6 100644 --- a/hive-c0re/src/agent_sockets.rs +++ b/hive-c0re/src/agent_sockets.rs @@ -81,7 +81,8 @@ pub fn socket_path_for(name: &str) -> PathBuf { #[must_use] pub fn build_map(names: &[String]) -> BTreeMap { build_map_with(names, |name| { - ready_marker_for(name).exists() || agent_dir_for(name).join(READY_MARKER_LEGACY).exists() + ready_marker_for(name).exists() + || agent_dir_for(name).join(READY_MARKER_LEGACY).exists() }) } @@ -145,10 +146,12 @@ pub fn write(names: &[String]) -> Result<()> { return Ok(()); } if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?; + std::fs::create_dir_all(parent) + .with_context(|| format!("create {}", parent.display()))?; } let tmp = path.with_extension("json.tmp"); - std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?; + std::fs::write(&tmp, &body) + .with_context(|| format!("write {}", tmp.display()))?; std::fs::rename(&tmp, &path).with_context(|| { format!( "rename {} -> {} (atomic publish)", @@ -302,30 +305,24 @@ mod tests { let marker = ready_marker_for("iris"); let socket = socket_path_for("iris"); assert_eq!(marker.parent(), socket.parent()); - assert_eq!( - marker, - Path::new("/run/hive-agent/iris/hyperhive-socket-bound") - ); + assert_eq!(marker, Path::new("/run/hive-agent/iris/hyperhive-socket-bound")); } #[test] fn render_is_pretty_and_sorted() { let mut map = BTreeMap::new(); - map.insert( - "zeta".to_owned(), - PathBuf::from("/run/hive-agent/zeta/web.sock"), - ); - map.insert( - "alpha".to_owned(), - PathBuf::from("/run/hive-agent/alpha/web.sock"), - ); + map.insert("zeta".to_owned(), PathBuf::from("/run/hive-agent/zeta/web.sock")); + map.insert("alpha".to_owned(), PathBuf::from("/run/hive-agent/alpha/web.sock")); let body = render(&map); // Pretty-print = newlines between keys + indentation. assert!(body.contains('\n')); // BTreeMap sorts → alpha before zeta in output. let alpha_pos = body.find("alpha").expect("alpha in output"); let zeta_pos = body.find("zeta").expect("zeta in output"); - assert!(alpha_pos < zeta_pos, "sorted order broken:\n{body}"); + assert!( + alpha_pos < zeta_pos, + "sorted order broken:\n{body}" + ); } #[test] @@ -335,12 +332,10 @@ mod tests { // gateway-side reader can deserialise into String values // without nested struct logic. let mut map = BTreeMap::new(); - map.insert( - "iris".to_owned(), - PathBuf::from("/run/hive-agent/iris/web.sock"), - ); + map.insert("iris".to_owned(), PathBuf::from("/run/hive-agent/iris/web.sock")); let body = render(&map); assert!(body.contains("\"iris\"")); assert!(body.contains("\"/run/hive-agent/iris/web.sock\"")); } } + diff --git a/hive-c0re/src/auto_update.rs b/hive-c0re/src/auto_update.rs index 856b1caf..1ff3d67b 100644 --- a/hive-c0re/src/auto_update.rs +++ b/hive-c0re/src/auto_update.rs @@ -209,19 +209,16 @@ pub async fn ensure_manager(coord: &Arc) -> Result<()> { /// Sort `names` in-place so parents precede their children in the topology. /// Uses BFS from root agents (depth 0). Agents absent from `topo` sort last, /// alphabetically within their tier. Stable within each depth tier. -pub fn topology_sort( - names: &mut Vec, - topo: &std::collections::BTreeMap>, -) { +pub fn topology_sort(names: &mut [String], topo: &std::collections::BTreeMap>) { use std::collections::{HashMap, VecDeque}; // Build depth map using owned clones so the borrow on `names` is released // before the sort_by mutable borrow. - let name_set: Vec = names.clone(); + let name_set: Vec = names.to_vec(); let mut depth: HashMap = HashMap::new(); let mut queue: VecDeque = VecDeque::new(); // Seed roots: entries with no parent, or names not present in topo at all. for name in &name_set { - if topo.get(name).map_or(true, |p| p.is_none()) { + if topo.get(name).is_none_or(Option::is_none) { depth.insert(name.clone(), 0); queue.push_back(name.clone()); } @@ -303,3 +300,4 @@ pub async fn run(coord: Arc) -> Result<()> { coord.emit_rebuild_queue_snapshot(); Ok(()) } + diff --git a/hive-c0re/src/bash_tasks_vacuum.rs b/hive-c0re/src/bash_tasks_vacuum.rs index 8bdb3a00..402dafb7 100644 --- a/hive-c0re/src/bash_tasks_vacuum.rs +++ b/hive-c0re/src/bash_tasks_vacuum.rs @@ -23,7 +23,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use crate::coordinator::Coordinator; -const VACUUM_INTERVAL: Duration = Duration::from_secs(3600); +const VACUUM_INTERVAL: Duration = Duration::from_hours(1); /// Keep completed task files for 48 hours before sweeping them. const KEEP_SECS: i64 = 48 * 3600; @@ -65,9 +65,7 @@ fn sweep_once() { /// files removed (each represents one task; `.out`/`.err` deletions /// are not counted separately). fn vacuum_dir(dir: &Path, cutoff: i64) -> u64 { - let Ok(rd) = std::fs::read_dir(dir) else { - return 0; - }; + let Ok(rd) = std::fs::read_dir(dir) else { return 0 }; let mut removed: u64 = 0; for entry in rd.flatten() { let path = entry.path(); @@ -99,10 +97,7 @@ fn should_delete(json_path: &Path, cutoff: i64) -> bool { if !TERMINAL_STATUSES.contains(&status) { return false; } - let completed_at = v - .get("completed_at") - .and_then(|t| t.as_i64()) - .unwrap_or(i64::MAX); + let completed_at = v.get("completed_at").and_then(serde_json::Value::as_i64).unwrap_or(i64::MAX); completed_at < cutoff } @@ -111,10 +106,10 @@ fn should_delete(json_path: &Path, cutoff: i64) -> bool { fn delete_trio(dir: &Path, stem: &str) { for ext in ["json", "out", "err"] { let path = dir.join(format!("{stem}.{ext}")); - if path.exists() { - if let Err(e) = std::fs::remove_file(&path) { - tracing::warn!(path = %path.display(), error = ?e, "bash-tasks vacuum: remove failed"); - } + if path.exists() + && let Err(e) = std::fs::remove_file(&path) + { + tracing::warn!(path = %path.display(), error = ?e, "bash-tasks vacuum: remove failed"); } } } diff --git a/hive-c0re/src/bin/hivectl.rs b/hive-c0re/src/bin/hivectl.rs index f5734b8b..f7632820 100644 --- a/hive-c0re/src/bin/hivectl.rs +++ b/hive-c0re/src/bin/hivectl.rs @@ -15,8 +15,6 @@ //! dirs) and reuse the `forge` / `matrix` modules from the //! `hive-c0re` lib — single source of truth, no duplication. -use std::path::{Path, PathBuf}; - use anyhow::{Context as _, Result, bail}; use clap::{Parser, Subcommand}; use hive_c0re::coordinator::Coordinator; @@ -56,14 +54,6 @@ enum Cmd { #[command(subcommand)] cmd: MatrixCmd, }, - /// Gateway htpasswd user management. Add, remove, or list users in - /// an htpasswd file used by the gateway's HTTP Basic auth - /// (`services.hyperhive.gateway.auth`). Credentials are stored as - /// BCrypt hashes — no extra service or PAM required. - Gateway { - #[command(subcommand)] - cmd: GatewayCmd, - }, } #[derive(Subcommand)] @@ -147,48 +137,6 @@ enum MatrixCmd { }, } -#[derive(Subcommand)] -enum GatewayCmd { - /// Add a new user or update the password of an existing user in an - /// htpasswd file. The password is hashed with BCrypt (cost 12). - /// - /// Pass `--password-stdin` when scripting or when you don't want the - /// password visible in shell history. The file is created if it does - /// not exist; its parent directory must already exist. - CreateUser { - /// Path to the htpasswd file (the value of - /// `services.hyperhive.gateway.auth.htpasswdFile`). - #[arg(long, short = 'f')] - file: PathBuf, - /// Username to add or update. - username: String, - /// Set the password inline. WARNING: visible in shell history and - /// process listings — prefer `--password-stdin` for sensitive input. - /// Mutually exclusive with `--password-stdin`. - #[arg(long, conflicts_with = "password_stdin")] - password: Option, - /// Read the password from stdin (single line, trailing newline - /// stripped). Mutually exclusive with `--password`. - #[arg(long)] - password_stdin: bool, - }, - /// Remove a user from an htpasswd file. Exits with an error when the - /// user is not found so callers can detect the no-op case. - DeleteUser { - /// Path to the htpasswd file. - #[arg(long, short = 'f')] - file: PathBuf, - /// Username to remove. - username: String, - }, - /// List all usernames in an htpasswd file, one per line. - ListUsers { - /// Path to the htpasswd file. - #[arg(long, short = 'f')] - file: PathBuf, - }, -} - #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() @@ -213,16 +161,6 @@ async fn main() -> Result<()> { password_stdin, } => matrix_create_user(&name, password.as_deref(), password_stdin).await, }, - Cmd::Gateway { cmd } => match cmd { - GatewayCmd::CreateUser { - file, - username, - password, - password_stdin, - } => gateway_create_user(&file, &username, password.as_deref(), password_stdin), - GatewayCmd::DeleteUser { file, username } => gateway_delete_user(&file, &username), - GatewayCmd::ListUsers { file } => gateway_list_users(&file), - }, } } @@ -235,7 +173,11 @@ fn is_agent(name: &str) -> bool { Coordinator::agent_state_root(name).exists() } -async fn forge_create_user(name: &str, password: Option<&str>, password_stdin: bool) -> Result<()> { +async fn forge_create_user( + name: &str, + password: Option<&str>, + password_stdin: bool, +) -> Result<()> { if !hive_c0re::forge::is_present().await { bail!( "hive-forge container not running — start it (services.hyperhive.forge.enable = true) before provisioning forge users" @@ -332,18 +274,13 @@ async fn matrix_create_user( } else { let effective_password = match user_password { Some(p) => p, - None => { - hive_c0re::matrix::random_password().context("generate random matrix password")? - } + None => hive_c0re::matrix::random_password() + .context("generate random matrix password")?, }; - let token = hive_c0re::matrix::provision_user_token( - &client, - name, - ®ister_token, - &effective_password, - ) - .await - .with_context(|| format!("matrix create-user {name}"))?; + let token = + hive_c0re::matrix::provision_user_token(&client, name, ®ister_token, &effective_password) + .await + .with_context(|| format!("matrix create-user {name}"))?; println!("matrix: provisioned user '{name}' (not an agent — token not persisted)"); println!("token: {token}"); if password.is_some() || password_stdin { @@ -356,115 +293,3 @@ async fn matrix_create_user( } Ok(()) } - -// --------------------------------------------------------------------------- -// Gateway htpasswd helpers -// --------------------------------------------------------------------------- - -/// Read an htpasswd file into a list of lines, or return an empty list -/// if the file does not exist yet. -fn htpasswd_read(path: &Path) -> Result> { - if !path.exists() { - return Ok(vec![]); - } - let content = std::fs::read_to_string(path) - .with_context(|| format!("read htpasswd file {}", path.display()))?; - Ok(content.lines().map(str::to_owned).collect()) -} - -/// Write lines back to `path` atomically (write to `.tmp`, then -/// rename). A trailing newline is always appended to the last line. -fn htpasswd_write(path: &Path, lines: &[String]) -> Result<()> { - let tmp = path.with_extension("htpasswd.tmp"); - let content = if lines.is_empty() { - String::new() - } else { - let mut s = lines.join("\n"); - s.push('\n'); - s - }; - std::fs::write(&tmp, &content) - .with_context(|| format!("write htpasswd tmp {}", tmp.display()))?; - std::fs::rename(&tmp, path) - .with_context(|| format!("rename {} → {}", tmp.display(), path.display()))?; - Ok(()) -} - -/// Add or update `username` in the htpasswd file at `file`, hashing -/// `password` with BCrypt (cost 12). Creates the file when absent. -fn gateway_create_user( - file: &Path, - username: &str, - password: Option<&str>, - password_stdin: bool, -) -> Result<()> { - let pw = resolve_password(password, password_stdin)?.ok_or_else(|| { - anyhow::anyhow!("a password is required — pass --password or --password-stdin") - })?; - validate_htpasswd_username(username)?; - let raw_hash = bcrypt::hash(&pw, 12).context("bcrypt hash")?; - // nginx auth_basic only recognises $2a$/$2x$/$2y$ — not $2b$. The two - // prefixes are algorithmically identical; remap so nginx accepts the hash. - let hash = raw_hash.replacen("$2b$", "$2y$", 1); - let entry = format!("{username}:{hash}"); - let mut lines = htpasswd_read(file)?; - let prefix = format!("{username}:"); - if let Some(pos) = lines.iter().position(|l| l.starts_with(&prefix)) { - lines[pos] = entry; - htpasswd_write(file, &lines)?; - println!( - "gateway: updated password for '{username}' in {}", - file.display() - ); - } else { - lines.push(entry); - htpasswd_write(file, &lines)?; - println!("gateway: added user '{username}' to {}", file.display()); - } - Ok(()) -} - -/// Remove `username` from the htpasswd file. Errors when the user is -/// not present so callers can detect the no-op case. -fn gateway_delete_user(file: &Path, username: &str) -> Result<()> { - let mut lines = htpasswd_read(file)?; - let prefix = format!("{username}:"); - let before = lines.len(); - lines.retain(|l| !l.starts_with(&prefix)); - if lines.len() == before { - bail!("gateway: user '{username}' not found in {}", file.display()); - } - htpasswd_write(file, &lines)?; - println!("gateway: removed user '{username}' from {}", file.display()); - Ok(()) -} - -/// Print one username per line from the htpasswd file. -fn gateway_list_users(file: &Path) -> Result<()> { - let lines = htpasswd_read(file)?; - for line in &lines { - // Skip blank lines and comments. - if line.is_empty() || line.starts_with('#') { - continue; - } - if let Some((name, _)) = line.split_once(':') { - println!("{name}"); - } - } - Ok(()) -} - -/// Reject usernames containing `:` (field separator) or control chars -/// that would corrupt the htpasswd file format. -fn validate_htpasswd_username(username: &str) -> Result<()> { - if username.is_empty() { - bail!("username must not be empty"); - } - if username.contains(':') { - bail!("username must not contain ':' (htpasswd field separator)"); - } - if username.chars().any(|c| c.is_control()) { - bail!("username must not contain control characters"); - } - Ok(()) -} diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index 60f3dc1a..e28a698e 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -268,7 +268,12 @@ impl Broker { /// broker. Used by the scheduler to skip re-delivery of the same /// scheduled prompt without blocking distinct schedules whose /// bodies differ. - pub fn has_pending_with_body(&self, recipient: &str, sender: &str, body: &str) -> Result { + pub fn has_pending_with_body( + &self, + recipient: &str, + sender: &str, + body: &str, + ) -> Result { let conn = self.conn.lock().unwrap(); let n: i64 = conn.query_row( "SELECT COUNT(*) FROM messages @@ -386,13 +391,7 @@ impl Broker { )?; let rows: Vec<(i64, String, String, String, Option)> = stmt .query_map(params![recipient, max_i], |row| { - Ok(( - row.get(0)?, - row.get(1)?, - row.get(2)?, - row.get(3)?, - row.get(4)?, - )) + Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?)) })? .collect::>()?; drop(stmt); @@ -674,11 +673,7 @@ impl Broker { /// Reminder rollup stats for an agent over a time window. Returns /// counts of scheduled, delivered, and pending reminders created /// in the last `since_secs` seconds (0 = all reminders). - pub fn reminder_rollup_for( - &self, - agent: &str, - since_secs: u64, - ) -> Result { + pub fn reminder_rollup_for(&self, agent: &str, since_secs: u64) -> Result { let conn = self.conn.lock().unwrap(); let cutoff_time = if since_secs > 0 { let now = std::time::SystemTime::now() @@ -745,7 +740,9 @@ impl Broker { || canceller == hive_sh4re::OPERATOR_RECIPIENT || canceller == hive_sh4re::MANAGER_AGENT; if !authorised { - anyhow::bail!("reminder {id}: '{canceller}' not allowed to cancel (owner = '{owner}')"); + anyhow::bail!( + "reminder {id}: '{canceller}' not allowed to cancel (owner = '{owner}')" + ); } let n = conn.execute( "DELETE FROM reminders WHERE id = ?1 AND sent_at IS NULL", @@ -865,9 +862,7 @@ impl Broker { } drop(conn); // Emit per-row Sent events (only for rows that succeeded). - for (((id, agent, body), result), msg_id) in - items.iter().zip(results.iter()).zip(msg_ids.iter()) - { + for (((id, agent, body), result), msg_id) in items.iter().zip(results.iter()).zip(msg_ids.iter()) { if result.is_ok() { let _ = self.events.send(MessageEvent::Sent { id: *msg_id, @@ -1034,7 +1029,10 @@ mod tests { assert_eq!(broker.requeue_inflight("b").unwrap(), 1); let d2 = pop_one(broker, "b").expect("popped again"); assert_eq!(d2.message.body, "hi"); - assert!(d2.redelivered, "second pop should be tagged redelivered"); + assert!( + d2.redelivered, + "second pop should be tagged redelivered" + ); assert_eq!(broker.ack_turn("b").unwrap(), 1); } @@ -1277,3 +1275,4 @@ mod tests { assert!(pop_one(broker, "bob").is_none()); } } + diff --git a/hive-c0re/src/build_logs.rs b/hive-c0re/src/build_logs.rs index 5b3cdf31..f518be59 100644 --- a/hive-c0re/src/build_logs.rs +++ b/hive-c0re/src/build_logs.rs @@ -274,8 +274,14 @@ impl BuildLogs { match row { None => Ok(None), Some((stdout, stderr, finished_at, status)) => { - let stdout_append = stdout.get(stdout_cursor..).unwrap_or("").to_string(); - let stderr_append = stderr.get(stderr_cursor..).unwrap_or("").to_string(); + let stdout_append = stdout + .get(stdout_cursor..) + .unwrap_or("") + .to_string(); + let stderr_append = stderr + .get(stderr_cursor..) + .unwrap_or("") + .to_string(); Ok(Some(BuildLogProgress { stdout_append, stderr_append, @@ -290,7 +296,11 @@ impl BuildLogs { /// Headers only (no stdout/stderr blobs) — keeps `/api/state` /// payloads light. Limit is hard-clamped to 50 to bound worst-case /// payload regardless of caller input. - pub fn list_recent_for_agent(&self, agent: &str, limit: usize) -> Result> { + pub fn list_recent_for_agent( + &self, + agent: &str, + limit: usize, + ) -> Result> { let limit = limit.min(50); let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( @@ -300,10 +310,7 @@ impl BuildLogs { ORDER BY started_at DESC LIMIT ?2", )?; - let rows = stmt.query_map( - params![agent, i64::try_from(limit).unwrap_or(50)], - row_to_header, - )?; + let rows = stmt.query_map(params![agent, i64::try_from(limit).unwrap_or(50)], row_to_header)?; let mut out = Vec::new(); for r in rows { out.push(r?); @@ -463,9 +470,7 @@ mod tests { #[test] fn start_appends_finish_flow() { let (_d, db) = tmpdb(); - let id = db - .start("alice", "prebuild", "nix build foo") - .expect("start"); + let id = db.start("alice", "prebuild", "nix build foo").expect("start"); db.append_stdout(id, "building '/nix/store/abc.drv'"); db.append_stderr(id, "error: line 12"); db.append_stderr(id, " at /nix/store/.../module.nix:5"); @@ -504,7 +509,8 @@ mod tests { // but list_recent already orders by `started_at DESC` then // sqlite's natural insertion-order tiebreak. We rely only on // both IDs being present + correct count + agent isolation. - let ids: std::collections::HashSet = alice_rows.iter().map(|h| h.id).collect(); + let ids: std::collections::HashSet = + alice_rows.iter().map(|h| h.id).collect(); assert!(ids.contains(&id_a1)); assert!(ids.contains(&id_a2)); diff --git a/hive-c0re/src/container_view.rs b/hive-c0re/src/container_view.rs index 378d05ca..214d08cc 100644 --- a/hive-c0re/src/container_view.rs +++ b/hive-c0re/src/container_view.rs @@ -115,9 +115,7 @@ pub async fn build_all(coord: &Coordinator) -> Vec { } else { continue; }; - let deployed_full = locked - .get(&format!("agent-{logical}")) - .map(std::string::String::as_str); + let deployed_full = locked.get(&format!("agent-{logical}")).map(std::string::String::as_str); let needs_update = crate::auto_update::agent_config_pending(&logical, deployed_full); let deployed_sha = deployed_full.map(|s| s[..s.len().min(12)].to_owned()); // Recipient name the broker uses for this agent — sub-agents @@ -145,40 +143,27 @@ pub async fn build_all(coord: &Coordinator) -> Vec { // Static / declared fields (extra_links, deployed_sha, // pending_reminders, needs_update, parent) stay populated // regardless of run state. - let ( - needs_login, - ctx_tokens, - context_window_tokens, - rate_limited, - status_text, - status_set_at, - ) = if running { - // needs_login fires when EITHER the claude session dir is - // missing (boot-time / fresh container) OR the harness wrote - // the auth-failed sentinel because a turn hit 401. The - // manager has its own session lifecycle and never - // participates in needs_login. - let needs_login = !is_manager - && (!claude_has_session(&Coordinator::agent_claude_dir(&logical)) - || auth_failed_sentinel(&logical)); - let last_turn = read_last_turn(&logical); - let ctx_tokens = last_turn.as_ref().map(|(toks, _)| *toks); - let context_window_tokens = last_turn - .as_ref() - .and_then(|(_, model)| resolve_ctx_window(model, &coord.context_window_tokens)); - let rate_limited = is_rate_limited(&logical); - let (status_text, status_set_at) = read_status(&logical); - ( - needs_login, - ctx_tokens, - context_window_tokens, - rate_limited, - status_text, - status_set_at, - ) - } else { - (false, None, None, false, None, None) - }; + let (needs_login, ctx_tokens, context_window_tokens, rate_limited, status_text, status_set_at) = + if running { + // needs_login fires when EITHER the claude session dir is + // missing (boot-time / fresh container) OR the harness wrote + // the auth-failed sentinel because a turn hit 401. The + // manager has its own session lifecycle and never + // participates in needs_login. + let needs_login = !is_manager + && (!claude_has_session(&Coordinator::agent_claude_dir(&logical)) + || auth_failed_sentinel(&logical)); + let last_turn = read_last_turn(&logical); + let ctx_tokens = last_turn.as_ref().map(|(toks, _)| *toks); + let context_window_tokens = last_turn + .as_ref() + .and_then(|(_, model)| resolve_ctx_window(model, &coord.context_window_tokens)); + let rate_limited = is_rate_limited(&logical); + let (status_text, status_set_at) = read_status(&logical); + (needs_login, ctx_tokens, context_window_tokens, rate_limited, status_text, status_set_at) + } else { + (false, None, None, false, None, None) + }; out.push(ContainerView { port: lifecycle::agent_web_port(&logical), running, @@ -232,18 +217,12 @@ fn read_dashboard_links(name: &str) -> Vec { /// don't lose state during the transition window. fn read_harness_flags(name: &str) -> (bool, bool) { let dir = Coordinator::agent_notes_dir(name); - if let Ok(raw) = std::fs::read_to_string(dir.join("hyperhive-harness.json")) { - if let Ok(v) = serde_json::from_str::(&raw) { - let rl = v - .get("rate_limited") - .and_then(|x| x.as_bool()) - .unwrap_or(false); - let nl = v - .get("needs_login") - .and_then(|x| x.as_bool()) - .unwrap_or(false); - return (rl, nl); - } + if let Ok(raw) = std::fs::read_to_string(dir.join("hyperhive-harness.json")) + && let Ok(v) = serde_json::from_str::(&raw) + { + let rl = v.get("rate_limited").and_then(serde_json::Value::as_bool).unwrap_or(false); + let nl = v.get("needs_login").and_then(serde_json::Value::as_bool).unwrap_or(false); + return (rl, nl); } // Legacy fallback: presence of individual sentinel files. let rate_limited = dir.join("hyperhive-rate-limited").exists(); @@ -270,23 +249,14 @@ pub fn read_agent_status(name: &str) -> (Option, Option) { let path = Coordinator::agent_notes_dir(name).join("hyperhive-status"); let meta = std::fs::metadata(&path).ok(); let s = std::fs::read_to_string(&path).ok(); - let text = s - .as_deref() - .map(str::trim) - .filter(|t| !t.is_empty()) - .map(str::to_owned); + let text = s.as_deref().map(str::trim).filter(|t| !t.is_empty()).map(str::to_owned); let mtime = meta.and_then(|m| { m.modified().ok().and_then(|t| { - t.duration_since(std::time::UNIX_EPOCH) - .ok() + t.duration_since(std::time::UNIX_EPOCH).ok() .and_then(|d| i64::try_from(d.as_secs()).ok()) }) }); - if text.is_none() { - (None, None) - } else { - (text, mtime) - } + if text.is_none() { (None, None) } else { (text, mtime) } } fn read_status(name: &str) -> (Option, Option) { @@ -334,7 +304,9 @@ pub async fn read_agent_status_live(name: &str) -> (Option, Option, /// corresponding env var is unset or empty. #[must_use] pub fn hive_swarm_names() -> (Option, Option) { - let read = |var: &str| -> Option { std::env::var(var).ok().filter(|s| !s.is_empty()) }; + let read = |var: &str| -> Option { + std::env::var(var).ok().filter(|s| !s.is_empty()) + }; (read("HYPERHIVE_HIVE_NAME"), read("HYPERHIVE_SWARM_NAME")) } @@ -349,8 +321,11 @@ pub fn hive_swarm_names() -> (Option, Option) { /// mirroring `hive_ag3nt::events::TokenUsage::context_tokens`. fn read_last_turn(name: &str) -> Option<(u64, String)> { let path = Coordinator::agent_notes_dir(name).join("hyperhive-turn-stats.sqlite"); - let conn = - Connection::open_with_flags(&path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY).ok()?; + let conn = Connection::open_with_flags( + &path, + rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY, + ) + .ok()?; conn.query_row( "SELECT last_input_tokens + last_cache_read_input_tokens + last_cache_creation_input_tokens, model \ FROM turn_stats ORDER BY started_at DESC LIMIT 1", @@ -434,26 +409,14 @@ mod tests { #[test] fn resolves_family_substring() { - assert_eq!( - resolve_ctx_window("claude-3-5-haiku-20241022", &cfg()), - Some(200_000) - ); - assert_eq!( - resolve_ctx_window("claude-sonnet-4-5", &cfg()), - Some(1_000_000) - ); - assert_eq!( - resolve_ctx_window("claude-opus-4-1", &cfg()), - Some(1_000_000) - ); + assert_eq!(resolve_ctx_window("claude-3-5-haiku-20241022", &cfg()), Some(200_000)); + assert_eq!(resolve_ctx_window("claude-sonnet-4-5", &cfg()), Some(1_000_000)); + assert_eq!(resolve_ctx_window("claude-opus-4-1", &cfg()), Some(1_000_000)); } #[test] fn resolution_is_case_insensitive() { - assert_eq!( - resolve_ctx_window("Claude-Sonnet-4", &cfg()), - Some(1_000_000) - ); + assert_eq!(resolve_ctx_window("Claude-Sonnet-4", &cfg()), Some(1_000_000)); } #[test] @@ -463,10 +426,7 @@ mod tests { #[test] fn empty_config_yields_none() { - assert_eq!( - resolve_ctx_window("claude-3-5-haiku", &HashMap::new()), - None - ); + assert_eq!(resolve_ctx_window("claude-3-5-haiku", &HashMap::new()), None); } #[test] diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 7bea60bc..5bafe818 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -708,16 +708,11 @@ impl Coordinator { /// whose stop the crash watcher should NOT classify as a crash. /// Lazily reaps entries older than `grace` so the map stays /// bounded by the active agent count. - pub fn recent_transient_within( - &self, - grace: std::time::Duration, - ) -> HashMap { + pub fn recent_transient_within(&self, grace: std::time::Duration) -> HashMap { let now = std::time::Instant::now(); let mut map = self.recent_transient.lock().unwrap(); map.retain(|_, (_, ts)| now.duration_since(*ts) <= grace); - map.iter() - .map(|(k, (kind, _))| (k.clone(), *kind)) - .collect() + map.iter().map(|(k, (kind, _))| (k.clone(), *kind)).collect() } /// Set a transient state and return a guard that clears it on drop. diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index d4e79eb0..b66586da 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -472,15 +472,9 @@ async fn api_state(headers: HeaderMap, State(state): State) -> axum::J let s = v.to_string_lossy().to_ascii_lowercase(); matches!(s.as_str(), "1" | "true" | "yes") }), - forge_public_url: std::env::var("HIVE_FORGE_PUBLIC_URL") - .ok() - .filter(|s| !s.is_empty()), - hive_name: std::env::var("HYPERHIVE_HIVE_NAME") - .ok() - .filter(|s| !s.is_empty()), - swarm_name: std::env::var("HYPERHIVE_SWARM_NAME") - .ok() - .filter(|s| !s.is_empty()), + forge_public_url: std::env::var("HIVE_FORGE_PUBLIC_URL").ok().filter(|s| !s.is_empty()), + hive_name: std::env::var("HYPERHIVE_HIVE_NAME").ok().filter(|s| !s.is_empty()), + swarm_name: std::env::var("HYPERHIVE_SWARM_NAME").ok().filter(|s| !s.is_empty()), peer_hives: parse_peer_hives(), }) } @@ -1757,7 +1751,9 @@ async fn get_build_log_full( ) -> Response { match state.coord.build_logs.get_full(id) { Ok(Some(log)) => axum::Json(log).into_response(), - Ok(None) => (StatusCode::NOT_FOUND, format!("build log #{id} not found")).into_response(), + Ok(None) => { + (StatusCode::NOT_FOUND, format!("build log #{id} not found")).into_response() + } Err(e) => error_response(&format!("build-log {id}: {e:#}")), } } @@ -1874,7 +1870,10 @@ async fn get_build_log_stream( /// separator (same layout the JS side-panel renders). The /// `Content-Disposition` header triggers a browser download with a /// descriptive filename so the operator can save and share the log. -async fn get_build_log_raw(State(state): State, AxumPath(id): AxumPath) -> Response { +async fn get_build_log_raw( + State(state): State, + AxumPath(id): AxumPath, +) -> Response { match state.coord.build_logs.get_full(id) { Ok(Some(log)) => { let mut text = log.stdout; @@ -1898,7 +1897,9 @@ async fn get_build_log_raw(State(state): State, AxumPath(id): AxumPath ) .into_response() } - Ok(None) => (StatusCode::NOT_FOUND, format!("build log #{id} not found")).into_response(), + Ok(None) => { + (StatusCode::NOT_FOUND, format!("build log #{id} not found")).into_response() + } Err(e) => error_response(&format!("build-log {id}: {e:#}")), } } @@ -2521,10 +2522,7 @@ async fn get_tool_groups(State(_state): State) -> axum::Json Result> { // `inputs = { }`. Any other git failure (permission denied, // ref-not-found, etc.) propagates as a hard error rather than // being silently swallowed. - if stderr.contains("does not exist") || stderr.contains("exists on disk, but not in") { + if stderr.contains("does not exist") + || stderr.contains("exists on disk, but not in") + { return Ok(None); } anyhow::bail!("git show {spec} failed: {}", stderr.trim()); @@ -128,7 +130,10 @@ pub fn duplicate_groups(raw: &str) -> Result> { }); entry.keys.push(name.clone()); } - let mut dups: Vec = groups.into_values().filter(|g| g.keys.len() > 1).collect(); + let mut dups: Vec = groups + .into_values() + .filter(|g| g.keys.len() > 1) + .collect(); for g in &mut dups { g.keys.sort(); } @@ -254,7 +259,12 @@ async fn lock_in_sync_inner(worktree: &Path) -> Result<()> { async fn remove_worktree(repo: &Path, worktree: &Path) -> Result<()> { let out = git_command() .current_dir(repo) - .args(["worktree", "remove", "--force", &worktree.to_string_lossy()]) + .args([ + "worktree", + "remove", + "--force", + &worktree.to_string_lossy(), + ]) .output() .await .with_context(|| format!("git worktree remove {}", worktree.display()))?; diff --git a/hive-c0re/src/forge.rs b/hive-c0re/src/forge.rs index 509f521c..9f48939b 100644 --- a/hive-c0re/src/forge.rs +++ b/hive-c0re/src/forge.rs @@ -217,14 +217,7 @@ async fn ensure_user_exists(name: &str, admin: bool, password: Option<&str>) -> /// from the operator's point of view: same password input → same final /// account state. async fn change_user_password(name: &str, password: &str) -> Result<()> { - let args = [ - "user", - "change-password", - "--username", - name, - "--password", - password, - ]; + let args = ["user", "change-password", "--username", name, "--password", password]; forge_admin(&args) .await .with_context(|| format!("forgejo admin user change-password {name}"))?; @@ -558,8 +551,9 @@ pub async fn ensure_shared_docs_repo(core_token: &str) -> Result<()> { /// Mirrors `meta_read_access` so agents can clone the shared docs repo /// without authentication hassle. pub async fn shared_docs_access(name: &str, core_token: &str) -> Result<()> { - let url = - format!("{FORGE_HTTP}/api/v1/repos/{SHARED_ORG}/{SHARED_DOCS_REPO}/collaborators/{name}"); + let url = format!( + "{FORGE_HTTP}/api/v1/repos/{SHARED_ORG}/{SHARED_DOCS_REPO}/collaborators/{name}" + ); let body = r#"{"permission":"read"}"#; let out = Command::new("curl") .args([ diff --git a/hive-c0re/src/lib.rs b/hive-c0re/src/lib.rs index 833cd7c8..85b4c811 100644 --- a/hive-c0re/src/lib.rs +++ b/hive-c0re/src/lib.rs @@ -14,14 +14,15 @@ pub mod actions; pub mod agent_ports; +pub mod capabilities; pub mod agent_server; pub mod agent_sockets; +pub mod gateway_nginx; pub mod approvals; pub mod auto_update; pub mod bash_tasks_vacuum; pub mod broker; pub mod build_logs; -pub mod capabilities; pub mod client; pub mod container_view; pub mod coordinator; @@ -31,7 +32,6 @@ pub mod dashboard_events; pub mod events_vacuum; pub mod flake_check; pub mod forge; -pub mod gateway_nginx; pub mod lifecycle; pub mod limits; pub mod loose_ends; diff --git a/hive-c0re/src/lifecycle.rs b/hive-c0re/src/lifecycle.rs index 516ae0db..a2029253 100644 --- a/hive-c0re/src/lifecycle.rs +++ b/hive-c0re/src/lifecycle.rs @@ -449,12 +449,11 @@ pub async fn rebuild_no_meta( "kill before cold-start retry failed (ignored)" ); }); - run(&["start", &container]).await.map_err(|e| { - anyhow::anyhow!( + run(&["start", &container]).await + .map_err(|e| anyhow::anyhow!( "cold-start fallback also failed: {e:#} \ (original start error: {start_err:#})" - ) - }) + )) } else { Ok(()) } @@ -495,7 +494,9 @@ async fn prebuild_toplevel(name: &str, flake_ref: &str) -> Result<()> { // pair (no current callsite does, but the pair is redundant // and worth checking once). if fragment != name { - anyhow::bail!("prebuild_toplevel: flake_ref fragment '{fragment}' ≠ agent name '{name}'"); + anyhow::bail!( + "prebuild_toplevel: flake_ref fragment '{fragment}' ≠ agent name '{name}'" + ); } let attr = format!("{flake_root}#nixosConfigurations.{name}.config.system.build.toplevel"); let args = vec![ @@ -1134,7 +1135,8 @@ fn set_nspawn_flags( ); } let own_config = format!("{HOST_AGENTS_ROOT}/{agent_name}/config"); - std::fs::create_dir_all(&own_config).with_context(|| format!("create {own_config}"))?; + std::fs::create_dir_all(&own_config) + .with_context(|| format!("create {own_config}"))?; let _ = write!(binds, " --bind-ro={own_config}:/agents/{agent_name}/config"); } @@ -1335,9 +1337,9 @@ async fn run(args: &[&str]) -> Result<()> { // every notification with the eval-error verbatim. let journal = container_journal_tail(args).await; match log_id { - Some(id) => { - bail!("nixos-container {cmdline} failed ({status}); see build log #{id}{journal}") - } + Some(id) => bail!( + "nixos-container {cmdline} failed ({status}); see build log #{id}{journal}" + ), None => bail!("nixos-container {cmdline} failed ({status}){journal}"), } } @@ -1441,3 +1443,4 @@ mod tests { ); } } + diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index 6def7c87..6cd903de 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -16,7 +16,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; use anyhow::Result; -use hive_sh4re::{LooseEnd, MANAGER_AGENT}; +use hive_sh4re::{MANAGER_AGENT, LooseEnd}; use crate::coordinator::Coordinator; diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 6d149abf..f4ee0b62 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -12,8 +12,8 @@ use hive_sh4re::{HostRequest, HostResponse}; // explicit (any new daemon entry point reads off the next add). use hive_c0re::coordinator::Coordinator; use hive_c0re::{ - agent_sockets, auto_update, bash_tasks_vacuum, broker, client, crash_watch, dashboard, - dashboard_events, events_vacuum, forge, manager_server, matrix, migrate, rebuild_queue, + agent_sockets, auto_update, broker, client, crash_watch, dashboard, dashboard_events, + bash_tasks_vacuum, events_vacuum, forge, manager_server, matrix, migrate, rebuild_queue, reminder_scheduler, scheduled_prompts_worker, server, stats_vacuum, }; @@ -51,10 +51,7 @@ enum Cmd { /// short name to token count. Threaded into each container as /// `HIVE_CONTEXT_WINDOW_TOKENS_` env vars. Set via the /// `services.hive-c0re.contextWindowTokens` NixOS option. - #[arg( - long, - default_value = r#"{"haiku":200000,"sonnet":1000000,"opus":1000000}"# - )] + #[arg(long, default_value = r#"{"haiku":200000,"sonnet":1000000,"opus":1000000}"#)] context_window_tokens: String, }, /// Spawn a new agent container directly (`hive-agent-`). Bypasses @@ -122,17 +119,7 @@ async fn main() -> Result<()> { dashboard_port, operator_pronouns, context_window_tokens, - } => { - cmd_serve( - hyperhive_flake, - db, - dashboard_port, - operator_pronouns, - context_window_tokens, - &cli.socket, - ) - .await - } + } => cmd_serve(hyperhive_flake, db, dashboard_port, operator_pronouns, context_window_tokens, &cli.socket).await, Cmd::Spawn { name } => { render(client::request(&cli.socket, HostRequest::Spawn { name }).await?) } @@ -161,7 +148,11 @@ async fn main() -> Result<()> { } => { let new_parent = if root { None } else { parent }; render( - client::request(&cli.socket, HostRequest::SetParent { child, new_parent }).await?, + client::request( + &cli.socket, + HostRequest::SetParent { child, new_parent }, + ) + .await?, ) } } @@ -178,8 +169,9 @@ async fn cmd_serve( context_window_tokens: String, socket: &std::path::Path, ) -> Result<()> { - let cwt: std::collections::HashMap = serde_json::from_str(&context_window_tokens) - .context("--context-window-tokens: invalid JSON")?; + let cwt: std::collections::HashMap = + serde_json::from_str(&context_window_tokens) + .context("--context-window-tokens: invalid JSON")?; let coord = Arc::new(Coordinator::open( &db, hyperhive_flake, @@ -343,14 +335,7 @@ fn spawn_broker_to_dashboard_forwarder(coord: Arc) { tokio::spawn(async move { loop { match rx.recv().await { - Ok(MessageEvent::Sent { - id, - from, - to, - body, - at, - in_reply_to, - }) => { + Ok(MessageEvent::Sent { id, from, to, body, at, in_reply_to }) => { let file_refs = dashboard::scan_validated_paths(&body); coord.emit_dashboard_event(DashboardEvent::Sent { seq: coord.next_seq(), @@ -363,14 +348,7 @@ fn spawn_broker_to_dashboard_forwarder(coord: Arc) { file_refs, }); } - Ok(MessageEvent::Delivered { - id, - from, - to, - body, - at, - in_reply_to, - }) => { + Ok(MessageEvent::Delivered { id, from, to, body, at, in_reply_to }) => { let file_refs = dashboard::scan_validated_paths(&body); coord.emit_dashboard_event(DashboardEvent::Delivered { seq: coord.next_seq(), diff --git a/hive-c0re/src/meta.rs b/hive-c0re/src/meta.rs index 8bd21f32..7810ddfa 100644 --- a/hive-c0re/src/meta.rs +++ b/hive-c0re/src/meta.rs @@ -572,10 +572,7 @@ where // Emit `capabilities = "cap1,cap2"` when the operator has // granted capabilities to this agent. Absent entry = null = no // capability env var injected, capability-gated tools hidden. - let caps = capabilities_map - .get(&spec.name) - .cloned() - .unwrap_or_default(); + let caps = capabilities_map.get(&spec.name).cloned().unwrap_or_default(); let capabilities_attr = if caps.is_empty() { "null".to_owned() } else { diff --git a/hive-c0re/src/migrate.rs b/hive-c0re/src/migrate.rs index 8b37643d..29f7e569 100644 --- a/hive-c0re/src/migrate.rs +++ b/hive-c0re/src/migrate.rs @@ -147,9 +147,7 @@ fn migrate_harness_files(name: &str) { } 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") - } + Err(e) => tracing::warn!(%name, %file, error = ?e, "migration: move to harness dir failed"), } } } @@ -192,11 +190,7 @@ async fn rename_manager_container(coord: &Arc) { // Stop the old container. Abort if stop fails — continuing with a // running `root` and then starting `h-root` risks two manager // instances racing for the same broker / state files. - match Command::new("nixos-container") - .args(["stop", "root"]) - .status() - .await - { + match Command::new("nixos-container").args(["stop", "root"]).status().await { Ok(s) if s.success() => {} Ok(s) => { tracing::warn!(status = %s, "migration phase 5: nixos-container stop root failed — aborting"); @@ -224,20 +218,12 @@ async fn rename_manager_container(coord: &Arc) { } // Daemon reload so systemd picks up the new container@h-root unit. - if let Err(e) = Command::new("systemctl") - .args(["daemon-reload"]) - .status() - .await - { + if let Err(e) = Command::new("systemctl").args(["daemon-reload"]).status().await { tracing::warn!(error = ?e, "migration phase 5: systemctl daemon-reload failed"); } // Start the renamed container. - if let Err(e) = Command::new("nixos-container") - .args(["start", "h-root"]) - .status() - .await - { + if let Err(e) = Command::new("nixos-container").args(["start", "h-root"]).status().await { tracing::warn!(error = ?e, "migration phase 5: nixos-container start h-root failed"); return; } diff --git a/hive-c0re/src/operator_questions.rs b/hive-c0re/src/operator_questions.rs index 428145e1..5440c115 100644 --- a/hive-c0re/src/operator_questions.rs +++ b/hive-c0re/src/operator_questions.rs @@ -209,7 +209,11 @@ impl OperatorQuestions { /// /// Not the target — that's covered by `answer` (responding with /// an actual reply, sentinel or otherwise). - pub fn cancel(&self, id: i64, canceller: &str) -> Result<(String, String, Option)> { + pub fn cancel( + &self, + id: i64, + canceller: &str, + ) -> Result<(String, String, Option)> { let conn = self.conn.lock().unwrap(); let row: Option<(String, String, Option, Option)> = conn .query_row( @@ -228,7 +232,9 @@ impl OperatorQuestions { || canceller == hive_sh4re::OPERATOR_RECIPIENT || canceller == hive_sh4re::MANAGER_AGENT; if !authorised { - bail!("question {id}: '{canceller}' not allowed to cancel (asker = '{asker}')"); + bail!( + "question {id}: '{canceller}' not allowed to cancel (asker = '{asker}')" + ); } let sentinel = format!("[cancelled by {canceller}]"); conn.execute( @@ -283,6 +289,7 @@ impl OperatorQuestions { rows.collect::>>() .map_err(Into::into) } + } fn row_to_question(row: &rusqlite::Row<'_>) -> rusqlite::Result { diff --git a/hive-c0re/src/priv_client.rs b/hive-c0re/src/priv_client.rs index 81ef6b44..b4e65a12 100644 --- a/hive-c0re/src/priv_client.rs +++ b/hive-c0re/src/priv_client.rs @@ -7,7 +7,7 @@ //! a persistent connection. use anyhow::{Context as _, Result, bail}; -use hive_sh4re::priv_proto::{BindMount, PRIV_SOCK, PrivRequest, PrivResponse}; +use hive_sh4re::priv_proto::{PRIV_SOCK, BindMount, PrivRequest, PrivResponse}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; @@ -31,49 +31,27 @@ pub async fn call(req: &PrivRequest) -> Result { } pub async fn start_container(name: &str) -> Result<()> { - ok(call(&PrivRequest::StartContainer { - name: name.to_owned(), - }) - .await?) + ok(call(&PrivRequest::StartContainer { name: name.to_owned() }).await?) } pub async fn stop_container(name: &str) -> Result<()> { - ok(call(&PrivRequest::StopContainer { - name: name.to_owned(), - }) - .await?) + ok(call(&PrivRequest::StopContainer { name: name.to_owned() }).await?) } pub async fn kill_container(name: &str) -> Result<()> { - ok(call(&PrivRequest::KillContainer { - name: name.to_owned(), - }) - .await?) + ok(call(&PrivRequest::KillContainer { name: name.to_owned() }).await?) } pub async fn update_container(name: &str) -> Result<(String, String)> { - check( - call(&PrivRequest::UpdateContainer { - name: name.to_owned(), - }) - .await?, - ) + check(call(&PrivRequest::UpdateContainer { name: name.to_owned() }).await?) } pub async fn create_container(name: &str) -> Result<(String, String)> { - check( - call(&PrivRequest::CreateContainer { - name: name.to_owned(), - }) - .await?, - ) + check(call(&PrivRequest::CreateContainer { name: name.to_owned() }).await?) } pub async fn destroy_container(name: &str) -> Result<()> { - ok(call(&PrivRequest::DestroyContainer { - name: name.to_owned(), - }) - .await?) + ok(call(&PrivRequest::DestroyContainer { name: name.to_owned() }).await?) } pub async fn list_containers() -> Result { @@ -85,8 +63,7 @@ pub async fn write_nspawn_flags(container: &str, binds: &[BindMount]) -> Result< ok(call(&PrivRequest::WriteNspawnFlags { container: container.to_owned(), binds: binds.to_vec(), - }) - .await?) + }).await?) } pub async fn write_resource_limits( @@ -98,15 +75,13 @@ pub async fn write_resource_limits( container: container.to_owned(), memory_max: memory_max.to_owned(), cpu_quota: cpu_quota.to_owned(), - }) - .await?) + }).await?) } pub async fn remove_service_dropin(container: &str) -> Result<()> { ok(call(&PrivRequest::RemoveServiceDropin { container: container.to_owned(), - }) - .await?) + }).await?) } pub async fn daemon_reload() -> Result<()> { @@ -122,26 +97,21 @@ pub async fn chown_socket_dir(agent_name: &str, uid: u32, gid: u32) -> Result<() agent_name: agent_name.to_owned(), uid, gid, - }) - .await?) + }).await?) } pub async fn chmod_socket_dir(agent_name: &str, mode: u32) -> Result<()> { ok(call(&PrivRequest::ChmodSocketDir { agent_name: agent_name.to_owned(), mode, - }) - .await?) + }).await?) } fn check(resp: PrivResponse) -> Result<(String, String)> { if resp.ok { Ok((resp.stdout, resp.stderr)) } else { - bail!( - "{}", - resp.error.as_deref().unwrap_or("hive-priv returned error") - ) + bail!("{}", resp.error.as_deref().unwrap_or("hive-priv returned error")) } } diff --git a/hive-c0re/src/questions.rs b/hive-c0re/src/questions.rs index 35eaa1c8..64ca7218 100644 --- a/hive-c0re/src/questions.rs +++ b/hive-c0re/src/questions.rs @@ -236,7 +236,8 @@ mod tests { // exact MANAGER_AGENT constant passes). assert!(check_approval_canceller_is_manager("").is_err()); assert!( - check_approval_canceller_is_manager(hive_sh4re::OPERATOR_RECIPIENT).is_err(), + check_approval_canceller_is_manager(hive_sh4re::OPERATOR_RECIPIENT) + .is_err(), "operator surface uses the dashboard cancel path, not this dispatcher", ); } diff --git a/hive-c0re/src/rebuild_queue.rs b/hive-c0re/src/rebuild_queue.rs index fe8d090f..fd162e9e 100644 --- a/hive-c0re/src/rebuild_queue.rs +++ b/hive-c0re/src/rebuild_queue.rs @@ -1228,3 +1228,4 @@ mod tests { ); } } + diff --git a/hive-c0re/src/reminder_scheduler.rs b/hive-c0re/src/reminder_scheduler.rs index a626d3d3..2f25ff71 100644 --- a/hive-c0re/src/reminder_scheduler.rs +++ b/hive-c0re/src/reminder_scheduler.rs @@ -130,7 +130,8 @@ pub fn write_payload(agent: &str, host_path: &Path, message: &str) -> Result<(), let Some(parent) = host_path.parent() else { return Err("internal: host path has no parent".to_owned()); }; - std::fs::create_dir_all(parent).map_err(|e| format!("parent dir create failed: {e}"))?; + std::fs::create_dir_all(parent) + .map_err(|e| format!("parent dir create failed: {e}"))?; // Resolve symlinks in the parent chain, then re-verify the // canonical form still lives under the agent's host state root — // catches `ln -s /etc state/escape` style attacks. diff --git a/hive-c0re/src/scheduled_prompts_worker.rs b/hive-c0re/src/scheduled_prompts_worker.rs index 03440cef..1028f231 100644 --- a/hive-c0re/src/scheduled_prompts_worker.rs +++ b/hive-c0re/src/scheduled_prompts_worker.rs @@ -101,10 +101,7 @@ fn fire_schedule(coord: &Arc, schedule: &Schedule, now: i64) { // scheduled prompt from stacking up when an agent is slow or // briefly offline, while still allowing distinct scheduled // messages (different body) to enqueue independently. - match coord - .broker - .has_pending_with_body(target, "scheduled", &schedule.body) - { + match coord.broker.has_pending_with_body(target, "scheduled", &schedule.body) { Ok(true) => { tracing::debug!( schedule = schedule.id, @@ -387,3 +384,4 @@ async fn known_agents_async() -> std::collections::HashSet { } out } + diff --git a/hive-c0re/src/stats_vacuum.rs b/hive-c0re/src/stats_vacuum.rs index 7072a319..54e7ec44 100644 --- a/hive-c0re/src/stats_vacuum.rs +++ b/hive-c0re/src/stats_vacuum.rs @@ -39,7 +39,8 @@ pub fn spawn(coord: &Arc) { fn sweep_once() { for name in Coordinator::kept_state_names() { - let path = Coordinator::agent_harness_dir(&name).join("hyperhive-turn-stats.sqlite"); + let path = + Coordinator::agent_harness_dir(&name).join("hyperhive-turn-stats.sqlite"); if !path.exists() { continue; } @@ -59,9 +60,7 @@ fn vacuum_file(path: &Path) -> Result { .and_then(|d| i64::try_from(d.as_secs()).ok()) .unwrap_or(0); let cutoff = now - KEEP_SECS; - let removed = conn.execute( - "DELETE FROM turn_stats WHERE started_at < ?1", - params![cutoff], - )?; + let removed = + conn.execute("DELETE FROM turn_stats WHERE started_at < ?1", params![cutoff])?; Ok(u64::try_from(removed).unwrap_or(0)) } diff --git a/hive-c0re/src/topology.rs b/hive-c0re/src/topology.rs index 7f3e319f..d02d30bb 100644 --- a/hive-c0re/src/topology.rs +++ b/hive-c0re/src/topology.rs @@ -59,7 +59,10 @@ pub fn children_of(name: &str) -> Vec { /// Pure form of [`children_of`] for unit tests. #[must_use] -pub fn children_of_in(topo: &BTreeMap>, name: &str) -> Vec { +pub fn children_of_in( + topo: &BTreeMap>, + name: &str, +) -> Vec { topo.iter() .filter_map(|(agent, parent)| { if parent.as_deref() == Some(name) { @@ -87,13 +90,7 @@ pub fn top_level_agents() -> Vec { #[must_use] pub fn top_level_agents_in(topo: &BTreeMap>) -> Vec { topo.iter() - .filter_map(|(name, parent)| { - if parent.is_none() { - Some(name.clone()) - } else { - None - } - }) + .filter_map(|(name, parent)| if parent.is_none() { Some(name.clone()) } else { None }) .collect() } @@ -504,12 +501,8 @@ mod tests { // `alice` who lives under the manager) would close the loop. // The general cycle walk catches this; no separate manager // guard needed. - let err = apply_set_parent( - &topo_three_level(), - crate::lifecycle::MANAGER_NAME, - Some("bob"), - ) - .unwrap_err(); + let err = apply_set_parent(&topo_three_level(), crate::lifecycle::MANAGER_NAME, Some("bob")) + .unwrap_err(); assert!(err.contains("cycle"), "err = {err}"); } @@ -685,32 +678,20 @@ mod tests { "alice".to_owned(), vec![ROLE_CAN_MANAGE_TOP_LEVEL_AGENTS.to_owned()], ); - assert!(has_role_in( - &roles, - "alice", - ROLE_CAN_MANAGE_TOP_LEVEL_AGENTS - )); + assert!(has_role_in(&roles, "alice", ROLE_CAN_MANAGE_TOP_LEVEL_AGENTS)); } #[test] fn has_role_in_returns_false_for_absent_agent() { let roles: BTreeMap> = BTreeMap::new(); - assert!(!has_role_in( - &roles, - "alice", - ROLE_CAN_MANAGE_TOP_LEVEL_AGENTS - )); + assert!(!has_role_in(&roles, "alice", ROLE_CAN_MANAGE_TOP_LEVEL_AGENTS)); } #[test] fn has_role_in_returns_false_for_empty_list() { let mut roles = BTreeMap::new(); roles.insert("alice".to_owned(), vec![]); - assert!(!has_role_in( - &roles, - "alice", - ROLE_CAN_MANAGE_TOP_LEVEL_AGENTS - )); + assert!(!has_role_in(&roles, "alice", ROLE_CAN_MANAGE_TOP_LEVEL_AGENTS)); } /// Revoking a role must leave the key present with an empty list so @@ -720,10 +701,7 @@ mod tests { let mgr = crate::lifecycle::MANAGER_NAME; // Build an in-memory roles map as set_role would see it after granting. let mut roles: BTreeMap> = BTreeMap::new(); - roles.insert( - mgr.to_owned(), - vec![ROLE_CAN_MANAGE_TOP_LEVEL_AGENTS.to_owned()], - ); + roles.insert(mgr.to_owned(), vec![ROLE_CAN_MANAGE_TOP_LEVEL_AGENTS.to_owned()]); // Simulate the revoke path of set_role (in-memory, no disk). let list = roles.entry(mgr.to_owned()).or_default(); @@ -749,10 +727,7 @@ mod tests { let mgr_present = agent_names.iter().any(|n| n == mgr); let should_seed = mgr_present && !roles.contains_key(mgr); // should_seed must be false because manager key is present (tombstone). - assert!( - !should_seed, - "reconcile_roles must not re-seed an explicit revoke" - ); + assert!(!should_seed, "reconcile_roles must not re-seed an explicit revoke"); } /// `reconcile_roles` seeds the manager on first appearance (no prior entry). diff --git a/hive-forge/Cargo.toml b/hive-forge/Cargo.toml index 3b4eb86c..def1915e 100644 --- a/hive-forge/Cargo.toml +++ b/hive-forge/Cargo.toml @@ -10,12 +10,7 @@ path = "src/main.rs" [dependencies] anyhow = { workspace = true } clap = { workspace = true } -reqwest = { workspace = true, features = [ - "json", - "rustls-tls", - "blocking", - "multipart", -] } +reqwest = { workspace = true, features = ["json", "rustls-tls", "blocking", "multipart"] } serde = { workspace = true } serde_json = { workspace = true } diff --git a/hive-forge/src/body.rs b/hive-forge/src/body.rs index 0025c572..39341ade 100644 --- a/hive-forge/src/body.rs +++ b/hive-forge/src/body.rs @@ -23,8 +23,8 @@ pub fn resolve(body: Option<&str>, file: Option<&str>) -> Result> if path == "-" { return Ok(Some(read_stdin().context("read stdin for --body-file -")?)); } - let s = - std::fs::read_to_string(path).with_context(|| format!("read --body-file {path}"))?; + let s = std::fs::read_to_string(path) + .with_context(|| format!("read --body-file {path}"))?; return Ok(Some(s)); } if !std::io::stdin().is_terminal() { diff --git a/hive-forge/src/client.rs b/hive-forge/src/client.rs index cff4c592..6e7d47e4 100644 --- a/hive-forge/src/client.rs +++ b/hive-forge/src/client.rs @@ -180,12 +180,7 @@ impl Client { let form = reqwest::blocking::multipart::Form::new() .file("attachment", file) .with_context(|| format!("read {}", file.display()))?; - let resp = self - .http - .post(&url) - .multipart(form) - .send() - .context("POST")?; + let resp = self.http.post(&url).multipart(form).send().context("POST")?; decode_json(resp, &format!("POST {url}")) } } diff --git a/hive-forge/src/main.rs b/hive-forge/src/main.rs index 07b7ae25..f8dcd7a5 100644 --- a/hive-forge/src/main.rs +++ b/hive-forge/src/main.rs @@ -103,7 +103,8 @@ enum Verb { fn main() -> Result<()> { let cli = Cli::parse(); - let client = client::Client::from_env(cli.repo, cli.json).context("initialize forge client")?; + let client = + client::Client::from_env(cli.repo, cli.json).context("initialize forge client")?; match cli.verb { Verb::View(a) => verbs::view::run(&client, a), Verb::Issue(a) => verbs::issue::run(&client, a), diff --git a/hive-forge/src/verbs/comments.rs b/hive-forge/src/verbs/comments.rs index 73c1a0bc..944614d0 100644 --- a/hive-forge/src/verbs/comments.rs +++ b/hive-forge/src/verbs/comments.rs @@ -105,7 +105,10 @@ fn fetch_tail(client: &Client, repo: &str, number: u64, n: usize) -> Result Result<()> { let repo = client.repo(); - let diff = client.get_text( - &format!("/repos/{repo}/pulls/{}.diff", args.number), - "text/plain", - )?; + let diff = client.get_text(&format!("/repos/{repo}/pulls/{}.diff", args.number), "text/plain")?; let out = if args.full { diff } else { @@ -221,9 +218,7 @@ mod tests { assert!(is_autogenerated("flake.lock")); assert!(is_autogenerated("a/flake.lock")); assert!(is_autogenerated("hive-c0re/Cargo.lock")); - assert!(is_autogenerated( - "frontend/packages/dashboard/package-lock.json" - )); + assert!(is_autogenerated("frontend/packages/dashboard/package-lock.json")); assert!(!is_autogenerated("src/main.rs")); assert!(!is_autogenerated("Cargo.toml")); // Suffix-only files we deliberately don't match — keep @@ -351,7 +346,10 @@ index 1111..2222 100644 #[test] fn parse_diff_git_path_picks_b_side() { - assert_eq!(parse_diff_git_path("a/foo b/foo"), Some("foo".to_owned())); + assert_eq!( + parse_diff_git_path("a/foo b/foo"), + Some("foo".to_owned()) + ); assert_eq!( parse_diff_git_path("a/old.txt b/new.txt"), Some("new.txt".to_owned()) diff --git a/hive-forge/src/verbs/labels.rs b/hive-forge/src/verbs/labels.rs index 20db93b1..a0346750 100644 --- a/hive-forge/src/verbs/labels.rs +++ b/hive-forge/src/verbs/labels.rs @@ -36,8 +36,7 @@ pub fn run(client: &Client, args: Args) -> Result<()> { let repo = client.repo(); match args.action.unwrap_or(Action::List) { Action::List => { - let labels = - client.get_json(&format!("/repos/{repo}/issues/{}/labels", args.number))?; + let labels = client.get_json(&format!("/repos/{repo}/issues/{}/labels", args.number))?; print_label_names(&labels); } Action::Add { labels } => { @@ -65,8 +64,7 @@ pub fn run(client: &Client, args: Args) -> Result<()> { ); } } - let labels = - client.get_json(&format!("/repos/{repo}/issues/{}/labels", args.number))?; + let labels = client.get_json(&format!("/repos/{repo}/issues/{}/labels", args.number))?; print_label_names(&labels); } } diff --git a/hive-forge/src/verbs/list.rs b/hive-forge/src/verbs/list.rs index 49a78bc9..fecc5036 100644 --- a/hive-forge/src/verbs/list.rs +++ b/hive-forge/src/verbs/list.rs @@ -167,7 +167,9 @@ fn print_row(item: &Value) { // forge always populates `pull_request` but sets it to `null` for // issues; we check non-null specifically rather than just-present // (which would render every issue as a PR). - let is_pr = item.get("pull_request").is_some_and(|v| !v.is_null()); + let is_pr = item + .get("pull_request") + .is_some_and(|v| !v.is_null()); let kind = if is_pr { "PR" } else { " " }; println!("#{number:>4} {kind} [{author}] {title}"); } diff --git a/hive-forge/src/verbs/milestone.rs b/hive-forge/src/verbs/milestone.rs index ee92a28f..5ce9000a 100644 --- a/hive-forge/src/verbs/milestone.rs +++ b/hive-forge/src/verbs/milestone.rs @@ -41,7 +41,8 @@ pub fn run(client: &Client, args: Args) -> Result<()> { let repo = client.repo(); match args.action.unwrap_or(Action::List) { Action::List => { - let v = client.get_json(&format!("/repos/{repo}/milestones?state=open&limit=50"))?; + let v = + client.get_json(&format!("/repos/{repo}/milestones?state=open&limit=50"))?; let trimmed: Vec = v .as_array() .map(|a| { diff --git a/hive-forge/src/verbs/pr_create.rs b/hive-forge/src/verbs/pr_create.rs index a537a1c4..92d1376f 100644 --- a/hive-forge/src/verbs/pr_create.rs +++ b/hive-forge/src/verbs/pr_create.rs @@ -222,9 +222,7 @@ To http://localhost:3000/hyperhive/hyperhive.git\n\ assert!(is_pr_hint_opener( "remote: Create a new pull request for 'x':\n" )); - assert!(is_pr_hint_opener( - "remote: Visit the existing pull request:\n" - )); + assert!(is_pr_hint_opener("remote: Visit the existing pull request:\n")); assert!(!is_pr_hint_opener("remote: some other thing\n")); assert!(!is_pr_hint_opener("To http://example.com\n")); // Trailing-CRLF safety on windows-cloned forge clones. diff --git a/hive-forge/src/verbs/subscription.rs b/hive-forge/src/verbs/subscription.rs index f204f10f..8fc16d63 100644 --- a/hive-forge/src/verbs/subscription.rs +++ b/hive-forge/src/verbs/subscription.rs @@ -29,11 +29,7 @@ pub fn run(client: &Client, args: Args) -> Result<()> { return Ok(()); } if args.watch || args.ignore { - let (subscribed, ignored) = if args.ignore { - (false, true) - } else { - (true, false) - }; + let (subscribed, ignored) = if args.ignore { (false, true) } else { (true, false) }; let resp = client.post_json( &format!("/repos/{repo}/subscription"), &json!({ "subscribed": subscribed, "ignored": ignored }), diff --git a/hive-forge/src/verbs/timeline.rs b/hive-forge/src/verbs/timeline.rs index e9c7cca6..6c7cb288 100644 --- a/hive-forge/src/verbs/timeline.rs +++ b/hive-forge/src/verbs/timeline.rs @@ -170,10 +170,7 @@ fn format_event(ev: &Value) -> String { } } "commit_ref" => { - let sha = ev - .get("ref_commit_sha") - .and_then(Value::as_str) - .unwrap_or(""); + let sha = ev.get("ref_commit_sha").and_then(Value::as_str).unwrap_or(""); let short: String = sha.chars().take(7).collect(); if short.is_empty() { "referenced from a commit".to_owned() @@ -220,10 +217,7 @@ mod tests { "created_at": "2026-05-31T12:00:00Z", "body": "looks good to me", }); - assert_eq!( - format_event(&ev), - "**iris @ 2026-05-31T12:00:00Z**: looks good to me" - ); + assert_eq!(format_event(&ev), "**iris @ 2026-05-31T12:00:00Z**: looks good to me"); } #[test] diff --git a/hive-matrix-mcp/src/bin/mcp.rs b/hive-matrix-mcp/src/bin/mcp.rs index 38f4f715..427f30db 100644 --- a/hive-matrix-mcp/src/bin/mcp.rs +++ b/hive-matrix-mcp/src/bin/mcp.rs @@ -139,25 +139,21 @@ impl MatrixBridge { Returns the new event id." )] async fn send_message(&self, Parameters(args): Parameters) -> String { - render( - round_trip(DaemonRequest::SendMessage { - room: args.room, - body: args.body, - }) - .await, - ) + render(round_trip(DaemonRequest::SendMessage { + room: args.room, + body: args.body, + }).await) } - #[tool(description = "Open (or reuse) a direct message room with `user_id` \ - (@user:server) and post `body` to it.")] + #[tool( + description = "Open (or reuse) a direct message room with `user_id` \ + (@user:server) and post `body` to it." + )] async fn send_dm(&self, Parameters(args): Parameters) -> String { - render( - round_trip(DaemonRequest::SendDm { - user_id: args.user_id, - body: args.body, - }) - .await, - ) + render(round_trip(DaemonRequest::SendDm { + user_id: args.user_id, + body: args.body, + }).await) } #[tool( @@ -166,40 +162,35 @@ impl MatrixBridge { standard clients." )] async fn send_reaction(&self, Parameters(args): Parameters) -> String { - render( - round_trip(DaemonRequest::SendReaction { - room: args.room, - event_id: args.event_id, - key: args.key, - }) - .await, - ) + render(round_trip(DaemonRequest::SendReaction { + room: args.room, + event_id: args.event_id, + key: args.key, + }).await) } - #[tool(description = "Reply to a specific matrix event in a room, threaded \ - via m.in_reply_to. Returns the reply's event id.")] + #[tool( + description = "Reply to a specific matrix event in a room, threaded \ + via m.in_reply_to. Returns the reply's event id." + )] async fn send_reply(&self, Parameters(args): Parameters) -> String { - render( - round_trip(DaemonRequest::SendReply { - room: args.room, - event_id: args.event_id, - body: args.body, - }) - .await, - ) + render(round_trip(DaemonRequest::SendReply { + room: args.room, + event_id: args.event_id, + body: args.body, + }).await) } - #[tool(description = "Mark a specific event as read for this agent. Updates \ + #[tool( + description = "Mark a specific event as read for this agent. Updates \ the room's unread indicator + sends a read receipt other \ - participants can see.")] + participants can see." + )] async fn mark_read(&self, Parameters(args): Parameters) -> String { - render( - round_trip(DaemonRequest::MarkRead { - room: args.room, - event_id: args.event_id, - }) - .await, - ) + render(round_trip(DaemonRequest::MarkRead { + room: args.room, + event_id: args.event_id, + }).await) } #[tool( @@ -218,27 +209,28 @@ impl MatrixBridge { render(round_trip(DaemonRequest::ListRoomMembers { room: args.room }).await) } - #[tool(description = "Read the most recent N events from a matrix room \ + #[tool( + description = "Read the most recent N events from a matrix room \ (default 50, max 200). Returns each event's id, sender, timestamp, \ - type, and best-effort plain-text body.")] + type, and best-effort plain-text body." + )] async fn read_room(&self, Parameters(args): Parameters) -> String { - render( - round_trip(DaemonRequest::ReadRoom { - room: args.room, - limit: args.limit, - }) - .await, - ) + render(round_trip(DaemonRequest::ReadRoom { + room: args.room, + limit: args.limit, + }).await) } } -#[tool_handler(instructions = "Matrix client for an agent on a hyperhive swarm. Use \ +#[tool_handler( + instructions = "Matrix client for an agent on a hyperhive swarm. Use \ `send_message` to post in a joined room, `send_dm` to message a \ specific user, `send_reaction` to react with an emoji, `send_reply` \ to thread a reply, `mark_read` to acknowledge an event. Discover \ rooms with `list_rooms`, members with `list_room_members`, recent \ timeline with `read_room`. Room references accept ids (!abc:server) \ - or aliases (#name:server); user references use @user:server.")] + or aliases (#name:server); user references use @user:server." +)] impl ServerHandler for MatrixBridge {} #[tokio::main] @@ -266,13 +258,7 @@ async fn main() -> Result<()> { } let bridge = MatrixBridge::new(); - let service = bridge - .serve(stdio()) - .await - .context("serve MCP over stdio")?; - service - .waiting() - .await - .context("MCP service exited unexpectedly")?; + let service = bridge.serve(stdio()).await.context("serve MCP over stdio")?; + service.waiting().await.context("MCP service exited unexpectedly")?; Ok(()) } diff --git a/hive-matrix-mcp/src/client.rs b/hive-matrix-mcp/src/client.rs index 97a6fc75..50bf92a3 100644 --- a/hive-matrix-mcp/src/client.rs +++ b/hive-matrix-mcp/src/client.rs @@ -57,7 +57,10 @@ pub async fn build_and_restore( .trim() .to_owned(); if token.is_empty() { - return Err(anyhow!("matrix token at {} is empty", token_file.display())); + return Err(anyhow!( + "matrix token at {} is empty", + token_file.display() + )); } let (user_id, device_id) = whoami(homeserver, &token).await?; diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index 98026452..387b4726 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -16,8 +16,8 @@ use matrix_sdk::{ OwnedEventId, OwnedRoomId, OwnedUserId, RoomOrAliasId, api::client::receipt::create_receipt::v3::ReceiptType, events::{ - reaction::ReactionEventContent, receipt::ReceiptThread, + reaction::ReactionEventContent, relation::Annotation, room::message::{MessageType, RoomMessageEventContent}, }, @@ -64,20 +64,17 @@ async fn resolve_room( client: &Client, reference: &str, ) -> Result { - let parsed: &RoomOrAliasId = reference - .try_into() - .map_err(|e| DaemonResponse::error(format!("invalid room reference {reference}: {e}")))?; + let parsed: &RoomOrAliasId = reference.try_into().map_err(|e| { + DaemonResponse::error(format!("invalid room reference {reference}: {e}")) + })?; let room_id: OwnedRoomId = if parsed.is_room_id() { OwnedRoomId::try_from(reference) .map_err(|e| DaemonResponse::error(format!("invalid room_id: {e}")))? } else { client - .resolve_room_alias( - parsed - .as_str() - .try_into() - .map_err(|e| DaemonResponse::error(format!("invalid alias: {e}")))?, - ) + .resolve_room_alias(parsed.as_str().try_into().map_err(|e| { + DaemonResponse::error(format!("invalid alias: {e}")) + })?) .await .map(|r| r.room_id) .map_err(|e| DaemonResponse::error(format!("resolve_room_alias {reference}: {e}")))? @@ -96,14 +93,16 @@ async fn resolve_room( fn extract_body(event: &matrix_sdk::ruma::events::AnyTimelineEvent) -> String { use matrix_sdk::ruma::events::{AnyMessageLikeEvent, AnyTimelineEvent}; match event { - AnyTimelineEvent::MessageLike(AnyMessageLikeEvent::RoomMessage(ev)) => ev - .as_original() - .map_or_else(String::new, |orig| match &orig.content.msgtype { - MessageType::Text(t) => t.body.clone(), - MessageType::Notice(n) => n.body.clone(), - MessageType::Emote(e) => format!("* {}", e.body), - _ => String::new(), - }), + AnyTimelineEvent::MessageLike(AnyMessageLikeEvent::RoomMessage(ev)) => { + ev.as_original().map_or_else(String::new, |orig| { + match &orig.content.msgtype { + MessageType::Text(t) => t.body.clone(), + MessageType::Notice(n) => n.body.clone(), + MessageType::Emote(e) => format!("* {}", e.body), + _ => String::new(), + } + }) + } _ => String::new(), } } @@ -129,13 +128,14 @@ pub async fn send_dm(client: &Client, user_id: &str, body: &str) -> DaemonRespon Err(e) => return DaemonResponse::error(format!("invalid user_id {user_id}: {e}")), }; // Find existing DM or create one. - let room = client.joined_rooms().into_iter().find(|r| { - // is_direct() is async; check direct_targets() instead which - // reads from cached state. - r.direct_targets() - .iter() - .any(|t| t.as_str() == uid.as_str()) - }); + let room = client + .joined_rooms() + .into_iter() + .find(|r| { + // is_direct() is async; check direct_targets() instead which + // reads from cached state. + r.direct_targets().iter().any(|t| t.as_str() == uid.as_str()) + }); let room = match room { Some(r) => r, None => match client.create_dm(&uid).await { @@ -268,18 +268,16 @@ pub async fn list_room_members(client: &Client, room_ref: &str) -> DaemonRespons } pub async fn read_room(client: &Client, room_ref: &str, limit: Option) -> DaemonResponse { - use matrix_sdk::ruma::api::Direction; use matrix_sdk::ruma::api::client::message::get_message_events; + use matrix_sdk::ruma::api::Direction; let room = match resolve_room(client, room_ref).await { Ok(r) => r, Err(e) => return e, }; let limit = limit.unwrap_or(50).min(200); - let mut req = - get_message_events::v3::Request::new(room.room_id().to_owned(), Direction::Backward); - req.limit = matrix_sdk::ruma::UInt::try_from(limit as u64) - .unwrap_or(matrix_sdk::ruma::UInt::from(50u32)); + let mut req = get_message_events::v3::Request::new(room.room_id().to_owned(), Direction::Backward); + req.limit = matrix_sdk::ruma::UInt::try_from(limit as u64).unwrap_or(matrix_sdk::ruma::UInt::from(50u32)); let resp = match client.send(req).await { Ok(r) => r, Err(e) => return DaemonResponse::error(format!("get_message_events: {e}")), diff --git a/hive-matrix-mcp/src/paths.rs b/hive-matrix-mcp/src/paths.rs index 39913087..3b020646 100644 --- a/hive-matrix-mcp/src/paths.rs +++ b/hive-matrix-mcp/src/paths.rs @@ -45,8 +45,7 @@ pub fn homeserver_url() -> String { /// `HIVE_MATRIX_SOCKET`; default is `/run/hive-matrix/socket`. #[must_use] pub fn daemon_socket() -> PathBuf { - std::env::var_os("HIVE_MATRIX_SOCKET") - .map_or_else(|| PathBuf::from(DEFAULT_DAEMON_SOCKET), PathBuf::from) + std::env::var_os("HIVE_MATRIX_SOCKET").map_or_else(|| PathBuf::from(DEFAULT_DAEMON_SOCKET), PathBuf::from) } /// Persistent sqlite store directory for matrix-sdk's state (event @@ -63,6 +62,5 @@ pub fn matrix_state_dir() -> PathBuf { /// Mirrors the path `forge_notify` writes to. #[must_use] pub fn hyperhive_socket() -> PathBuf { - std::env::var_os("HIVE_CONTROL_SOCKET") - .map_or_else(|| PathBuf::from("/run/hive/mcp.sock"), PathBuf::from) + std::env::var_os("HIVE_CONTROL_SOCKET").map_or_else(|| PathBuf::from("/run/hive/mcp.sock"), PathBuf::from) } diff --git a/hive-matrix-mcp/src/wake.rs b/hive-matrix-mcp/src/wake.rs index 34e09dcd..39099157 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -84,10 +84,7 @@ mod tests { #[test] fn format_wake_body_short_passes_through() { let body = format_wake_body("@iris:matrix.darkest.space", "#general", "hi all"); - assert_eq!( - body, - "[matrix] @iris:matrix.darkest.space in #general: hi all" - ); + assert_eq!(body, "[matrix] @iris:matrix.darkest.space in #general: hi all"); } #[test] diff --git a/hive-priv/src/main.rs b/hive-priv/src/main.rs index da557ec6..5c727a97 100644 --- a/hive-priv/src/main.rs +++ b/hive-priv/src/main.rs @@ -17,13 +17,11 @@ //! `LISTEN_FDS=1` + `LISTEN_PID=`, the inherited fd 3 is used //! instead of binding a fresh socket. +use std::fmt::Write as _; use std::path::{Path, PathBuf}; use anyhow::{Context as _, Result, bail}; -use hive_sh4re::priv_proto::{ - AGENT_PREFIX, BindMount, MANAGER_NAME, META_DIR, PRIV_SOCK, PrivRequest, PrivResponse, - SIBLING_CONTAINERS, -}; +use hive_sh4re::priv_proto::{AGENT_PREFIX, MANAGER_NAME, META_DIR, PRIV_SOCK, SIBLING_CONTAINERS, BindMount, PrivRequest, PrivResponse}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; use tokio::process::Command; @@ -35,7 +33,8 @@ const SOCKET_DIR_ROOT: &str = "/run/hive-agent"; async fn main() -> Result<()> { tracing_subscriber::fmt() .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()), + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "info".into()), ) .init(); @@ -55,6 +54,7 @@ async fn main() -> Result<()> { } fn socket_listener() -> Result { + use std::os::unix::fs::PermissionsExt as _; // Socket activation: systemd passes the socket as fd 3 when // LISTEN_FDS >= 1 and LISTEN_PID matches our pid. let listen_fds: Option = std::env::var("LISTEN_FDS") @@ -64,32 +64,32 @@ fn socket_listener() -> Result { .ok() .and_then(|s| s.parse().ok()); - if let (Some(n), Some(p)) = (listen_fds, listen_pid) { - if n >= 1 && p == std::process::id() { - // SAFETY: systemd has passed us a ready UnixListener on fd 3. - let std_listener = unsafe { - use std::os::unix::io::FromRawFd; - std::os::unix::net::UnixListener::from_raw_fd(3) - }; - std_listener - .set_nonblocking(true) - .context("set socket non-blocking")?; - let listener = - tokio::net::UnixListener::from_std(std_listener).context("wrap systemd socket")?; - tracing::info!("using systemd-activated socket"); - return Ok(listener); - } + if let (Some(n), Some(p)) = (listen_fds, listen_pid) + && n >= 1 && p == std::process::id() + { + // SAFETY: systemd has passed us a ready UnixListener on fd 3. + let std_listener = unsafe { + use std::os::unix::io::FromRawFd; + std::os::unix::net::UnixListener::from_raw_fd(3) + }; + std_listener + .set_nonblocking(true) + .context("set socket non-blocking")?; + let listener = + tokio::net::UnixListener::from_std(std_listener).context("wrap systemd socket")?; + tracing::info!("using systemd-activated socket"); + return Ok(listener); } // Fallback: bind the socket ourselves. let path = Path::new(PRIV_SOCK); if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?; + std::fs::create_dir_all(parent) + .with_context(|| format!("create {}", parent.display()))?; } let _ = std::fs::remove_file(path); let listener = UnixListener::bind(path).with_context(|| format!("bind {PRIV_SOCK}"))?; // Mode 0660: only the hive-core group can connect. - use std::os::unix::fs::PermissionsExt as _; std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o660)) .context("chmod priv.sock")?; tracing::info!(path = PRIV_SOCK, "bound priv socket"); @@ -138,6 +138,7 @@ async fn dispatch(line: &str) -> PrivResponse { } /// Execute a validated `PrivRequest`. Returns `(stdout, stderr)` on success. +#[allow(clippy::too_many_lines)] async fn exec(req: PrivRequest) -> Result<(String, String)> { match req { PrivRequest::StartContainer { ref name } => { @@ -158,25 +159,13 @@ async fn exec(req: PrivRequest) -> Result<(String, String)> { PrivRequest::UpdateContainer { ref name } => { validate_container_name(name)?; let flake_ref = agent_flake_ref(name); - container_run(&[ - "update", - &container_system_name(name), - "--flake", - &flake_ref, - ]) - .await + container_run(&["update", &container_system_name(name), "--flake", &flake_ref]).await } PrivRequest::CreateContainer { ref name } => { validate_container_name(name)?; let flake_ref = agent_flake_ref(name); - container_run(&[ - "create", - &container_system_name(name), - "--flake", - &flake_ref, - ]) - .await + container_run(&["create", &container_system_name(name), "--flake", &flake_ref]).await } PrivRequest::DestroyContainer { ref name } => { @@ -186,10 +175,7 @@ async fn exec(req: PrivRequest) -> Result<(String, String)> { PrivRequest::ListContainers => container_run(&["list"]).await, - PrivRequest::WriteNspawnFlags { - ref container, - ref binds, - } => { + PrivRequest::WriteNspawnFlags { ref container, ref binds } => { validate_container_system_name(container)?; for bind in binds { validate_bind_path(&bind.host_path)?; @@ -217,7 +203,8 @@ async fn exec(req: PrivRequest) -> Result<(String, String)> { validate_container_system_name(container)?; let dir = format!("/run/systemd/system/container@{container}.service.d"); if Path::new(&dir).exists() { - std::fs::remove_dir_all(&dir).with_context(|| format!("remove {dir}"))?; + std::fs::remove_dir_all(&dir) + .with_context(|| format!("remove {dir}"))?; } Ok((String::new(), String::new())) } @@ -240,14 +227,7 @@ async fn exec(req: PrivRequest) -> Result<(String, String)> { PrivRequest::ReloadGatewayNginx => { let out = Command::new("systemd-run") - .args([ - "--machine=hive-gateway", - "--quiet", - "--", - "nginx", - "-s", - "reload", - ]) + .args(["--machine=hive-gateway", "--quiet", "--", "nginx", "-s", "reload"]) .output() .await .context("invoke systemd-run for gateway nginx reload")?; @@ -261,11 +241,7 @@ async fn exec(req: PrivRequest) -> Result<(String, String)> { Ok((String::new(), String::new())) } - PrivRequest::ChownSocketDir { - ref agent_name, - uid, - gid, - } => { + PrivRequest::ChownSocketDir { ref agent_name, uid, gid } => { validate_agent_name(agent_name)?; let path = socket_dir_path(agent_name); std::os::unix::fs::chown(&path, Some(uid), Some(gid)) @@ -273,13 +249,10 @@ async fn exec(req: PrivRequest) -> Result<(String, String)> { Ok((String::new(), String::new())) } - PrivRequest::ChmodSocketDir { - ref agent_name, - mode, - } => { + PrivRequest::ChmodSocketDir { ref agent_name, mode } => { + use std::os::unix::fs::PermissionsExt as _; validate_agent_name(agent_name)?; let path = socket_dir_path(agent_name); - use std::os::unix::fs::PermissionsExt as _; std::fs::set_permissions(&path, std::fs::Permissions::from_mode(mode)) .with_context(|| format!("chmod {:o} {}", mode, path.display()))?; Ok((String::new(), String::new())) @@ -388,9 +361,7 @@ fn agent_flake_ref(name: &str) -> String { fn validate_bind_path(path: &str) -> Result<()> { if path.is_empty() || !path.starts_with('/') - || path - .bytes() - .any(|b| b == 0 || b == b'\n' || b == b'"' || b == b':') + || path.bytes().any(|b| b == 0 || b == b'\n' || b == b'"' || b == b':') { bail!( "invalid bind path {path:?}: must be an absolute path with no colons, newlines, null bytes, or double-quotes" @@ -405,7 +376,8 @@ fn validate_bind_path(path: &str) -> Result<()> { /// then appends `EXTRA_NSPAWN_FLAGS=""`. fn write_nspawn_flags(container: &str, binds: &[BindMount]) -> Result<()> { let path = format!("/etc/nixos-containers/{container}.conf"); - let original = std::fs::read_to_string(&path).with_context(|| format!("read {path}"))?; + let original = std::fs::read_to_string(&path) + .with_context(|| format!("read {path}"))?; let lines: Vec<&str> = original .lines() .filter(|line| { @@ -429,14 +401,11 @@ fn write_nspawn_flags(container: &str, binds: &[BindMount]) -> Result<()> { out.push_str("HOST_ADDRESS6=\n"); out.push_str("LOCAL_ADDRESS6=\n"); out.push_str("HOST_BRIDGE=\n"); - let flags: Vec = binds - .iter() - .map(|b| { - let flag = if b.read_only { "--bind-ro" } else { "--bind" }; - format!("{flag}={}:{}", b.host_path, b.container_path) - }) - .collect(); + let flags: Vec = binds.iter().map(|b| { + let flag = if b.read_only { "--bind-ro" } else { "--bind" }; + format!("{flag}={}:{}", b.host_path, b.container_path) + }).collect(); let flags_joined = flags.join(" "); - out.push_str(&format!("EXTRA_NSPAWN_FLAGS=\"{flags_joined}\"\n")); + writeln!(out, "EXTRA_NSPAWN_FLAGS=\"{flags_joined}\"").unwrap(); std::fs::write(&path, out).with_context(|| format!("write {path}")) } diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index d0b53c04..31b0576c 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -463,6 +463,7 @@ pub enum Request { }, // ---- privileged (manager socket only for now) --------------------------- + /// *(privileged)* Initialise a brand-new agent's proposed config repo /// and queue an approval for the operator to review. RequestInitConfig { @@ -723,6 +724,7 @@ pub enum HelperEvent { }, } + /// Submission payload for `RequestSchedulePrompt`. Lives outside the /// enum so it can also serialize into the approval row's `commit_ref` /// (the dispatcher re-parses it on approve and inserts the schedule). @@ -974,3 +976,4 @@ pub struct WireScheduleTarget { #[serde(default, skip_serializing_if = "Option::is_none")] pub last_result: Option, } + diff --git a/hive-sh4re/src/priv_proto.rs b/hive-sh4re/src/priv_proto.rs index 2bc9b87e..37ebf686 100644 --- a/hive-sh4re/src/priv_proto.rs +++ b/hive-sh4re/src/priv_proto.rs @@ -42,6 +42,7 @@ pub struct BindMount { #[serde(tag = "op", rename_all = "snake_case")] pub enum PrivRequest { // --- Container lifecycle --- + /// `nixos-container start ` StartContainer { name: String }, @@ -66,13 +67,11 @@ pub enum PrivRequest { ListContainers, // --- Config file writes --- + /// Update `/etc/nixos-containers/.conf`: strip network-isolation /// vars, force `PRIVATE_NETWORK=0`, and set `EXTRA_NSPAWN_FLAGS` from the /// provided bind-mount list. Written by `lifecycle::set_nspawn_flags`. - WriteNspawnFlags { - container: String, - binds: Vec, - }, + WriteNspawnFlags { container: String, binds: Vec }, /// Write `/run/systemd/system/container@.service.d/hyperhive-limits.conf` /// with `[Service]\nMemoryMax=\nCPUQuota=\n`. @@ -88,6 +87,7 @@ pub enum PrivRequest { RemoveServiceDropin { container: String }, // --- System --- + /// Run `systemctl daemon-reload`. DaemonReload, @@ -96,13 +96,10 @@ pub enum PrivRequest { ReloadGatewayNginx, // --- Socket dir ownership --- + /// Set ownership of `/run/hive-agent//` to `uid:gid`. /// Called by `lifecycle::set_nspawn_flags` after `create_dir_all`. - ChownSocketDir { - agent_name: String, - uid: u32, - gid: u32, - }, + ChownSocketDir { agent_name: String, uid: u32, gid: u32 }, /// Set mode of `/run/hive-agent//`. /// Fallback when uid lookup returns `None` on first spawn. diff --git a/nix/modules/hive-gateway.nix b/nix/modules/hive-gateway.nix index 9925aeb9..ef7c5c28 100644 --- a/nix/modules/hive-gateway.nix +++ b/nix/modules/hive-gateway.nix @@ -198,50 +198,6 @@ in ''; }; - auth = { - enable = lib.mkEnableOption '' - HTTP basic auth on the gateway using an htpasswd file. When - enabled, every request to the gateway's main vhost requires a - valid username and password from the htpasswd file at - `services.hyperhive.gateway.auth.htpasswdFile`. nginx's built-in - `auth_basic` module handles credential validation — no extra - service or host-side daemon required. Off by default. - - Manage users with `hivectl gateway create-user`, `delete-user`, - and `list-users` — see `hivectl gateway --help` for usage. - ''; - - htpasswdFile = lib.mkOption { - type = lib.types.nullOr lib.types.path; - default = null; - example = "/etc/hyperhive/gateway.htpasswd"; - description = '' - Path on the **host** to an htpasswd-format file whose - `username:hashed-password` entries nginx uses for Basic auth. - The parent directory is bind-mounted read-only into the gateway - container at `/run/gateway-auth/`. The file must be readable by - the `nginx` user inside the container (mode 0644 recommended). - - Manage with: `hivectl gateway create-user --file `. - BCrypt (cost 12) is used by default; no external `htpasswd` binary - required. - - Required when `enable = true`. - ''; - }; - - realm = lib.mkOption { - type = lib.types.strMatching "[^\"$]*"; - default = "hyperhive"; - example = "my-hive"; - description = '' - HTTP Basic auth `realm` value sent in the `WWW-Authenticate` - header when credentials are absent or rejected. Must not - contain `"` or `$` (nginx string metacharacters). - ''; - }; - }; - }; config = lib.mkIf cfg.enable { @@ -254,14 +210,6 @@ in or leave `localHostsEntry` at its default of false. ''; } - { - assertion = !cfg.auth.enable || cfg.auth.htpasswdFile != null; - message = '' - services.hyperhive.gateway.auth.enable = true requires - services.hyperhive.gateway.auth.htpasswdFile to be set. - Create an htpasswd file with: hivectl gateway create-user --file /path/to/file - ''; - } ]; # Ensure bind-mount sources exist at host boot before the gateway @@ -310,16 +258,6 @@ in hostPath = "/var/lib/hyperhive/gateway"; isReadOnly = true; }; - # When auth is enabled, bind-mount the parent directory of the - # htpasswd file read-only into the container at /run/gateway-auth/. - # nginx's `auth_basic_user_file` points at the file inside that dir. - # Using the parent directory (not the file itself) because nspawn - # bind-mounts need a pre-existing destination — binding a directory - # is always safe; nginx picks the file up by name inside. - bindMounts."/run/gateway-auth" = lib.mkIf (cfg.auth.enable && cfg.auth.htpasswdFile != null) { - hostPath = builtins.dirOf cfg.auth.htpasswdFile; - isReadOnly = true; - }; config = { pkgs, ... }: let @@ -573,20 +511,12 @@ in # headers stay set so SSE (`/dashboard/stream`, # `/events/stream`) + websocket (`/screen/ws`) # endpoints keep working transparently. - # When auth is enabled, nginx's built-in `auth_basic` - # validates credentials against the htpasswd file - # bind-mounted at `/run/gateway-auth/`. No extra - # service or host-side daemon required. "/" = { proxyPass = "http://${cfg.upstreamHost}:${toString cfg.upstreamPort}"; proxyWebsockets = true; extraConfig = '' proxy_buffering off; proxy_read_timeout 1d; - ${lib.optionalString (cfg.auth.enable && cfg.auth.htpasswdFile != null) '' - auth_basic "${cfg.auth.realm}"; - auth_basic_user_file /run/gateway-auth/${builtins.baseNameOf cfg.auth.htpasswdFile}; - ''} ''; }; };