harness: unify hive-ag3nt + hive-m1nd into one binary picking role from HIVE_ROLE (#598)

hive-ag3nt/src/bin/hive.rs

@@ -0,0 +1,698 @@
+//! Unified hyperhive harness binary (#598).
+//!
+//! Replaces the pre-#598 `hive-ag3nt` + `hive-m1nd` split. One binary,
+//! picks its role at startup from `HIVE_ROLE` (set by post-#676
+//! `harness-base.nix` from `hyperhive.role` — `"agent"` or
+//! `"manager"`). Both agent + manager wire surfaces live in the same
+//! binary because the privilege boundary is enforced server-side at
+//! the socket (`/run/hive/mcp.sock`): an agent socket refuses
+//! `ManagerRequest` calls regardless of who sends them, so there's no
+//! escalation risk in shipping the same code to both.
+//!
+//! `HIVE_ROLE` defaults to `"agent"` if unset to keep the standalone
+//! `nix run .#hive-ag3nt` shape working without env plumbing.
+
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use hive_ag3nt::web_ui::TurnLock;
+
+use anyhow::{Result, bail};
+use clap::{Parser, Subcommand};
+use hive_ag3nt::events::{Bus, LiveEvent, TurnState};
+use hive_ag3nt::login::{self, LoginState};
+use hive_ag3nt::turn_stats::TurnStats;
+use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, plugins, serve_common, turn, web_ui};
+use hive_sh4re::{
+    AgentRequest, AgentResponse, HelperEvent, ManagerRequest, ManagerResponse, SYSTEM_SENDER,
+};
+
+#[derive(Parser)]
+#[command(
+    name = "hive",
+    about = "hyperhive harness — role from $HIVE_ROLE (agent|manager)"
+)]
+struct Cli {
+    /// Path to the per-agent MCP socket (bind-mounted from the host).
+    #[arg(long, global = true, default_value = DEFAULT_SOCKET)]
+    socket: PathBuf,
+
+    #[command(subcommand)]
+    cmd: Cmd,
+}
+
+#[derive(Subcommand)]
+enum Cmd {
+    /// Run the long-lived harness loop. Polls inbox; replies via
+    /// `claude --print` when available.
+    Serve {
+        /// Inbox poll interval in milliseconds.
+        #[arg(long, default_value_t = 1000)]
+        poll_ms: u64,
+    },
+    /// Run this role's MCP server on stdio. Spawned by `claude` via
+    /// `--mcp-config`; tools dispatch through `/run/hive/mcp.sock` back
+    /// into the hyperhive broker.
+    Mcp,
+    /// Agent-only: inject a wake-up event into this agent's inbox so
+    /// the next turn fires with the given body. Intended for extra MCP
+    /// servers / helpers (matrix bridge, scraper, webhook listener)
+    /// that need to nudge claude on external events. Refused when
+    /// `HIVE_ROLE=manager` — the manager surface has no `Wake`
+    /// equivalent.
+    Wake {
+        #[arg(long)]
+        from: String,
+        /// Body of the wake message. Pass `-` to read from stdin.
+        #[arg(long)]
+        body: String,
+    },
+}
+
+#[derive(Copy, Clone)]
+enum Role {
+    Agent,
+    Manager,
+}
+
+fn resolve_role() -> Result<Role> {
+    match std::env::var("HIVE_ROLE").as_deref() {
+        Ok("agent") | Err(_) => Ok(Role::Agent),
+        Ok("manager") => Ok(Role::Manager),
+        Ok(other) => bail!("unknown HIVE_ROLE={other:?}; expected 'agent' or 'manager'"),
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
+        )
+        .init();
+
+    let cli = Cli::parse();
+    let role = resolve_role()?;
+
+    match (role, cli.cmd) {
+        (Role::Agent, Cmd::Serve { poll_ms }) => agent_serve_main(&cli.socket, poll_ms).await,
+        (Role::Manager, Cmd::Serve { poll_ms }) => manager_serve_main(&cli.socket, poll_ms).await,
+        (Role::Agent, Cmd::Mcp) => mcp::serve_agent_stdio(cli.socket).await,
+        (Role::Manager, Cmd::Mcp) => mcp::serve_manager_stdio(cli.socket).await,
+        (Role::Agent, Cmd::Wake { from, body }) => agent_wake(&cli.socket, from, body).await,
+        (Role::Manager, Cmd::Wake { .. }) => {
+            bail!("wake is agent-only — manager has no equivalent surface")
+        }
+    }
+}
+
+// ---------- agent role ----------
+
+async fn agent_serve_main(socket: &Path, poll_ms: u64) -> Result<()> {
+    let port = std::env::var("HIVE_PORT")
+        .ok()
+        .and_then(|s| s.parse::<u16>().ok())
+        .unwrap_or(DEFAULT_WEB_PORT);
+    let label = std::env::var("HIVE_LABEL").unwrap_or_else(|_| "hive-ag3nt".into());
+    let claude_dir = login::default_dir();
+    let initial = LoginState::from_dir(&claude_dir);
+    tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "harness boot");
+    let login_state = Arc::new(Mutex::new(initial));
+    let bus = Bus::new();
+    let stats = TurnStats::open_default();
+    if let Some(s) = &stats {
+        let (ctx, cost) = s.last_usage();
+        if ctx.is_some() || cost.is_some() {
+            bus.seed_usage(ctx, cost);
+        }
+    }
+    let files = turn::TurnFiles::prepare(socket, &label, mcp::Flavor::Agent).await?;
+    let turn_lock: TurnLock = Arc::new(tokio::sync::Mutex::new(()));
+    plugins::install_configured(socket, Some("manager")).await;
+    tokio::spawn(hive_ag3nt::forge_notify::run(socket.to_path_buf(), false));
+    tokio::spawn(web_ui::serve(
+        label.clone(),
+        port,
+        login_state.clone(),
+        bus.clone(),
+        socket.to_path_buf(),
+        files.clone(),
+        turn_lock.clone(),
+    ));
+    match initial {
+        LoginState::Online => {
+            agent_serve_loop(
+                socket,
+                Duration::from_millis(poll_ms),
+                login_state,
+                claude_dir,
+                bus,
+                stats,
+                &files,
+                turn_lock,
+                &label,
+            )
+            .await
+        }
+        LoginState::NeedsLogin => {
+            turn::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await;
+            agent_serve_loop(
+                socket,
+                Duration::from_millis(poll_ms),
+                login_state,
+                claude_dir,
+                bus,
+                stats,
+                &files,
+                turn_lock,
+                &label,
+            )
+            .await
+        }
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn agent_serve_loop(
+    socket: &Path,
+    interval: Duration,
+    login_state: Arc<Mutex<LoginState>>,
+    claude_dir: std::path::PathBuf,
+    bus: Bus,
+    stats: Option<TurnStats>,
+    files: &turn::TurnFiles,
+    turn_lock: TurnLock,
+    label: &str,
+) -> Result<()> {
+    tracing::info!(socket = %socket.display(), "hive-ag3nt serve");
+    agent_requeue_inflight(socket).await;
+    loop {
+        let recv: Result<AgentResponse> = client::request(
+            socket,
+            &AgentRequest::Recv {
+                wait_seconds: Some(180),
+                max: None,
+            },
+        )
+        .await;
+        match recv {
+            Ok(AgentResponse::Messages { messages }) if !messages.is_empty() => {
+                let first = messages.into_iter().next().expect("checked non-empty");
+                let auth_failed =
+                    handle_agent_turn(socket, &bus, stats.as_ref(), files, &turn_lock, label, first)
+                        .await;
+                if auth_failed {
+                    *login_state.lock().unwrap() = LoginState::NeedsLogin;
+                    turn::wait_for_login(
+                        &claude_dir,
+                        login_state.clone(),
+                        &bus,
+                        u64::try_from(interval.as_millis()).unwrap_or(2000),
+                    )
+                    .await;
+                }
+            }
+            Ok(AgentResponse::Messages { .. }) => {
+                tokio::time::sleep(interval).await;
+            }
+            Ok(
+                AgentResponse::Ok
+                | AgentResponse::Status { .. }
+                | AgentResponse::Recent { .. }
+                | AgentResponse::QuestionQueued { .. }
+                | AgentResponse::LooseEnds { .. }
+                | AgentResponse::PendingRemindersCount { .. }
+                | AgentResponse::ReminderRollup { .. }
+                | AgentResponse::AgentMeta { .. },
+            ) => {
+                tracing::warn!("recv produced unexpected response kind");
+            }
+            Ok(AgentResponse::Err { message }) => {
+                tracing::warn!(%message, "recv error");
+            }
+            Err(e) => {
+                tracing::warn!(error = ?e, "recv failed; retrying");
+            }
+        }
+    }
+}
+
+async fn handle_agent_turn(
+    socket: &Path,
+    bus: &Bus,
+    stats: Option<&TurnStats>,
+    files: &turn::TurnFiles,
+    turn_lock: &TurnLock,
+    label: &str,
+    first: hive_sh4re::DeliveredMessage,
+) -> bool {
+    let from = first.from;
+    let body = first.body;
+    let redelivered = first.redelivered;
+    tracing::info!(%from, %body, %redelivered, "inbox");
+    let unread = agent_inbox_unread(socket).await;
+    bus.emit(LiveEvent::TurnStart { from: from.clone(), body: body.clone(), unread });
+    bus.set_state(TurnState::Thinking);
+    let started_at = serve_common::now_unix();
+    let started_instant = std::time::Instant::now();
+    let model_at_start = bus.model();
+    let prompt = serve_common::format_wake_prompt(&from, &body, unread, redelivered);
+    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);
+    if matches!(outcome, turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted) {
+        agent_ack_turn(socket).await;
+    }
+    if matches!(outcome, turn::TurnOutcome::RateLimited) {
+        let secs = turn::rate_limit_sleep_secs();
+        bus.emit_status("rate_limited");
+        bus.emit(LiveEvent::Note {
+            text: format!("API rate-limited — sleeping {secs}s before retry"),
+        });
+        tracing::warn!(sleep_secs = secs, "rate-limited; parking");
+        tokio::time::sleep(Duration::from_secs(secs)).await;
+        agent_requeue_inflight(socket).await;
+        bus.emit_status("online");
+    }
+    if matches!(outcome, turn::TurnOutcome::AuthFailed) {
+        bus.emit_status("needs_login_idle");
+        bus.emit(LiveEvent::Note {
+            text: "API 401 — waiting for re-login via web UI".into(),
+        });
+        tracing::warn!("auth-failed; parking until re-login");
+        agent_requeue_inflight(socket).await;
+    }
+    if let turn::TurnOutcome::Failed(e) = &outcome {
+        agent_notify_manager_of_failure(socket, label, e).await;
+    }
+    if let Some(stats) = stats {
+        let ended_at = serve_common::now_unix();
+        let duration_ms =
+            i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
+        let (open_threads, open_reminders) = agent_post_turn_counts(socket).await;
+        let row = serve_common::build_row(
+            started_at,
+            ended_at,
+            duration_ms,
+            model_at_start,
+            from.clone(),
+            &outcome,
+            bus,
+            open_threads,
+            open_reminders,
+        );
+        stats.record(&row);
+    }
+    let pending = agent_inbox_unread(socket).await;
+    if pending > 0 {
+        tracing::info!(%pending, "pending messages after turn; fetching next");
+    }
+    agent_check_and_inject_continue(socket, label).await;
+    matches!(outcome, turn::TurnOutcome::AuthFailed)
+}
+
+async fn agent_ack_turn(socket: &Path) {
+    match client::request::<_, AgentResponse>(socket, &AgentRequest::AckTurn).await {
+        Ok(AgentResponse::Ok) => {}
+        Ok(AgentResponse::Err { message }) => {
+            tracing::warn!(%message, "ack_turn rejected by broker");
+        }
+        Ok(other) => {
+            tracing::warn!(?other, "ack_turn unexpected response");
+        }
+        Err(e) => tracing::warn!(error = ?e, "ack_turn transport error"),
+    }
+}
+
+async fn agent_requeue_inflight(socket: &Path) {
+    match client::request::<_, AgentResponse>(socket, &AgentRequest::RequeueInflight).await {
+        Ok(AgentResponse::Ok) => {}
+        Ok(AgentResponse::Err { message }) => {
+            tracing::warn!(%message, "requeue_inflight rejected by broker");
+        }
+        Ok(other) => {
+            tracing::warn!(?other, "requeue_inflight unexpected response");
+        }
+        Err(e) => tracing::warn!(error = ?e, "requeue_inflight transport error"),
+    }
+}
+
+async fn agent_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 {
+            to: "manager".into(),
+            body,
+            in_reply_to: None,
+        },
+    )
+    .await;
+    if let Err(e) = res {
+        tracing::warn!(error = ?e, "failed to notify manager of turn failure");
+    }
+}
+
+async fn agent_inbox_unread(socket: &Path) -> u64 {
+    match client::request::<_, AgentResponse>(socket, &AgentRequest::Status).await {
+        Ok(AgentResponse::Status { unread }) => unread,
+        _ => 0,
+    }
+}
+
+async fn agent_post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) {
+    let threads =
+        match client::request::<_, AgentResponse>(socket, &AgentRequest::GetLooseEnds).await {
+            Ok(AgentResponse::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(),
+            _ => None,
+        };
+    let reminders = match client::request::<_, AgentResponse>(
+        socket,
+        &AgentRequest::CountPendingReminders,
+    )
+    .await
+    {
+        Ok(AgentResponse::PendingRemindersCount { count }) => Some(count),
+        _ => None,
+    };
+    (threads, reminders)
+}
+
+async fn agent_check_and_inject_continue(socket: &Path, label: &str) {
+    let sentinel = hive_ag3nt::paths::state_dir().join("hyperhive-continue");
+    if !sentinel.exists() {
+        return;
+    }
+    if let Err(e) = std::fs::remove_file(&sentinel) {
+        tracing::warn!(error = %e, "check_and_inject_continue: remove sentinel failed");
+        return;
+    }
+    let res = client::request::<_, AgentResponse>(
+        socket,
+        &AgentRequest::Wake {
+            from: "self".into(),
+            body: "continue".into(),
+        },
+    )
+    .await;
+    match res {
+        Ok(AgentResponse::Ok) => {
+            tracing::info!(%label, "request_next_turn: injected self-continue wake");
+        }
+        Ok(AgentResponse::Err { message }) => {
+            tracing::warn!(%message, "check_and_inject_continue: wake rejected");
+        }
+        Err(e) => {
+            tracing::warn!(error = ?e, "check_and_inject_continue: wake transport error");
+        }
+        _ => {}
+    }
+}
+
+async fn agent_wake(socket: &Path, from: String, body: String) -> Result<()> {
+    let body = if body == "-" {
+        let mut buf = String::new();
+        std::io::Read::read_to_string(&mut std::io::stdin(), &mut buf)?;
+        buf
+    } else {
+        body
+    };
+    let resp: AgentResponse =
+        client::request(socket, &AgentRequest::Wake { from, body }).await?;
+    match resp {
+        AgentResponse::Ok => Ok(()),
+        AgentResponse::Err { message } => anyhow::bail!("wake: {message}"),
+        other => anyhow::bail!("wake: unexpected response {other:?}"),
+    }
+}
+
+// ---------- manager role ----------
+
+async fn manager_serve_main(socket: &Path, poll_ms: u64) -> Result<()> {
+    let port = std::env::var("HIVE_PORT")
+        .ok()
+        .and_then(|s| s.parse::<u16>().ok())
+        .unwrap_or(DEFAULT_WEB_PORT);
+    let label = std::env::var("HIVE_LABEL").unwrap_or_else(|_| "hm1nd".into());
+    let claude_dir = login::default_dir();
+    let initial = LoginState::from_dir(&claude_dir);
+    tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "hm1nd boot");
+    let login_state = Arc::new(Mutex::new(initial));
+    let bus = Bus::new();
+    let stats = TurnStats::open_default();
+    if let Some(s) = &stats {
+        let (ctx, cost) = s.last_usage();
+        if ctx.is_some() || cost.is_some() {
+            bus.seed_usage(ctx, cost);
+        }
+    }
+    let files = turn::TurnFiles::prepare(socket, &label, mcp::Flavor::Manager).await?;
+    let turn_lock: TurnLock = Arc::new(tokio::sync::Mutex::new(()));
+    plugins::install_configured(socket, None).await;
+    tokio::spawn(web_ui::serve(
+        label,
+        port,
+        login_state.clone(),
+        bus.clone(),
+        socket.to_path_buf(),
+        files.clone(),
+        turn_lock.clone(),
+    ));
+    tokio::spawn(hive_ag3nt::forge_notify::run(socket.to_path_buf(), true));
+    match initial {
+        LoginState::Online => {
+            manager_serve_loop(
+                socket,
+                Duration::from_millis(poll_ms),
+                login_state,
+                claude_dir,
+                bus,
+                stats,
+                &files,
+                turn_lock,
+            )
+            .await
+        }
+        LoginState::NeedsLogin => {
+            turn::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await;
+            manager_serve_loop(
+                socket,
+                Duration::from_millis(poll_ms),
+                login_state,
+                claude_dir,
+                bus,
+                stats,
+                &files,
+                turn_lock,
+            )
+            .await
+        }
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn manager_serve_loop(
+    socket: &Path,
+    interval: Duration,
+    login_state: Arc<Mutex<LoginState>>,
+    claude_dir: std::path::PathBuf,
+    bus: Bus,
+    stats: Option<TurnStats>,
+    files: &turn::TurnFiles,
+    turn_lock: TurnLock,
+) -> Result<()> {
+    tracing::info!(socket = %socket.display(), "hive-m1nd serve");
+    manager_requeue_inflight(socket).await;
+    loop {
+        let recv: Result<ManagerResponse> = client::request(
+            socket,
+            &ManagerRequest::Recv {
+                wait_seconds: Some(180),
+                max: None,
+            },
+        )
+        .await;
+        match recv {
+            Ok(ManagerResponse::Messages { messages }) if !messages.is_empty() => {
+                let first = messages.into_iter().next().expect("checked non-empty");
+                let auth_failed =
+                    handle_manager_turn(socket, &bus, stats.as_ref(), files, &turn_lock, first)
+                        .await;
+                if auth_failed {
+                    *login_state.lock().unwrap() = LoginState::NeedsLogin;
+                    turn::wait_for_login(
+                        &claude_dir,
+                        login_state.clone(),
+                        &bus,
+                        u64::try_from(interval.as_millis()).unwrap_or(2000),
+                    )
+                    .await;
+                }
+            }
+            Ok(ManagerResponse::Messages { .. }) => {
+                tokio::time::sleep(interval).await;
+            }
+            Ok(
+                ManagerResponse::Ok
+                | ManagerResponse::Status { .. }
+                | ManagerResponse::QuestionQueued { .. }
+                | ManagerResponse::Recent { .. }
+                | ManagerResponse::Logs { .. }
+                | ManagerResponse::LooseEnds { .. }
+                | ManagerResponse::PendingRemindersCount { .. }
+                | ManagerResponse::ReminderRollup { .. }
+                | ManagerResponse::AgentMeta { .. }
+                | ManagerResponse::Schedules { .. },
+            ) => {
+                tracing::warn!("recv produced unexpected response kind");
+            }
+            Ok(ManagerResponse::Err { message }) => {
+                tracing::warn!(%message, "recv error");
+            }
+            Err(e) => {
+                tracing::warn!(error = ?e, "recv failed; retrying");
+            }
+        }
+    }
+}
+
+async fn handle_manager_turn(
+    socket: &Path,
+    bus: &Bus,
+    stats: Option<&TurnStats>,
+    files: &turn::TurnFiles,
+    turn_lock: &TurnLock,
+    first: hive_sh4re::DeliveredMessage,
+) -> bool {
+    let from = first.from;
+    let body = first.body;
+    let redelivered = first.redelivered;
+    if from == SYSTEM_SENDER {
+        let parsed = serde_json::from_str::<HelperEvent>(&body).ok();
+        if let Some(event) = parsed {
+            tracing::info!(?event, "helper event");
+        } else {
+            tracing::info!(%from, %body, "system message");
+        }
+        bus.emit(LiveEvent::Note { text: format!("[system] {body}") });
+    }
+    tracing::info!(%from, %body, %redelivered, "manager inbox");
+    let unread = manager_inbox_unread(socket).await;
+    bus.emit(LiveEvent::TurnStart { from: from.clone(), body: body.clone(), unread });
+    let prompt = serve_common::format_wake_prompt(&from, &body, unread, redelivered);
+    bus.set_state(TurnState::Thinking);
+    let started_at = serve_common::now_unix();
+    let started_instant = std::time::Instant::now();
+    let model_at_start = bus.model();
+    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);
+    if matches!(outcome, turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted) {
+        manager_ack_turn(socket).await;
+    }
+    if matches!(outcome, turn::TurnOutcome::RateLimited) {
+        let secs = turn::rate_limit_sleep_secs();
+        bus.emit_status("rate_limited");
+        bus.emit(LiveEvent::Note {
+            text: format!("API rate-limited — sleeping {secs}s before retry"),
+        });
+        tracing::warn!(sleep_secs = secs, "rate-limited; parking");
+        tokio::time::sleep(Duration::from_secs(secs)).await;
+        manager_requeue_inflight(socket).await;
+        bus.emit_status("online");
+    }
+    if matches!(outcome, turn::TurnOutcome::AuthFailed) {
+        bus.emit_status("needs_login_idle");
+        bus.emit(LiveEvent::Note {
+            text: "API 401 — waiting for re-login via web UI".into(),
+        });
+        tracing::warn!("auth-failed; parking until re-login");
+        manager_requeue_inflight(socket).await;
+    }
+    if let Some(stats) = stats {
+        let ended_at = serve_common::now_unix();
+        let duration_ms =
+            i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
+        let (open_threads, open_reminders) = manager_post_turn_counts(socket).await;
+        let row = serve_common::build_row(
+            started_at,
+            ended_at,
+            duration_ms,
+            model_at_start,
+            from.clone(),
+            &outcome,
+            bus,
+            open_threads,
+            open_reminders,
+        );
+        stats.record(&row);
+    }
+    let pending = manager_inbox_unread(socket).await;
+    if pending > 0 {
+        tracing::info!(%pending, "pending messages after turn; fetching next");
+    }
+    matches!(outcome, turn::TurnOutcome::AuthFailed)
+}
+
+async fn manager_ack_turn(socket: &Path) {
+    match client::request::<_, ManagerResponse>(socket, &ManagerRequest::AckTurn).await {
+        Ok(ManagerResponse::Ok) => {}
+        Ok(ManagerResponse::Err { message }) => {
+            tracing::warn!(%message, "ack_turn rejected by broker");
+        }
+        Ok(other) => {
+            tracing::warn!(?other, "ack_turn unexpected response");
+        }
+        Err(e) => tracing::warn!(error = ?e, "ack_turn transport error"),
+    }
+}
+
+async fn manager_requeue_inflight(socket: &Path) {
+    match client::request::<_, ManagerResponse>(socket, &ManagerRequest::RequeueInflight).await {
+        Ok(ManagerResponse::Ok) => {}
+        Ok(ManagerResponse::Err { message }) => {
+            tracing::warn!(%message, "requeue_inflight rejected by broker");
+        }
+        Ok(other) => {
+            tracing::warn!(?other, "requeue_inflight unexpected response");
+        }
+        Err(e) => tracing::warn!(error = ?e, "requeue_inflight transport error"),
+    }
+}
+
+async fn manager_inbox_unread(socket: &Path) -> u64 {
+    match client::request::<_, ManagerResponse>(socket, &ManagerRequest::Status).await {
+        Ok(ManagerResponse::Status { unread }) => unread,
+        _ => 0,
+    }
+}
+
+async fn manager_post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) {
+    let threads = match client::request::<_, ManagerResponse>(
+        socket,
+        &ManagerRequest::GetLooseEnds { agent: None },
+    )
+    .await
+    {
+        Ok(ManagerResponse::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(),
+        _ => None,
+    };
+    let reminders = match client::request::<_, ManagerResponse>(
+        socket,
+        &ManagerRequest::CountPendingReminders { agent: None },
+    )
+    .await
+    {
+        Ok(ManagerResponse::PendingRemindersCount { count }) => Some(count),
+        _ => None,
+    };
+    (threads, reminders)
+}