diff --git a/docs/forge.md b/docs/forge.md index f3bf9d00..a2dfd793 100644 --- a/docs/forge.md +++ b/docs/forge.md @@ -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) diff --git a/hive-ag3nt/src/forge_notify.rs b/hive-ag3nt/src/forge_notify.rs index 86e0753b..b99026a1 100644 --- a/hive-ag3nt/src/forge_notify.rs +++ b/hive-ag3nt/src/forge_notify.rs @@ -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 = 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 = 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, + delivered: &mut HashMap, 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 = 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, 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