feat(#2642): a github.com notification poller alongside the forge one
hive-forge-notify grows a second binary, hive-github-notify. The two share the notification half of the job — tolerant parse, classification, formatting, dedupe, todo delivery — and nothing else: each binary owns its host's protocol outright. Two binaries rather than one multi-source daemon, and rather than a cargo feature. A feature would unify across the workspace and cost every crate its build cache. Two binaries keep the decision in nix: forge.nix installs the forge unit, github.nix installs the github one under hyperhive.github.enable, so a hive built without that module has no github poller in its closure at all — GitHub access is separable (a tier, a policy boundary), not merely switched off. Both binaries ship from the existing derivation, so packages.nix is untouched. The split is real at the code level too, not just at the unit level. source.rs is a trait; the impls live in the binaries that use them, so neither binary links the other's protocol code and the library names no host at all. The forge-only assigned-issue rollup moves into the forge binary for the same reason: it asks the forge what is assigned to this agent, which is not a notification-protocol concern. At runtime the github unit needs a PAT at <state>/github-token, the same dashboard-provisioned token the gh wrapper and the git credential helper already use. No PAT: it logs why and exits 0, which is why the unit is Restart=on-failure and not always. Forgejo's notifications API is modelled on GitHub's, so one tolerant parse serves both — the differences (string thread ids, PullRequest vs Pull) are absorbed by lenient deserializers rather than a second parse path. Thread ids normalise to String at the parse boundary; they are only ever opaque keys. Todo keys gain a per-source prefix so the two hosts cannot collide, and the forge's is deliberately empty to keep existing forge todo keys stable across the deploy that lands this. The github loop honours the server's X-Poll-Interval, re-arming only when the server asks for a slower cadence than ours; the hint is read before the status check, because it arrives on error and empty pages too and that is exactly when it matters. Reading the notification stream needs the notifications scope on the PAT, which a token minted for push access typically lacks; the failure mode is silence, so docs/github.md says so explicitly.
This commit is contained in:
atlas
mara
parent
3059523172
commit
0db83c40a0
15 changed files with 971 additions and 412 deletions
|
|
@ -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 <kind>` 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<u64, String> = 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<S: Source>(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<serde_json::Value> {
|
||||
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<S: Source>(
|
||||
client: &reqwest::Client,
|
||||
url: &str,
|
||||
source: &S,
|
||||
) -> Option<serde_json::Value> {
|
||||
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<S: Source>(
|
||||
client: &reqwest::Client,
|
||||
token: &str,
|
||||
source: &S,
|
||||
notif: &PolledNotification,
|
||||
own_login: &str,
|
||||
) -> Option<String> {
|
||||
|
|
@ -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<S: Source>(
|
||||
client: &reqwest::Client,
|
||||
token: &str,
|
||||
source: &S,
|
||||
meta: &NotifMeta<'_>,
|
||||
comment_api_url: &str,
|
||||
comment_html_url: &str,
|
||||
own_login: &str,
|
||||
) -> Option<String> {
|
||||
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<S: Source>(
|
||||
client: &reqwest::Client,
|
||||
token: &str,
|
||||
source: &S,
|
||||
comment_api_url: &str,
|
||||
subject: Option<&serde_json::Value>,
|
||||
own_login: &str,
|
||||
) -> Option<String> {
|
||||
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<OffsetDateTime> {
|
|||
/// break notification read-state again.
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct NotificationThread {
|
||||
#[serde(default)]
|
||||
id: Option<i64>,
|
||||
/// 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<String>,
|
||||
#[serde(default, deserialize_with = "de_opt_rfc3339")]
|
||||
updated_at: Option<OffsetDateTime>,
|
||||
#[serde(default)]
|
||||
|
|
@ -909,7 +787,10 @@ where
|
|||
{
|
||||
Ok(
|
||||
Option::<String>::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<Option<String>, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
Ok(match Option::<serde_json::Value>::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<Url>` would reject).
|
||||
|
|
@ -984,58 +882,26 @@ fn parse_notification(value: serde_json::Value) -> Option<PolledNotification> {
|
|||
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<S: Source, H: std::hash::BuildHasher>(
|
||||
source: &S,
|
||||
client: &reqwest::Client,
|
||||
token: &str,
|
||||
socket: &Path,
|
||||
delivered: &mut HashMap<u64, String>,
|
||||
delivered: &mut HashMap<String, String, H>,
|
||||
own_login: &str,
|
||||
) {
|
||||
// Fetch the page as raw JSON (`response_type::<String>`) instead of the
|
||||
// crate's `Vec<NotificationThread>`: 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::<String>();
|
||||
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<serde_json::Value> = match serde_json::from_str(&raw) {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
warn!("forge_notify: response parse error: {e}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
) -> Option<u64> {
|
||||
// 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<u64> = 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<u64, String>, 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::<String>();
|
||||
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<u64> {
|
||||
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::<u64>().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<H: std::hash::BuildHasher>(
|
||||
delivered: &HashMap<String, String, H>,
|
||||
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
|
||||
|
|
|
|||
Loading…
Reference in a new issue