From 25508d7399912d8cf0c8267bccfa11f304626168 Mon Sep 17 00:00:00 2001 From: damocles Date: Sat, 16 May 2026 19:53:50 +0200 Subject: [PATCH 1/4] fix manager loop: pending wake + move sleep into Empty arm only --- hive-ag3nt/src/bin/hive-m1nd.rs | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/hive-ag3nt/src/bin/hive-m1nd.rs b/hive-ag3nt/src/bin/hive-m1nd.rs index f0b0cc09..85f70920 100644 --- a/hive-ag3nt/src/bin/hive-m1nd.rs +++ b/hive-ag3nt/src/bin/hive-m1nd.rs @@ -134,8 +134,18 @@ async fn serve( 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 { .. } @@ -151,7 +161,6 @@ async fn serve( tracing::warn!(error = ?e, "recv failed; retrying"); } } - tokio::time::sleep(interval).await; } } From fca480b86ed0ef76a691839f08458c64e7cf41ca Mon Sep 17 00:00:00 2001 From: damocles Date: Sat, 16 May 2026 19:57:03 +0200 Subject: [PATCH 2/4] add turn lock to prevent /compact racing with in-flight turns --- hive-ag3nt/src/bin/hive-ag3nt.rs | 12 +++++++++++- hive-ag3nt/src/bin/hive-m1nd.rs | 14 +++++++++++--- hive-ag3nt/src/web_ui.rs | 22 ++++++++++++++++++++++ 3 files changed, 44 insertions(+), 4 deletions(-) diff --git a/hive-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs index d486e803..ee6fd0cb 100644 --- a/hive-ag3nt/src/bin/hive-ag3nt.rs +++ b/hive-ag3nt/src/bin/hive-ag3nt.rs @@ -2,6 +2,8 @@ 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}; @@ -71,6 +73,7 @@ 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, @@ -79,6 +82,7 @@ async fn main() -> Result<()> { bus.clone(), cli.socket.clone(), files.clone(), + turn_lock.clone(), )); match initial { LoginState::Online => { @@ -88,6 +92,7 @@ async fn main() -> Result<()> { login_state, bus, &files, + turn_lock, ) .await } @@ -102,6 +107,7 @@ async fn main() -> Result<()> { login_state, bus, &files, + turn_lock, ) .await } @@ -136,6 +142,7 @@ async fn serve( state: Arc>, bus: Bus, files: &turn::TurnFiles, + turn_lock: TurnLock, ) -> Result<()> { tracing::info!(socket = %socket.display(), "hive-ag3nt serve"); let _ = state; // reserved for future state transitions (turn-loop -> needs-login) @@ -163,7 +170,10 @@ async fn serve( }); bus.set_state(TurnState::Thinking); let prompt = format_wake_prompt(&from, &body, unread); - let outcome = turn::drive_turn(&prompt, files, &bus).await; + let outcome = { + let _guard = turn_lock.lock().await; + 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 diff --git a/hive-ag3nt/src/bin/hive-m1nd.rs b/hive-ag3nt/src/bin/hive-m1nd.rs index 85f70920..9d890389 100644 --- a/hive-ag3nt/src/bin/hive-m1nd.rs +++ b/hive-ag3nt/src/bin/hive-m1nd.rs @@ -6,6 +6,8 @@ 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}; @@ -61,6 +63,7 @@ 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, @@ -69,14 +72,15 @@ 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).await + serve(&cli.socket, Duration::from_millis(poll_ms), bus, &files, turn_lock).await } LoginState::NeedsLogin => { turn::wait_for_login(&claude_dir, login_state, poll_ms).await; - serve(&cli.socket, Duration::from_millis(poll_ms), bus, &files).await + serve(&cli.socket, Duration::from_millis(poll_ms), bus, &files, turn_lock).await } } } @@ -89,6 +93,7 @@ async fn serve( interval: Duration, bus: Bus, files: &turn::TurnFiles, + turn_lock: TurnLock, ) -> Result<()> { tracing::info!(socket = %socket.display(), "hive-m1nd serve"); loop { @@ -131,7 +136,10 @@ async fn serve( }); let prompt = format_wake_prompt(&from, &body, unread); bus.set_state(TurnState::Thinking); - let outcome = turn::drive_turn(&prompt, files, &bus).await; + let outcome = { + let _guard = turn_lock.lock().await; + 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 diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs index 931ec2d6..02a9df2e 100644 --- a/hive-ag3nt/src/web_ui.rs +++ b/hive-ag3nt/src/web_ui.rs @@ -36,6 +36,12 @@ use crate::turn::TurnFiles; /// render. pub type LoginStateCell = Arc>; +/// 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>; + #[derive(Clone)] struct AppState { label: String, @@ -48,6 +54,8 @@ 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 { @@ -70,6 +78,7 @@ pub async fn serve( bus: Bus, socket: PathBuf, files: TurnFiles, + turn_lock: TurnLock, ) -> Result<()> { let state = AppState { label, @@ -78,6 +87,7 @@ pub async fn serve( bus, socket, files, + turn_lock, }; let app = Router::new() .route("/", get(serve_index)) @@ -406,9 +416,21 @@ async fn post_set_model(State(state): State, Form(form): Form) -> 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(), )); From 1023acf69f2577395d3e7d6ecd104160e6516fc3 Mon Sep 17 00:00:00 2001 From: damocles Date: Sat, 16 May 2026 20:28:42 +0200 Subject: [PATCH 3/4] add get_logs tool to manager mcp surface --- hive-ag3nt/src/mcp.rs | 46 +++++++++++++++++++++++++++++++++ hive-c0re/src/manager_server.rs | 29 +++++++++++++++++++++ hive-sh4re/src/lib.rs | 15 +++++++++++ 3 files changed, 90 insertions(+) diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index e252d105..9fc3ddb2 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -39,6 +39,7 @@ pub enum SocketReply { Status(u64), QuestionQueued(i64), Recent(Vec), + Logs(String), } impl From for SocketReply { @@ -65,6 +66,7 @@ impl From 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), } } } @@ -351,6 +353,15 @@ pub struct RequestApplyCommitArgs { pub description: Option, } +#[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, +} + #[derive(Debug, Clone)] pub struct ManagerServer { socket: PathBuf, @@ -580,6 +591,40 @@ 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) -> 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( @@ -635,6 +680,7 @@ pub fn allowed_mcp_tools(flavor: Flavor) -> Vec { "update", "request_apply_commit", "ask_operator", + "get_logs", ], }; let mut out: Vec = names diff --git a/hive-c0re/src/manager_server.rs b/hive-c0re/src/manager_server.rs index 1146b9fb..5f90a573 100644 --- a/hive-c0re/src/manager_server.rs +++ b/hive-c0re/src/manager_server.rs @@ -273,6 +273,35 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> 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, diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 5a6fd0ce..314aba60 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -475,6 +475,17 @@ pub enum ManagerRequest { #[serde(default)] ttl_seconds: Option, }, + /// Fetch recent journal lines for a sub-agent container. hive-c0re + /// runs `journalctl -M -n --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, + }, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -502,4 +513,8 @@ pub enum ManagerResponse { Recent { rows: Vec, }, + /// `GetLogs` result: journal lines for the requested container. + Logs { + content: String, + }, } From 824acee134c784211acf056c986dd8e9f65d3946 Mon Sep 17 00:00:00 2001 From: damocles Date: Sat, 16 May 2026 20:28:45 +0200 Subject: [PATCH 4/4] include agent label in turn failure notification body --- hive-ag3nt/src/bin/hive-ag3nt.rs | 23 ++++++++++++++--------- hive-ag3nt/src/bin/hive-m1nd.rs | 3 ++- 2 files changed, 16 insertions(+), 10 deletions(-) diff --git a/hive-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs index ee6fd0cb..48def743 100644 --- a/hive-ag3nt/src/bin/hive-ag3nt.rs +++ b/hive-ag3nt/src/bin/hive-ag3nt.rs @@ -76,7 +76,7 @@ async fn main() -> Result<()> { let turn_lock: TurnLock = Arc::new(tokio::sync::Mutex::new(())); plugins::install_configured(&cli.socket, Some("manager")).await; tokio::spawn(web_ui::serve( - label, + label.clone(), port, login_state.clone(), bus.clone(), @@ -93,6 +93,7 @@ async fn main() -> Result<()> { bus, &files, turn_lock, + &label, ) .await } @@ -108,6 +109,7 @@ async fn main() -> Result<()> { bus, &files, turn_lock, + &label, ) .await } @@ -143,6 +145,7 @@ async fn serve( 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) @@ -182,7 +185,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, e).await; + notify_manager_of_failure(socket, label, e).await; } // After turn completes, check if there are pending messages waiting. @@ -235,13 +238,15 @@ 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 -/// 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:#}"); +/// 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:#}"); let res = client::request::<_, AgentResponse>( socket, &AgentRequest::Send { diff --git a/hive-ag3nt/src/bin/hive-m1nd.rs b/hive-ag3nt/src/bin/hive-m1nd.rs index 9d890389..89eab9ff 100644 --- a/hive-ag3nt/src/bin/hive-m1nd.rs +++ b/hive-ag3nt/src/bin/hive-m1nd.rs @@ -158,7 +158,8 @@ async fn serve( ManagerResponse::Ok | ManagerResponse::Status { .. } | ManagerResponse::QuestionQueued { .. } - | ManagerResponse::Recent { .. }, + | ManagerResponse::Recent { .. } + | ManagerResponse::Logs { .. }, ) => { tracing::warn!("recv produced unexpected response kind"); }