hyperhive/hive-c0re/src/loose_ends.rs

//! Loose-ends aggregator. Walks the `approvals` table once per call and
//! assembles a `Vec<LooseEnd>` for either a single agent (`for_agent`) or
//! the whole hive (`hive_wide`). `Request::GetLooseEnds` from either the
//! agent or manager socket lands here so the routing logic + age-seconds
//! derivation stay in one place. Reminders AND questions are agent-local
//! (in-container stores, `hive-agent::reminders` / `hive-agent::questions`)
//! and no longer sourced from here (loose-ends-v2's questions phase) —
//! c0re remains the `Ask`/`Answer` routing + delivery rendezvous
//! (`coord.questions`), it just isn't asked for the *pending-view*
//! rendering anymore. The
//! operator dashboard's questions pane is unaffected: it reads
//! `coord.questions.pending_all()` directly (`dashboard/state_snapshot.rs`),
//! independent of this module.
//!
//! Call frequency is low (an agent doing self-introspection between
//! turns), so the sweep happens fresh every time — no caching, no
//! mutation events. If the sweep ever shows up in a profile, the sqlite
//! queries already filter on the same index (`idx_approvals_pending`)
//! that the dashboard uses, so the bottleneck would be json
//! (de)serialisation, not the read.

use anyhow::Result;
use chrono::Utc;
use hive_sh4re::LooseEnd;

use crate::coordinator::Coordinator;

/// Open threads pending against `agent`:
/// - undelivered inbox messages this agent still owes itself a `recv`
///   for (only when the count is non-zero);
/// - pending approvals where this agent is the submitter (a parent
///   agent with the `approvals` group submits for its children; the
///   root submits for top-level agents). Legacy rows with no recorded
///   submitter count as the root's.
///
/// Ordered `pending_messages` (when non-zero) → approvals within the
/// returned vector. Within each kind, source-of-truth ordering (sqlite's
/// `pending()` query returns newest-first within its index).
///
/// # Errors
///
/// Propagates errors from `count_pending` and the pending-approval
/// sqlite query.
pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<LooseEnd>> {
    let now = Utc::now().timestamp();
    let mut out = Vec::new();
    // Undelivered inbox messages this agent still owes itself a `recv`
    // for. Surfaced first (most actionable) and only when non-zero so a
    // clean inbox doesn't add noise. The wake-message that drove the
    // current turn is already delivered, so this counts only what's
    // genuinely still queued.
    let pending_messages = coord.broker.count_pending(agent)?;
    if pending_messages > 0 {
        out.push(LooseEnd::PendingMessages {
            count: pending_messages,
        });
    }
    // Show each pending approval to the agent that submitted it. The
    // submitter column is NULL for rows predating it; those count as
    // operator-initiated (no agent tracking predates the column).
    for a in coord.approvals.pending()? {
        let submitter = coord
            .approvals
            .submitter_of(a.id)?
            .unwrap_or_else(|| "operator".to_owned());
        if submitter != agent {
            continue;
        }
        out.push(LooseEnd::Approval {
            id: a.id,
            agent: a.agent.to_string(),
            commit_ref: a.commit_ref,
            description: a.description,
            age_seconds: saturating_age(now, a.requested_at.timestamp()),
        });
    }
    Ok(out)
}

/// Hive-wide loose-ends view: EVERY pending approval. Manager surface
/// only; sub-agents can't see each other's threads via the agent surface
/// (`for_agent` filters by name).
pub fn hive_wide(coord: &Coordinator) -> Result<Vec<LooseEnd>> {
    let now = Utc::now().timestamp();
    let mut out = Vec::new();
    for a in coord.approvals.pending()? {
        out.push(LooseEnd::Approval {
            id: a.id,
            agent: a.agent.to_string(),
            commit_ref: a.commit_ref,
            description: a.description,
            age_seconds: saturating_age(now, a.requested_at.timestamp()),
        });
    }
    Ok(out)
}

fn saturating_age(now: i64, then: i64) -> u64 {
    let delta = now.saturating_sub(then);
    u64::try_from(delta).unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn saturating_age_handles_clock_back_step() {
        // `now` < `then`: caller's clock went backwards between rows.
        // We saturate to 0 rather than returning a negative or
        // wrapping around to ~u64::MAX (which would render as "27
        // billion years ago" in the wake prompt).
        assert_eq!(saturating_age(100, 200), 0);
    }

    #[test]
    fn saturating_age_normal_case() {
        assert_eq!(saturating_age(1_000_000, 999_990), 10);
    }

    #[test]
    fn saturating_age_zero_when_equal() {
        assert_eq!(saturating_age(42, 42), 0);
    }
}