1e67f56249
`age_seconds` is documented on the LooseEnd enum as saturating to zero on any clock anomaly, but the derivation was in three places: hive-c0re had a named `saturating_age` helper with tests, and the in-agent socket server hand-rolled the same two lines twice, untested. Move the helper to hive-sh4re::inbox, beside the enum whose contract it implements and inside the one crate both producers already depend on. Its three tests move with it (not dropped) and gain two arms: the whole-i64 range, where the saturating_sub is what stops the subtraction overflowing, and a far-past control so those zeros are the clamp firing rather than the function bottoming out on large inputs. The two clamps are not redundant, which is what `to_loose_end`'s doc got wrong: it credited "saturating" for the zero, but saturating_sub bottoms out at i64::MIN, still negative. The try_from is what yields 0. Also cover the two projections themselves, which is the part the shared helper cannot: that a reminder ages from created_at rather than due_at, and a todo from updated_at, with a future timestamp reading 0 through both and a past-timestamp control on each.
95 lines
4 KiB
Rust
95 lines
4 KiB
Rust
//! 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 are agent-local (in-container
|
|
//! `hive-agent::reminders` store) and never sourced from here. The
|
|
//! `ask`/`answer` MCP tools, their wire protocol (`hive-c0re::questions`,
|
|
//! `stores::operator_questions`), and the operator dashboard's questions
|
|
//! pane have all been removed entirely — this file never had a
|
|
//! `Question` loose-end path to begin with (it only ever emitted
|
|
//! `PendingMessages`/`Approval`), so nothing here changed shape when
|
|
//! that removal landed.
|
|
//!
|
|
//! 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::inbox::{LooseEnd, saturating_age};
|
|
|
|
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)
|
|
}
|