hyperhive/hive-ag3nt/src/bin/hive.rs

//! Unified hyperhive harness binary. Picks role from `HIVE_ROLE`
//! (`"agent"` | `"manager"`), dispatches one of three subcommands
//! (`serve` / `mcp` / `wake`), and runs the turn loop through a
//! generic `Surface` trait so both wire surfaces stay in lockstep.
//!
//! Architecture (single-binary rationale, Surface-trait + zero-sized
//! type tags, boot wiring, turn-outcome branch) lives in
//! [`docs/turn-loop.md::Harness binary shape`](../../../docs/turn-loop.md).

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,
    /// Inject a wake-up event into this harness's inbox so the next
    /// turn fires with the given body. Intended for extra MCP servers
    /// / helpers (matrix bridge, scraper, webhook listener, etc.) that
    /// need to nudge claude on external events. Available on both
    /// agent and manager roles; mirrors the `AgentRequest::Wake` /
    /// `ManagerRequest::Wake` pair already on the wire.
    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()?;

    // Generic dispatch: one `serve_main` / `wake` body, two
    // monomorphisations driven by the `Surface` type parameter. See
    // `docs/turn-loop.md::Surface trait + zero-sized type tags`.
    match (role, cli.cmd) {
        (Role::Agent, Cmd::Serve { poll_ms }) => {
            serve_main::<AgentSurface>(&cli.socket, poll_ms).await
        }
        (Role::Manager, Cmd::Serve { poll_ms }) => {
            serve_main::<ManagerSurface>(&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 }) => {
            wake::<AgentSurface>(&cli.socket, from, body).await
        }
        (Role::Manager, Cmd::Wake { from, body }) => {
            wake::<ManagerSurface>(&cli.socket, from, body).await
        }
    }
}

// ---------- shared turn helpers ----------

/// Surface a `SYSTEM_SENDER` message in the live event bus + tracing
/// log. Both agents and the manager receive `QuestionAnswered`,
/// `ContainerCrash`, reparent notifications, and friends; the parse
/// + log path is identical. Quiet no-op when `from` isn't
/// `SYSTEM_SENDER`.
fn log_system_event(bus: &Bus, from: &str, body: &str) {
    if from != SYSTEM_SENDER {
        return;
    }
    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}") });
}

/// Body string for the turn-failure notification we route to
/// `<parent>` on `TurnOutcome::Failed`. Reads the hive-qualified
/// identity so the receiver sees `agent@hive` rather than relying on
/// the caller threading a `label` through every turn-handling layer.
/// Falls back to `<unknown>` when `HIVE_LABEL` is missing so a
/// misconfigured harness still produces a parseable line.
fn format_turn_failure(err: &anyhow::Error) -> String {
    let who = hive_ag3nt::identity::qualified_label();
    let who = if who.is_empty() { "<unknown>".to_owned() } else { who };
    format!("[system] `{who}` claude turn failed:\n{err:#}")
}

/// Check for the `hyperhive-continue` sentinel under the state dir
/// (dropped by the `request_next_turn` MCP tool). Returns true and
/// consumes the file when present; false otherwise. Caller fires
/// the role-specific `Wake` request — the sentinel itself is wire-
/// agnostic so this helper lives outside both surfaces.
fn consume_continue_sentinel() -> bool {
    let sentinel = hive_ag3nt::paths::state_dir().join("hyperhive-continue");
    if !sentinel.exists() {
        return false;
    }
    if let Err(e) = std::fs::remove_file(&sentinel) {
        tracing::warn!(error = %e, "consume_continue_sentinel: remove sentinel failed");
        return false;
    }
    true
}

// ---------- surface trait ----------

/// What a `Recv` long-poll returned. Decoupled from the per-role
/// Response enum so `serve_loop` can pattern-match without seeing
/// either AgentResponse or ManagerResponse directly.
enum RecvOutcome {
    /// Long-poll returned at least one message; first one is detached.
    Message(hive_sh4re::DeliveredMessage),
    /// Long-poll timed out cleanly (empty `Messages` response). Caller
    /// sleeps then retries.
    Empty,
    /// Wire returned an error / unexpected variant. Caller logs +
    /// retries; the surface impl is responsible for tracing the
    /// detail before returning this.
    TransportError,
}

/// Per-role wire surface. Two impls — `AgentSurface`, `ManagerSurface`
/// — wrap the disjoint `Request`/`Response` enums plus a handful of
/// boot-time constants that vary by role. Every other function in this
/// binary that talks to the broker goes through this trait so the turn
/// loop itself has zero per-role branches.
trait Surface {
    /// MCP flavor passed to `TurnFiles::prepare`. Picks which static
    /// system-prompt block + tool registration goes into the spawned
    /// `claude` process.
    const FLAVOR: mcp::Flavor;
    /// Ack the in-flight turn. Logs warnings on transport/broker
    /// errors but never propagates — turn loop continues either way.
    fn ack_turn(socket: &Path) -> impl Future<Output = ()>;

    /// Requeue any messages that were "in-flight" (delivered but not
    /// ack'd) — fires on harness boot to recover from a crash mid-turn.
    fn requeue_inflight(socket: &Path) -> impl Future<Output = ()>;

    /// Current inbox unread count via `Status`. Returns 0 on any
    /// transport/wire error so the caller falls through cleanly.
    fn inbox_unread(socket: &Path) -> impl Future<Output = u64>;

    /// `(open_threads, open_reminders)` for the post-turn stats row.
    /// Either field is `None` when the underlying request errors.
    fn post_turn_counts(
        socket: &Path,
    ) -> impl Future<Output = (Option<u64>, Option<u64>)>;

    /// Send a message addressed to `<parent>` (broker resolves the
    /// sentinel via `topology::parent_of` at delivery time; root
    /// agents/manager fall through to operator).
    fn send_to_parent(socket: &Path, body: String) -> impl Future<Output = ()>;

    /// Fire a `Wake { from: "self", body: "continue" }` at our own
    /// inbox — the request_next_turn sentinel pickup.
    fn self_wake(socket: &Path) -> impl Future<Output = ()>;

    /// Long-poll the broker for the next message. Wraps the
    /// `Messages`/empty/error trichotomy in `RecvOutcome` so the
    /// generic `serve_loop` doesn't need the per-role Response enum
    /// at all.
    fn recv_next(socket: &Path) -> impl Future<Output = RecvOutcome>;

    /// External `wake` subcommand (the `hive wake` CLI command, used
    /// by co-process daemons like matrix to push events into the
    /// harness inbox). Errors out via `anyhow::bail!` so the calling
    /// binary surfaces them on stderr.
    fn wake_external(
        socket: &Path,
        from: String,
        body: String,
    ) -> impl Future<Output = Result<()>>;
}

// ---------- AgentSurface ----------

/// Zero-sized type tag for the sub-agent wire surface.
/// Talks `AgentRequest` / `AgentResponse`.
struct AgentSurface;

impl Surface for AgentSurface {
    const FLAVOR: mcp::Flavor = mcp::Flavor::Agent;

    async fn 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 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 inbox_unread(socket: &Path) -> u64 {
        match client::request::<_, AgentResponse>(socket, &AgentRequest::Status).await {
            Ok(AgentResponse::Status { unread }) => unread,
            _ => 0,
        }
    }

    async fn post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) {
        let threads =
            match client::request::<_, AgentResponse>(socket, &AgentRequest::GetLooseEnds { agent: None }).await {
                Ok(AgentResponse::LooseEnds { loose_ends }) => {
                    u64::try_from(loose_ends.len()).ok()
                }
                _ => None,
            };
        let reminders = match client::request::<_, AgentResponse>(
            socket,
            &AgentRequest::CountPendingReminders { agent: None },
        )
        .await
        {
            Ok(AgentResponse::PendingRemindersCount { count }) => Some(count),
            _ => None,
        };
        (threads, reminders)
    }

    async fn send_to_parent(socket: &Path, body: String) {
        let res = client::request::<_, AgentResponse>(
            socket,
            &AgentRequest::Send {
                to: hive_sh4re::PARENT_RECIPIENT.into(),
                body,
                in_reply_to: None,
            },
        )
        .await;
        if let Err(e) = res {
            tracing::warn!(error = ?e, "failed to notify parent of turn failure");
        }
    }

    async fn self_wake(socket: &Path) {
        let res = client::request::<_, AgentResponse>(
            socket,
            &AgentRequest::Wake {
                from: "self".into(),
                body: "continue".into(),
            },
        )
        .await;
        match res {
            Ok(AgentResponse::Ok) => {
                tracing::info!("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 recv_next(socket: &Path) -> RecvOutcome {
        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");
                RecvOutcome::Message(first)
            }
            Ok(AgentResponse::Messages { .. }) => RecvOutcome::Empty,
            Ok(AgentResponse::Err { message }) => {
                tracing::warn!(%message, "recv error");
                RecvOutcome::TransportError
            }
            Ok(other) => {
                tracing::warn!(?other, "recv produced unexpected response kind");
                RecvOutcome::TransportError
            }
            Err(e) => {
                tracing::warn!(error = ?e, "recv failed; retrying");
                RecvOutcome::TransportError
            }
        }
    }

    async fn wake_external(socket: &Path, from: String, body: String) -> Result<()> {
        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:?}"),
        }
    }
}

// ---------- ManagerSurface ----------

/// Zero-sized type tag for the manager wire surface.
/// Talks `ManagerRequest` / `ManagerResponse`.
struct ManagerSurface;

impl Surface for ManagerSurface {
    const FLAVOR: mcp::Flavor = mcp::Flavor::Manager;

    async fn 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 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 inbox_unread(socket: &Path) -> u64 {
        match client::request::<_, ManagerResponse>(socket, &ManagerRequest::Status).await {
            Ok(ManagerResponse::Status { unread }) => unread,
            _ => 0,
        }
    }

    async fn 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)
    }

    async fn send_to_parent(socket: &Path, body: String) {
        let res = client::request::<_, ManagerResponse>(
            socket,
            &ManagerRequest::Send {
                to: hive_sh4re::PARENT_RECIPIENT.into(),
                body,
                in_reply_to: None,
            },
        )
        .await;
        if let Err(e) = res {
            tracing::warn!(error = ?e, "failed to notify parent of turn failure");
        }
    }

    async fn self_wake(socket: &Path) {
        let res = client::request::<_, ManagerResponse>(
            socket,
            &ManagerRequest::Wake {
                from: "self".into(),
                body: "continue".into(),
            },
        )
        .await;
        match res {
            Ok(ManagerResponse::Ok) => {
                tracing::info!("request_next_turn: injected self-continue wake");
            }
            Ok(ManagerResponse::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 recv_next(socket: &Path) -> RecvOutcome {
        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");
                RecvOutcome::Message(first)
            }
            Ok(ManagerResponse::Messages { .. }) => RecvOutcome::Empty,
            Ok(ManagerResponse::Err { message }) => {
                tracing::warn!(%message, "recv error");
                RecvOutcome::TransportError
            }
            Ok(other) => {
                tracing::warn!(?other, "recv produced unexpected response kind");
                RecvOutcome::TransportError
            }
            Err(e) => {
                tracing::warn!(error = ?e, "recv failed; retrying");
                RecvOutcome::TransportError
            }
        }
    }

    async fn wake_external(socket: &Path, from: String, body: String) -> Result<()> {
        let resp: ManagerResponse =
            client::request(socket, &ManagerRequest::Wake { from, body }).await?;
        match resp {
            ManagerResponse::Ok => Ok(()),
            ManagerResponse::Err { message } => anyhow::bail!("wake: {message}"),
            other => anyhow::bail!("wake: unexpected response {other:?}"),
        }
    }
}

// ---------- generic turn loop ----------

/// Per-role boot — wires up the web UI, login state, stats, plugins,
/// forge notifier, and either drops into `serve_loop` directly
/// (`Online`) or parks on the login flow first (`NeedsLogin`). See
/// `docs/turn-loop.md::Boot wiring`.
async fn serve_main<S: Surface>(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);
    // `HIVE_LABEL` is set unconditionally by the meta-flake envelope
    // for any container-deployed agent; the `"hive"` fallback here
    // covers standalone `nix run .#hive` invocations and pre-meta
    // dev shells. Role-independent: no semantic reason for the
    // fallback to differ when the env var is missing.
    let label = std::env::var("HIVE_LABEL").unwrap_or_else(|_| "hive".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, S::FLAVOR).await?;
    let turn_lock: TurnLock = Arc::new(tokio::sync::Mutex::new(()));
    // Plugin install runs role-agnostic: failures come back as a
    // Vec<String> and we route each through `<parent>` via the same
    // `send_to_parent` failure-notify path the turn loop uses. The
    // broker resolves `<parent>` per `topology::parent_of`; root
    // agents and the manager fall through to operator.
    for failure in plugins::install_configured(socket).await {
        S::send_to_parent(socket, failure).await;
    }
    tokio::spawn(hive_ag3nt::forge_notify::run(socket.to_path_buf()));
    // Log web_ui::serve's error instead of dropping it. A bare
    // `tokio::spawn(web_ui::serve(...))` discards the JoinHandle, so
    // any Err (e.g. EACCES from `bind_unix` when HIVE_WEB_SOCKET points
    // at a dir the agent user can't write) vanishes — leaving an
    // operator with no log line and no socket, debuggable only by
    // staring at lifecycle.rs.
    let web_ui_args = (
        label.clone(),
        port,
        login_state.clone(),
        bus.clone(),
        socket.to_path_buf(),
        files.clone(),
        turn_lock.clone(),
    );
    tokio::spawn(async move {
        let (label, port, login_state, bus, socket, files, turn_lock) = web_ui_args;
        if let Err(e) =
            web_ui::serve(label, port, login_state, bus, socket, files, turn_lock).await
        {
            tracing::error!(error = %e, "web_ui::serve exited with error");
        }
    });
    if matches!(initial, LoginState::NeedsLogin) {
        turn::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await;
    } else {
        // Clear any stale `hyperhive-needs-login` sentinel left over
        // from a prior boot — `online` status writes the sentinel
        // cleanup in `Bus::emit_status`.
        bus.emit_status("online");
    }
    serve_loop::<S>(
        socket,
        Duration::from_millis(poll_ms),
        login_state,
        claude_dir,
        bus,
        stats,
        &files,
        turn_lock,
    )
    .await
}

/// The long-running message loop. Long-polls the broker via
/// `S::recv_next`, drives a turn per message, parks on auth-failed,
/// otherwise retries.
#[allow(clippy::too_many_arguments)]
async fn serve_loop<S: Surface>(
    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(), "harness serve");
    S::requeue_inflight(socket).await;
    loop {
        match S::recv_next(socket).await {
            RecvOutcome::Message(first) => {
                let auth_failed =
                    handle_turn::<S>(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;
                }
            }
            RecvOutcome::Empty => {
                tokio::time::sleep(interval).await;
            }
            RecvOutcome::TransportError => {
                // `recv_next` already logged the detail; just retry.
                // No backoff: the long-poll wait is itself the throttle.
            }
        }
    }
}

/// Drive a single turn: emit boot-of-turn events, run claude, ack on
/// success / requeue on rate-limit-or-401 / notify parent on failure,
/// record stats, then pick up the `request_next_turn` sentinel if it's
/// been dropped during the turn. Returns true iff the outcome was
/// `AuthFailed` — the caller flips the harness to needs-login.
async fn handle_turn<S: Surface>(
    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;
    log_system_event(bus, &from, &body);
    tracing::info!(%from, %body, %redelivered, "inbox");
    let unread = S::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) {
        S::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;
        S::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");
        S::requeue_inflight(socket).await;
    }
    if let turn::TurnOutcome::Failed(e) = &outcome {
        S::send_to_parent(socket, format_turn_failure(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) = S::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 = S::inbox_unread(socket).await;
    if pending > 0 {
        tracing::info!(%pending, "pending messages after turn; fetching next");
    }
    if consume_continue_sentinel() {
        S::self_wake(socket).await;
    }
    matches!(outcome, turn::TurnOutcome::AuthFailed)
}

/// External `hive wake` subcommand — push a message into our own
/// inbox so the next turn fires with the given body. Reads the body
/// from stdin when `body == "-"`.
async fn wake<S: Surface>(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
    };
    S::wake_external(socket, from, body).await
}