diff --git a/Cargo.lock b/Cargo.lock index 73b73db2..42d70f8e 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", + "crypto-common 0.1.7", "generic-array", ] @@ -25,7 +25,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" dependencies = [ "cfg-if", - "cipher", + "cipher 0.4.4", "cpufeatures 0.2.17", ] @@ -303,6 +303,19 @@ 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" @@ -350,12 +363,28 @@ 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" @@ -374,7 +403,7 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "26b52a9543ae338f279b96b0b9fed9c8093744685043739079ce85cd58f289a6" dependencies = [ - "cipher", + "cipher 0.4.4", ] [[package]] @@ -406,7 +435,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3613f74bd2eac03dad61bd53dbe620703d4371614fe0bc3b9f04dd36fe4e818" dependencies = [ "cfg-if", - "cipher", + "cipher 0.4.4", "cpufeatures 0.2.17", ] @@ -418,7 +447,7 @@ checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35" dependencies = [ "aead", "chacha20", - "cipher", + "cipher 0.4.4", "poly1305", "zeroize", ] @@ -443,11 +472,21 @@ version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ - "crypto-common", - "inout", + "crypto-common 0.1.7", + "inout 0.1.4", "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" @@ -591,6 +630,15 @@ 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" @@ -742,7 +790,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer", - "crypto-common", + "crypto-common 0.1.7", "subtle", ] @@ -1200,6 +1248,7 @@ dependencies = [ "anyhow", "axum", "base64", + "bcrypt", "clap", "hive-sh4re", "listenfd", @@ -1353,6 +1402,15 @@ 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" @@ -1618,6 +1676,15 @@ 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" @@ -3740,7 +3807,7 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" dependencies = [ - "crypto-common", + "crypto-common 0.1.7", "subtle", ] diff --git a/Cargo.toml b/Cargo.toml index b8ac38ca..cfcfd132 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,13 @@ [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" @@ -18,6 +25,7 @@ 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"] } @@ -45,6 +53,13 @@ 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 5d8631d5..ebb0cf2f 100644 --- a/docs/gateway.md +++ b/docs/gateway.md @@ -356,16 +356,27 @@ services.hyperhive.gateway.auth = { }; ``` -Create the htpasswd file on the host: +Manage users with `hivectl gateway`: ```sh -# Create new file with first user (BCrypt, recommended): -htpasswd -Bc /etc/hyperhive/gateway.htpasswd alice +# Add or update a user (prompted for password): +hivectl gateway create-user --file /etc/hyperhive/gateway.htpasswd alice --password-stdin -# Add subsequent users: -htpasswd -B /etc/hyperhive/gateway.htpasswd bob +# 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 diff --git a/hive-ag3nt/src/bash_runner.rs b/hive-ag3nt/src/bash_runner.rs index 60fd0179..b048a705 100644 --- a/hive-ag3nt/src/bash_runner.rs +++ b/hive-ag3nt/src/bash_runner.rs @@ -60,10 +60,7 @@ 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") } @@ -136,9 +133,8 @@ 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)?; @@ -229,7 +225,7 @@ async fn run_loop(socket: PathBuf) { let claimed: Arc>> = Arc::new(Mutex::new(HashSet::new())); loop { - poll_once(&socket, &claimed); + poll_once(&socket, &claimed).await; tokio::time::sleep(POLL_INTERVAL).await; } } @@ -237,14 +233,20 @@ 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; } @@ -258,14 +260,18 @@ async fn mark_interrupted(socket: &Path) { } } -fn poll_once(socket: &Path, claimed: &Arc>>) { - let Ok(rd) = std::fs::read_dir(tasks_dir()) else { return }; +async 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) { @@ -322,7 +328,11 @@ 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()); @@ -344,7 +354,12 @@ 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") @@ -396,7 +411,9 @@ 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") + } } } @@ -415,12 +432,7 @@ 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 d2ad6ce1..cf2d3c50 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -18,7 +18,9 @@ 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, }; @@ -129,7 +131,9 @@ 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 @@ -140,7 +144,11 @@ 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:#}") } @@ -202,9 +210,7 @@ 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 @@ -225,11 +231,8 @@ 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 ---------- @@ -271,13 +274,15 @@ 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 }, @@ -415,9 +420,7 @@ 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>( @@ -569,8 +572,7 @@ 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"); } @@ -658,7 +660,11 @@ 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(); @@ -670,7 +676,10 @@ 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) { @@ -697,8 +706,7 @@ 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 d511ab28..700c7fcc 100644 --- a/hive-ag3nt/src/events.rs +++ b/hive-ag3nt/src/events.rs @@ -80,12 +80,18 @@ 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()) - && 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); + 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); + } } // 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 0784aaea..829e3584 100644 --- a/hive-ag3nt/src/forge_notify.rs +++ b/hive-ag3nt/src/forge_notify.rs @@ -244,10 +244,7 @@ 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| { @@ -914,7 +911,10 @@ 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 14acb7f2..56759aee 100644 --- a/hive-ag3nt/src/identity.rs +++ b/hive-ag3nt/src/identity.rs @@ -122,7 +122,9 @@ 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 f9b26d58..4d79930a 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -437,31 +437,30 @@ 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 - && let Some(s) = task.started_at - { - let _ = write!(out, ", took {}s", t - 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(ref stdout) = task.stdout_tail - && !stdout.trim().is_empty() - { - let _ = write!(out, "\n\nstdout:\n```\n{}\n```", stdout.trim()); + 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 stderr) = task.stderr_tail - && !stderr.trim().is_empty() - { - let _ = write!(out, "\n\nstderr:\n```\n{}\n```", stderr.trim()); + if let Some(ref stderr) = task.stderr_tail { + if !stderr.trim().is_empty() { + let _ = write!(out, "\n\nstderr:\n```\n{}\n```", stderr.trim()); + } } out } @@ -668,7 +667,9 @@ 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. @@ -678,8 +679,11 @@ 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 @@ -830,9 +834,11 @@ 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 } @@ -875,10 +881,7 @@ 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 @@ -1913,14 +1916,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 @@ -1976,12 +1979,13 @@ 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). - 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"); + 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 groups.is_empty() { diff --git a/hive-ag3nt/src/stats.rs b/hive-ag3nt/src/stats.rs index ae98ca7e..301d1c50 100644 --- a/hive-ag3nt/src/stats.rs +++ b/hive-ag3nt/src/stats.rs @@ -391,7 +391,11 @@ 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; @@ -471,7 +475,15 @@ 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(); } @@ -485,7 +497,14 @@ 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 a513ab5a..6e588f29 100644 --- a/hive-ag3nt/src/web_ui.rs +++ b/hive-ag3nt/src/web_ui.rs @@ -561,7 +561,9 @@ 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(), }) @@ -613,8 +615,7 @@ 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) @@ -662,7 +663,10 @@ 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 { @@ -956,7 +960,9 @@ 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 e0ff4406..3446e8f2 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -9,6 +9,7 @@ workspace = true anyhow.workspace = true axum.workspace = true base64.workspace = true +bcrypt.workspace = true reqwest.workspace = true clap.workspace = true hive-sh4re.workspace = true diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index dfdb298c..1905ac0b 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -76,8 +76,7 @@ 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 f86ae25f..de1032c4 100644 --- a/hive-c0re/src/agent_ports.rs +++ b/hive-c0re/src/agent_ports.rs @@ -40,8 +40,7 @@ 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 @@ -61,12 +60,10 @@ 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)", @@ -132,9 +129,6 @@ 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 c220f020..92b7f6cf 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -129,9 +129,11 @@ 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 @@ -223,8 +225,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 { @@ -242,11 +244,9 @@ 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,7 +294,15 @@ 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. @@ -331,7 +339,10 @@ 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) { @@ -547,10 +558,7 @@ 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 \ @@ -604,8 +612,9 @@ 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() @@ -634,19 +643,26 @@ 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 d7edbfd6..a18a1bbd 100644 --- a/hive-c0re/src/agent_sockets.rs +++ b/hive-c0re/src/agent_sockets.rs @@ -81,8 +81,7 @@ 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() }) } @@ -146,12 +145,10 @@ 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)", @@ -305,24 +302,30 @@ 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] @@ -332,10 +335,12 @@ 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 1ff3d67b..856b1caf 100644 --- a/hive-c0re/src/auto_update.rs +++ b/hive-c0re/src/auto_update.rs @@ -209,16 +209,19 @@ 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 [String], topo: &std::collections::BTreeMap>) { +pub fn topology_sort( + names: &mut Vec, + 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.to_vec(); + let name_set: Vec = names.clone(); 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).is_none_or(Option::is_none) { + if topo.get(name).map_or(true, |p| p.is_none()) { depth.insert(name.clone(), 0); queue.push_back(name.clone()); } @@ -300,4 +303,3 @@ 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 402dafb7..8bdb3a00 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_hours(1); +const VACUUM_INTERVAL: Duration = Duration::from_secs(3600); /// Keep completed task files for 48 hours before sweeping them. const KEEP_SECS: i64 = 48 * 3600; @@ -65,7 +65,9 @@ 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(); @@ -97,7 +99,10 @@ 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(serde_json::Value::as_i64).unwrap_or(i64::MAX); + let completed_at = v + .get("completed_at") + .and_then(|t| t.as_i64()) + .unwrap_or(i64::MAX); completed_at < cutoff } @@ -106,10 +111,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() - && let Err(e) = std::fs::remove_file(&path) - { - tracing::warn!(path = %path.display(), error = ?e, "bash-tasks vacuum: remove failed"); + if path.exists() { + if 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 f7632820..f5734b8b 100644 --- a/hive-c0re/src/bin/hivectl.rs +++ b/hive-c0re/src/bin/hivectl.rs @@ -15,6 +15,8 @@ //! 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; @@ -54,6 +56,14 @@ 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)] @@ -137,6 +147,48 @@ 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() @@ -161,6 +213,16 @@ 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), + }, } } @@ -173,11 +235,7 @@ 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" @@ -274,13 +332,18 @@ 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 { @@ -293,3 +356,115 @@ 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 e28a698e..60f3dc1a 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -268,12 +268,7 @@ 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 @@ -391,7 +386,13 @@ 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); @@ -673,7 +674,11 @@ 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() @@ -740,9 +745,7 @@ 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", @@ -862,7 +865,9 @@ 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, @@ -1029,10 +1034,7 @@ 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); } @@ -1275,4 +1277,3 @@ 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 f518be59..5b3cdf31 100644 --- a/hive-c0re/src/build_logs.rs +++ b/hive-c0re/src/build_logs.rs @@ -274,14 +274,8 @@ 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, @@ -296,11 +290,7 @@ 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( @@ -310,7 +300,10 @@ 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?); @@ -470,7 +463,9 @@ 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"); @@ -509,8 +504,7 @@ 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 214d08cc..378d05ca 100644 --- a/hive-c0re/src/container_view.rs +++ b/hive-c0re/src/container_view.rs @@ -115,7 +115,9 @@ 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 @@ -143,27 +145,40 @@ 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, @@ -217,12 +232,18 @@ 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")) - && 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); + 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); + } } // Legacy fallback: presence of individual sentinel files. let rate_limited = dir.join("hyperhive-rate-limited").exists(); @@ -249,14 +270,23 @@ 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) { @@ -304,9 +334,7 @@ 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")) } @@ -321,11 +349,8 @@ 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", @@ -409,14 +434,26 @@ 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] @@ -426,7 +463,10 @@ 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 5bafe818..7bea60bc 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -708,11 +708,16 @@ 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 b66586da..d4e79eb0 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -472,9 +472,15 @@ 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(), }) } @@ -1751,9 +1757,7 @@ 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:#}")), } } @@ -1870,10 +1874,7 @@ 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; @@ -1897,9 +1898,7 @@ async fn get_build_log_raw( ) .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:#}")), } } @@ -2522,7 +2521,10 @@ 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()); @@ -130,10 +128,7 @@ 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(); } @@ -259,12 +254,7 @@ 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 9f48939b..509f521c 100644 --- a/hive-c0re/src/forge.rs +++ b/hive-c0re/src/forge.rs @@ -217,7 +217,14 @@ 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}"))?; @@ -551,9 +558,8 @@ 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 85b4c811..833cd7c8 100644 --- a/hive-c0re/src/lib.rs +++ b/hive-c0re/src/lib.rs @@ -14,15 +14,14 @@ 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; @@ -32,6 +31,7 @@ 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 a2029253..516ae0db 100644 --- a/hive-c0re/src/lifecycle.rs +++ b/hive-c0re/src/lifecycle.rs @@ -449,11 +449,12 @@ 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(()) } @@ -494,9 +495,7 @@ 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![ @@ -1135,8 +1134,7 @@ 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"); } @@ -1337,9 +1335,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}"), } } @@ -1443,4 +1441,3 @@ mod tests { ); } } - diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index 6cd903de..6def7c87 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::{MANAGER_AGENT, LooseEnd}; +use hive_sh4re::{LooseEnd, MANAGER_AGENT}; use crate::coordinator::Coordinator; diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index f4ee0b62..6d149abf 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, broker, client, crash_watch, dashboard, dashboard_events, - bash_tasks_vacuum, events_vacuum, forge, manager_server, matrix, migrate, rebuild_queue, + agent_sockets, auto_update, bash_tasks_vacuum, broker, client, crash_watch, dashboard, + dashboard_events, events_vacuum, forge, manager_server, matrix, migrate, rebuild_queue, reminder_scheduler, scheduled_prompts_worker, server, stats_vacuum, }; @@ -51,7 +51,10 @@ 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 @@ -119,7 +122,17 @@ 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?) } @@ -148,11 +161,7 @@ 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?, ) } } @@ -169,9 +178,8 @@ 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, @@ -335,7 +343,14 @@ 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(), @@ -348,7 +363,14 @@ 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 7810ddfa..8bd21f32 100644 --- a/hive-c0re/src/meta.rs +++ b/hive-c0re/src/meta.rs @@ -572,7 +572,10 @@ 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 29f7e569..8b37643d 100644 --- a/hive-c0re/src/migrate.rs +++ b/hive-c0re/src/migrate.rs @@ -147,7 +147,9 @@ 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") + } } } } @@ -190,7 +192,11 @@ 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"); @@ -218,12 +224,20 @@ 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 5440c115..428145e1 100644 --- a/hive-c0re/src/operator_questions.rs +++ b/hive-c0re/src/operator_questions.rs @@ -209,11 +209,7 @@ 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( @@ -232,9 +228,7 @@ 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( @@ -289,7 +283,6 @@ 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 b4e65a12..81ef6b44 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::{PRIV_SOCK, BindMount, PrivRequest, PrivResponse}; +use hive_sh4re::priv_proto::{BindMount, PRIV_SOCK, PrivRequest, PrivResponse}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; @@ -31,27 +31,49 @@ 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 { @@ -63,7 +85,8 @@ 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( @@ -75,13 +98,15 @@ 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<()> { @@ -97,21 +122,26 @@ 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 64ca7218..35eaa1c8 100644 --- a/hive-c0re/src/questions.rs +++ b/hive-c0re/src/questions.rs @@ -236,8 +236,7 @@ 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 fd162e9e..fe8d090f 100644 --- a/hive-c0re/src/rebuild_queue.rs +++ b/hive-c0re/src/rebuild_queue.rs @@ -1228,4 +1228,3 @@ mod tests { ); } } - diff --git a/hive-c0re/src/reminder_scheduler.rs b/hive-c0re/src/reminder_scheduler.rs index 2f25ff71..a626d3d3 100644 --- a/hive-c0re/src/reminder_scheduler.rs +++ b/hive-c0re/src/reminder_scheduler.rs @@ -130,8 +130,7 @@ 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 1028f231..03440cef 100644 --- a/hive-c0re/src/scheduled_prompts_worker.rs +++ b/hive-c0re/src/scheduled_prompts_worker.rs @@ -101,7 +101,10 @@ 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, @@ -384,4 +387,3 @@ 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 54e7ec44..7072a319 100644 --- a/hive-c0re/src/stats_vacuum.rs +++ b/hive-c0re/src/stats_vacuum.rs @@ -39,8 +39,7 @@ 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; } @@ -60,7 +59,9 @@ 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 d02d30bb..7f3e319f 100644 --- a/hive-c0re/src/topology.rs +++ b/hive-c0re/src/topology.rs @@ -59,10 +59,7 @@ 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) { @@ -90,7 +87,13 @@ 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() } @@ -501,8 +504,12 @@ 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}"); } @@ -678,20 +685,32 @@ 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 @@ -701,7 +720,10 @@ 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(); @@ -727,7 +749,10 @@ 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 def1915e..3b4eb86c 100644 --- a/hive-forge/Cargo.toml +++ b/hive-forge/Cargo.toml @@ -10,7 +10,12 @@ 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 39341ade..0025c572 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 6e7d47e4..cff4c592 100644 --- a/hive-forge/src/client.rs +++ b/hive-forge/src/client.rs @@ -180,7 +180,12 @@ 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 f8dcd7a5..07b7ae25 100644 --- a/hive-forge/src/main.rs +++ b/hive-forge/src/main.rs @@ -103,8 +103,7 @@ 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 944614d0..73c1a0bc 100644 --- a/hive-forge/src/verbs/comments.rs +++ b/hive-forge/src/verbs/comments.rs @@ -105,10 +105,7 @@ 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 { @@ -218,7 +221,9 @@ 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 @@ -346,10 +351,7 @@ 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 a0346750..20db93b1 100644 --- a/hive-forge/src/verbs/labels.rs +++ b/hive-forge/src/verbs/labels.rs @@ -36,7 +36,8 @@ 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 } => { @@ -64,7 +65,8 @@ 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 fecc5036..49a78bc9 100644 --- a/hive-forge/src/verbs/list.rs +++ b/hive-forge/src/verbs/list.rs @@ -167,9 +167,7 @@ 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 5ce9000a..ee92a28f 100644 --- a/hive-forge/src/verbs/milestone.rs +++ b/hive-forge/src/verbs/milestone.rs @@ -41,8 +41,7 @@ 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 92d1376f..a537a1c4 100644 --- a/hive-forge/src/verbs/pr_create.rs +++ b/hive-forge/src/verbs/pr_create.rs @@ -222,7 +222,9 @@ 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 8fc16d63..f204f10f 100644 --- a/hive-forge/src/verbs/subscription.rs +++ b/hive-forge/src/verbs/subscription.rs @@ -29,7 +29,11 @@ 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 6c7cb288..e9c7cca6 100644 --- a/hive-forge/src/verbs/timeline.rs +++ b/hive-forge/src/verbs/timeline.rs @@ -170,7 +170,10 @@ 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() @@ -217,7 +220,10 @@ 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 427f30db..38f4f715 100644 --- a/hive-matrix-mcp/src/bin/mcp.rs +++ b/hive-matrix-mcp/src/bin/mcp.rs @@ -139,21 +139,25 @@ 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( @@ -162,35 +166,40 @@ 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( @@ -209,28 +218,27 @@ 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] @@ -258,7 +266,13 @@ 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 50bf92a3..97a6fc75 100644 --- a/hive-matrix-mcp/src/client.rs +++ b/hive-matrix-mcp/src/client.rs @@ -57,10 +57,7 @@ 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 387b4726..98026452 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::{ - receipt::ReceiptThread, reaction::ReactionEventContent, + receipt::ReceiptThread, relation::Annotation, room::message::{MessageType, RoomMessageEventContent}, }, @@ -64,17 +64,20 @@ 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}")))? @@ -93,16 +96,14 @@ 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(), } } @@ -128,14 +129,13 @@ 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,16 +268,18 @@ 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::client::message::get_message_events; use matrix_sdk::ruma::api::Direction; + use matrix_sdk::ruma::api::client::message::get_message_events; 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 3b020646..39913087 100644 --- a/hive-matrix-mcp/src/paths.rs +++ b/hive-matrix-mcp/src/paths.rs @@ -45,7 +45,8 @@ 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 @@ -62,5 +63,6 @@ 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 39099157..34e09dcd 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -84,7 +84,10 @@ 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 5c727a97..da557ec6 100644 --- a/hive-priv/src/main.rs +++ b/hive-priv/src/main.rs @@ -17,11 +17,13 @@ //! `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, MANAGER_NAME, META_DIR, PRIV_SOCK, SIBLING_CONTAINERS, BindMount, PrivRequest, PrivResponse}; +use hive_sh4re::priv_proto::{ + AGENT_PREFIX, BindMount, MANAGER_NAME, META_DIR, PRIV_SOCK, PrivRequest, PrivResponse, + SIBLING_CONTAINERS, +}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; use tokio::process::Command; @@ -33,8 +35,7 @@ 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(); @@ -54,7 +55,6 @@ 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) - && 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) { + 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); + } } // 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,7 +138,6 @@ 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 } => { @@ -159,13 +158,25 @@ 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 } => { @@ -175,7 +186,10 @@ 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)?; @@ -203,8 +217,7 @@ 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())) } @@ -227,7 +240,14 @@ 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")?; @@ -241,7 +261,11 @@ 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)) @@ -249,10 +273,13 @@ async fn exec(req: PrivRequest) -> Result<(String, String)> { Ok((String::new(), String::new())) } - PrivRequest::ChmodSocketDir { ref agent_name, mode } => { - use std::os::unix::fs::PermissionsExt as _; + PrivRequest::ChmodSocketDir { + ref agent_name, + mode, + } => { 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())) @@ -361,7 +388,9 @@ 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" @@ -376,8 +405,7 @@ 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| { @@ -401,11 +429,14 @@ 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(" "); - writeln!(out, "EXTRA_NSPAWN_FLAGS=\"{flags_joined}\"").unwrap(); + out.push_str(&format!("EXTRA_NSPAWN_FLAGS=\"{flags_joined}\"\n")); 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 31b0576c..d0b53c04 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -463,7 +463,6 @@ 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 { @@ -724,7 +723,6 @@ 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). @@ -976,4 +974,3 @@ 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 37ebf686..2bc9b87e 100644 --- a/hive-sh4re/src/priv_proto.rs +++ b/hive-sh4re/src/priv_proto.rs @@ -42,7 +42,6 @@ pub struct BindMount { #[serde(tag = "op", rename_all = "snake_case")] pub enum PrivRequest { // --- Container lifecycle --- - /// `nixos-container start ` StartContainer { name: String }, @@ -67,11 +66,13 @@ 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`. @@ -87,7 +88,6 @@ pub enum PrivRequest { RemoveServiceDropin { container: String }, // --- System --- - /// Run `systemctl daemon-reload`. DaemonReload, @@ -96,10 +96,13 @@ 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 995edf49..a70daee2 100644 --- a/nix/modules/hive-gateway.nix +++ b/nix/modules/hive-gateway.nix @@ -213,7 +213,8 @@ in ''; htpasswdFile = lib.mkOption { - type = lib.types.path; + 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 @@ -254,16 +255,15 @@ in ''; } { - assertion = !cfg.auth.enable || cfg.auth ? htpasswdFile; + 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: htpasswd -Bc /path/to/file + Create an htpasswd file with: hivectl gateway create-user --file /path/to/file ''; } ]; - # Ensure bind-mount sources exist at host boot before the gateway # container's first start. nspawn would auto-create missing dirs # tmpfiles rules make the intent explicit @@ -316,7 +316,7 @@ in # 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 { + bindMounts."/run/gateway-auth" = lib.mkIf (cfg.auth.enable && cfg.auth.htpasswdFile != null) { hostPath = builtins.dirOf cfg.auth.htpasswdFile; isReadOnly = true; }; @@ -583,7 +583,7 @@ in extraConfig = '' proxy_buffering off; proxy_read_timeout 1d; - ${lib.optionalString cfg.auth.enable '' + ${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}; ''}