diff --git a/docs/forge.md b/docs/forge.md index 5895d659..b0de1a4c 100644 --- a/docs/forge.md +++ b/docs/forge.md @@ -91,66 +91,48 @@ 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 on read, not on delivery +### Mark-read 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. +On a **successful** broker delivery, `forge_notify` marks the thread +read on forge straight away (`PATCH /notifications/threads/{id}`). The +broker inbox is the durable work queue now — each delivered wake is a +sqlite row with its own ack lifecycle — so the forge unread flag no +longer needs to track whether the agent has *processed* a +notification. Clearing it on delivery keeps forge's unread set **tiny +by construction**: at rest it holds only threads that failed to +deliver plus whatever arrived since the last 30s poll. -Because a delivered thread stays unread, it reappears in every -`?all=false` poll. A **delivery-dedupe cursor** (thread id → -last-delivered `updated_at`) 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. -Each poll prunes it 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. +That size property is the whole point. A container rebuild starts the +poller with no memory of what it delivered, re-scans `?all=false`, and +finds nothing stale — the delivered threads are already read on forge. +Forge's own read-state is thus the durable, cross-rebuild record of +what's been delivered; there is **no persisted cursor**. (This +replaced an earlier design that left threads unread and leaned on a +persisted dedup cursor: a rebuild that lost the cursor re-delivered the +entire still-unread backlog as fresh wakes — the notification flood of +#2593 / #2106.) -**Size bound:** the per-poll prune retains only ids present in the -single `limit=UNREAD_FETCH_LIMIT` (50) fetch page, so the cursor never -exceeds that many entries — it tracks the unread _window_, not the -all-time notification count. The fetch limit and the bound are the -same constant in `forge_notify.rs` (with a debug assertion), so a -future pagination change grows the ceiling visibly rather than -silently. This is why the cursor stays a small JSON field rather than -a db table — see the storage discussion on the tracker (issue 2117). +**Read-before-comment coupling, dropped on purpose.** The old design +left threads unread so the hive-forge read-before-comment guard (which +keys off forge unread-state) would force the agent to view a thread +before commenting. That coupling is gone: the broker wake already +carries the notification body, so *delivery is the read*. An agent that +wants the full thread still runs `hive-forge comments` / `view`; the +guard no longer blocks a first comment on a freshly-delivered thread. -The cursor is **persisted** as the `forge_cursor` field of the -harness's consolidated `hyperhive-harness.json` state file (atomic -tmp+rename, flushed only when it changed) and reloaded on boot, so a -container rebuild/restart doesn't re-deliver the whole currently-unread -backlog (#2106 — previously the in-memory-only cursor was lost on -restart and every old still-unread thread re-fired a wake). The poller -runs in the same harness process that owns that file, so it's one -daemon → one state file rather than a second json; both writers -(turn-loop fields + this cursor) go read-modify-write under a shared -lock so neither clobbers the other's fields. This is safe because a -thread is recorded **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. Crucially the cursor is -a private dedup mirror, **not** forge's read-state: it does not -reintroduce the read-before-comment coupling that ruled out the old -mark-read-on-delivery approach. A missing (first boot) or malformed -cursor degrades to empty — re-deliver the unread set once — never an -abort. +**In-process dedupe (tiny, ephemeral).** A single-process map (thread +id → last-delivered `updated_at`) guards the narrow window where a +mark-read call *transiently fails* and the thread reappears unread in +the next poll before its `updated_at` bumps — so a flaky PATCH doesn't +re-fire the wake. It is **not persisted** and resets on restart (forge +read-state covers the durable case). Each poll prunes it to the ids in +the single `limit=UNREAD_FETCH_LIMIT` (50) fetch page, so it can never +exceed that many entries (a debug assertion pins the invariant; the +fetch limit and the bound are the same constant). A failed *delivery* +is left unread and out of the map, so it resurfaces next tick. -Self-echo notifications (the agent's own writes, see below) are the -one path still marked-read directly (no read-before-comment value). - -> 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 is a -> separate follow-up — explicit subscription management via a -> hive-forge CLI verb, rather than the poller second-guessing which -> repo watches to drop. +Self-echo notifications (the agent's own writes, see below) are marked +read directly without a delivery — same `mark_read` call, no wake. ### Activation gates (graceful no-ops) diff --git a/hive-agent/src/forge_notify.rs b/hive-agent/src/forge_notify.rs index e701e4b7..2ecae583 100644 --- a/hive-agent/src/forge_notify.rs +++ b/hive-agent/src/forge_notify.rs @@ -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 = 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 = HashMap::new(); loop { interval.tick().await; @@ -1054,21 +1040,16 @@ async fn poll_once( let notifications: Vec = 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 ¬ifications { 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 = 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, 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 diff --git a/hive-agent/src/harness_state.rs b/hive-agent/src/harness_state.rs index a7b49968..18f02168 100644 --- a/hive-agent/src/harness_state.rs +++ b/hive-agent/src/harness_state.rs @@ -140,7 +140,7 @@ pub(crate) fn read_harness_state() -> (bool, bool, Option) { } /// 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 { - 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 { - 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( - cursor: &std::collections::HashMap, -) { - 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() {