vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

Compare commits

base: vinzenz:824acee134c784211acf056c986dd8e9f65d3946
Branches Tags
vinzenz:main
..
compare: vinzenz:f2a0dc410768e3ff5126b7180055243578491632
Branches Tags
vinzenz:main
6 changed files with 16 additions and 161 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -2,8 +2,6 @@ use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use hive_ag3nt::web_ui::TurnLock;
use anyhow::Result;
use clap::{Parser, Subcommand};
use hive_ag3nt::events::{Bus, LiveEvent, TurnState};
@ -73,16 +71,14 @@ async fn main() -> Result<()> {
let login_state = Arc::new(Mutex::new(initial));
let bus = Bus::new();
let files = turn::TurnFiles::prepare(&cli.socket, &label, mcp::Flavor::Agent).await?;
let turn_lock: TurnLock = Arc::new(tokio::sync::Mutex::new(()));
plugins::install_configured(&cli.socket, Some("manager")).await;
tokio::spawn(web_ui::serve(
label.clone(),
label,
port,
login_state.clone(),
bus.clone(),
cli.socket.clone(),
files.clone(),
turn_lock.clone(),
));
match initial {
LoginState::Online => {
@ -92,8 +88,6 @@ async fn main() -> Result<()> {
login_state,
bus,
&files,
turn_lock,
&label,
)
.await
}
@ -108,8 +102,6 @@ async fn main() -> Result<()> {
login_state,
bus,
&files,
turn_lock,
&label,
)
.await
}
@ -144,8 +136,6 @@ async fn serve(
state: Arc<Mutex<LoginState>>,
bus: Bus,
files: &turn::TurnFiles,
turn_lock: TurnLock,
label: &str,
) -> Result<()> {
tracing::info!(socket = %socket.display(), "hive-ag3nt serve");
let _ = state; // reserved for future state transitions (turn-loop -> needs-login)
@ -173,10 +163,7 @@ async fn serve(
});
bus.set_state(TurnState::Thinking);
let prompt = format_wake_prompt(&from, &body, unread);
let outcome = {
let _guard = turn_lock.lock().await;
turn::drive_turn(&prompt, files, &bus).await
};
let outcome = turn::drive_turn(&prompt, files, &bus).await;
turn::emit_turn_end(&bus, &outcome);
bus.set_state(TurnState::Idle);
// Failures are unhandled by definition — PromptTooLong is
@ -185,7 +172,7 @@ async fn serve(
// manager so it can investigate / restart / page the
// operator; best-effort, swallow the send error.
if let turn::TurnOutcome::Failed(e) = &outcome {
notify_manager_of_failure(socket, label, e).await;
notify_manager_of_failure(socket, e).await;
}
// After turn completes, check if there are pending messages waiting.
@ -238,15 +225,13 @@ fn format_wake_prompt(from: &str, body: &str, unread: u64) -> String {
/// Best-effort: tell the manager that this agent's last turn crashed
/// (claude exited non-zero, compaction didn't help, etc.). Routed
/// through the normal send path so the manager's inbox surfaces it
/// as a system-style event; `label` is included explicitly in the
/// body so the manager can identify the failing agent without having
/// to look at the `from` field (which is broker-stamped and may
/// differ from what the operator sees in the dashboard). Swallows
/// transport errors — we just logged the failure, the worst case is
/// the manager learns about the crash from the dashboard instead of
/// inbox.
async fn notify_manager_of_failure(socket: &Path, label: &str, err: &anyhow::Error) {
let body = format!("[system] agent `{label}` claude turn failed:\n{err:#}");
/// like any other message; the agent's label is what the broker
/// stamps as `from`, so the message body doesn't need to repeat it.
/// Swallows transport errors — we just logged the failure, the worst
/// case is the manager learns about the crash from the dashboard
/// instead of inbox.
async fn notify_manager_of_failure(socket: &Path, err: &anyhow::Error) {
let body = format!("claude turn failed:\n{err:#}");
let res = client::request::<_, AgentResponse>(
socket,
&AgentRequest::Send {

30
hive-ag3nt/src/bin/hive-m1nd.rs
View file

@ -6,8 +6,6 @@ use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use hive_ag3nt::web_ui::TurnLock;
use anyhow::Result;
use clap::{Parser, Subcommand};
use hive_ag3nt::events::{Bus, LiveEvent, TurnState};
@ -63,7 +61,6 @@ async fn main() -> Result<()> {
let login_state = Arc::new(Mutex::new(initial));
let bus = Bus::new();
let files = turn::TurnFiles::prepare(&cli.socket, &label, mcp::Flavor::Manager).await?;
let turn_lock: TurnLock = Arc::new(tokio::sync::Mutex::new(()));
plugins::install_configured(&cli.socket, None).await;
tokio::spawn(web_ui::serve(
label,
@ -72,15 +69,14 @@ async fn main() -> Result<()> {
bus.clone(),
cli.socket.clone(),
files.clone(),
turn_lock.clone(),
));
match initial {
LoginState::Online => {
serve(&cli.socket, Duration::from_millis(poll_ms), bus, &files, turn_lock).await
serve(&cli.socket, Duration::from_millis(poll_ms), bus, &files).await
}
LoginState::NeedsLogin => {
turn::wait_for_login(&claude_dir, login_state, poll_ms).await;
serve(&cli.socket, Duration::from_millis(poll_ms), bus, &files, turn_lock).await
serve(&cli.socket, Duration::from_millis(poll_ms), bus, &files).await
}
}
}
@ -93,7 +89,6 @@ async fn serve(
interval: Duration,
bus: Bus,
files: &turn::TurnFiles,
turn_lock: TurnLock,
) -> Result<()> {
tracing::info!(socket = %socket.display(), "hive-m1nd serve");
loop {
@ -136,30 +131,16 @@ async fn serve(
});
let prompt = format_wake_prompt(&from, &body, unread);
bus.set_state(TurnState::Thinking);
let outcome = {
let _guard = turn_lock.lock().await;
turn::drive_turn(&prompt, files, &bus).await
};
let outcome = turn::drive_turn(&prompt, files, &bus).await;
turn::emit_turn_end(&bus, &outcome);
bus.set_state(TurnState::Idle);
// Check for messages that arrived during the turn and loop
// immediately if any are waiting — mirrors hive-ag3nt behaviour.
let pending = inbox_unread(socket).await;
if pending > 0 {
tracing::info!(%pending, "pending messages after turn; fetching next");
continue;
}
}
Ok(ManagerResponse::Empty) => {
// Idle: sleep briefly before next long-poll attempt.
tokio::time::sleep(interval).await;
}
Ok(ManagerResponse::Empty) => {}
Ok(
ManagerResponse::Ok
| ManagerResponse::Status { .. }
| ManagerResponse::QuestionQueued { .. }
| ManagerResponse::Recent { .. }
| ManagerResponse::Logs { .. },
| ManagerResponse::Recent { .. },
) => {
tracing::warn!("recv produced unexpected response kind");
}
@ -170,6 +151,7 @@ async fn serve(
tracing::warn!(error = ?e, "recv failed; retrying");
}
}
tokio::time::sleep(interval).await;
}
}

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

@ -39,7 +39,6 @@ pub enum SocketReply {
Status(u64),
QuestionQueued(i64),
Recent(Vec<hive_sh4re::InboxRow>),
Logs(String),
}
impl From<hive_sh4re::AgentResponse> for SocketReply {
@ -66,7 +65,6 @@ impl From<hive_sh4re::ManagerResponse> for SocketReply {
hive_sh4re::ManagerResponse::Status { unread } => Self::Status(unread),
hive_sh4re::ManagerResponse::QuestionQueued { id } => Self::QuestionQueued(id),
hive_sh4re::ManagerResponse::Recent { rows } => Self::Recent(rows),
hive_sh4re::ManagerResponse::Logs { content } => Self::Logs(content),
}
}
}
@ -353,15 +351,6 @@ pub struct RequestApplyCommitArgs {
pub description: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct GetLogsArgs {
/// Logical name of the sub-agent container to fetch logs for.
pub agent: String,
/// How many journal lines to return (default: 50, max: 500).
#[serde(default)]
pub lines: Option<u32>,
}
#[derive(Debug, Clone)]
pub struct ManagerServer {
socket: PathBuf,
@ -591,40 +580,6 @@ impl ManagerServer {
})
.await
}
#[tool(
description = "Fetch recent journal log lines for a sub-agent container. Useful \
for diagnosing MCP server registration failures, startup crashes, plugin install \
errors, or any harness issue you can't see from inside the container. `lines` \
defaults to 50 (max capped at 500 on the host side)."
)]
async fn get_logs(&self, Parameters(args): Parameters<GetLogsArgs>) -> String {
let log = format!("{args:?}");
let agent = args.agent.clone();
run_tool_envelope("get_logs", log, async move {
let lines = args.lines.map(|n| n.min(500));
let (resp, retries) = self
.dispatch(hive_sh4re::ManagerRequest::GetLogs {
agent: agent.clone(),
lines,
})
.await;
let s = match resp {
Ok(SocketReply::Logs(content)) => {
if content.is_empty() {
format!("(no journal output for {agent})")
} else {
content
}
}
Ok(SocketReply::Err(m)) => format!("get_logs failed: {m}"),
Ok(other) => format!("get_logs unexpected response: {other:?}"),
Err(e) => format!("get_logs transport error: {e:#}"),
};
annotate_retries(s, retries)
})
.await
}
}
#[tool_handler(
@ -680,7 +635,6 @@ pub fn allowed_mcp_tools(flavor: Flavor) -> Vec<String> {
"update",
"request_apply_commit",
"ask_operator",
"get_logs",
],
};
let mut out: Vec<String> = names

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

@ -36,12 +36,6 @@ use crate::turn::TurnFiles;
/// render.
pub type LoginStateCell = Arc<Mutex<LoginState>>;
/// Shared turn lock. The serve loop acquires this (as an async mutex) for the
/// duration of every `drive_turn` call. The `/api/compact` handler tries
/// `try_lock()` and rejects immediately if a turn is in flight, preventing
/// concurrent access to the claude session.
pub type TurnLock = Arc<tokio::sync::Mutex<()>>;
#[derive(Clone)]
struct AppState {
label: String,
@ -54,8 +48,6 @@ struct AppState {
/// settings claude saw on the last regular turn — keeps the
/// session shape identical across compact + normal turns.
files: TurnFiles,
/// Prevents `/api/compact` from racing with an in-flight normal turn.
turn_lock: TurnLock,
}
impl AppState {
@ -78,7 +70,6 @@ pub async fn serve(
bus: Bus,
socket: PathBuf,
files: TurnFiles,
turn_lock: TurnLock,
) -> Result<()> {
let state = AppState {
label,
@ -87,7 +78,6 @@ pub async fn serve(
bus,
socket,
files,
turn_lock,
};
let app = Router::new()
.route("/", get(serve_index))
@ -416,21 +406,9 @@ async fn post_set_model(State(state): State<AppState>, Form(form): Form<ModelFor
}
async fn post_compact(State(state): State<AppState>) -> Response {
// Clone the Arc before locking so the guard's lifetime is tied to the
// clone (which we can move into the spawn) rather than to `state`.
let lock = state.turn_lock.clone();
// Reject immediately if a normal turn is in flight — concurrent access
// to the claude session is unsafe and produces garbled output.
let guard = match lock.try_lock_owned() {
Ok(g) => g,
Err(_) => {
return error_response("turn in flight — wait for it to finish before compacting");
}
};
let bus = state.bus.clone();
let files = state.files.clone();
tokio::spawn(async move {
let _guard = guard; // keep lock alive for the duration of compaction
bus.emit(crate::events::LiveEvent::Note(
"operator: /compact — running on persistent session".into(),
));

29
hive-c0re/src/manager_server.rs
View file

@ -273,35 +273,6 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
},
}
}
ManagerRequest::GetLogs { agent, lines } => {
let n = lines.unwrap_or(50);
tracing::info!(%agent, %n, "manager: get_logs");
match tokio::process::Command::new("journalctl")
.args([
"-M",
agent,
"-n",
&n.to_string(),
"--no-pager",
"--output=short",
])
.output()
.await
{
Ok(out) => {
let content = if out.status.success() || !out.stdout.is_empty() {
String::from_utf8_lossy(&out.stdout).into_owned()
} else {
let stderr = String::from_utf8_lossy(&out.stderr);
format!("journalctl exited {}: {stderr}", out.status)
};
ManagerResponse::Logs { content }
}
Err(e) => ManagerResponse::Err {
message: format!("journalctl spawn failed: {e:#}"),
},
}
}
ManagerRequest::RequestApplyCommit {
agent,
commit_ref,

15
hive-sh4re/src/lib.rs
View file

@ -475,17 +475,6 @@ pub enum ManagerRequest {
#[serde(default)]
ttl_seconds: Option<u64>,
},
/// Fetch recent journal lines for a sub-agent container. hive-c0re
/// runs `journalctl -M <agent> -n <lines> --no-pager` and returns
/// the output as a string. Useful for diagnosing MCP registration
/// failures, startup crashes, and harness errors.
///
/// `lines` defaults to 50 when omitted.
GetLogs {
agent: String,
#[serde(default)]
lines: Option<u32>,
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@ -513,8 +502,4 @@ pub enum ManagerResponse {
Recent {
rows: Vec<InboxRow>,
},
/// `GetLogs` result: journal lines for the requested container.
Logs {
content: String,
},
}