vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

feat(gateway): hivectl gateway user management + fix htpasswdFile assertion

Browse source 
Add `hivectl gateway {create-user,delete-user,list-users}` subcommands for
managing htpasswd files used by gateway Basic auth. Pure Rust bcrypt
(cost 12, $2y$ prefix nginx accepts). No external htpasswd binary required.

Also fix the NixOS module assertion: `cfg.auth ? htpasswdFile` is always
true in the module system (declared options always exist as keys); switch
to `nullOr path; default = null` + `!= null` check so the assertion
actually fires with a useful error when enable=true but no file is set.
Guard bind-mount and nginx config against null to prevent eval errors.

Update docs/gateway.md to show hivectl commands instead of raw htpasswd.
This commit is contained in:
atlas 2026-06-01 23:00:38 +02:00
commit 4bff450343
61 changed files with 1084 additions and 547 deletions
Download patch file Download diff file Expand all files Collapse all files

85
Cargo.lock generated
View file

@ -14,7 +14,7 @@ version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0" checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0"
dependencies = [ dependencies = [
"crypto-common", "crypto-common 0.1.7",
"generic-array", "generic-array",
] ]
@ -25,7 +25,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0"
dependencies = [ dependencies = [
"cfg-if", "cfg-if",
"cipher", "cipher 0.4.4",
"cpufeatures 0.2.17", "cpufeatures 0.2.17",
] ]
@ -303,6 +303,19 @@ version = "1.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" 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]] [[package]]
name = "bitflags" name = "bitflags"
version = "2.11.1" version = "2.11.1"
@ -350,12 +363,28 @@ dependencies = [
"generic-array", "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]] [[package]]
name = "bumpalo" name = "bumpalo"
version = "3.20.3" version = "3.20.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
[[package]]
name = "byteorder"
version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
[[package]] [[package]]
name = "bytes" name = "bytes"
version = "1.11.1" version = "1.11.1"
@ -374,7 +403,7 @@ version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "26b52a9543ae338f279b96b0b9fed9c8093744685043739079ce85cd58f289a6" checksum = "26b52a9543ae338f279b96b0b9fed9c8093744685043739079ce85cd58f289a6"
dependencies = [ dependencies = [
"cipher", "cipher 0.4.4",
] ]
[[package]] [[package]]
@ -406,7 +435,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c3613f74bd2eac03dad61bd53dbe620703d4371614fe0bc3b9f04dd36fe4e818" checksum = "c3613f74bd2eac03dad61bd53dbe620703d4371614fe0bc3b9f04dd36fe4e818"
dependencies = [ dependencies = [
"cfg-if", "cfg-if",
"cipher", "cipher 0.4.4",
"cpufeatures 0.2.17", "cpufeatures 0.2.17",
] ]
@ -418,7 +447,7 @@ checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35"
dependencies = [ dependencies = [
"aead", "aead",
"chacha20", "chacha20",
"cipher", "cipher 0.4.4",
"poly1305", "poly1305",
"zeroize", "zeroize",
] ]
@ -443,11 +472,21 @@ version = "0.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad"
dependencies = [ dependencies = [
"crypto-common", "crypto-common 0.1.7",
"inout", "inout 0.1.4",
"zeroize", "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]] [[package]]
name = "clap" name = "clap"
version = "4.6.1" version = "4.6.1"
@ -591,6 +630,15 @@ dependencies = [
"typenum", "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]] [[package]]
name = "curve25519-dalek" name = "curve25519-dalek"
version = "4.1.3" version = "4.1.3"
@ -742,7 +790,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
dependencies = [ dependencies = [
"block-buffer", "block-buffer",
"crypto-common", "crypto-common 0.1.7",
"subtle", "subtle",
] ]
@ -1200,6 +1248,7 @@ dependencies = [
"anyhow", "anyhow",
"axum", "axum",
"base64", "base64",
"bcrypt",
"clap", "clap",
"hive-sh4re", "hive-sh4re",
"listenfd", "listenfd",
@ -1353,6 +1402,15 @@ version = "1.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" 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]] [[package]]
name = "hyper" name = "hyper"
version = "1.10.1" version = "1.10.1"
@ -1618,6 +1676,15 @@ dependencies = [
"generic-array", "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]] [[package]]
name = "ipnet" name = "ipnet"
version = "2.12.0" version = "2.12.0"
@ -3740,7 +3807,7 @@ version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea"
dependencies = [ dependencies = [
"crypto-common", "crypto-common 0.1.7",
"subtle", "subtle",
] ]

21
Cargo.toml
View file

@ -1,6 +1,13 @@
[workspace] [workspace]
resolver = "3" 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] [workspace.package]
edition = "2024" edition = "2024"
@ -18,6 +25,7 @@ must_use_candidate = "allow"
anyhow = "1" anyhow = "1"
axum = { version = "0.8", features = ["ws"] } axum = { version = "0.8", features = ["ws"] }
base64 = "0.22" base64 = "0.22"
bcrypt = "0.19"
clap = { version = "4", features = ["derive"] } clap = { version = "4", features = ["derive"] }
hive-sh4re = { path = "hive-sh4re" } hive-sh4re = { path = "hive-sh4re" }
tower-http = { version = "0.6", features = ["fs"] } tower-http = { version = "0.6", features = ["fs"] }
@ -45,6 +53,13 @@ tokio = { version = "1", features = [
tokio-stream = { version = "0.1", features = ["sync"] } tokio-stream = { version = "0.1", features = ["sync"] }
tracing = "0.1" tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] }
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } reqwest = { version = "0.12", default-features = false, features = [
matrix-sdk = { version = "0.14", default-features = false, features = ["rustls-tls", "sqlite", "markdown"] } "json",
"rustls-tls",
] }
matrix-sdk = { version = "0.14", default-features = false, features = [
"rustls-tls",
"sqlite",
"markdown",
] }
futures-util = "0.3" futures-util = "0.3"

21
docs/gateway.md
View file

@ -356,16 +356,27 @@ services.hyperhive.gateway.auth = {
}; };
``` ```
Create the htpasswd file on the host: Manage users with `hivectl gateway`:
```sh ```sh
# Create new file with first user (BCrypt, recommended): # Add or update a user (prompted for password):
htpasswd -Bc /etc/hyperhive/gateway.htpasswd alice hivectl gateway create-user --file /etc/hyperhive/gateway.htpasswd alice --password-stdin
# Add subsequent users: # Add with inline password (visible in shell history — avoid for sensitive creds):
htpasswd -B /etc/hyperhive/gateway.htpasswd bob 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 The file must be readable by the `nginx` user inside the container
(`chmod 0644`). The module bind-mounts the file's parent directory (`chmod 0644`). The module bind-mounts the file's parent directory
read-only into the container at `/run/gateway-auth/`; nginx reads read-only into the container at `/run/gateway-auth/`; nginx reads

58
hive-ag3nt/src/bash_runner.rs
View file

@ -60,10 +60,7 @@ fn tasks_dir() -> PathBuf {
} else { } else {
// Pre-split fallback: derive harness/ as a sibling of state/. // Pre-split fallback: derive harness/ as a sibling of state/.
let state = crate::paths::state_dir(); let state = crate::paths::state_dir();
state state.parent().map(|p| p.join("harness")).unwrap_or(state)
.parent()
.map(|p| p.join("harness"))
.unwrap_or(state)
}; };
base.join("bash-tasks") base.join("bash-tasks")
} }
@ -136,9 +133,8 @@ impl TaskFile {
/// Write a task file atomically (tmp + rename). /// Write a task file atomically (tmp + rename).
fn write_task(task: &TaskFile) -> std::io::Result<()> { fn write_task(task: &TaskFile) -> std::io::Result<()> {
let json = serde_json::to_string_pretty(task).map_err(|e| { let json = serde_json::to_string_pretty(task)
std::io::Error::new(std::io::ErrorKind::InvalidData, e) .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
})?;
let dest = task_json(&task.id); let dest = task_json(&task.id);
let tmp = dest.with_extension("json.tmp"); let tmp = dest.with_extension("json.tmp");
std::fs::write(&tmp, json)?; std::fs::write(&tmp, json)?;
@ -229,7 +225,7 @@ async fn run_loop(socket: PathBuf) {
let claimed: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new())); let claimed: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new()));
loop { loop {
poll_once(&socket, &claimed); poll_once(&socket, &claimed).await;
tokio::time::sleep(POLL_INTERVAL).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 /// On boot, find any task files in `running` state and flip them to
/// `interrupted`, then fire a wake so the agent unblocks. /// `interrupted`, then fire a wake so the agent unblocks.
async fn mark_interrupted(socket: &Path) { 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() { for entry in rd.flatten() {
let path = entry.path(); let path = entry.path();
if path.extension().and_then(|e| e.to_str()) != Some("json") { if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue; 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 {
let Some(mut task) = read_task(&id) else { continue }; continue;
};
let Some(mut task) = read_task(&id) else {
continue;
};
if task.status != TaskStatus::Running { if task.status != TaskStatus::Running {
continue; continue;
} }
@ -258,14 +260,18 @@ async fn mark_interrupted(socket: &Path) {
} }
} }
fn poll_once(socket: &Path, claimed: &Arc<Mutex<HashSet<String>>>) { async fn poll_once(socket: &Path, claimed: &Arc<Mutex<HashSet<String>>>) {
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() { for entry in rd.flatten() {
let path = entry.path(); let path = entry.path();
if path.extension().and_then(|e| e.to_str()) != Some("json") { if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue; 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(); let guard = claimed.lock().unwrap();
if guard.contains(&id) { 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 stdout_tail = tail_file(&out_path, SUMMARY_BYTES);
let stderr_tail = tail_file(&err_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.completed_at = Some(crate::serve_common::now_unix());
task.exit_code = exit_code; task.exit_code = exit_code;
task.stdout_tail = stdout_tail.clone().filter(|s| !s.is_empty()); 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)`. /// Run `sh -c cmd`, streaming output to files. Returns `(exit_code, timed_out)`.
/// On timeout the child process is explicitly killed before returning. /// 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; use tokio::process::Command;
let mut child = Command::new("sh") let mut child = Command::new("sh")
.arg("-c") .arg("-c")
@ -396,7 +411,9 @@ where
let _ = tokio::io::copy(&mut reader, &mut f).await; let _ = tokio::io::copy(&mut reader, &mut f).await;
let _ = f.flush().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<String> {
// Wake delivery // Wake delivery
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
async fn send_wake( async fn send_wake(socket: &Path, id: &str, summary: &str, output: Option<(&str, &str)>) {
socket: &Path,
id: &str,
summary: &str,
output: Option<(&str, &str)>,
) {
let mut body = format!("bash task `{id}` finished: {summary}"); let mut body = format!("bash task `{id}` finished: {summary}");
if let Some((stdout, stderr)) = output { if let Some((stdout, stderr)) = output {
if !stdout.is_empty() { if !stdout.is_empty() {

62
hive-ag3nt/src/bin/hive.rs
View file

@ -18,7 +18,9 @@ use clap::{Parser, Subcommand};
use hive_ag3nt::events::{Bus, LiveEvent, TurnState}; use hive_ag3nt::events::{Bus, LiveEvent, TurnState};
use hive_ag3nt::login::{self, LoginState}; use hive_ag3nt::login::{self, LoginState};
use hive_ag3nt::turn_stats::TurnStats; 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::{ use hive_sh4re::{
AgentRequest, AgentResponse, HelperEvent, ManagerRequest, ManagerResponse, SYSTEM_SENDER, AgentRequest, AgentResponse, HelperEvent, ManagerRequest, ManagerResponse, SYSTEM_SENDER,
}; };
@ -129,7 +131,9 @@ fn log_system_event(bus: &Bus, from: &str, body: &str) {
} else { } else {
tracing::info!(%from, %body, "system message"); 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 /// 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. /// misconfigured harness still produces a parseable line.
fn format_turn_failure(err: &anyhow::Error) -> String { fn format_turn_failure(err: &anyhow::Error) -> String {
let who = hive_ag3nt::identity::qualified_label(); let who = hive_ag3nt::identity::qualified_label();
let who = if who.is_empty() { "<unknown>".to_owned() } else { who }; let who = if who.is_empty() {
"<unknown>".to_owned()
} else {
who
};
format!("[system] `{who}` claude turn failed:\n{err:#}") format!("[system] `{who}` claude turn failed:\n{err:#}")
} }
@ -202,9 +210,7 @@ trait Surface {
/// `(open_threads, open_reminders)` for the post-turn stats row. /// `(open_threads, open_reminders)` for the post-turn stats row.
/// Either field is `None` when the underlying request errors. /// Either field is `None` when the underlying request errors.
fn post_turn_counts( fn post_turn_counts(socket: &Path) -> impl Future<Output = (Option<u64>, Option<u64>)>;
socket: &Path,
) -> impl Future<Output = (Option<u64>, Option<u64>)>;
/// Send a message addressed to `<parent>` (broker resolves the /// Send a message addressed to `<parent>` (broker resolves the
/// sentinel via `topology::parent_of` at delivery time; root /// 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 /// by co-process daemons like matrix to push events into the
/// harness inbox). Errors out via `anyhow::bail!` so the calling /// harness inbox). Errors out via `anyhow::bail!` so the calling
/// binary surfaces them on stderr. /// binary surfaces them on stderr.
fn wake_external( fn wake_external(socket: &Path, from: String, body: String)
socket: &Path, -> impl Future<Output = Result<()>>;
from: String,
body: String,
) -> impl Future<Output = Result<()>>;
} }
// ---------- AgentSurface ---------- // ---------- AgentSurface ----------
@ -271,13 +274,15 @@ impl Surface for AgentSurface {
} }
async fn post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) { async fn post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) {
let threads = let threads = match client::request::<_, AgentResponse>(
match client::request::<_, AgentResponse>(socket, &AgentRequest::GetLooseEnds { agent: None }).await { socket,
Ok(AgentResponse::LooseEnds { loose_ends }) => { &AgentRequest::GetLooseEnds { agent: None },
u64::try_from(loose_ends.len()).ok() )
} .await
_ => None, {
}; Ok(AgentResponse::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(),
_ => None,
};
let reminders = match client::request::<_, AgentResponse>( let reminders = match client::request::<_, AgentResponse>(
socket, socket,
&AgentRequest::CountPendingReminders { agent: None }, &AgentRequest::CountPendingReminders { agent: None },
@ -415,9 +420,7 @@ impl Surface for ManagerSurface {
) )
.await .await
{ {
Ok(ManagerResponse::LooseEnds { loose_ends }) => { Ok(ManagerResponse::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(),
u64::try_from(loose_ends.len()).ok()
}
_ => None, _ => None,
}; };
let reminders = match client::request::<_, ManagerResponse>( let reminders = match client::request::<_, ManagerResponse>(
@ -569,8 +572,7 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
); );
tokio::spawn(async move { tokio::spawn(async move {
let (label, port, login_state, bus, socket, files, turn_lock) = web_ui_args; let (label, port, login_state, bus, socket, files, turn_lock) = web_ui_args;
if let Err(e) = if let Err(e) = web_ui::serve(label, port, login_state, bus, socket, files, turn_lock).await
web_ui::serve(label, port, login_state, bus, socket, files, turn_lock).await
{ {
tracing::error!(error = %e, "web_ui::serve exited with error"); tracing::error!(error = %e, "web_ui::serve exited with error");
} }
@ -658,7 +660,11 @@ async fn handle_turn<S: Surface>(
log_system_event(bus, &from, &body); log_system_event(bus, &from, &body);
tracing::info!(%from, %body, %redelivered, "inbox"); tracing::info!(%from, %body, %redelivered, "inbox");
let unread = S::inbox_unread(socket).await; 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); bus.set_state(TurnState::Thinking);
let started_at = serve_common::now_unix(); let started_at = serve_common::now_unix();
let started_instant = std::time::Instant::now(); let started_instant = std::time::Instant::now();
@ -670,7 +676,10 @@ async fn handle_turn<S: Surface>(
}; };
turn::emit_turn_end(bus, &outcome); turn::emit_turn_end(bus, &outcome);
bus.set_state(TurnState::Idle); 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; S::ack_turn(socket).await;
} }
if matches!(outcome, turn::TurnOutcome::RateLimited) { if matches!(outcome, turn::TurnOutcome::RateLimited) {
@ -697,8 +706,7 @@ async fn handle_turn<S: Surface>(
} }
if let Some(stats) = stats { if let Some(stats) = stats {
let ended_at = serve_common::now_unix(); let ended_at = serve_common::now_unix();
let duration_ms = let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
let (open_threads, open_reminders) = S::post_turn_counts(socket).await; let (open_threads, open_reminders) = S::post_turn_counts(socket).await;
let row = serve_common::build_row( let row = serve_common::build_row(
started_at, started_at,

18
hive-ag3nt/src/events.rs
View file

@ -80,12 +80,18 @@ fn harness_json_path() -> PathBuf {
fn read_harness_state() -> (bool, bool) { fn read_harness_state() -> (bool, bool) {
// Try the new consolidated file first. // Try the new consolidated file first.
if let Ok(raw) = std::fs::read_to_string(harness_json_path()) if let Ok(raw) = std::fs::read_to_string(harness_json_path()) {
&& let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw) if let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw) {
{ let rate_limited = v
let rate_limited = v.get("rate_limited").and_then(serde_json::Value::as_bool).unwrap_or(false); .get("rate_limited")
let needs_login = v.get("needs_login").and_then(serde_json::Value::as_bool).unwrap_or(false); .and_then(|x| x.as_bool())
return (rate_limited, needs_login); .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. // Fall back to legacy sentinel files written by older harness builds.
let state_dir = crate::paths::state_dir(); let state_dir = crate::paths::state_dir();

10
hive-ag3nt/src/forge_notify.rs
View file

@ -244,10 +244,7 @@ fn is_username_byte(b: u8) -> bool {
/// window so addressed agents never silently miss a mention on a long /// window so addressed agents never silently miss a mention on a long
/// body. See `docs/forge.md::Body excerpt + truncation + heading /// body. See `docs/forge.md::Body excerpt + truncation + heading
/// escape` for the truncate-before-escape ordering rule. /// escape` for the truncate-before-escape ordering rule.
fn extract_truncated_mention_lines<'a>( fn extract_truncated_mention_lines<'a>(full_body: &'a str, included_excerpt: &str) -> Vec<&'a str> {
full_body: &'a str,
included_excerpt: &str,
) -> Vec<&'a str> {
full_body full_body
.lines() .lines()
.filter(|line| { .filter(|line| {
@ -914,7 +911,10 @@ mod tests {
let full = "# @argus check this\nmore body\n"; let full = "# @argus check this\nmore body\n";
let raw_excerpt = full; // fits entirely let raw_excerpt = full; // fits entirely
let lines = extract_truncated_mention_lines(full, raw_excerpt); 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] #[test]

4
hive-ag3nt/src/identity.rs
View file

@ -122,7 +122,9 @@ mod tests {
swarm_name: Option<&str>, swarm_name: Option<&str>,
f: F, 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_label = env::var("HIVE_LABEL").ok();
let prev_domain = env::var("HYPERHIVE_HIVE_DOMAIN").ok(); let prev_domain = env::var("HYPERHIVE_HIVE_DOMAIN").ok();
let prev_hive_name = env::var("HYPERHIVE_HIVE_NAME").ok(); let prev_hive_name = env::var("HYPERHIVE_HIVE_NAME").ok();

74
hive-ag3nt/src/mcp.rs
View file

@ -437,31 +437,30 @@ fn format_bash_status(id: &str) -> String {
let Some(task) = crate::bash_runner::read_task(id) else { let Some(task) = crate::bash_runner::read_task(id) else {
return format!("bash_status: unknown task id `{id}`"); return format!("bash_status: unknown task id `{id}`");
}; };
let mut out = format!( let mut out = format!("task `{id}`: status={status:?}", status = task.status);
"task `{id}`: status={status:?}",
status = task.status
);
if let Some(code) = task.exit_code { if let Some(code) = task.exit_code {
let _ = write!(out, ", 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 age = crate::serve_common::now_unix() - t;
let _ = write!(out, ", running for {age}s"); let _ = write!(out, ", running for {age}s");
} }
if let Some(t) = task.completed_at if let Some(t) = task.completed_at {
&& let Some(s) = task.started_at if let Some(s) = task.started_at {
{ let _ = write!(out, ", took {}s", t - s);
let _ = write!(out, ", took {}s", t - s); }
} }
if let Some(ref stdout) = task.stdout_tail if let Some(ref stdout) = task.stdout_tail {
&& !stdout.trim().is_empty() if !stdout.trim().is_empty() {
{ let _ = write!(out, "\n\nstdout:\n```\n{}\n```", stdout.trim());
let _ = write!(out, "\n\nstdout:\n```\n{}\n```", stdout.trim()); }
} }
if let Some(ref stderr) = task.stderr_tail if let Some(ref stderr) = task.stderr_tail {
&& !stderr.trim().is_empty() if !stderr.trim().is_empty() {
{ let _ = write!(out, "\n\nstderr:\n```\n{}\n```", stderr.trim());
let _ = write!(out, "\n\nstderr:\n```\n{}\n```", stderr.trim()); }
} }
out out
} }
@ -668,7 +667,9 @@ impl AgentServer {
)] )]
async fn get_loose_ends(&self, Parameters(args): Parameters<AgentGetLooseEndsArgs>) -> String { async fn get_loose_ends(&self, Parameters(args): Parameters<AgentGetLooseEndsArgs>) -> String {
run_tool_envelope("get_loose_ends", String::new(), async move { 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); let mut out = annotate_retries(format_loose_ends(resp), retries);
// Append any local bash tasks still in pending/running state so // Append any local bash tasks still in pending/running state so
// the agent sees all outstanding work in one call. // 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()); let _ = write!(out, "\n\n{} active bash task(s):", active.len());
for task in &active { for task in &active {
let age = crate::serve_common::now_unix() - task.created_at; let age = crate::serve_common::now_unix() - task.created_at;
let _ = write!(out, "\n- `{}` status={:?}, cmd: `{}`, age {}s", let _ = write!(
task.id, task.status, task.cmd, age); out,
"\n- `{}` status={:?}, cmd: `{}`, age {}s",
task.id, task.status, task.cmd, age
);
} }
} }
out out
@ -830,9 +834,11 @@ impl AgentServer {
)] )]
async fn bash_status(&self, Parameters(args): Parameters<BashStatusArgs>) -> String { async fn bash_status(&self, Parameters(args): Parameters<BashStatusArgs>) -> String {
let log = format!("{args:?}"); let log = format!("{args:?}");
run_tool_envelope("bash_status", log, async move { run_tool_envelope(
format_bash_status(&args.id) "bash_status",
}) log,
async move { format_bash_status(&args.id) },
)
.await .await
} }
@ -875,10 +881,7 @@ impl AgentServer {
`since`: show entries on or newer than this (e.g. `-1h`, `2024-01-01 12:00:00`). \ `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." `until`: show entries on or older than this."
)] )]
async fn get_host_journal( async fn get_host_journal(&self, Parameters(args): Parameters<GetHostJournalArgs>) -> String {
&self,
Parameters(args): Parameters<GetHostJournalArgs>,
) -> String {
let log = format!("{args:?}"); let log = format!("{args:?}");
run_tool_envelope("get_host_journal", log, async move { run_tool_envelope("get_host_journal", log, async move {
let (resp, retries) = self let (resp, retries) = self
@ -1913,14 +1916,14 @@ pub enum Flavor {
} }
/// Env var written by the meta renderer with a comma-separated list of /// 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 /// 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`. /// instead of using the hardcoded flavor default. See `docs/conventions.md::Tool groups`.
const TOOL_GROUPS_ENV: &str = "HIVE_TOOL_GROUPS"; const TOOL_GROUPS_ENV: &str = "HIVE_TOOL_GROUPS";
/// `HIVE_CAPABILITIES` env var injected by `meta::render_flake` when the /// `HIVE_CAPABILITIES` env var injected by `meta::render_flake` when the
/// operator grants capabilities to this agent. Comma-separated /// 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"; const CAPABILITIES_ENV: &str = "HIVE_CAPABILITIES";
/// Returns the MCP tool names (without `mcp__hyperhive__` prefix) that are /// Returns the MCP tool names (without `mcp__hyperhive__` prefix) that are
@ -1976,12 +1979,13 @@ fn effective_tool_groups(flavor: Flavor) -> Vec<hive_sh4re::ToolGroup> {
for token in raw.split(',') { for token in raw.split(',') {
let t = token.trim().to_ascii_lowercase(); let t = token.trim().to_ascii_lowercase();
// Parse via serde_json (the canonical deserialization path). // Parse via serde_json (the canonical deserialization path).
if let Ok(g) = serde_json::from_value::<hive_sh4re::ToolGroup>( match serde_json::from_value::<hive_sh4re::ToolGroup>(serde_json::Value::String(t.clone()))
serde_json::Value::String(t.clone()), {
) { Ok(g) => groups.push(g),
groups.push(g); Err(_) => tracing::warn!(
} else { token = %t,
tracing::warn!(token = %t, "{TOOL_GROUPS_ENV}: unknown tool group, skipping"); "{TOOL_GROUPS_ENV}: unknown tool group, skipping"
),
} }
} }
if groups.is_empty() { if groups.is_empty() {

25
hive-ag3nt/src/stats.rs
View file

@ -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 { fn percentile(sorted: &[i64], pct: u8) -> f64 {
if sorted.is_empty() { if sorted.is_empty() {
return 0.0; return 0.0;
@ -471,7 +475,15 @@ mod tests {
(started_at, ended_at, duration_ms, model, wake_from, (started_at, ended_at, duration_ms, model, wake_from,
last_input_tokens, tool_call_breakdown_json, result_kind) last_input_tokens, tool_call_breakdown_json, result_kind)
VALUES (?1, ?2, ?3, ?4, ?5, 1000, ?6, ?7)", 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(); .unwrap();
} }
@ -485,7 +497,14 @@ mod tests {
seed_db( seed_db(
&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 - 300, 10_000, "opus", "recv", "ok", r#"{"Read":3}"#),
(now - 100, 20_000, "sonnet", "operator", "failed", "{}"), (now - 100, 20_000, "sonnet", "operator", "failed", "{}"),
], ],

16
hive-ag3nt/src/web_ui.rs
View file

@ -561,7 +561,9 @@ async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> {
ctx_usage, ctx_usage,
cost_usage, cost_usage,
links: agent_links(&state.label, state.gui_vnc_port.is_some()), 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(), hive_name: crate::identity::hive_name(),
swarm_name: crate::identity::swarm_name(), swarm_name: crate::identity::swarm_name(),
}) })
@ -613,8 +615,7 @@ fn agent_links(label: &str, gui_enabled: bool) -> Vec<AgentLink> {
// `{state_dir}/hyperhive-dashboard-links.json`). Shape on disk // `{state_dir}/hyperhive-dashboard-links.json`). Shape on disk
// is `{label, icon, url}` with absolute URLs — those become // is `{label, icon, url}` with absolute URLs — those become
// `kind = External` links, passed through verbatim. // `kind = External` links, passed through verbatim.
let extras_path = let extras_path = crate::paths::state_dir().join("hyperhive-dashboard-links.json");
crate::paths::state_dir().join("hyperhive-dashboard-links.json");
if let Ok(text) = std::fs::read_to_string(&extras_path) if let Ok(text) = std::fs::read_to_string(&extras_path)
&& !text.trim().is_empty() && !text.trim().is_empty()
&& let Ok(extras) = serde_json::from_str::<Vec<ExtraLink>>(&text) && let Ok(extras) = serde_json::from_str::<Vec<ExtraLink>>(&text)
@ -662,7 +663,10 @@ async fn recent_inbox(socket: &std::path::Path) -> Vec<hive_sh4re::InboxRow> {
/// Fetch reminder activity stats from the broker via the per-agent / /// Fetch reminder activity stats from the broker via the per-agent /
/// manager socket. Returns None on any transport / decode failure — the /// manager socket. Returns None on any transport / decode failure — the
/// stats are decorative, not authoritative. /// stats are decorative, not authoritative.
async fn fetch_reminder_stats(socket: &std::path::Path, window_secs: u64) -> Option<hive_sh4re::ReminderStats> { async fn fetch_reminder_stats(
socket: &std::path::Path,
window_secs: u64,
) -> Option<hive_sh4re::ReminderStats> {
match client::request::<_, hive_sh4re::Response>( match client::request::<_, hive_sh4re::Response>(
socket, socket,
&hive_sh4re::Request::ReminderRollup { &hive_sh4re::Request::ReminderRollup {
@ -956,7 +960,9 @@ async fn post_cancel_turn(State(state): State<AppState>) -> Response {
), ),
Err(e) => format!("operator: /cancel — pkill failed: {e}"), 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() (axum::http::StatusCode::OK, "ok").into_response()
} }

1
hive-c0re/Cargo.toml
View file

@ -9,6 +9,7 @@ workspace = true
anyhow.workspace = true anyhow.workspace = true
axum.workspace = true axum.workspace = true
base64.workspace = true base64.workspace = true
bcrypt.workspace = true
reqwest.workspace = true reqwest.workspace = true
clap.workspace = true clap.workspace = true
hive-sh4re.workspace = true hive-sh4re.workspace = true

3
hive-c0re/src/actions.rs
View file

@ -76,8 +76,7 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
// Pre-enqueue cascade rebuilds in topological order so // Pre-enqueue cascade rebuilds in topological order so
// agents depending on updated inputs are rebuilt after the // agents depending on updated inputs are rebuilt after the
// lock bump, matching the dashboard post_meta_update path. // lock bump, matching the dashboard post_meta_update path.
let cascade_agents = let cascade_agents = crate::rebuild_queue::meta_update_cascade_agents(&inputs).await;
crate::rebuild_queue::meta_update_cascade_agents(&inputs).await;
let cascade_reason = format!("approval #{id} meta input cascade"); let cascade_reason = format!("approval #{id} meta input cascade");
for name in cascade_agents { for name in cascade_agents {
coord.rebuild_queue.enqueue( coord.rebuild_queue.enqueue(

14
hive-c0re/src/agent_ports.rs
View file

@ -40,8 +40,7 @@ fn render(map: &BTreeMap<String, u16>) -> String {
// BTreeMap → serde_json::to_string_pretty preserves key order, // BTreeMap → serde_json::to_string_pretty preserves key order,
// so the output is deterministic across calls with the same // so the output is deterministic across calls with the same
// agent set. // agent set.
serde_json::to_string_pretty(map) serde_json::to_string_pretty(map).expect("BTreeMap<String, u16> is always serialisable")
.expect("BTreeMap<String, u16> is always serialisable")
} }
/// Atomically write the JSON for `names` to /// Atomically write the JSON for `names` to
@ -61,12 +60,10 @@ pub fn write(names: &[String]) -> Result<()> {
return Ok(()); return Ok(());
} }
if let Some(parent) = path.parent() { if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent) std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
.with_context(|| format!("create {}", parent.display()))?;
} }
let tmp = path.with_extension("json.tmp"); let tmp = path.with_extension("json.tmp");
std::fs::write(&tmp, &body) std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
.with_context(|| format!("write {}", tmp.display()))?;
std::fs::rename(&tmp, &path).with_context(|| { std::fs::rename(&tmp, &path).with_context(|| {
format!( format!(
"rename {} -> {} (atomic publish)", "rename {} -> {} (atomic publish)",
@ -132,9 +129,6 @@ mod tests {
// BTreeMap sorts → alpha before zeta in output. // BTreeMap sorts → alpha before zeta in output.
let alpha_pos = body.find("alpha").expect("alpha in output"); let alpha_pos = body.find("alpha").expect("alpha in output");
let zeta_pos = body.find("zeta").expect("zeta in output"); let zeta_pos = body.find("zeta").expect("zeta in output");
assert!( assert!(alpha_pos < zeta_pos, "sorted order broken:\n{body}");
alpha_pos < zeta_pos,
"sorted order broken:\n{body}"
);
} }
} }

58
hive-c0re/src/agent_server.rs
View file

@ -129,9 +129,11 @@ pub(crate) async fn dispatch_shared(
) -> Option<hive_sh4re::Response> { ) -> Option<hive_sh4re::Response> {
let broker = &coord.broker; let broker = &coord.broker;
Some(match req { Some(match req {
hive_sh4re::Request::Send { to, body, in_reply_to } => { hive_sh4re::Request::Send {
handle_send(coord, agent, to, body, *in_reply_to) to,
} body,
in_reply_to,
} => handle_send(coord, agent, to, body, *in_reply_to),
hive_sh4re::Request::Recv { wait_seconds, max } => { hive_sh4re::Request::Recv { wait_seconds, max } => {
let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize; let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize;
match broker match broker
@ -223,8 +225,8 @@ pub(crate) async fn dispatch_shared(
if let Err(message) = crate::limits::check_status_text(text) { if let Err(message) = crate::limits::check_status_text(text) {
return Some(hive_sh4re::Response::Err { message }); return Some(hive_sh4re::Response::Err { message });
} }
let path = crate::coordinator::Coordinator::agent_notes_dir(agent) let path =
.join("hyperhive-status"); crate::coordinator::Coordinator::agent_notes_dir(agent).join("hyperhive-status");
let result = if text.trim().is_empty() { let result = if text.trim().is_empty() {
std::fs::remove_file(&path).or_else(|e| { std::fs::remove_file(&path).or_else(|e| {
if e.kind() == std::io::ErrorKind::NotFound { 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 }); tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
hive_sh4re::Response::Ok hive_sh4re::Response::Ok
} }
Err(e) => { Err(e) => hive_sh4re::Response::Err {
hive_sh4re::Response::Err { message: format!("set_status write failed: {e}"),
message: format!("set_status write failed: {e}"), },
}
}
} }
} }
hive_sh4re::Request::GetAgentMeta { name } => { hive_sh4re::Request::GetAgentMeta { name } => {
@ -294,7 +294,15 @@ pub(crate) async fn dispatch_shared(
message: format!("{e:#}"), 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 dispatch_host_journal(agent, unit, container, lines, priority, grep, since, until).await
} }
// Not a shared variant. // Not a shared variant.
@ -331,7 +339,10 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
Err(message) => AgentResponse::Err { message }, 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()); let name = resolve_agent_state_target(agent, target.as_deref());
match name { match name {
Ok(name) => match coord.broker.reminder_rollup_for(name, *since_secs) { Ok(name) => match coord.broker.reminder_rollup_for(name, *since_secs) {
@ -547,10 +558,7 @@ pub(crate) fn store_remind(
) -> Result<(), String> { ) -> Result<(), String> {
let max = remind_max_pending(); let max = remind_max_pending();
if max > 0 { if max > 0 {
let pending = coord let pending = coord.broker.count_pending_reminders_for(agent).unwrap_or(0);
.broker
.count_pending_reminders_for(agent)
.unwrap_or(0);
if pending >= max { if pending >= max {
return Err(format!( return Err(format!(
"reminder rejected: agent `{agent}` already has {pending} pending \ "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) let host_path = crate::reminder_scheduler::resolve_host_path(agent, &req_path)
.map_err(|reason| format!("auto-save path `{req_path}` rejected: {reason}"))?; .map_err(|reason| format!("auto-save path `{req_path}` rejected: {reason}"))?;
crate::reminder_scheduler::write_payload(agent, &host_path, message) crate::reminder_scheduler::write_payload(agent, &host_path, message).map_err(|reason| {
.map_err(|reason| format!("auto-save of large reminder body to `{req_path}` failed: {reason}"))?; format!("auto-save of large reminder body to `{req_path}` failed: {reason}")
})?;
let hint = format!( let hint = format!(
"[reminder body of {} bytes auto-saved to `{req_path}`; read with your filesystem tools]", "[reminder body of {} bytes auto-saved to `{req_path}`; read with your filesystem tools]",
message.len() message.len()
@ -634,19 +643,26 @@ fn auto_reminder_path(agent: &str) -> String {
/// - `Some("<other>")` where other is not a child → requires the /// - `Some("<other>")` where other is not a child → requires the
/// `query_agent_state` capability; returns an error otherwise. /// `query_agent_state` capability; returns an error otherwise.
/// - `Some("*")` → always rejected (hive-wide scans are manager-only). /// - `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 { match target {
None => Ok(caller), None => Ok(caller),
Some("*") => Err( Some("*") => Err(
"hive-wide query (agent=\"*\") is not available on the agent socket; \ "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) => { Some(name) => {
if name == caller { if name == caller {
return Ok(caller); return Ok(caller);
} }
// Direct children are visible to their parent without extra capability. // 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); return Ok(name);
} }
if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) { if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) {

35
hive-c0re/src/agent_sockets.rs
View file

@ -81,8 +81,7 @@ pub fn socket_path_for(name: &str) -> PathBuf {
#[must_use] #[must_use]
pub fn build_map(names: &[String]) -> BTreeMap<String, PathBuf> { pub fn build_map(names: &[String]) -> BTreeMap<String, PathBuf> {
build_map_with(names, |name| { build_map_with(names, |name| {
ready_marker_for(name).exists() ready_marker_for(name).exists() || agent_dir_for(name).join(READY_MARKER_LEGACY).exists()
|| agent_dir_for(name).join(READY_MARKER_LEGACY).exists()
}) })
} }
@ -146,12 +145,10 @@ pub fn write(names: &[String]) -> Result<()> {
return Ok(()); return Ok(());
} }
if let Some(parent) = path.parent() { if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent) std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
.with_context(|| format!("create {}", parent.display()))?;
} }
let tmp = path.with_extension("json.tmp"); let tmp = path.with_extension("json.tmp");
std::fs::write(&tmp, &body) std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
.with_context(|| format!("write {}", tmp.display()))?;
std::fs::rename(&tmp, &path).with_context(|| { std::fs::rename(&tmp, &path).with_context(|| {
format!( format!(
"rename {} -> {} (atomic publish)", "rename {} -> {} (atomic publish)",
@ -305,24 +302,30 @@ mod tests {
let marker = ready_marker_for("iris"); let marker = ready_marker_for("iris");
let socket = socket_path_for("iris"); let socket = socket_path_for("iris");
assert_eq!(marker.parent(), socket.parent()); 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] #[test]
fn render_is_pretty_and_sorted() { fn render_is_pretty_and_sorted() {
let mut map = BTreeMap::new(); let mut map = BTreeMap::new();
map.insert("zeta".to_owned(), PathBuf::from("/run/hive-agent/zeta/web.sock")); map.insert(
map.insert("alpha".to_owned(), PathBuf::from("/run/hive-agent/alpha/web.sock")); "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); let body = render(&map);
// Pretty-print = newlines between keys + indentation. // Pretty-print = newlines between keys + indentation.
assert!(body.contains('\n')); assert!(body.contains('\n'));
// BTreeMap sorts → alpha before zeta in output. // BTreeMap sorts → alpha before zeta in output.
let alpha_pos = body.find("alpha").expect("alpha in output"); let alpha_pos = body.find("alpha").expect("alpha in output");
let zeta_pos = body.find("zeta").expect("zeta in output"); let zeta_pos = body.find("zeta").expect("zeta in output");
assert!( assert!(alpha_pos < zeta_pos, "sorted order broken:\n{body}");
alpha_pos < zeta_pos,
"sorted order broken:\n{body}"
);
} }
#[test] #[test]
@ -332,10 +335,12 @@ mod tests {
// gateway-side reader can deserialise into String values // gateway-side reader can deserialise into String values
// without nested struct logic. // without nested struct logic.
let mut map = BTreeMap::new(); 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); let body = render(&map);
assert!(body.contains("\"iris\"")); assert!(body.contains("\"iris\""));
assert!(body.contains("\"/run/hive-agent/iris/web.sock\"")); assert!(body.contains("\"/run/hive-agent/iris/web.sock\""));
} }
} }

10
hive-c0re/src/auto_update.rs
View file

@ -209,16 +209,19 @@ pub async fn ensure_manager(coord: &Arc<Coordinator>) -> Result<()> {
/// Sort `names` in-place so parents precede their children in the topology. /// 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, /// Uses BFS from root agents (depth 0). Agents absent from `topo` sort last,
/// alphabetically within their tier. Stable within each depth tier. /// alphabetically within their tier. Stable within each depth tier.
pub fn topology_sort(names: &mut [String], topo: &std::collections::BTreeMap<String, Option<String>>) { pub fn topology_sort(
names: &mut Vec<String>,
topo: &std::collections::BTreeMap<String, Option<String>>,
) {
use std::collections::{HashMap, VecDeque}; use std::collections::{HashMap, VecDeque};
// Build depth map using owned clones so the borrow on `names` is released // Build depth map using owned clones so the borrow on `names` is released
// before the sort_by mutable borrow. // before the sort_by mutable borrow.
let name_set: Vec<String> = names.to_vec(); let name_set: Vec<String> = names.clone();
let mut depth: HashMap<String, usize> = HashMap::new(); let mut depth: HashMap<String, usize> = HashMap::new();
let mut queue: VecDeque<String> = VecDeque::new(); let mut queue: VecDeque<String> = VecDeque::new();
// Seed roots: entries with no parent, or names not present in topo at all. // Seed roots: entries with no parent, or names not present in topo at all.
for name in &name_set { 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); depth.insert(name.clone(), 0);
queue.push_back(name.clone()); queue.push_back(name.clone());
} }
@ -300,4 +303,3 @@ pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
Ok(()) Ok(())
} }

19
hive-c0re/src/bash_tasks_vacuum.rs
View file

@ -23,7 +23,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
use crate::coordinator::Coordinator; 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. /// Keep completed task files for 48 hours before sweeping them.
const KEEP_SECS: i64 = 48 * 3600; const KEEP_SECS: i64 = 48 * 3600;
@ -65,7 +65,9 @@ fn sweep_once() {
/// files removed (each represents one task; `.out`/`.err` deletions /// files removed (each represents one task; `.out`/`.err` deletions
/// are not counted separately). /// are not counted separately).
fn vacuum_dir(dir: &Path, cutoff: i64) -> u64 { 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; let mut removed: u64 = 0;
for entry in rd.flatten() { for entry in rd.flatten() {
let path = entry.path(); let path = entry.path();
@ -97,7 +99,10 @@ fn should_delete(json_path: &Path, cutoff: i64) -> bool {
if !TERMINAL_STATUSES.contains(&status) { if !TERMINAL_STATUSES.contains(&status) {
return false; 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 completed_at < cutoff
} }
@ -106,10 +111,10 @@ fn should_delete(json_path: &Path, cutoff: i64) -> bool {
fn delete_trio(dir: &Path, stem: &str) { fn delete_trio(dir: &Path, stem: &str) {
for ext in ["json", "out", "err"] { for ext in ["json", "out", "err"] {
let path = dir.join(format!("{stem}.{ext}")); let path = dir.join(format!("{stem}.{ext}"));
if path.exists() if path.exists() {
&& let Err(e) = std::fs::remove_file(&path) if let Err(e) = std::fs::remove_file(&path) {
{ tracing::warn!(path = %path.display(), error = ?e, "bash-tasks vacuum: remove failed");
tracing::warn!(path = %path.display(), error = ?e, "bash-tasks vacuum: remove failed"); }
} }
} }
} }

197
hive-c0re/src/bin/hivectl.rs
View file

@ -15,6 +15,8 @@
//! dirs) and reuse the `forge` / `matrix` modules from the //! dirs) and reuse the `forge` / `matrix` modules from the
//! `hive-c0re` lib — single source of truth, no duplication. //! `hive-c0re` lib — single source of truth, no duplication.
use std::path::{Path, PathBuf};
use anyhow::{Context as _, Result, bail}; use anyhow::{Context as _, Result, bail};
use clap::{Parser, Subcommand}; use clap::{Parser, Subcommand};
use hive_c0re::coordinator::Coordinator; use hive_c0re::coordinator::Coordinator;
@ -54,6 +56,14 @@ enum Cmd {
#[command(subcommand)] #[command(subcommand)]
cmd: MatrixCmd, 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)] #[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<String>,
/// 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] #[tokio::main]
async fn main() -> Result<()> { async fn main() -> Result<()> {
tracing_subscriber::fmt() tracing_subscriber::fmt()
@ -161,6 +213,16 @@ async fn main() -> Result<()> {
password_stdin, password_stdin,
} => matrix_create_user(&name, password.as_deref(), password_stdin).await, } => 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() Coordinator::agent_state_root(name).exists()
} }
async fn forge_create_user( async fn forge_create_user(name: &str, password: Option<&str>, password_stdin: bool) -> Result<()> {
name: &str,
password: Option<&str>,
password_stdin: bool,
) -> Result<()> {
if !hive_c0re::forge::is_present().await { if !hive_c0re::forge::is_present().await {
bail!( bail!(
"hive-forge container not running — start it (services.hyperhive.forge.enable = true) before provisioning forge users" "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 { } else {
let effective_password = match user_password { let effective_password = match user_password {
Some(p) => p, Some(p) => p,
None => hive_c0re::matrix::random_password() None => {
.context("generate random matrix password")?, hive_c0re::matrix::random_password().context("generate random matrix password")?
}
}; };
let token = let token = hive_c0re::matrix::provision_user_token(
hive_c0re::matrix::provision_user_token(&client, name, &register_token, &effective_password) &client,
.await name,
.with_context(|| format!("matrix create-user {name}"))?; &register_token,
&effective_password,
)
.await
.with_context(|| format!("matrix create-user {name}"))?;
println!("matrix: provisioned user '{name}' (not an agent — token not persisted)"); println!("matrix: provisioned user '{name}' (not an agent — token not persisted)");
println!("token: {token}"); println!("token: {token}");
if password.is_some() || password_stdin { if password.is_some() || password_stdin {
@ -293,3 +356,115 @@ async fn matrix_create_user(
} }
Ok(()) 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<Vec<String>> {
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 `<path>.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(())
}

35
hive-c0re/src/broker.rs
View file

@ -268,12 +268,7 @@ impl Broker {
/// broker. Used by the scheduler to skip re-delivery of the same /// broker. Used by the scheduler to skip re-delivery of the same
/// scheduled prompt without blocking distinct schedules whose /// scheduled prompt without blocking distinct schedules whose
/// bodies differ. /// bodies differ.
pub fn has_pending_with_body( pub fn has_pending_with_body(&self, recipient: &str, sender: &str, body: &str) -> Result<bool> {
&self,
recipient: &str,
sender: &str,
body: &str,
) -> Result<bool> {
let conn = self.conn.lock().unwrap(); let conn = self.conn.lock().unwrap();
let n: i64 = conn.query_row( let n: i64 = conn.query_row(
"SELECT COUNT(*) FROM messages "SELECT COUNT(*) FROM messages
@ -391,7 +386,13 @@ impl Broker {
)?; )?;
let rows: Vec<(i64, String, String, String, Option<i64>)> = stmt let rows: Vec<(i64, String, String, String, Option<i64>)> = stmt
.query_map(params![recipient, max_i], |row| { .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::<rusqlite::Result<_>>()?; .collect::<rusqlite::Result<_>>()?;
drop(stmt); drop(stmt);
@ -673,7 +674,11 @@ impl Broker {
/// Reminder rollup stats for an agent over a time window. Returns /// Reminder rollup stats for an agent over a time window. Returns
/// counts of scheduled, delivered, and pending reminders created /// counts of scheduled, delivered, and pending reminders created
/// in the last `since_secs` seconds (0 = all reminders). /// in the last `since_secs` seconds (0 = all reminders).
pub fn reminder_rollup_for(&self, agent: &str, since_secs: u64) -> Result<hive_sh4re::ReminderStats> { pub fn reminder_rollup_for(
&self,
agent: &str,
since_secs: u64,
) -> Result<hive_sh4re::ReminderStats> {
let conn = self.conn.lock().unwrap(); let conn = self.conn.lock().unwrap();
let cutoff_time = if since_secs > 0 { let cutoff_time = if since_secs > 0 {
let now = std::time::SystemTime::now() let now = std::time::SystemTime::now()
@ -740,9 +745,7 @@ impl Broker {
|| canceller == hive_sh4re::OPERATOR_RECIPIENT || canceller == hive_sh4re::OPERATOR_RECIPIENT
|| canceller == hive_sh4re::MANAGER_AGENT; || canceller == hive_sh4re::MANAGER_AGENT;
if !authorised { if !authorised {
anyhow::bail!( anyhow::bail!("reminder {id}: '{canceller}' not allowed to cancel (owner = '{owner}')");
"reminder {id}: '{canceller}' not allowed to cancel (owner = '{owner}')"
);
} }
let n = conn.execute( let n = conn.execute(
"DELETE FROM reminders WHERE id = ?1 AND sent_at IS NULL", "DELETE FROM reminders WHERE id = ?1 AND sent_at IS NULL",
@ -862,7 +865,9 @@ impl Broker {
} }
drop(conn); drop(conn);
// Emit per-row Sent events (only for rows that succeeded). // 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() { if result.is_ok() {
let _ = self.events.send(MessageEvent::Sent { let _ = self.events.send(MessageEvent::Sent {
id: *msg_id, id: *msg_id,
@ -1029,10 +1034,7 @@ mod tests {
assert_eq!(broker.requeue_inflight("b").unwrap(), 1); assert_eq!(broker.requeue_inflight("b").unwrap(), 1);
let d2 = pop_one(broker, "b").expect("popped again"); let d2 = pop_one(broker, "b").expect("popped again");
assert_eq!(d2.message.body, "hi"); assert_eq!(d2.message.body, "hi");
assert!( assert!(d2.redelivered, "second pop should be tagged redelivered");
d2.redelivered,
"second pop should be tagged redelivered"
);
assert_eq!(broker.ack_turn("b").unwrap(), 1); assert_eq!(broker.ack_turn("b").unwrap(), 1);
} }
@ -1275,4 +1277,3 @@ mod tests {
assert!(pop_one(broker, "bob").is_none()); assert!(pop_one(broker, "bob").is_none());
} }
} }

Some files were not shown because too many files have changed in this diff Show more