hyperhive/hive-ag3nt/src/serve_common.rs

//! Pure helpers factored out of the `hive` serve loop (`bin/hive.rs`).
//! Only functions with no wire-type dependency live here;
//! request/response-flavored helpers (`requeue_inflight`, `ack_turn`, etc.)
//! stay in the binary because they touch the request enum variants directly.

use crate::events::Bus;
use crate::mcp::REDELIVERY_HINT;
use crate::turn::{TurnError, TurnOutcome};
use crate::turn_stats::TurnStatRow;

/// Assemble the per-turn wake prompt string. The role/tools/etc. live in the
/// system prompt; this is just the wake signal body. `id` is the broker row
/// id, rendered as a `[msg #<id>]` marker so the agent can reference it in
/// `ack_until`. `unread` is the inbox depth after this message was popped.
/// `redelivered` prepends a "may already be handled" banner.
#[must_use]
pub fn format_wake_prompt(
    id: i64,
    from: &str,
    body: &str,
    unread: u64,
    redelivered: bool,
) -> String {
    let banner = if redelivered { REDELIVERY_HINT } else { "" };
    let tag = if id > 0 {
        format!("[msg #{id}] ")
    } else {
        String::new()
    };
    let pending = if unread == 0 {
        String::new()
    } else {
        // Suggested batch size is clamped to the server-side recv cap
        // so the hint never asks for more than one round-trip can
        // deliver.
        let batch = unread.min(u64::from(hive_sh4re::RECV_BATCH_MAX));
        format!(
            "\n\n({unread} more message(s) pending in your inbox — call `mcp__hyperhive__recv` \
             with `max: {batch}` to drain the next batch before acting. If the \
             backlog is stale/already handled, `ack_until(up_to: <highest [msg #N] seen>)` \
             clears everything up to that id in one call instead.)"
        )
    };
    format!("{banner}{tag}Incoming message from `{from}`:\n---\n{body}\n---{pending}")
}

/// Current time as a Unix timestamp (seconds). Returns 0 on any error.
#[must_use]
pub fn now_unix() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .ok()
        .and_then(|d| i64::try_from(d.as_secs()).ok())
        .unwrap_or(0)
}

/// Field-named args for [`build_row`]. Mirrors the turn-stats row
/// columns; `outcome` and `bus` borrow for the duration of the call.
pub struct TurnRowArgs<'a> {
    pub started_at: i64,
    pub ended_at: i64,
    pub duration_ms: i64,
    pub model: String,
    pub wake_from: String,
    pub outcome: &'a TurnOutcome,
    pub bus: &'a Bus,
    pub open_threads_count: Option<u64>,
    pub open_reminders_count: Option<u64>,
}

/// Assemble a `TurnStatRow` from the harness's per-turn state. Lives here
/// (rather than inline in the `hive` serve loop) so it stays wire-type-free
/// and unit-testable; the binary just feeds it the post-turn counts.
#[must_use]
pub fn build_row(args: TurnRowArgs<'_>) -> TurnStatRow {
    let TurnRowArgs {
        started_at,
        ended_at,
        duration_ms,
        model,
        wake_from,
        outcome,
        bus,
        open_threads_count,
        open_reminders_count,
    } = args;
    // Prefer the API-resolved model id (e.g. `claude-opus-4-8`) captured
    // from this turn's assistant events over the requested `--model`
    // name/alias, so the model-mix + cost rollup label the concrete
    // version that ran. Falls back to the requested name on a degenerate
    // turn that produced no assistant event.
    let model = bus.last_resolved_model().unwrap_or(model);
    let cost = bus.last_cost_usage().unwrap_or_default();
    let ctx = bus.last_ctx_usage().unwrap_or(cost);
    let tool_calls = bus.take_tool_calls();
    let tool_call_count: u64 = tool_calls.values().copied().sum();
    let tool_call_breakdown_json = if tool_calls.is_empty() {
        None
    } else {
        serde_json::to_string(&tool_calls).ok()
    };
    let (result_kind, note) = match outcome {
        Ok(false) => ("ok", None),
        Ok(true) => ("compacted", None),
        Err(TurnError::PromptTooLong) => ("prompt_too_long", None),
        Err(TurnError::RateLimited) => ("rate_limited", None),
        Err(TurnError::AuthFailed) => ("auth_failed", None),
        Err(TurnError::Failed(e)) => ("failed", Some(format!("{e:#}"))),
    };
    let wake_from = if wake_from.starts_with("bash-task-") {
        "bash-task".to_owned()
    } else {
        wake_from
    };
    TurnStatRow {
        started_at,
        ended_at,
        duration_ms,
        model,
        wake_from,
        input_tokens: cost.input_tokens,
        output_tokens: cost.output_tokens,
        cache_read_input_tokens: cost.cache_read_input_tokens,
        cache_creation_input_tokens: cost.cache_creation_input_tokens,
        last_input_tokens: ctx.input_tokens,
        last_output_tokens: ctx.output_tokens,
        last_cache_read_input_tokens: ctx.cache_read_input_tokens,
        last_cache_creation_input_tokens: ctx.cache_creation_input_tokens,
        tool_call_count,
        tool_call_breakdown_json,
        open_threads_count,
        open_reminders_count,
        result_kind,
        note,
        session_id: bus.current_session_id(),
    }
}