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(#2106): persist forge-notify dedupe cursor to disk so rebuilds don't re-deliver the unread backlog

Browse source
This commit is contained in:
damocles 2026-07-01 18:37:08 +02:00 • committed by mara
commit 2549cc51ba
2 changed files with 164 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

33
docs/forge.md
View file

@ -92,17 +92,28 @@ 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.
`?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.
The cursor is **persisted** to
`$HYPERHIVE_STATE_DIR/forge-notify-cursor.json` (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). 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
file 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 corrupt
cursor file degrades to empty — re-deliver the unread set once — never
an abort.
Self-echo notifications (the agent's own writes, see below) are the
one path still marked-read directly (no read-before-comment value).

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

@ -4,10 +4,14 @@
//! 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.
//! 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 and drop-listed notifications
//! are still marked read directly. The cursor is persisted to
//! `$HYPERHIVE_STATE_DIR/forge-notify-cursor.json` (atomic tmp+rename) so
//! a container rebuild/restart doesn't re-deliver the whole currently-
//! unread backlog — the on-disk cursor is a private dedup mirror, NOT
//! forge's read-state, so the read-before-comment guard is untouched.
//!
//! Activation gates, self-notification filtering, body excerpt +
//! truncation + heading escape, wrapper formats (comment / review /
@ -123,15 +127,35 @@ pub async fn run(socket: PathBuf) {
// 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();
// 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 to `$HYPERHIVE_STATE_DIR/forge-notify-cursor.json`
// and reloaded here on boot so a container rebuild/restart doesn't
// re-deliver the entire currently-unread backlog (#2106). 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 file 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 cursor_path: Option<PathBuf> = if state_dir.is_empty() {
None
} else {
Some(PathBuf::from(format!(
"{state_dir}/forge-notify-cursor.json"
)))
};
let mut delivered: HashMap<u64, String> =
cursor_path.as_deref().map(load_cursor).unwrap_or_default();
if !delivered.is_empty() {
info!(
entries = delivered.len(),
"forge_notify: restored delivery-dedupe cursor from disk"
);
}
loop {
interval.tick().await;
@ -141,12 +165,53 @@ pub async fn run(socket: PathBuf) {
&token,
&socket,
&mut delivered,
cursor_path.as_deref(),
&own_login,
)
.await;
}
}
/// Load the persisted delivery-dedupe cursor from disk. Returns an empty
/// map on any error — a missing file (first boot) or corrupt JSON both
/// degrade to the pre-persistence behaviour (re-deliver the currently-
/// unread set once) rather than aborting the poller.
fn load_cursor(path: &Path) -> HashMap<u64, String> {
match std::fs::read_to_string(path) {
Ok(s) => serde_json::from_str(&s).unwrap_or_else(|e| {
warn!(?path, error = %e, "forge_notify: cursor parse failed — starting empty");
HashMap::new()
}),
Err(e) => {
debug!(?path, error = %e, "forge_notify: no cursor file — starting empty");
HashMap::new()
}
}
}
/// Persist the delivery-dedupe cursor atomically (write tmp + rename).
/// Best-effort: logs on failure and lets the poll loop continue. Safe to
/// call after a successful broker delivery because the wake is already
/// durably queued in the broker sqlite inbox — recording "delivered" here
/// can never swallow a notification the agent hasn't received.
async fn persist_cursor(path: &Path, delivered: &HashMap<u64, String>) {
let json = match serde_json::to_string(delivered) {
Ok(j) => j,
Err(e) => {
warn!(?path, error = %e, "forge_notify: cursor serialize failed");
return;
}
};
let tmp = path.with_extension("json.tmp");
if let Err(e) = tokio::fs::write(&tmp, json.as_bytes()).await {
warn!(?tmp, error = %e, "forge_notify: cursor tmp write failed");
return;
}
if let Err(e) = tokio::fs::rename(&tmp, path).await {
warn!(?path, error = %e, "forge_notify: cursor rename failed");
}
}
/// Fetch a JSON value from a URL using the agent's forge token. Returns
/// `None` on any HTTP or parse error (best-effort enrichment).
async fn fetch_json(client: &reqwest::Client, url: &str, token: &str) -> Option<serde_json::Value> {
@ -723,6 +788,7 @@ async fn poll_once(
token: &str,
socket: &Path,
delivered: &mut HashMap<u64, String>,
cursor_path: Option<&Path>,
own_login: &str,
) {
let url = format!("{forge_url}/api/v1/notifications?all=false&limit=50");
@ -761,6 +827,11 @@ async fn poll_once(
"forge_notify: delivering notifications"
);
// 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["id"].as_u64() else {
continue;
@ -804,6 +875,7 @@ async fn poll_once(
// here in the Ok arm — a failed delivery leaves the cursor
// untouched, so it re-delivers next tick.
delivered.insert(id, updated_at);
cursor_dirty = true;
}
Err(e) => {
warn!(%id, error = ?e, "forge_notify: deliver failed — leaving unread");
@ -821,7 +893,17 @@ async fn poll_once(
.iter()
.filter_map(|n| n["id"].as_u64())
.collect();
let before_prune = delivered.len();
delivered.retain(|id, _| current_ids.contains(id));
if delivered.len() != before_prune {
cursor_dirty = true;
}
// Flush the cursor to disk only when it changed, so a rebuild/restart
// reloads it instead of re-delivering the whole unread backlog (#2106).
if cursor_dirty && let Some(path) = cursor_path {
persist_cursor(path, delivered).await;
}
}
/// Whether a notification should be delivered as a wake given the
@ -894,6 +976,53 @@ mod tests {
assert!(should_deliver(&delivered, 99, "2026-06-22T16:00:00Z"));
}
#[test]
fn cursor_serde_roundtrips_u64_keys() {
// serde_json stringifies integer map keys; make sure the
// round-trip preserves the u64 thread ids the cursor is keyed on.
let mut delivered = HashMap::new();
delivered.insert(42u64, "2026-06-22T16:00:00Z".to_owned());
delivered.insert(99u64, "2026-06-22T17:30:00Z".to_owned());
let json = serde_json::to_string(&delivered).unwrap();
let back: HashMap<u64, String> = serde_json::from_str(&json).unwrap();
assert_eq!(delivered, back);
}
#[test]
fn load_cursor_missing_file_is_empty() {
// First boot: no cursor file yet → empty map (re-deliver once).
let path = std::env::temp_dir().join("forge-notify-cursor-missing-xyzzy.json");
let _ = std::fs::remove_file(&path);
assert!(load_cursor(&path).is_empty());
}
#[test]
fn load_cursor_corrupt_file_is_empty() {
// A truncated / garbage cursor degrades to empty rather than
// aborting the poller.
let path = std::env::temp_dir().join(format!(
"forge-notify-cursor-corrupt-{}.json",
std::process::id()
));
std::fs::write(&path, "{not valid json").unwrap();
assert!(load_cursor(&path).is_empty());
let _ = std::fs::remove_file(&path);
}
#[test]
fn load_cursor_reads_written_map() {
// The happy path: a persisted cursor reloads to the same map.
let path = std::env::temp_dir().join(format!(
"forge-notify-cursor-rw-{}.json",
std::process::id()
));
let mut delivered = HashMap::new();
delivered.insert(7u64, "2026-06-22T16:00:00Z".to_owned());
std::fs::write(&path, serde_json::to_string(&delivered).unwrap()).unwrap();
assert_eq!(load_cursor(&path), delivered);
let _ = std::fs::remove_file(&path);
}
#[test]
fn escape_md_headings_escapes_top_level_atx() {
// Argus reviews start with `## argus review`, which would