//! Loose-ends aggregator. Walks the `approvals` + `operator_questions` //! tables once per call and assembles a `Vec` for either //! a single agent (`for_agent`) or the whole hive (`hive_wide`). Both //! `AgentRequest::GetLooseEnds` and `ManagerRequest::GetLooseEnds` //! land here so the routing logic + age-seconds derivation stay in //! one place. //! //! 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 indexes //! (`idx_approvals_pending` + `idx_operator_questions_pending`) that //! the dashboard uses, so the bottleneck would be json //! (de)serialisation, not the read. use anyhow::Result; use hive_sh4re::LooseEnd; use crate::coordinator::Coordinator; use hive_sh4re::wire_time::now_unix; /// 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; /// - unanswered questions where `agent` is the asker (waiting on /// someone) OR the target (owes a reply); /// - pending reminders this agent scheduled (`owner == self`). /// /// Ordered `pending_messages` (when non-zero) → approvals → questions → /// reminders within the returned vector. Within each kind, /// source-of-truth ordering (sqlite's `pending()` queries return /// newest-first within their indexes). /// /// # Errors /// /// Propagates errors from `count_pending` and the pending-approval / /// question / reminder sqlite queries. pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { let now = now_unix(); 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, commit_ref: a.commit_ref, description: a.description, age_seconds: saturating_age(now, a.requested_at.timestamp()), }); } for q in coord.questions.pending_all()? { let role_match = q.asker == agent || q.target.as_deref() == Some(agent); if !role_match { continue; } out.push(LooseEnd::Question { id: q.id, asker: q.asker, target: q.target, question: q.question, age_seconds: saturating_age(now, q.asked_at.timestamp()), }); } for r in coord.broker.list_pending_reminders()? { if r.agent != agent { continue; } out.push(LooseEnd::Reminder { id: r.id, owner: r.agent, message: r.message, due_at: r.due_at, age_seconds: saturating_age(now, r.created_at.timestamp()), }); } Ok(out) } /// Hive-wide loose-ends view: EVERY pending approval + EVERY /// unanswered question + EVERY pending reminder. 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> { let now = now_unix(); let mut out = Vec::new(); for a in coord.approvals.pending()? { out.push(LooseEnd::Approval { id: a.id, agent: a.agent, commit_ref: a.commit_ref, description: a.description, age_seconds: saturating_age(now, a.requested_at.timestamp()), }); } for q in coord.questions.pending_all()? { out.push(LooseEnd::Question { id: q.id, asker: q.asker, target: q.target, question: q.question, age_seconds: saturating_age(now, q.asked_at.timestamp()), }); } for r in coord.broker.list_pending_reminders()? { out.push(LooseEnd::Reminder { id: r.id, owner: r.agent, message: r.message, due_at: r.due_at, age_seconds: saturating_age(now, r.created_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); } }