vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(#2593): mark forge notifications read on broker-delivery

Browse source
This commit is contained in:
damocles 2026-07-19 18:21:04 +02:00 • committed by mara
commit da056d0043
3 changed files with 104 additions and 212 deletions
Download patch file Download diff file Expand all files Collapse all files

148
hive-agent/src/forge_notify.rs
View file

@ -1,18 +1,19 @@
//! Background Forgejo notification poller. Polls
//! `GET /notifications?all=false` every 30s, formats each unread
//! notification as a broker `Wake { from: "forge" }` message, and
//! delivers it to the agent's inbox. Delivered threads are deliberately
//! left UNREAD in forge — the hive-forge read-before-comment guard keys
//! off forge's own unread-state, and the agent reading the thread via the
//! CLI is what marks it read. A delivery-dedupe cursor (thread id →
//! last-delivered `updated_at`) stops the still-unread notification from
//! re-firing a wake every poll; self-echo notifications (the agent's own
//! writes) are still marked read directly. The cursor is persisted as the
//! `forge_cursor` field of the harness's consolidated `hyperhive-harness.json`
//! (via [`crate::events`]) and reloaded on boot so a container
//! rebuild/restart doesn't re-deliver the whole currently-unread backlog —
//! it is a private dedup mirror, NOT forge's read-state, so the
//! read-before-comment guard is untouched.
//! notification as a broker `Wake { from: "forge" }` message, delivers it
//! to the agent's inbox, and — on a successful delivery — marks the thread
//! read on forge straight away. The broker inbox is the durable work queue
//! (each row has its own ack lifecycle), so the forge unread flag no longer
//! needs to track agent processing: clearing it on delivery keeps forge's
//! unread set tiny by construction, so a container rebuild's re-scan of
//! `?all=false` finds nothing stale and cannot re-deliver a backlog. Forge's
//! own read-state is thus the durable, cross-rebuild record of what's been
//! delivered — there is no persisted cursor. A small in-process dedupe map
//! (thread id → last-delivered `updated_at`) only guards the narrow window
//! where a mark-read call transiently fails and the thread reappears unread
//! before its `updated_at` bumps; it is ephemeral and reset on restart.
//! Self-echo notifications (the agent's own writes) are marked read without
//! a delivery.
//!
//! Activation gates, self-notification filtering, body excerpt +
//! truncation + heading escape, wrapper formats (comment / review /
@ -37,10 +38,9 @@ const POLL_INTERVAL_SECS: u64 = 30;
/// `forgejo-api` client (which exposes no timeout knob of its own).
const HTTP_TIMEOUT_SECS: u64 = 10;
/// Page size of the unread-notifications fetch. This is also the hard
/// bound on the persisted delivery-dedupe cursor: each poll prunes the
/// cursor to the ids in this window, so the `forge_cursor` field in
/// `hyperhive-harness.json` can never exceed this many entries. Keep
/// the two coupled — bumping the fetch limit grows the cursor's ceiling
/// bound on the in-process dedupe map: each poll prunes the map to the
/// ids in this window, so it can never exceed this many entries. Keep
/// the two coupled — bumping the fetch limit grows the map's ceiling
/// with it, deliberately and visibly.
const UNREAD_FETCH_LIMIT: usize = 50;
/// Maximum characters of a body/comment to include in the wake message.
@ -160,31 +160,17 @@ pub async fn run(socket: PathBuf) {
info!(forge_url = %forge_url, "forge_notify: polling started");
// Delivery-dedupe cursor: notification thread id -> the `updated_at`
// of the version we last woke the agent for. We no longer mark a
// thread read on delivery (that would consume the unread signal the
// hive-forge read-before-comment guard relies on), so this map is what
// stops the same unread notification from re-firing a wake every poll.
// A new comment bumps `updated_at`, so the thread re-delivers. This is
// purely anti-spam, NOT a correctness oracle.
//
// It is persisted as the `forge_cursor` field of the harness's
// consolidated state file and reloaded here on boot so a container
// rebuild/restart doesn't re-deliver the entire currently-unread
// backlog. Persisting is safe because we only record a thread AFTER a
// successful broker delivery, and the broker inbox is durable sqlite —
// so a persisted "delivered" entry can never swallow a wake the agent
// never received. The cursor is a private dedup mirror, decoupled from
// forge's own read-state, so it doesn't reintroduce the
// read-before-comment coupling that the mark-read-on-delivery approach
// suffered.
let mut delivered: HashMap<u64, String> = crate::harness_state::read_forge_cursor();
if !delivered.is_empty() {
info!(
entries = delivered.len(),
"forge_notify: restored delivery-dedupe cursor"
);
}
// In-process delivery-dedupe map: notification thread id -> the
// `updated_at` of the version we last woke the agent for. A delivered
// thread is marked read on forge (in `poll_once`), so it drops out of
// `?all=false` next poll and never re-fires; this map only guards the
// narrow window where a mark-read call transiently fails and the thread
// reappears unread before its `updated_at` bumps. It is deliberately
// ephemeral (NOT persisted): forge's own read-state is the durable,
// cross-rebuild source of truth for what's been delivered, so a rebuild
// starts with an empty map and re-scans only the genuinely-still-unread
// set — which is tiny by construction because delivery marks read.
let mut delivered: HashMap<u64, String> = HashMap::new();
loop {
interval.tick().await;
@ -1054,21 +1040,16 @@ async fn poll_once(
let notifications: Vec<PolledNotification> =
values.into_iter().filter_map(parse_notification).collect();
// Tracks whether the dedupe cursor changed this poll (a new delivery
// recorded, or the prune below dropped now-read threads) so we only
// rewrite the on-disk cursor when there's something to persist.
let mut cursor_dirty = false;
for notif in &notifications {
let Some(id) = notif.thread.id.and_then(|id| u64::try_from(id).ok()) else {
continue;
};
// Delivery-dedupe: we no longer mark threads read on delivery, so
// an unread notification reappears in every `?all=false` poll.
// Skip it silently unless its `updated_at` advanced since the
// version we last delivered a wake for (i.e. genuinely new
// activity). See the `delivered` cursor note in `run`.
// In-process delivery-dedupe: guards against re-firing a wake for a
// thread whose mark-read (below) transiently failed and so still
// shows up unread in the next `?all=false` poll. Skip unless its
// `updated_at` advanced since the version we last delivered (i.e.
// genuinely new activity). See the `delivered` note in `run`.
let updated_at = notif.updated_at.clone();
if !should_deliver(delivered, id, &updated_at) {
debug!(%id, "forge_notify: skipping (already delivered this version)");
@ -1093,15 +1074,19 @@ async fn poll_once(
match deliver_result {
Ok(()) => {
debug!(%id, "forge_notify: delivered");
// Record the delivered version in the dedupe cursor INSTEAD
// of marking the thread read. Leaving it unread is
// deliberate: the hive-forge read-before-comment guard keys
// off forge's own unread-state, and the agent reading the
// thread via the CLI is what marks it read. Recorded only
// here in the Ok arm — a failed delivery leaves the cursor
// untouched, so it re-delivers next tick.
// Mark the thread read on forge immediately after a
// successful broker delivery. The broker inbox is the
// durable work queue now (each row has its own ack
// lifecycle), so the forge unread flag no longer needs to
// track agent processing — clearing it on delivery keeps
// forge's unread set tiny by construction, so a container
// rebuild re-scan finds nothing stale to re-deliver. The
// in-memory `delivered` entry below is only a within-process
// guard so a transient mark-read failure doesn't re-fire the
// wake next tick; it is deliberately NOT persisted — forge's
// own read-state is the cross-rebuild source of truth.
mark_read(forge, id).await;
delivered.insert(id, updated_at);
cursor_dirty = true;
}
Err(e) => {
warn!(%id, error = ?e, "forge_notify: deliver failed — leaving unread");
@ -1109,39 +1094,24 @@ async fn poll_once(
}
}
// Prune the dedupe cursor down to the threads still present in this
// poll's unread set. Once the agent reads a thread (marking it read
// via the CLI) it drops out of `?all=false`, so its cursor entry is
// dead weight; dropping it bounds the map to the current unread size.
// If such a thread later goes unread again it carries a fresh
// `updated_at` and re-delivers correctly.
// Prune the in-process dedupe map to the threads still present in this
// poll's unread set. A delivered thread is marked read above, so it
// drops out of `?all=false` next poll and its entry becomes dead
// weight; dropping it bounds the map to the current unread size.
//
// This retain IS the cursor's size bound: `current_ids` comes from a
// single `limit=UNREAD_FETCH_LIMIT` page, so the persisted cursor can
// never exceed that many entries — it tracks the unread *window*, not
// the all-time notification count. The assert makes the invariant
// loud in tests/dev if a future pagination change silently breaks it.
// `current_ids` comes from a single `limit=UNREAD_FETCH_LIMIT` page, so
// the map can never exceed that many entries — the assert makes that
// invariant loud in tests/dev if a future pagination change breaks it.
let current_ids: HashSet<u64> = notifications
.iter()
.filter_map(|n| n.thread.id.and_then(|id| u64::try_from(id).ok()))
.collect();
let before_prune = delivered.len();
delivered.retain(|id, _| current_ids.contains(id));
debug_assert!(
delivered.len() <= UNREAD_FETCH_LIMIT,
"dedupe cursor exceeded the fetch window ({} > {UNREAD_FETCH_LIMIT})",
"in-process dedupe map exceeded the fetch window ({} > {UNREAD_FETCH_LIMIT})",
delivered.len(),
);
if delivered.len() != before_prune {
cursor_dirty = true;
}
// Flush the cursor to the consolidated state file only when it
// changed, so a rebuild/restart reloads it instead of re-delivering
// the whole unread backlog.
if cursor_dirty {
crate::harness_state::write_forge_cursor(delivered);
}
}
/// Whether a notification should be delivered as a wake given the
@ -1153,11 +1123,13 @@ fn should_deliver(delivered: &HashMap<u64, String>, id: u64, updated_at: &str) -
}
/// Mark a notification thread as read. Best-effort — logs on failure but
/// does not abort the poll loop. Called only on the self-echo path (the
/// agent's own comment/review/creation writes) — delivered threads are
/// deliberately left unread for the read-before-comment guard, and a failed
/// delivery is left unread + out of the dedupe cursor so it resurfaces on the
/// next poll tick.
/// does not abort the poll loop. Called on two paths: right after a
/// successful broker delivery (so forge's unread set stays tiny and a
/// rebuild can't re-deliver), and on the self-echo path (the agent's own
/// comment/review/creation writes, delivered nowhere). A failed delivery
/// leaves the thread unread + out of the in-process dedupe map so it
/// resurfaces on the next poll tick; a transient mark-read failure after a
/// good delivery is caught by that same dedupe map (no duplicate wake).
async fn mark_read(forge: &Forgejo, id: u64) {
let Ok(thread_id) = i64::try_from(id) else {
// Thread ids originate from `i64` in the poll parse, so an

66
hive-agent/src/harness_state.rs
View file

@ -140,7 +140,7 @@ pub(crate) fn read_harness_state() -> (bool, bool, Option<String>) {
}
/// Write the turn-loop's harness state fields via a read-modify-write so
/// any other writer's fields (e.g. `forge_notify`'s `forge_cursor`) survive.
/// any other writer's fields in the consolidated state file survive.
/// Pass `active_model: Some(s)` to update the resolved model (surfaced in
/// the dashboard badge); `None` leaves the stored value untouched.
pub(crate) fn write_harness_state(
@ -160,40 +160,6 @@ pub(crate) fn write_harness_state(
write_harness_json(&v);
}
/// Parse the `forge_notify` delivery-dedupe cursor (notification thread id
/// -> last-delivered `updated_at`) out of a harness-state JSON value.
/// Empty when the field is absent (first boot) or malformed.
fn forge_cursor_from_json(v: &serde_json::Value) -> std::collections::HashMap<u64, String> {
v.get("forge_cursor")
.cloned()
.and_then(|c| serde_json::from_value(c).ok())
.unwrap_or_default()
}
/// Restore the `forge_notify` delivery-dedupe cursor from the consolidated
/// state file so a container rebuild/restart doesn't re-deliver the whole
/// currently-unread backlog.
pub fn read_forge_cursor() -> std::collections::HashMap<u64, String> {
forge_cursor_from_json(&read_harness_json())
}
/// Persist the `forge_notify` delivery-dedupe cursor into the consolidated
/// state file, read-modify-write under the shared lock so the turn-loop's
/// own fields survive. Best-effort: a serialize failure is a no-op.
pub fn write_forge_cursor<S: std::hash::BuildHasher>(
cursor: &std::collections::HashMap<u64, String, S>,
) {
let Ok(value) = serde_json::to_value(cursor) else {
return;
};
let _guard = HARNESS_JSON_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let mut v = read_harness_json();
v["forge_cursor"] = value;
write_harness_json(&v);
}
/// Compiled-in fallback model used when neither `HIVE_DEFAULT_MODEL` nor a
/// persisted runtime override is present.
pub const DEFAULT_MODEL: &str = "haiku";
@ -280,35 +246,7 @@ pub fn context_window_tokens(model: &str) -> u64 {
#[cfg(test)]
mod tests {
use super::{DEFAULT_EFFORT, EFFORT_LEVELS, forge_cursor_from_json, is_valid_effort};
use serde_json::json;
#[test]
fn forge_cursor_absent_field_is_empty() {
// First boot / a state file that only carries the turn-loop fields:
// no cursor yet, so we re-deliver the currently-unread set once.
assert!(forge_cursor_from_json(&json!({ "rate_limited": false })).is_empty());
}
#[test]
fn forge_cursor_malformed_field_is_empty() {
// A wrong-typed / corrupt cursor degrades to empty rather than
// aborting the poller.
assert!(forge_cursor_from_json(&json!({ "forge_cursor": "nonsense" })).is_empty());
}
#[test]
fn forge_cursor_roundtrips_u64_keys() {
// serde_json stringifies integer map keys; confirm the u64 thread
// ids the cursor is keyed on survive the JSON round-trip.
let v = json!({
"forge_cursor": { "42": "2026-06-22T16:00:00Z", "99": "2026-06-22T17:30:00Z" }
});
let cursor = forge_cursor_from_json(&v);
assert_eq!(cursor.len(), 2);
assert_eq!(cursor.get(&42), Some(&"2026-06-22T16:00:00Z".to_owned()));
assert_eq!(cursor.get(&99), Some(&"2026-06-22T17:30:00Z".to_owned()));
}
use super::{DEFAULT_EFFORT, EFFORT_LEVELS, is_valid_effort};
#[test]
fn effort_validation_accepts_only_known_levels() {