forge_notify: auto-unsubscribe from repo watches on subscribed notifications

hive-ag3nt/src/forge_notify.rs

@@ -17,6 +17,7 @@
 //! `PATCH /notifications/threads/{id}` so it does not re-fire. If delivery
 //! fails the thread is left unread so it resurfaces next tick.
 
+use std::collections::HashSet;
 use std::path::{Path, PathBuf};
 use std::time::Duration;
 
@@ -80,9 +81,13 @@ pub async fn run(socket: PathBuf, is_manager: bool) {
     // socket becoming available right at boot.
     interval.tick().await;
 
+    // Repos we have already unsubscribed this process lifetime. Persists
+    // across polls so we don't hammer DELETE on every cycle.
+    let mut unsubbed_repos: HashSet<String> = HashSet::new();
+
     loop {
         interval.tick().await;
-        poll_once(&client, &forge_url, &token, &socket, is_manager).await;
+        poll_once(&client, &forge_url, &token, &socket, is_manager, &mut unsubbed_repos).await;
     }
 }
 
@@ -222,7 +227,14 @@ async fn format_notification(
     }
 }
 
-async fn poll_once(client: &reqwest::Client, forge_url: &str, token: &str, socket: &Path, is_manager: bool) {
+async fn poll_once(
+    client: &reqwest::Client,
+    forge_url: &str,
+    token: &str,
+    socket: &Path,
+    is_manager: bool,
+    unsubbed_repos: &mut HashSet<String>,
+) {
     let url = format!("{forge_url}/api/v1/notifications?all=false&limit=50");
     let resp = match client
         .get(&url)
@@ -310,5 +322,37 @@ async fn poll_once(client: &reqwest::Client, forge_url: &str, token: &str, socke
                 debug!(%id, "forge_notify: marked read");
             }
         }
+
+        // Auto-unsubscribe from broad repo watches. If the notification
+        // reason is "subscribed" (agent is watching the whole repo) and
+        // the agent has no personal stake (was just watching), drop the
+        // watch subscription so we don't accumulate firehose noise from
+        // repos we contributed to but are no longer actively working in.
+        // Thread-level subscriptions (specific issue/PR) are unaffected.
+        let reason = notif["reason"].as_str().unwrap_or("");
+        if reason == "subscribed" {
+            if let Some(repo) = notif["repository"]["full_name"].as_str() {
+                if !unsubbed_repos.contains(repo) {
+                    let unsub_url = format!("{forge_url}/api/v1/repos/{repo}/subscription");
+                    match client
+                        .delete(&unsub_url)
+                        .header("Authorization", format!("token {token}"))
+                        .send()
+                        .await
+                    {
+                        Ok(r) if r.status().is_success() || r.status().as_u16() == 404 => {
+                            debug!(%repo, "forge_notify: unsubscribed from repo watch");
+                            unsubbed_repos.insert(repo.to_owned());
+                        }
+                        Ok(r) => {
+                            debug!(%repo, status = %r.status(), "forge_notify: unsub non-2xx (ignored)");
+                        }
+                        Err(e) => {
+                            debug!(%repo, error = ?e, "forge_notify: unsub request failed (ignored)");
+                        }
+                    }
+                }
+            }
+        }
     }
 }