diff --git a/hive-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs index 48def743..d486e803 100644 --- a/hive-ag3nt/src/bin/hive-ag3nt.rs +++ b/hive-ag3nt/src/bin/hive-ag3nt.rs @@ -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>, 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 { diff --git a/hive-ag3nt/src/bin/hive-m1nd.rs b/hive-ag3nt/src/bin/hive-m1nd.rs index 89eab9ff..f0b0cc09 100644 --- a/hive-ag3nt/src/bin/hive-m1nd.rs +++ b/hive-ag3nt/src/bin/hive-m1nd.rs @@ -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; } } diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 9fc3ddb2..e252d105 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -39,7 +39,6 @@ pub enum SocketReply { Status(u64), QuestionQueued(i64), Recent(Vec), - Logs(String), } impl From for SocketReply { @@ -66,7 +65,6 @@ 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), } } } @@ -353,15 +351,6 @@ 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, @@ -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) -> 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 { "update", "request_apply_commit", "ask_operator", - "get_logs", ], }; let mut out: Vec = names diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs index 02a9df2e..931ec2d6 100644 --- a/hive-ag3nt/src/web_ui.rs +++ b/hive-ag3nt/src/web_ui.rs @@ -36,12 +36,6 @@ 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, @@ -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, 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(), )); diff --git a/hive-c0re/src/manager_server.rs b/hive-c0re/src/manager_server.rs index 5f90a573..1146b9fb 100644 --- a/hive-c0re/src/manager_server.rs +++ b/hive-c0re/src/manager_server.rs @@ -273,35 +273,6 @@ 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 314aba60..5a6fd0ce 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -475,17 +475,6 @@ 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)] @@ -513,8 +502,4 @@ pub enum ManagerResponse { Recent { rows: Vec, }, - /// `GetLogs` result: journal lines for the requested container. - Logs { - content: String, - }, }