diff --git a/CLAUDE.md b/CLAUDE.md index 8d1ca614..1c72fa66 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -62,10 +62,12 @@ hand-maintained per-file tree drifts out of sync with the code. `docs/boundary.md`. - **`hive-forge/`** — `hive-forge` Forgejo CLI wrapper; one module per verb under `src/verbs/`. -- **`hive-forge-notify/`** — per-agent Forgejo notification poller daemon - (`hive-forge-notify`); turns unread notification threads into todos on - the harness's in-agent socket. Was a task inside the `hive-agent` serve - loop; own process since it needs nothing else from the harness. +- **`hive-forge-notify/`** — per-agent notification poller daemons; turns + unread notification threads into todos on the harness's in-agent + socket. Two binaries from one crate: `hive-forge-notify` (the hive's + Forgejo) and `hive-github-notify` (github.com, installed by + `nix/agent-modules/github.nix`). Was a task inside the `hive-agent` + serve loop; own process since it needs nothing else from the harness. - **`hive-matrix-mcp/`** — per-agent matrix-sdk daemon (`hive-matrix-daemon`); serves its MCP tools (`send_message`, `read_room`, …) directly over streamable-http (no stdio bridge), diff --git a/docs/forge.md b/docs/forge.md index 46699cb5..6eaacbd7 100644 --- a/docs/forge.md +++ b/docs/forge.md @@ -85,14 +85,24 @@ reqwest. ## Notification poller (`hive-forge-notify/src/notify.rs`) -Its own long-running per-agent daemon (`hive-forge-notify`, one systemd -unit, a sibling of `hive-bash-daemon` / `hive-matrix-daemon`) — it used +Its own long-running per-agent daemon (`hive-forge-notify`, a sibling of +`hive-bash-daemon` / `hive-matrix-daemon`) — it used to be a background task inside the `hive-agent` serve loop. 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. +The crate builds a second, independent binary for a different host — see +[github.md](github.md#notifications). + +The host-specific calls — list unread, mark read, resolve own login — +live behind `Source` in `hive-forge-notify/src/source.rs`; +classification, formatting, dedupe and todo delivery are shared. Todo +keys here are the bare thread ids, and must stay that way: renaming them +would orphan every in-flight forge todo on the first restart after a +deploy. + ### Mark-read on delivery On a **successful** broker delivery, `forge_notify` marks the thread diff --git a/docs/github.md b/docs/github.md index 0f6c14cd..8e69915f 100644 --- a/docs/github.md +++ b/docs/github.md @@ -70,6 +70,63 @@ file `0600` owned by the agent (so the container can read it) — the same credential-injection path as forge/matrix tokens. See [hivectl → GitHub](tools/hivectl.md#github). +## Notifications + +`hive-github-notify` polls github.com for the agent, turning each unread +notification thread into a todo. It is a **separate binary and a +separate systemd unit** from the internal forge's poller +(`hive-forge-notify`, see [forge.md](forge.md#notification-poller-hive-forge-notifysrcnotifyrs)), +installed by `nix/agent-modules/github.nix` under +`hyperhive.github.enable`. Both binaries ship from the one +`hive-forge-notify` derivation, so the unit is a second `ExecStart` +path, not a new package. + +Two units rather than one daemon with two loops, because it puts the +decision in nix: a hive built without this module has **no github poller +in its closure at all**, which is what makes GitHub access separable +rather than merely switched off. It is also why this is not a cargo +feature — a feature would unify across the workspace and cost every +crate its build cache. + +At runtime the poller needs the PAT above. No PAT, no polling: the unit +logs why and exits 0, which is why it is `Restart = on-failure` and +never `always` — a clean exit on a PAT-less agent must not become a +restart loop. + +Forgejo's notifications API is modelled on GitHub's, so one tolerant +parse serves both: `id`, `repository.full_name`, +`subject {title,url,latest_comment_url}` and `updated_at` line up field +for field. The two real differences are absorbed by lenient +deserializers — GitHub sends the thread id as a *string* where Forgejo +sends a number, and says `PullRequest` where Forgejo says `Pull`. Todo +keys are prefixed `gh:` so a github thread id cannot collide with a +forge one. + +Two host differences worth knowing before touching this code: + +- **Auth scheme, not just value.** Forgejo takes + `Authorization: token `; GitHub wants `Bearer ` plus `Accept: + application/vnd.github+json`, `X-GitHub-Api-Version` and a + `User-Agent`. Sending Forgejo's form to GitHub does not error — it + authenticates as *nobody* and silently drops to the unauthenticated + rate limit. The cheap way to tell the two apart is the rate-limit + header: `x-ratelimit-remaining` near 5000 is an authenticated user, + near 60 is anonymous. +- **GitHub sets the cadence.** It returns `X-Poll-Interval` (60s in + practice, slower than our own tick) and rate-limits callers who ignore + it, so the loop re-arms to the server's interval whenever that is + *slower* than ours. A hint faster than our own tick is not a reason to + poll harder. + +⚠️ **This needs the `notifications` scope on the PAT.** A token minted +for `gh` + `git push` typically carries `repo` only, which is enough to +push and open PRs but **not** to read the notification stream (nor to +mark a thread read, which is the same scope). A PAT without it doesn't +break anything: the poller logs the refusal and stays quiet, and the +agent simply never gets GitHub wakes. If an agent's GitHub +notifications never arrive, check the token's scopes first — the +symptom is silence, not an error. + ## Security - Use a **dedicated bot account**, never a human's. diff --git a/flake.nix b/flake.nix index 8141c9bc..d32a796a 100644 --- a/flake.nix +++ b/flake.nix @@ -89,6 +89,7 @@ hive-bash-daemon hive-forge hive-forge-notify + hive-github-notify hive-matrix-daemon hive-metric hive-screen-mcp diff --git a/hive-forge-notify/src/bin/hive-forge-notify/main.rs b/hive-forge-notify/src/bin/hive-forge-notify/main.rs new file mode 100644 index 00000000..9eedfbcc --- /dev/null +++ b/hive-forge-notify/src/bin/hive-forge-notify/main.rs @@ -0,0 +1,206 @@ +//! Notification poller for the hive's internal Forgejo. +//! +//! Polls unread threads, turns each into a todo on the harness's in-agent +//! socket, and marks it read. The shared half — classification, wake +//! formatting, dedupe, delivery — lives in the library; this binary owns +//! the Forgejo protocol (`source.rs`) and the forge-only assigned-issue +//! rollup below. + +mod source; + +use std::collections::HashMap; +use std::path::Path; +use std::time::Duration; + +use hive_forge_notify::notify::{ + POLL_INTERVAL_SECS, TOKEN_RETRY_MAX, TOKEN_RETRY_SECS, poll_once, resolve_own_login, +}; +use hive_forge_notify::{HTTP_TIMEOUT_SECS, TODO_SOCKET_RETRY, agent_socket, init_tracing}; +use source::ForgejoSource; +use tracing::{debug, info, warn}; + +#[tokio::main] +async fn main() { + init_tracing(); + forgejo_loop(hive_forge_notify::state_dir(), agent_socket()).await; +} + +/// Returns — ending the process — when the forge is not configured for +/// this agent, which is a supported state and not a failure. The unit is +/// `Restart = on-failure` for exactly this reason. +async fn forgejo_loop(state_dir: String, socket: std::path::PathBuf) { + let forge_url = match std::env::var("HIVE_FORGE_URL") { + Ok(u) if !u.is_empty() => u, + _ => { + debug!("forge_notify: HIVE_FORGE_URL not set — disabled"); + return; + } + }; + + let token_path = format!("{state_dir}/forge-token"); + // Retry reading the token to handle races where hive-priv provisions + // it after the harness starts, or where a parent-container chown + // briefly makes the file unreadable. + let token = { + let mut attempts = 0u32; + loop { + match tokio::fs::read_to_string(&token_path).await { + Ok(t) => { + let t = t.trim().to_owned(); + if !t.is_empty() { + break t; + } + debug!("forge_notify: empty forge token at {token_path}"); + } + Err(e) => debug!("forge_notify: cannot read token at {token_path}: {e}"), + } + attempts += 1; + if attempts >= TOKEN_RETRY_MAX { + debug!( + "forge_notify: token not available after {TOKEN_RETRY_MAX} retries — disabled" + ); + return; + } + tokio::time::sleep(Duration::from_secs(TOKEN_RETRY_SECS)).await; + } + }; + + let Some(source) = ForgejoSource::new(&forge_url, &token) else { + return; + }; + + let client = match reqwest::Client::builder() + .timeout(Duration::from_secs(HTTP_TIMEOUT_SECS)) + .build() + { + Ok(c) => c, + Err(e) => { + warn!("forge_notify: failed to build HTTP client: {e}"); + return; + } + }; + + // Self-notification filtering degrades safely: an empty login means + // no filtering, and a boot-time failure (forge not yet reachable) is + // re-attempted on each tick rather than staying off for the process + // lifetime. + let mut own_login = resolve_own_login(&client, &source).await; + if own_login.is_empty() { + warn!( + "forge_notify: could not resolve own login yet — self-notification \ + filtering disabled until it resolves on a later poll" + ); + } else { + debug!(%own_login, "forge_notify: own login resolved"); + } + + let mut interval = tokio::time::interval(Duration::from_secs(POLL_INTERVAL_SECS)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + // First tick fires immediately — skip it so we don't race the broker + // socket becoming available right at boot. + interval.tick().await; + + info!(forge_url = %forge_url, "forge_notify: polling started"); + + // In-process delivery dedupe: thread id -> the `updated_at` we last + // woke the agent for. Deliberately ephemeral: forge's own read-state + // is the durable record of what's been delivered, so a rebuild starts + // empty 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; + if own_login.is_empty() { + own_login = resolve_own_login(&client, &source).await; + if !own_login.is_empty() { + debug!(%own_login, "forge_notify: own login resolved on retry"); + } + } + // Forgejo states no poll-interval hint, so the return is ignored + // and the configured cadence stands. + let _ = poll_once(&source, &client, &socket, &mut delivered, &own_login).await; + update_assigned_rollup(&client, &forge_url, &token, &socket).await; + } +} + +/// Count of open issues or PRs assigned to this agent, via Forgejo's +/// global search API. `issue_type` is `"issues"` or `"pulls"`. Reads the +/// `X-Total-Count` header rather than deserialising the body — only page 1 +/// with limit 1 is fetched, keeping the request cheap. `None` on any error. +async fn count_assigned( + client: &reqwest::Client, + forge_url: &str, + token: &str, + issue_type: &str, +) -> Option { + let url = format!( + "{forge_url}/api/v1/issues/search\ + ?type={issue_type}&state=open&assigned=true&limit=1&page=1" + ); + let resp = match client + .get(&url) + .header("Authorization", format!("token {token}")) + .send() + .await + { + Ok(r) if r.status().is_success() => r, + _ => return None, + }; + let count_str = resp.headers().get("x-total-count")?.to_str().ok()?; + count_str.parse::().ok() +} + +/// Keep a keyed `"rollup"` todo in sync with the count of open assigned +/// issues + PRs: a positive count summarises the breakdown, zero clears +/// the todo. The rollup key is distinct from per-thread keys so clearing +/// it never touches notification todos. +/// +/// Forge-only by nature — it asks the forge what is assigned to this +/// agent, which is not a notification-protocol concern and has no +/// github.com counterpart here. +async fn update_assigned_rollup( + client: &reqwest::Client, + forge_url: &str, + token: &str, + socket: &Path, +) { + let issues = count_assigned(client, forge_url, token, "issues") + .await + .unwrap_or(0); + let pulls = count_assigned(client, forge_url, token, "pulls") + .await + .unwrap_or(0); + let total = issues + pulls; + + let req = if total == 0 { + hive_agent_sock::Request::ClearTodo { + subsystem: "forge".to_owned(), + key: Some("rollup".to_owned()), + all: false, + } + } else { + let breakdown = match (issues, pulls) { + (i, 0) => format!("{i} issue{}", if i == 1 { "" } else { "s" }), + (0, p) => format!("{p} PR{}", if p == 1 { "" } else { "s" }), + (i, p) => format!( + "{i} issue{}, {p} PR{}", + if i == 1 { "" } else { "s" }, + if p == 1 { "" } else { "s" } + ), + }; + hive_agent_sock::Request::UpsertTodo { + subsystem: "forge".to_owned(), + key: Some("rollup".to_owned()), + summary: format!("{total} open assigned: {breakdown}"), + source: None, + } + }; + + match hive_sock_client::request::<_, hive_agent_sock::Response>(socket, &req, TODO_SOCKET_RETRY) + .await + { + Ok(_) => debug!(total, "forge_notify: assigned rollup todo updated"), + Err(e) => debug!("forge_notify: assigned rollup todo update failed: {e}"), + } +} diff --git a/hive-forge-notify/src/bin/hive-forge-notify/source.rs b/hive-forge-notify/src/bin/hive-forge-notify/source.rs new file mode 100644 index 00000000..6fa27d14 --- /dev/null +++ b/hive-forge-notify/src/bin/hive-forge-notify/source.rs @@ -0,0 +1,151 @@ +//! The internal Forgejo's half of the poller — the protocol code that +//! only this binary links. +//! +//! Uses the typed `forgejo-api` client for the calls with stable shapes +//! (identity probe, notification list, mark-read). Pages come back as an +//! opaque `String` and are parsed per-item upstream: one unrepresentable +//! field (a merged PR's `subject.state = "merged"`) must not poison the +//! whole page. + +use std::time::Duration; + +use forgejo_api::structs::{NotifyGetListQuery, NotifyReadThreadQuery}; +use forgejo_api::{Auth, Forgejo, ForgejoError}; +use hive_forge_notify::notify::{HTTP_TIMEOUT_SECS, UNREAD_FETCH_LIMIT}; +use hive_forge_notify::source::Source; +use tracing::{debug, warn}; + +/// The hive's internal Forgejo. +pub struct ForgejoSource { + forge: Box, + /// Base URL, kept for the assigned-issues rollup query, which goes + /// out over plain `reqwest`. + pub base_url: String, + pub token: String, +} + +impl ForgejoSource { + /// Build from the internal forge's URL + token. `None` when the URL + /// or client is unusable — the caller treats that as "not + /// configured", not as fatal. + pub fn new(base_url: &str, token: &str) -> Option { + let parsed = match url::Url::parse(base_url) { + Ok(u) => u, + Err(e) => { + warn!("forge_notify: invalid HIVE_FORGE_URL {base_url}: {e}"); + return None; + } + }; + match Forgejo::new(Auth::Token(token), parsed) { + Ok(forge) => Some(Self { + forge: Box::new(forge), + base_url: base_url.to_owned(), + token: token.to_owned(), + }), + Err(e) => { + warn!("forge_notify: failed to build forge client: {e}"); + None + } + } + } +} + +impl Source for ForgejoSource { + /// Deliberately empty: forge todo keys stay the bare thread ids they + /// have always been, so a deploy cannot orphan in-flight todos. + fn key_prefix(&self) -> &'static str { + "" + } + + fn name(&self) -> &'static str { + "forgejo" + } + + fn authorize(&self, rb: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + rb.header("Authorization", format!("token {}", self.token)) + } + + async fn own_login(&self, client: &reqwest::Client) -> String { + let url = format!("{}/api/v1/user", self.base_url.trim_end_matches('/')); + let Ok(resp) = self.authorize(client.get(&url)).send().await else { + return String::new(); + }; + if !resp.status().is_success() { + return String::new(); + } + resp.json::() + .await + .ok() + .and_then(|v| v["login"].as_str().map(str::to_owned)) + .unwrap_or_default() + } + + async fn list_unread( + &self, + _client: &reqwest::Client, + ) -> Option<(Vec, Option)> { + let query = NotifyGetListQuery { + all: Some(false), + ..NotifyGetListQuery::default() + }; + let request = self + .forge + .notify_get_list(query) + .page_size(u32::try_from(UNREAD_FETCH_LIMIT).unwrap_or(u32::MAX)) + .response_type::(); + let raw = match tokio::time::timeout(Duration::from_secs(HTTP_TIMEOUT_SECS), request.send()) + .await + { + Ok(Ok(raw)) => raw, + Ok(Err(ForgejoError::UnexpectedStatusCode(status))) => { + debug!("forge_notify: poll status {status}"); + return None; + } + Ok(Err(e)) => { + debug!("forge_notify: poll request failed: {e}"); + return None; + } + Err(_) => { + debug!("forge_notify: poll request failed: timed out"); + return None; + } + }; + + match serde_json::from_str(&raw) { + // Forgejo states no poll-interval hint, so the caller keeps + // its own cadence. + Ok(values) => Some((values, None)), + Err(e) => { + warn!("forge_notify: response parse error: {e}"); + None + } + } + } + + async fn mark_read(&self, _client: &reqwest::Client, id: &str) { + let Ok(thread_id) = id.parse::() else { + warn!(%id, "forge_notify: non-numeric forgejo thread id — cannot mark read"); + return; + }; + // `to_status: None` → Forgejo's default unread → read transition. + // The 205 body is the thread JSON, which can carry + // `subject.state = "merged"`; take it as an opaque `String` so a + // merged PR can't turn a successful mark-read into a parse error. + let request = self + .forge + .notify_read_thread(thread_id, NotifyReadThreadQuery { to_status: None }) + .response_type::(); + match tokio::time::timeout(Duration::from_secs(HTTP_TIMEOUT_SECS), request.send()).await { + Err(_) => { + warn!(%id, "forge_notify: mark-read request failed — notification will resurface"); + } + Ok(Err(e @ ForgejoError::ReqwestError(_))) => { + warn!(%id, error = ?e, "forge_notify: mark-read request failed — notification will resurface"); + } + Ok(Err(e)) => { + warn!(%id, error = %e, "forge_notify: mark-read returned non-2xx — notification will resurface"); + } + Ok(Ok(_)) => debug!(%id, "forge_notify: marked read"), + } + } +} diff --git a/hive-forge-notify/src/bin/hive-github-notify/main.rs b/hive-forge-notify/src/bin/hive-github-notify/main.rs new file mode 100644 index 00000000..947e914a --- /dev/null +++ b/hive-forge-notify/src/bin/hive-github-notify/main.rs @@ -0,0 +1,110 @@ +//! `hive-github-notify` binary — long-running per-agent **github.com** +//! notification poller. Same delivery path as its Forgejo sibling: unread +//! list, per-thread summary, todo upsert on the harness's in-agent socket, +//! mark read on the source. +//! +//! Takes no arguments. `HYPERHIVE_STATE_DIR` holds the PAT +//! (`github-token`, provisioned from the dashboard credentials tab — see +//! `docs/github.md`) and `HIVE_AGENT_SOCKET` is the harness's todo socket. +//! Having a PAT *is* the opt-in: with no token the poller logs why and +//! exits 0, so deploying this unit to an agent that never gets one costs a +//! settled process rather than a restart loop. +//! +//! ⚠️ Reading the notification stream needs the **`notifications` scope** +//! on the PAT — a token minted for `gh` + `git push` usually carries `repo` +//! only, which is enough to push and open PRs but not to read (or mark +//! read) notifications. A PAT without it is not fatal: the poller logs the +//! refusal and stays quiet, so the symptom is silence rather than an error. + +mod source; + +use std::collections::HashMap; +use std::time::Duration; + +use hive_forge_notify::notify::{ + POLL_INTERVAL_SECS, TOKEN_RETRY_MAX, TOKEN_RETRY_SECS, poll_once, resolve_own_login, +}; +use hive_forge_notify::{HTTP_TIMEOUT_SECS, agent_socket, init_tracing}; +use source::GithubSource; +use tracing::{debug, info, warn}; + +#[tokio::main] +async fn main() { + init_tracing(); + + let socket = agent_socket(); + info!(socket = %socket.display(), "hive-github-notify starting"); + + // Returns only when no PAT ever arrives; otherwise loops forever. + github_loop(hive_forge_notify::state_dir(), socket).await; +} + +/// Waits for the PAT to appear: the token is written out of band from the +/// dashboard and takes effect without a rebuild, so an agent that gains a +/// PAT mid-session starts getting notifications on the next tick rather +/// than after a restart. Gives up — returning, so the process exits 0 +/// rather than restart-looping — when no PAT ever arrives, which is the +/// common case for an agent that has the unit but no account. +async fn github_loop(state_dir: String, socket: std::path::PathBuf) { + let token_path = format!("{state_dir}/github-token"); + let mut attempts = 0u32; + let token = loop { + match tokio::fs::read_to_string(&token_path).await { + Ok(t) if !t.trim().is_empty() => break t.trim().to_owned(), + _ => debug!("forge_notify: no github token at {token_path} yet"), + } + attempts += 1; + if attempts >= TOKEN_RETRY_MAX { + debug!( + "forge_notify: no github token after {TOKEN_RETRY_MAX} retries — github disabled" + ); + return; + } + tokio::time::sleep(Duration::from_secs(TOKEN_RETRY_SECS)).await; + }; + + let client = match reqwest::Client::builder() + .timeout(Duration::from_secs(HTTP_TIMEOUT_SECS)) + .build() + { + Ok(c) => c, + Err(e) => { + warn!("forge_notify: failed to build github HTTP client: {e}"); + return; + } + }; + + let source = GithubSource::new(&token); + let mut own_login = resolve_own_login(&client, &source).await; + let mut delivered: HashMap = HashMap::new(); + + // GitHub tells callers how often it is willing to be polled + // (`X-Poll-Interval`, 60s in practice) and rate-limits those who + // ignore it. Start at our own cadence and re-arm to whatever the + // server asks for — never faster than it wants, never slower than we + // need. + let mut cadence = POLL_INTERVAL_SECS; + let mut interval = tokio::time::interval(Duration::from_secs(cadence)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + interval.tick().await; + + info!("forge_notify: github polling started"); + + loop { + interval.tick().await; + if own_login.is_empty() { + own_login = resolve_own_login(&client, &source).await; + } + let hint = poll_once(&source, &client, &socket, &mut delivered, &own_login).await; + if let Some(secs) = hint.filter(|s| *s > cadence) { + debug!( + secs, + "forge_notify: github asked for a slower poll — re-arming" + ); + cadence = secs; + interval = tokio::time::interval(Duration::from_secs(cadence)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + interval.tick().await; + } + } +} diff --git a/hive-forge-notify/src/bin/hive-github-notify/source.rs b/hive-forge-notify/src/bin/hive-github-notify/source.rs new file mode 100644 index 00000000..9f84dfef --- /dev/null +++ b/hive-forge-notify/src/bin/hive-github-notify/source.rs @@ -0,0 +1,132 @@ +//! github.com's half of the poller — the protocol code that only this +//! binary links. +//! +//! Plain `reqwest`: the typed Forgejo client cannot address a different +//! API surface, and GitHub's payloads differ enough (string thread ids, +//! its own `subject.type` vocabulary, no `subject.state`) that the +//! tolerant parse in the shared library is the right common layer, not +//! the client. + +use hive_forge_notify::notify::UNREAD_FETCH_LIMIT; +use hive_forge_notify::source::Source; +use tracing::{debug, warn}; + +/// GitHub's REST base. `docs/github.md` scopes the integration to +/// github.com only — the same constraint the `gh` wrapper and the git +/// credential helper already carry — so this is a constant rather than +/// another operator-entered URL. +const API_BASE: &str = "https://api.github.com"; + +/// Value of the `X-GitHub-Api-Version` header. GitHub dates its REST +/// versions; pinning one means a future default bump cannot silently +/// reshape the payloads this daemon parses. +const API_VERSION: &str = "2022-11-28"; + +/// `User-Agent` for GitHub calls. GitHub rejects requests without one. +const USER_AGENT: &str = "hyperhive-forge-notify"; + +/// github.com, via a personal access token. +pub struct GithubSource { + token: String, +} + +impl GithubSource { + pub fn new(token: &str) -> Self { + Self { + token: token.to_owned(), + } + } +} + +impl Source for GithubSource { + /// Namespaced, so a github thread id cannot collide with a forge one + /// on a shared todo key. + fn key_prefix(&self) -> &'static str { + "gh:" + } + + fn name(&self) -> &'static str { + "github" + } + + /// `Bearer`, not Forgejo's `token` — plus the version and user-agent + /// headers GitHub requires. Sending the wrong scheme does not error: + /// it authenticates as *nobody* and silently drops to the + /// unauthenticated rate limit, which is why the rate-limit header is + /// the only cheap way to tell the two apart. + fn authorize(&self, rb: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + rb.header("Authorization", format!("Bearer {}", self.token)) + .header("Accept", "application/vnd.github+json") + .header("X-GitHub-Api-Version", API_VERSION) + .header("User-Agent", USER_AGENT) + } + + async fn own_login(&self, client: &reqwest::Client) -> String { + let url = format!("{API_BASE}/user"); + let Ok(resp) = self.authorize(client.get(&url)).send().await else { + return String::new(); + }; + if !resp.status().is_success() { + return String::new(); + } + resp.json::() + .await + .ok() + .and_then(|v| v["login"].as_str().map(str::to_owned)) + .unwrap_or_default() + } + + async fn list_unread( + &self, + client: &reqwest::Client, + ) -> Option<(Vec, Option)> { + let url = format!("{API_BASE}/notifications?all=false&per_page={UNREAD_FETCH_LIMIT}"); + let resp = match self.authorize(client.get(&url)).send().await { + Ok(resp) => resp, + Err(e) => { + debug!("forge_notify: github poll request failed: {e}"); + return None; + } + }; + // Read the cadence hint before the status check: GitHub sends it + // on an empty page too, and that is exactly the tick where we + // most want to learn we are polling too fast. + let poll_interval = resp + .headers() + .get("x-poll-interval") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()); + let status = resp.status(); + if !status.is_success() { + // 403 with a rate-limit body and 401 on a revoked PAT are + // both "stay quiet and retry", not fatal: the unit must not + // restart-loop on a credential the operator fixes out of band. + debug!("forge_notify: github poll status {status}"); + return None; + } + let raw = resp.text().await.ok()?; + + match serde_json::from_str(&raw) { + Ok(values) => Some((values, poll_interval)), + Err(e) => { + warn!("forge_notify: github response parse error: {e}"); + None + } + } + } + + async fn mark_read(&self, client: &reqwest::Client, id: &str) { + let url = format!("{API_BASE}/notifications/threads/{id}"); + match self.authorize(client.patch(&url)).send().await { + Ok(resp) if resp.status().is_success() => { + debug!(%id, "forge_notify: marked read (github)"); + } + Ok(resp) => { + warn!(%id, status = %resp.status(), "forge_notify: github mark-read non-2xx — notification will resurface"); + } + Err(e) => { + warn!(%id, error = ?e, "forge_notify: github mark-read failed — notification will resurface"); + } + } + } +} diff --git a/hive-forge-notify/src/lib.rs b/hive-forge-notify/src/lib.rs new file mode 100644 index 00000000..1554b00c --- /dev/null +++ b/hive-forge-notify/src/lib.rs @@ -0,0 +1,68 @@ +//! Per-agent notification pollers — shared library half. +//! +//! Two binaries ship from this crate, one per notification host: +//! +//! - **`hive-forge-notify`** — the hive's internal Forgejo. Always +//! deployed; the behaviour predates this split and is unchanged. +//! - **`hive-github-notify`** — github.com, for agents that have a PAT. +//! +//! They are separate *binaries* rather than one process with two loops so +//! the deployment can choose: an agent module installs the GitHub unit or +//! it doesn't, and the decision lives in the module rather than in a cargo +//! feature. A feature flag would unify across the workspace — enabling it +//! for one consumer changes feature resolution for the whole graph and +//! stops the two builds sharing any cached crate — which is a permanent +//! cost for something a second binary expresses for free. +//! +//! Everything except the host-specific calls (list unread, mark read, +//! resolve own login) is shared and lives here: classification, wake +//! formatting, the tolerant parse, delivery dedupe, and the todo upsert. +//! The host differences live behind [`source::Source`]. + +pub mod notify; +pub mod source; + +/// Re-exported so a binary can spell it once at the crate root alongside +/// the other things it needs to build a client; it is a property of the +/// poller, not of the notification format. +pub use notify::HTTP_TIMEOUT_SECS; + +/// Retry policy for the harness's in-agent socket. Deliberately fail-fast: +/// both callers are inside the poll loop and both treat a failed request as +/// "leave the thread unread and try again next tick", so the poll interval +/// *is* the retry — a second, in-request backoff would only stack sleeps on +/// top of it and delay the rest of the batch. That is the opposite +/// trade-off from the serve loop's client, which rides out a hive-c0re +/// restart because its callers have no natural retry of their own. +pub const TODO_SOCKET_RETRY: hive_sock_client::Retry = hive_sock_client::Retry::None; + +/// Resolve the harness's in-agent todo socket from the environment, falling +/// back to the well-known path. Shared by both binaries so they can't drift +/// on where they deliver. +#[must_use] +pub fn agent_socket() -> std::path::PathBuf { + std::env::var_os("HIVE_AGENT_SOCKET").map_or_else( + || std::path::PathBuf::from(hive_agent_sock::DEFAULT_AGENT_SOCKET), + std::path::PathBuf::from, + ) +} + +/// Install the tracing subscriber both binaries use: `RUST_LOG` when set, +/// `info` otherwise. +pub fn init_tracing() { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_env("RUST_LOG") + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .init(); +} + +/// Read `HYPERHIVE_STATE_DIR`, the directory holding the agent's +/// credentials (`forge-token`, `github-token`). Empty when unset, which +/// makes the token paths relative and the read fail — the callers treat +/// that as "not configured" and settle. +#[must_use] +pub fn state_dir() -> String { + std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default() +} diff --git a/hive-forge-notify/src/main.rs b/hive-forge-notify/src/main.rs deleted file mode 100644 index 988d1c3e..00000000 --- a/hive-forge-notify/src/main.rs +++ /dev/null @@ -1,46 +0,0 @@ -//! `hive-forge-notify` binary — long-running per-agent Forgejo -//! notification poller. Polls the agent's unread notification list, -//! formats each thread into a short summary, and pushes it as a todo -//! (loose-ends v2) on the harness's in-agent socket so claude drives a -//! turn to handle it. -//! -//! Takes no arguments: everything comes from the environment the -//! per-agent systemd unit provides — `HIVE_FORGE_URL` (forwarded into -//! every container by the meta flake), `HYPERHIVE_STATE_DIR` (where the -//! agent's `forge-token` lives) and `HIVE_AGENT_SOCKET` (the harness's -//! todo socket). When the forge is not configured for this agent the -//! poller logs why and exits 0 — the unit is `Restart=on-failure`, so a -//! forge-less agent settles instead of restart-looping. - -mod notify; - -/// Retry policy for the harness's in-agent socket. Deliberately fail-fast: -/// both callers are inside the 30s poll loop and both treat a failed -/// request as "leave the thread unread and try again next tick", so the -/// poll interval *is* the retry — a second, in-request backoff would only -/// stack sleeps on top of it and delay the rest of the batch. That is the -/// opposite trade-off from the serve loop's client, which rides out a -/// hive-c0re restart because its callers have no natural retry of their own. -const TODO_SOCKET_RETRY: hive_sock_client::Retry = hive_sock_client::Retry::None; - -#[tokio::main] -async fn main() { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_env("RUST_LOG") - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - .init(); - - let socket = std::env::var_os("HIVE_AGENT_SOCKET").map_or_else( - || std::path::PathBuf::from(hive_agent_sock::DEFAULT_AGENT_SOCKET), - std::path::PathBuf::from, - ); - - tracing::info!(socket = %socket.display(), "hive-forge-notify starting"); - - // Returns only when the forge is not configured (or the token never - // arrives); otherwise loops forever. Either way there is nothing left - // for this process to do, so fall off the end and exit 0. - notify::run(socket).await; -} diff --git a/hive-forge-notify/src/notify.rs b/hive-forge-notify/src/notify.rs index 2dce745f..5aaa58a3 100644 --- a/hive-forge-notify/src/notify.rs +++ b/hive-forge-notify/src/notify.rs @@ -15,6 +15,11 @@ //! Self-echo notifications (the agent's own writes) are marked read without //! a delivery. //! +//! **Multi-source**: always the internal Forgejo, plus github.com when the +//! agent has a PAT. Each source polls independently behind +//! [`Source`](crate::source::Source); everything below is shared. Rationale +//! + host differences: [`docs/forge.md::Sources`](../../../docs/forge.md). +//! //! Activation gates, self-notification filtering, body excerpt + //! truncation + heading escape, wrapper formats (comment / review / //! new-item / state-change), meta suffix, and review-request override @@ -22,36 +27,37 @@ use std::collections::{HashMap, HashSet}; use std::fmt::Write as _; -use std::path::{Path, PathBuf}; +use std::path::Path; use std::time::Duration; -use forgejo_api::structs::{NotifyGetListQuery, NotifyReadThreadQuery, NotifySubjectType}; -use forgejo_api::{Auth, Forgejo, ForgejoError}; +use forgejo_api::structs::NotifySubjectType; use serde::{Deserialize, Deserializer}; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; -use tracing::{debug, info, warn}; +use tracing::{debug, warn}; -const POLL_INTERVAL_SECS: u64 = 30; +use crate::source::Source; + +pub const POLL_INTERVAL_SECS: u64 = 30; /// Per-request cap applied to every forge call — natively on the reqwest /// enrichment client, via `tokio::time::timeout` around the typed /// `forgejo-api` client (which exposes no timeout knob of its own). -const HTTP_TIMEOUT_SECS: u64 = 10; +pub const HTTP_TIMEOUT_SECS: u64 = 10; /// Page size of the unread-notifications fetch. This is also the hard /// 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; +pub const UNREAD_FETCH_LIMIT: usize = 50; /// Maximum characters of a body/comment to include in the wake message. const BODY_TRUNCATE: usize = 500; /// How long to wait between token-read retries when the token file is /// missing or unreadable at startup (e.g. hive-priv hasn't provisioned /// it yet, or a chown race left it temporarily root-owned). -const TOKEN_RETRY_SECS: u64 = 30; +pub const TOKEN_RETRY_SECS: u64 = 30; /// Give up waiting for the token after this many retries (~10 minutes). /// Avoids an infinite wait on agents that genuinely have no forge account. -const TOKEN_RETRY_MAX: u32 = 20; +pub const TOKEN_RETRY_MAX: u32 = 20; /// How close (seconds) the notification's event time must be to a /// subject's `created_at` for us to call it a genuine creation and emit /// a `new ` label. Later activity that lands on the state-change @@ -60,161 +66,28 @@ const TOKEN_RETRY_MAX: u32 = 20; /// claim it's "new" — see docs/forge.md, "new vs activity on". const NEW_ITEM_TOLERANCE_SECS: i64 = 120; -/// Spawn point: called once from the `hive-agent` serve loop. Returns immediately if the forge is not -/// configured. Otherwise loops forever, polling every -/// `POLL_INTERVAL_SECS` seconds. Errors are never fatal. -/// -/// `socket` is the harness's in-agent todo socket (`HIVE_AGENT_SOCKET`): -/// each forge notification is pushed as an `upsert_todo`, not a direct wake. -pub async fn run(socket: PathBuf) { - let forge_url = match std::env::var("HIVE_FORGE_URL") { - Ok(u) if !u.is_empty() => u, - _ => { - debug!("forge_notify: HIVE_FORGE_URL not set — disabled"); - return; - } - }; - - let state_dir = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default(); - let token_path = format!("{state_dir}/forge-token"); - // Retry reading the token to handle races where hive-priv provisions the - // token after the harness starts, or where a parent-container chown briefly - // makes the file unreadable. We wait up to - // TOKEN_RETRY_MAX * TOKEN_RETRY_SECS before giving up. - let token = { - let mut attempts = 0u32; - loop { - match tokio::fs::read_to_string(&token_path).await { - Ok(t) => { - let t = t.trim().to_owned(); - if !t.is_empty() { - break t; - } - debug!("forge_notify: empty forge token at {token_path}"); - } - Err(e) => { - debug!("forge_notify: cannot read token at {token_path}: {e}"); - } - } - attempts += 1; - if attempts >= TOKEN_RETRY_MAX { - debug!( - "forge_notify: token not available after {TOKEN_RETRY_MAX} retries — disabled" - ); - return; - } - tokio::time::sleep(Duration::from_secs(TOKEN_RETRY_SECS)).await; - } - }; - - // Typed Forgejo client for the API calls with stable shapes (identity - // probe, notification list, mark-read). The plain reqwest client below - // stays for the best-effort enrichment fetches of `subject.url` / - // `latest_comment_url`: those follow server-provided URLs whose payload - // shape is heterogeneous (issue vs comment vs review), which the typed - // client cannot express (its `Endpoint` trait is sealed). - let base_url = match url::Url::parse(&forge_url) { - Ok(u) => u, - Err(e) => { - warn!("forge_notify: invalid HIVE_FORGE_URL {forge_url}: {e}"); - return; - } - }; - let forge = match Forgejo::new(Auth::Token(&token), base_url) { - Ok(f) => f, - Err(e) => { - warn!("forge_notify: failed to build forge client: {e}"); - return; - } - }; - - let client = match reqwest::Client::builder() - .timeout(Duration::from_secs(HTTP_TIMEOUT_SECS)) - .build() - { - Ok(c) => c, - Err(e) => { - warn!("forge_notify: failed to build HTTP client: {e}"); - return; - } - }; - - // Fetch own login for self-notification filtering. Falls back to - // empty string on failure — no filtering (safe degradation; see - // `docs/forge.md::Self-notification filtering`). A boot-time failure - // (e.g. the forge not yet reachable) is re-attempted on each poll tick - // below rather than leaving filtering off for the whole process. - let mut own_login = resolve_own_login(&forge).await; - if own_login.is_empty() { - warn!( - "forge_notify: could not resolve own login yet — self-notification \ - filtering disabled until it resolves on a later poll" - ); - } else { - debug!(%own_login, "forge_notify: own login resolved"); - } - - let mut interval = tokio::time::interval(Duration::from_secs(POLL_INTERVAL_SECS)); - interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - // First tick fires immediately — skip it so we don't race the broker - // socket becoming available right at boot. - interval.tick().await; - - info!(forge_url = %forge_url, "forge_notify: polling started"); - - // 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; - // If own-login didn't resolve at boot (forge unreachable then), - // retry before this poll so self-echo filtering self-heals instead - // of staying off for the whole process lifetime. - if own_login.is_empty() { - own_login = resolve_own_login(&forge).await; - if !own_login.is_empty() { - debug!(%own_login, "forge_notify: own login resolved on retry"); - } - } - poll_once(&forge, &client, &token, &socket, &mut delivered, &own_login).await; - update_assigned_rollup(&client, &forge_url, &token, &socket).await; - } -} - -/// Fetch the agent's own forge login (`GET /api/v1/user`) for -/// self-notification filtering. Returns the empty string on any failure -/// (timeout, HTTP error, missing field); the caller treats empty as -/// "filtering disabled" and retries on the next poll tick. -async fn resolve_own_login(forge: &Forgejo) -> String { +/// Fetch the account's own login for self-notification filtering. +/// Returns the empty string on any failure (timeout, HTTP error, missing +/// field); the caller treats empty as "filtering disabled" and retries on +/// the next poll tick. +pub async fn resolve_own_login(client: &reqwest::Client, source: &S) -> String { tokio::time::timeout( Duration::from_secs(HTTP_TIMEOUT_SECS), - forge.user_get_current().send(), + source.own_login(client), ) .await - .ok() - .and_then(Result::ok) - .and_then(|u| u.login) .unwrap_or_default() } -/// 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 { - let resp = client - .get(url) - .header("Authorization", format!("token {token}")) - .send() - .await - .ok()?; +/// Fetch a JSON value from a URL, authenticated as the source that handed +/// us the URL. Returns `None` on any HTTP or parse error (best-effort +/// enrichment). +async fn fetch_json( + client: &reqwest::Client, + url: &str, + source: &S, +) -> Option { + let resp = source.authorize(client.get(url)).send().await.ok()?; if !resp.status().is_success() { return None; } @@ -382,9 +255,9 @@ fn review_state_label(state: &str) -> Option<&str> { /// delivery. Wrapper format table + meta-suffix shape + number/repo /// extraction live in `docs/forge.md::Wrapper format` + /// `::Meta suffix`. -async fn format_notification( +async fn format_notification( client: &reqwest::Client, - token: &str, + source: &S, notif: &PolledNotification, own_login: &str, ) -> Option { @@ -435,7 +308,7 @@ async fn format_notification( let subject = if subject_api_url.is_empty() { None } else { - fetch_json(client, subject_api_url, token).await + fetch_json(client, subject_api_url, source).await }; // Forgejo's notification `subject.type` is "Pull" / "Issue", never @@ -468,7 +341,7 @@ async fn format_notification( if has_comment && !is_fresh_state_change { format_comment_notification( client, - token, + source, &meta, comment_api_url, comment_html_url, @@ -487,7 +360,7 @@ async fn format_notification( let comment_tail = if has_comment { fresh_post_close_comment_tail( client, - token, + source, comment_api_url, meta.subject.as_ref(), own_login, @@ -554,15 +427,15 @@ fn build_meta_suffix(subject: Option<&serde_json::Value>, is_pr: bool) -> String } /// Format a notification triggered by a new comment or review submission. -async fn format_comment_notification( +async fn format_comment_notification( client: &reqwest::Client, - token: &str, + source: &S, meta: &NotifMeta<'_>, comment_api_url: &str, comment_html_url: &str, own_login: &str, ) -> Option { - let payload = fetch_json(client, comment_api_url, token).await; + let payload = fetch_json(client, comment_api_url, source).await; let actor_login = payload .as_ref() @@ -824,14 +697,14 @@ fn comment_is_after_close( /// agent's own write back at it), is empty/bodiless, or can't be fetched. /// This is the one extra fetch the merge/close path pays for the best of /// both worlds — cheap given how rare merge notifications are. -async fn fresh_post_close_comment_tail( +async fn fresh_post_close_comment_tail( client: &reqwest::Client, - token: &str, + source: &S, comment_api_url: &str, subject: Option<&serde_json::Value>, own_login: &str, ) -> Option { - let payload = fetch_json(client, comment_api_url, token).await?; + let payload = fetch_json(client, comment_api_url, source).await?; if !comment_is_after_close(&payload, subject) { return None; } @@ -868,8 +741,13 @@ fn parse_rfc3339(s: &str) -> Option { /// break notification read-state again. #[derive(Debug, Deserialize)] struct NotificationThread { - #[serde(default)] - id: Option, + /// Thread id as a **string**, whichever host it came from: Forgejo + /// sends a JSON number, GitHub sends a quoted string. It is only ever + /// used as an opaque key (dedupe map, todo key, mark-read path + /// segment), so normalising to `String` at the parse boundary is + /// cheaper than carrying the difference through every call site. + #[serde(default, deserialize_with = "de_opt_id")] + id: Option, #[serde(default, deserialize_with = "de_opt_rfc3339")] updated_at: Option, #[serde(default)] @@ -909,7 +787,10 @@ where { Ok( Option::::deserialize(d)?.and_then(|s| match s.as_str() { - "Pull" => Some(NotifySubjectType::Pull), + // Forgejo says `Pull`; GitHub says `PullRequest` for the same + // thing. Both map to the same label, so a GitHub PR doesn't + // render as the `?` unknown-type fallback. + "Pull" | "PullRequest" => Some(NotifySubjectType::Pull), "Issue" => Some(NotifySubjectType::Issue), "Commit" => Some(NotifySubjectType::Commit), "Repository" => Some(NotifySubjectType::Repository), @@ -918,6 +799,23 @@ where ) } +/// Deserialize an optional notification thread id from either a JSON +/// number (Forgejo) or a JSON string (GitHub), normalising both to +/// `String`. Anything else — including a `null` or an unexpected shape — +/// degrades to `None` rather than failing the parse, keeping the "one bad +/// field can never break notification read-state" property the rest of +/// this struct is built around. +fn de_opt_id<'de, D>(d: D) -> Result, D::Error> +where + D: Deserializer<'de>, +{ + Ok(match Option::::deserialize(d)? { + Some(serde_json::Value::String(s)) if !s.is_empty() => Some(s), + Some(serde_json::Value::Number(n)) => Some(n.to_string()), + _ => None, + }) +} + /// Deserialize an optional URL, mapping a blank or unparseable value to /// `None` (Forgejo marshals empty strings for absent URLs, which a plain /// `Option` would reject). @@ -984,58 +882,26 @@ fn parse_notification(value: serde_json::Value) -> Option { sequential 'fetch / classify / dispatch' rhythm and add helper \ functions for state shared across all three phases" )] -async fn poll_once( - forge: &Forgejo, +pub async fn poll_once( + source: &S, client: &reqwest::Client, - token: &str, socket: &Path, - delivered: &mut HashMap, + delivered: &mut HashMap, own_login: &str, -) { - // Fetch the page as raw JSON (`response_type::`) instead of the - // crate's `Vec`: one merged-PR notification - // (`subject.state = "merged"`, unrepresentable in `StateType`) would - // otherwise poison deserialization of the whole page. HTTP-level errors - // still surface as `ForgejoError` exactly like the fully-typed call; - // `parse_notification` below does the per-item typed parse. - let query = NotifyGetListQuery { - all: Some(false), - ..NotifyGetListQuery::default() - }; - let request = forge - .notify_get_list(query) - .page_size(u32::try_from(UNREAD_FETCH_LIMIT).unwrap_or(u32::MAX)) - .response_type::(); - let raw = - match tokio::time::timeout(Duration::from_secs(HTTP_TIMEOUT_SECS), request.send()).await { - Ok(Ok(raw)) => raw, - Ok(Err(ForgejoError::UnexpectedStatusCode(status))) => { - debug!("forge_notify: poll status {status}"); - return; - } - Ok(Err(e)) => { - debug!("forge_notify: poll request failed: {e}"); - return; - } - Err(_) => { - debug!("forge_notify: poll request failed: timed out"); - return; - } - }; - - let values: Vec = match serde_json::from_str(&raw) { - Ok(v) => v, - Err(e) => { - warn!("forge_notify: response parse error: {e}"); - return; - } - }; +) -> Option { + // The page comes back as raw JSON values rather than a typed page: + // one merged-PR notification (`subject.state = "merged"`, + // unrepresentable in the typed `StateType`) would otherwise poison + // deserialization of the whole page. `parse_notification` below does + // the per-item tolerant parse. See `Source::list_unread`. + let (values, poll_hint) = source.list_unread(client).await?; if values.is_empty() { - return; + return poll_hint; } debug!( + source = source.name(), count = values.len(), "forge_notify: delivering notifications" ); @@ -1044,7 +910,7 @@ async fn poll_once( values.into_iter().filter_map(parse_notification).collect(); for notif in ¬ifications { - let Some(id) = notif.thread.id.and_then(|id| u64::try_from(id).ok()) else { + let Some(id) = notif.thread.id.clone() else { continue; }; @@ -1054,27 +920,33 @@ async fn poll_once( // `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) { + if !should_deliver(delivered, &id, &updated_at) { debug!(%id, "forge_notify: skipping (already delivered this version)"); continue; } - let body_opt = format_notification(client, token, notif, own_login).await; + let body_opt = format_notification(client, source, notif, own_login).await; // None means self-echo — mark read silently, no delivery. let Some(body) = body_opt else { - mark_read(forge, id).await; + source.mark_read(client, &id).await; continue; }; // Upsert a *todo* (loose-ends v2) on the harness's in-agent socket, - // keyed by the forge thread id, instead of firing a direct wake. A + // keyed by the thread id, instead of firing a direct wake. A // new/changed summary makes the harness signal its turn loop; the // agent clears the todo (`mark_todo_done`) once it has handled the // thread. Re-scanning the same thread is an idempotent no-op. + // + // The key carries the source's prefix so two hosts handing out the + // same numeric thread id can't collide on one todo. The internal + // forge's prefix is deliberately EMPTY, keeping its keys the bare + // ids they have always been — renaming them would orphan every + // in-flight forge todo on the first restart after this lands. let req = hive_agent_sock::Request::UpsertTodo { subsystem: "forge".to_owned(), - key: Some(id.to_string()), + key: Some(format!("{}{id}", source.key_prefix())), summary: body, source: None, }; @@ -1100,7 +972,7 @@ async fn poll_once( // doesn't re-upsert next tick; it is deliberately NOT // persisted — forge's own read-state is the cross-rebuild // source of truth. - mark_read(forge, id).await; + source.mark_read(client, &id).await; delivered.insert(id, updated_at); } Err(e) => { @@ -1117,147 +989,30 @@ async fn poll_once( // `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 + let current_ids: HashSet<&str> = notifications .iter() - .filter_map(|n| n.thread.id.and_then(|id| u64::try_from(id).ok())) + .filter_map(|n| n.thread.id.as_deref()) .collect(); - delivered.retain(|id, _| current_ids.contains(id)); + delivered.retain(|id, _| current_ids.contains(id.as_str())); debug_assert!( delivered.len() <= UNREAD_FETCH_LIMIT, "in-process dedupe map exceeded the fetch window ({} > {UNREAD_FETCH_LIMIT})", delivered.len(), ); + + poll_hint } /// 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. 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 - // unrepresentable value can't actually reach here. - return; - }; - // `to_status: None` → Forgejo's default transition (unread → read), - // matching the old bare PATCH. The 205 response body is the thread - // JSON, which can carry `subject.state = "merged"` — take it as an - // opaque `String` (see `poll_once`) so a merged PR doesn't turn a - // successful mark-read into a spurious parse error. - let request = forge - .notify_read_thread(thread_id, NotifyReadThreadQuery { to_status: None }) - .response_type::(); - match tokio::time::timeout(Duration::from_secs(HTTP_TIMEOUT_SECS), request.send()).await { - Err(_) => { - warn!(%id, "forge_notify: mark-read request failed — notification will resurface"); - } - Ok(Err(e @ ForgejoError::ReqwestError(_))) => { - warn!(%id, error = ?e, "forge_notify: mark-read request failed — notification will resurface"); - } - Ok(Err(e)) => { - warn!(%id, error = %e, "forge_notify: mark-read returned non-2xx — notification will resurface"); - } - Ok(Ok(_)) => { - debug!(%id, "forge_notify: marked read"); - } - } -} - -/// Fetch the total count of open issues or PRs assigned to this agent via -/// Forgejo's global search API (`GET /api/v1/issues/search`). `issue_type` -/// is `"issues"` or `"pulls"`. Reads the `X-Total-Count` response header -/// rather than deserialising the full body — only page 1 with limit 1 is -/// fetched, keeping the request cheap. Returns `None` on any error (timeout, -/// HTTP error, missing or unparseable header). -async fn count_assigned( - client: &reqwest::Client, - forge_url: &str, - token: &str, - issue_type: &str, -) -> Option { - let url = format!( - "{forge_url}/api/v1/issues/search\ - ?type={issue_type}&state=open&assigned=true&limit=1&page=1" - ); - let resp = match client - .get(&url) - .header("Authorization", format!("token {token}")) - .send() - .await - { - Ok(r) if r.status().is_success() => r, - _ => return None, - }; - let count_str = resp.headers().get("x-total-count")?.to_str().ok()?; - count_str.parse::().ok() -} - -/// After each notification poll, query the forge for the count of open -/// issues and PRs assigned to this agent and keep a keyed `"rollup"` todo -/// in sync. When the count is positive the todo summarises the breakdown -/// (`N issues, M PRs`); when it reaches zero the todo is cleared. The rollup -/// key is distinct from per-thread numeric keys so clearing it never touches -/// notification todos. -async fn update_assigned_rollup( - client: &reqwest::Client, - forge_url: &str, - token: &str, - socket: &Path, -) { - let issues = count_assigned(client, forge_url, token, "issues") - .await - .unwrap_or(0); - let pulls = count_assigned(client, forge_url, token, "pulls") - .await - .unwrap_or(0); - let total = issues + pulls; - - let req = if total == 0 { - hive_agent_sock::Request::ClearTodo { - subsystem: "forge".to_owned(), - key: Some("rollup".to_owned()), - all: false, - } - } else { - let breakdown = match (issues, pulls) { - (i, 0) => format!("{i} issue{}", if i == 1 { "" } else { "s" }), - (0, p) => format!("{p} PR{}", if p == 1 { "" } else { "s" }), - (i, p) => format!( - "{i} issue{}, {p} PR{}", - if i == 1 { "" } else { "s" }, - if p == 1 { "" } else { "s" } - ), - }; - hive_agent_sock::Request::UpsertTodo { - subsystem: "forge".to_owned(), - key: Some("rollup".to_owned()), - summary: format!("{total} open assigned: {breakdown}"), - source: None, - } - }; - - match hive_sock_client::request::<_, hive_agent_sock::Response>( - socket, - &req, - crate::TODO_SOCKET_RETRY, - ) - .await - { - Ok(_) => debug!(total, "forge_notify: assigned rollup todo updated"), - Err(e) => debug!("forge_notify: assigned rollup todo update failed: {e}"), - } +fn should_deliver( + delivered: &HashMap, + id: &str, + updated_at: &str, +) -> bool { + delivered.get(id).is_none_or(|seen| seen != updated_at) } #[cfg(test)] @@ -1267,7 +1022,7 @@ mod tests { #[test] fn should_deliver_when_thread_never_seen() { let delivered = HashMap::new(); - assert!(should_deliver(&delivered, 42, "2026-06-22T16:00:00Z")); + assert!(should_deliver(&delivered, "42", "2026-06-22T16:00:00Z")); } #[test] @@ -1275,8 +1030,8 @@ mod tests { // 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")); + delivered.insert("42".to_owned(), "2026-06-22T16:00:00Z".to_owned()); + assert!(!should_deliver(&delivered, "42", "2026-06-22T16:00:00Z")); } #[test] @@ -1284,16 +1039,16 @@ mod tests { // 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")); + delivered.insert("42".to_owned(), "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")); + delivered.insert("42".to_owned(), "2026-06-22T16:00:00Z".to_owned()); + assert!(should_deliver(&delivered, "99", "2026-06-22T16:00:00Z")); } #[test] @@ -1568,7 +1323,7 @@ mod tests { assert_eq!(polled.state, "merged"); // Cursor string is the raw `updated_at`, byte-identical. assert_eq!(polled.updated_at, "2026-06-13T11:18:42+02:00"); - assert_eq!(polled.thread.id, Some(7)); + assert_eq!(polled.thread.id.as_deref(), Some("7")); assert_eq!( polled.thread.subject.as_ref().and_then(|s| s.r#type), Some(NotifySubjectType::Pull), @@ -1616,7 +1371,7 @@ mod tests { }, })) .expect("drifted notification must still parse"); - assert_eq!(polled.thread.id, Some(42)); + assert_eq!(polled.thread.id.as_deref(), Some("42")); // Unknown subject type degrades to None instead of failing. assert!( polled diff --git a/hive-forge-notify/src/source.rs b/hive-forge-notify/src/source.rs new file mode 100644 index 00000000..b1e9d813 --- /dev/null +++ b/hive-forge-notify/src/source.rs @@ -0,0 +1,70 @@ +//! The per-host half of the poller, as a trait. +//! +//! Everything except three host-specific calls — list unread, mark read, +//! resolve own login — is host-agnostic, so a host is described by this +//! trait and implemented **inside the binary that polls it**: the Forgejo +//! impl lives in `src/bin/hive-forge-notify/`, the GitHub one in +//! `src/bin/hive-github-notify/`. Neither binary links the other's +//! protocol code, and this library names no host at all. +//! +//! Implementors hand [`notify`](crate::notify) the same raw JSON either +//! way. Forgejo's notifications API is modelled on GitHub's, which is why +//! one shared parse works: `id` / `repository.full_name` / `subject` +//! `{title,url,latest_comment_url}` / `updated_at` line up field for +//! field, and the differences are absorbed by the lenient deserializers +//! in `notify` rather than a second parse path. + +use std::future::Future; + +/// A polled notification host. +/// +/// Methods return `impl Future` rather than being `async fn` so the +/// returned futures carry an explicit `Send` bound — the poll loops move +/// them across tasks. +pub trait Source { + /// Namespace for this host's todo keys, so two hosts handing out the + /// same numeric thread id cannot collide on one todo. + /// + /// The internal forge deliberately uses an **empty** namespace: its + /// keys stay the bare thread ids they have always been, because + /// renaming them would orphan every in-flight forge todo on the first + /// restart after a deploy. + fn key_prefix(&self) -> &'static str; + + /// Human-readable host name for log lines. + fn name(&self) -> &'static str; + + /// Apply this host's auth — and any headers it requires — to an + /// outgoing request. Used by the shared enrichment fetches, which + /// follow URLs the server handed us and so must authenticate as + /// whichever host produced them. + /// + /// Hosts differ in *scheme*, not just value, and getting it wrong is + /// silent: a request that authenticates as nobody still returns 200, + /// just on the unauthenticated rate limit. + fn authorize(&self, rb: reqwest::RequestBuilder) -> reqwest::RequestBuilder; + + /// Resolve the account's own login, for self-notification filtering. + /// Returns the empty string on any failure; the caller treats that as + /// "filtering disabled" and retries on the next tick. + fn own_login(&self, client: &reqwest::Client) -> impl Future + Send; + + /// Fetch the unread notification page as raw JSON values, plus the + /// server's requested minimum seconds between polls when it states + /// one (a host that says nothing returns `None` and the caller keeps + /// its own cadence). + /// + /// Raw rather than typed so one unrepresentable field in one item + /// cannot poison the whole page; per-item parsing happens in + /// [`notify::parse_notification`](crate::notify). `None` means the + /// fetch failed — the caller skips this tick. + fn list_unread( + &self, + client: &reqwest::Client, + ) -> impl Future, Option)>> + Send; + + /// Mark a notification thread read on this host. Best-effort: a + /// failure leaves the thread unread so it resurfaces next tick, which + /// the in-process dedupe map then suppresses from re-waking the agent. + fn mark_read(&self, client: &reqwest::Client, id: &str) -> impl Future + Send; +} diff --git a/nix/agent-modules/github.nix b/nix/agent-modules/github.nix index d1a9aa6a..1fe7a9bd 100644 --- a/nix/agent-modules/github.nix +++ b/nix/agent-modules/github.nix @@ -85,5 +85,46 @@ in username = x-access-token ''; }; + + # GitHub notification poller — the github.com sibling of + # `hive-forge-notify` (forge.nix). Same delivery path: unread list -> + # per-thread summary -> todo on the harness's in-agent socket. + # + # Its own binary and its own package, so the deployment decides + # whether an agent gets GitHub notifications at all — the unit is + # installed or it isn't. That keeps the choice here rather than in a + # cargo feature, which would unify across the workspace and stop the + # two builds sharing any cached crate. + # + # The separate package is what makes that real: the per-bin + # extractor copies exactly one binary, so an agent that installs + # only the Forgejo poller has no github.com poller anywhere in its + # closure — not merely an unstarted unit. + systemd.services.hive-github-notify = lib.mkIf config.hyperhive.github.enable { + description = "github.com notification poller for this agent"; + wantedBy = [ "multi-user.target" ]; + after = [ "network.target" ]; + environment = { + # Same in-agent todo socket the forge poller and the harness use + # (agent-service.nix) — one todo stream, two producers. + HIVE_AGENT_SOCKET = "/run/hive-agent/${userName}/agent.sock"; + RUST_LOG = "info"; + # HYPERHIVE_STATE_DIR comes from systemd.globalEnvironment; the + # poller reads the agent's `github-token` from under it. + }; + serviceConfig = { + ExecStart = "${config.hyperhive.packages.hive-github-notify}/bin/hive-github-notify"; + SyslogIdentifier = "hive-github-notify"; + # `on-failure`, NOT `always`, for the same reason as the forge + # poller: this unit ships on every agent, but most agents have no + # PAT. The poller reports that by logging why and exiting 0, which + # under `always` would become a restart loop on every PAT-less + # agent. A crash still restarts. + Restart = "on-failure"; + RestartSec = 5; + User = userName; + Group = userName; + }; + }; }; } diff --git a/nix/agent-modules/packages.nix b/nix/agent-modules/packages.nix index ca465fee..0e4fe36d 100644 --- a/nix/agent-modules/packages.nix +++ b/nix/agent-modules/packages.nix @@ -12,7 +12,8 @@ hyperhive package outputs consumed by the harness modules: the per-binary daemon/CLI packages (`hive-agent`, `hive-agent-mcp`, `hive-bash-daemon`, - `hive-forge`, `hive-forge-notify`, `hive-matrix-daemon`, + `hive-forge`, `hive-forge-notify`, `hive-github-notify`, + `hive-matrix-daemon`, `hive-metric`, `hive-screen-mcp`) plus the `assets`, `frontend`, `reference-docs` and `claude-plugins` trees. Wired by the flake's agent-base/ruth nixosModules to `hyperhive.packages..*`; override an diff --git a/nix/packages/default.nix b/nix/packages/default.nix index b3759bb9..29c14b4d 100644 --- a/nix/packages/default.nix +++ b/nix/packages/default.nix @@ -32,6 +32,7 @@ let hive-screen-mcp = "hyperhive screen MCP bridge (screenshot + input for GUI agents)"; hive-forge = "hyperhive Forgejo CLI"; hive-forge-notify = "hyperhive per-agent Forgejo notification poller daemon"; + hive-github-notify = "hyperhive per-agent github.com notification poller daemon"; }; # ONE compile of the whole workspace (every bin, sharing the