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

forge_notify: leave delivered threads unread, dedupe wakes in-memory

Browse source
This commit is contained in:
damocles 2026-06-22 16:23:04 +02:00
commit f5ac6d79e3
2 changed files with 143 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

43
docs/forge.md
View file

@ -70,9 +70,46 @@ Background task spawned once per harness boot. Polls
`GET /api/v1/notifications?all=false` every 30 seconds (Forgejo's
unread-only filter), formats each notification as a broker
`Wake { from: "forge" }` message, and delivers it to the agent's own
inbox so claude's normal turn loop picks it up. Mark-read happens
after successful delivery so a failed-delivery notification
resurfaces on the next tick.
inbox so claude's normal turn loop picks it up.
### Mark-read on read, not on delivery
Delivered conversation threads are deliberately left **unread** in
forge. The hive-forge read-before-comment guard keys off forge's own
notification read-state (`GET /notifications?all=false`) to refuse a
comment when a thread has unread activity by others — so the agent
reading the thread via the CLI (`hive-forge comments` / `view`,
which `PATCH`es `/notifications/threads/{id}`) is the single
mark-read point. If `forge_notify` marked threads read on delivery,
that unread signal would be consumed before the agent acts and the
guard could never fire.
Because a delivered thread stays unread, it reappears in every
`?all=false` poll. An in-memory **delivery-dedupe cursor** (thread
id → last-delivered `updated_at`, held in the poll loop) stops the
same version from re-firing a wake; a new comment bumps `updated_at`
so genuinely new activity re-delivers. The cursor is pure anti-spam,
not a correctness oracle: lost on harness restart it just
re-delivers currently-unread threads once (harmless — `recv`
tolerates redelivery), so it carries none of the persisted-mirror
fragility that ruled out an on-disk seen-cursor. Each poll prunes
the cursor to the threads still in the unread set. A failed wake
delivery is left unread **and** out of the cursor, so it resurfaces
next tick.
Two paths still mark-read directly (no read-before-comment value):
self-echo notifications (the agent's own writes, see below) and
`HIVE_FORGE_NOTIFY_SKIP_REASONS` drop-listed reasons.
> Note: the unread list grows for threads the agent never reads via
> the CLI, since nothing else trims it. This does not affect guard
> correctness (the guard does a per-thread, repo-scoped query) nor
> wake delivery (Forgejo orders unread newest-first, so new activity
> always lands in the polled window). Bounding the unread list via a
> reason-independent firehose-reduction is a separate follow-up — the
> existing auto-unsubscribe below is gated on a `reason` field that
> this Forgejo's notification API does not actually emit, so it never
> fires today.
### Activation gates (graceful no-ops)

116
hive-ag3nt/src/forge_notify.rs
View file

@ -1,7 +1,13 @@
//! Background Forgejo notification poller. Polls
//! `GET /notifications?all=false` every 30s, formats each unread
//! notification as a broker `Wake { from: "forge" }` message, and
//! marks it read after delivery so failures resurface next tick.
//! 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. An in-memory delivery-dedupe cursor
//! (thread id → last-delivered `updated_at`) stops the still-unread
//! notification from re-firing a wake every poll; self-echo and
//! drop-listed notifications are still marked read directly.
//!
//! Activation gates, self-notification filtering, body excerpt +
//! truncation + heading escape, wrapper formats (comment / review /
@ -9,7 +15,7 @@
//! reason drop-list, and auto-unsubscribe on broad watches all live
//! in [`docs/forge.md::Notification poller`](../../../docs/forge.md).
use std::collections::HashSet;
use std::collections::{HashMap, HashSet};
use std::fmt::Write as _;
use std::path::{Path, PathBuf};
use std::time::Duration;
@ -139,6 +145,19 @@ pub async fn run(socket: PathBuf) {
// across polls so we don't hammer DELETE on every cycle.
let mut unsubbed_repos: HashSet<String> = HashSet::new();
// 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 in-memory
// 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:
// lost on harness restart it just re-delivers currently-unread
// threads once (harmless — recv tolerates redelivery), so it carries
// none of the persisted-mirror fragility that sank the on-disk
// cursor approach.
let mut delivered: HashMap<u64, String> = HashMap::new();
loop {
interval.tick().await;
poll_once(
@ -148,6 +167,7 @@ pub async fn run(socket: PathBuf) {
&socket,
keep_subscriptions,
&mut unsubbed_repos,
&mut delivered,
&own_login,
&skip_reasons,
)
@ -723,9 +743,9 @@ fn days_from_civil(y: i64, m: i64, d: i64) -> i64 {
#[allow(
clippy::too_many_arguments,
reason = "the notification poll's config + mutable subscription state, \
wired once from the poll loop; a struct would just move the \
same fields one level out"
reason = "the notification poll's config + mutable subscription / \
delivery-dedupe state, wired once from the poll loop; a struct \
would just move the same fields one level out"
)]
#[allow(
clippy::too_many_lines,
@ -740,6 +760,7 @@ async fn poll_once(
socket: &Path,
keep_subscriptions: bool,
unsubbed_repos: &mut HashSet<String>,
delivered: &mut HashMap<u64, String>,
own_login: &str,
skip_reasons: &[String],
) {
@ -784,6 +805,17 @@ async fn poll_once(
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`.
let updated_at = notif["updated_at"].as_str().unwrap_or("").to_owned();
if !should_deliver(delivered, id, &updated_at) {
debug!(%id, "forge_notify: skipping (already delivered this version)");
continue;
}
// Reason drop-list: suppress noisy reasons; null/unknown pass
// through so directed signals stay deliverable (see
// `docs/forge.md::Reason drop-list`).
@ -809,10 +841,10 @@ async fn poll_once(
body,
transient: false,
};
let delivered = crate::client::request::<_, hive_sh4re::Response>(socket, &req)
let deliver_result = crate::client::request::<_, hive_sh4re::Response>(socket, &req)
.await
.map(|_| ());
match delivered {
match deliver_result {
Ok(()) => {
debug!(%id, "forge_notify: delivered");
}
@ -822,9 +854,13 @@ async fn poll_once(
}
}
// Mark as read only after successful delivery so a failed-delivery
// notification resurfaces on the next poll tick.
mark_read(client, forge_url, token, id).await;
// 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. A failed delivery (above) is left
// unrecorded so it re-delivers next tick.
delivered.insert(id, updated_at);
// Auto-unsubscribe from broad repo watches after delivering a
// `subscribed` notification. Gated by HIVE_FORGE_KEEP_SUBSCRIPTIONS
@ -856,12 +892,34 @@ 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.
let current_ids: HashSet<u64> = notifications
.iter()
.filter_map(|n| n["id"].as_u64())
.collect();
delivered.retain(|id, _| current_ids.contains(id));
}
/// Whether a notification should be delivered as a wake given the
/// delivery-dedupe cursor. Delivers when the thread has never been
/// delivered, or when its `updated_at` advanced since the last delivered
/// version (genuinely new activity). Pure for unit testing.
fn should_deliver(delivered: &HashMap<u64, String>, id: u64, updated_at: &str) -> bool {
delivered.get(&id).is_none_or(|seen| seen != updated_at)
}
/// Mark a notification thread as read. Best-effort — logs on failure but
/// does not abort the poll loop. A notification left unread will resurface
/// on the next poll tick (desirable for delivery failures; for self-echo
/// silencing we call this without prior delivery).
/// does not abort the poll loop. Called only on the self-echo and
/// drop-listed paths (the agent's own writes / explicitly-suppressed
/// reasons) — 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.
async fn mark_read(client: &reqwest::Client, forge_url: &str, token: &str, id: u64) {
let mark_url = format!("{forge_url}/api/v1/notifications/threads/{id}");
match client
@ -886,6 +944,38 @@ async fn mark_read(client: &reqwest::Client, forge_url: &str, token: &str, id: u
mod tests {
use super::*;
#[test]
fn should_deliver_when_thread_never_seen() {
let delivered = HashMap::new();
assert!(should_deliver(&delivered, 42, "2026-06-22T16:00:00Z"));
}
#[test]
fn should_not_deliver_same_version_again() {
// The dedupe case: an unread thread reappears every poll with the
// same `updated_at` — must not re-fire a wake.
let mut delivered = HashMap::new();
delivered.insert(42, "2026-06-22T16:00:00Z".to_owned());
assert!(!should_deliver(&delivered, 42, "2026-06-22T16:00:00Z"));
}
#[test]
fn should_deliver_when_updated_at_advanced() {
// A new comment bumps `updated_at` → genuinely new activity →
// deliver again.
let mut delivered = HashMap::new();
delivered.insert(42, "2026-06-22T16:00:00Z".to_owned());
assert!(should_deliver(&delivered, 42, "2026-06-22T16:05:00Z"));
}
#[test]
fn should_deliver_tracks_per_thread() {
// A cursor for one thread says nothing about another.
let mut delivered = HashMap::new();
delivered.insert(42, "2026-06-22T16:00:00Z".to_owned());
assert!(should_deliver(&delivered, 99, "2026-06-22T16:00:00Z"));
}
#[test]
fn escape_md_headings_escapes_top_level_atx() {
// Argus reviews start with `## argus review`, which would