7a826f9ee2
Six places in the tree hand-rolled the same connect / write one JSON line / read one JSON line back. Two of them — the harness serve loop's client and the MCP server's — were byte-identical apart from a six-line wrapper, ~145 lines of literal copy-paste. The other four each reimplemented a subset, and the subsets had drifted: some named the socket path in their errors and some did not, one classified transient against fatal failures and the rest retried nothing at all, two drained the response and two decoded it. That duplication was defended when the daemons were split out, on the grounds that a daemon's socket etiquette should stay visible in the crate that depends on it. The etiquette genuinely does differ. The code does not, and five copies is where "each daemon documents its own etiquette" stops paying for itself. `hive-sock-client` now owns the transport once, generic over the request and response types so it is protocol-agnostic: the host-served control socket and the harness's in-agent socket both use it with their own wire-type crates. The two real differences become values instead of forks. Retry is `Retry::RideOutRestart` (2/4/8/16/30s, sized to ride out a service restart) for callers with no natural retry of their own, or `Retry::None` for callers already inside a poll loop where the poll interval is the retry — and the reason each caller picked one is a comment at the call site rather than a reimplementation. The response is either decoded (`request`) or half-closed and drained (`notify`, where the drain exists so the server's write-back doesn't land on a closed socket). Whether a failure propagates or is logged and swallowed stays at the call site, because that is the caller's choice and not a property of the transport. Errors always name the socket path now, everywhere. That detail is load-bearing: a permission problem on a socket that reads as "is the daemon running?" sends the operator to fix the wrong thing. The transient-against-fatal enum is gone rather than moved. Serialising happens before the retry loop and deserialising after it, so only connect, I/O and short-read failures can reach the loop at all — a deterministic failure is now unretryable by construction instead of by classification. It is deliberately a new crate and not part of `hive-agent-sock`. The `*-sock` crates are pure wire types by convention — `hive-agent-sock` depends on serde and nothing else — and the two largest copies talk to the host socket, whose types live in a different crate entirely. A transport in either wire-type crate would drag tokio into it and point the wrong way besides. No wire-format change: same JSON line in, same line out.
1731 lines
69 KiB
Rust
1731 lines
69 KiB
Rust
//! Background Forgejo notification poller. Polls
|
|
//! `GET /notifications?all=false` every 30s, formats each unread
|
|
//! notification as a broker `Wake { from: "forge" }` message, delivers it
|
|
//! to the agent's inbox, and — on a successful delivery — marks the thread
|
|
//! read on forge straight away. The broker inbox is the durable work queue
|
|
//! (each row has its own ack lifecycle), so the forge unread flag no longer
|
|
//! needs to track agent processing: clearing it on delivery keeps forge's
|
|
//! unread set tiny by construction, so a container rebuild's re-scan of
|
|
//! `?all=false` finds nothing stale and cannot re-deliver a backlog. Forge's
|
|
//! own read-state is thus the durable, cross-rebuild record of what's been
|
|
//! delivered — there is no persisted cursor. A small in-process dedupe map
|
|
//! (thread id → last-delivered `updated_at`) only guards the narrow window
|
|
//! where a mark-read call transiently fails and the thread reappears unread
|
|
//! before its `updated_at` bumps; it is ephemeral and reset on restart.
|
|
//! Self-echo notifications (the agent's own writes) are marked read without
|
|
//! a delivery.
|
|
//!
|
|
//! Activation gates, self-notification filtering, body excerpt +
|
|
//! truncation + heading escape, wrapper formats (comment / review /
|
|
//! new-item / state-change), meta suffix, and review-request override
|
|
//! all live in [`docs/forge.md::Notification poller`](../../../docs/forge.md).
|
|
|
|
use std::collections::{HashMap, HashSet};
|
|
use std::fmt::Write as _;
|
|
use std::path::{Path, PathBuf};
|
|
use std::time::Duration;
|
|
|
|
use forgejo_api::structs::{NotifyGetListQuery, NotifyReadThreadQuery, NotifySubjectType};
|
|
use forgejo_api::{Auth, Forgejo, ForgejoError};
|
|
use serde::{Deserialize, Deserializer};
|
|
use time::OffsetDateTime;
|
|
use time::format_description::well_known::Rfc3339;
|
|
use tracing::{debug, info, warn};
|
|
|
|
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;
|
|
/// 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;
|
|
/// 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;
|
|
/// 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;
|
|
/// 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
|
|
/// path because it carries no `latest_comment_url` (e.g. a bodiless
|
|
/// review submission) fires well outside this window, so we must not
|
|
/// 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 {
|
|
tokio::time::timeout(
|
|
Duration::from_secs(HTTP_TIMEOUT_SECS),
|
|
forge.user_get_current().send(),
|
|
)
|
|
.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()?;
|
|
if !resp.status().is_success() {
|
|
return None;
|
|
}
|
|
resp.json().await.ok()
|
|
}
|
|
|
|
/// Map a Forgejo notification `subject.type` to a human-readable label.
|
|
/// `Commit` / `Repository` keep their API names, matching the old raw
|
|
/// pass-through of types we don't relabel; a missing type degrades to
|
|
/// `?` like every other absent field.
|
|
fn notif_type_label(t: Option<NotifySubjectType>) -> &'static str {
|
|
match t {
|
|
Some(NotifySubjectType::Pull) => "PR",
|
|
Some(NotifySubjectType::Issue) => "issue",
|
|
Some(NotifySubjectType::Commit) => "Commit",
|
|
Some(NotifySubjectType::Repository) => "Repository",
|
|
None => "?",
|
|
}
|
|
}
|
|
|
|
/// Escape strict-ATX markdown headings in a body before embedding it
|
|
/// inside the forge-notify wrapper, so a leading `## title` line
|
|
/// doesn't blow into an h2 in the dashboard render. See
|
|
/// `docs/forge.md::Body excerpt + truncation + heading escape` for
|
|
/// the strict-ATX-vs-`#tag` rationale and the `split_inclusive`
|
|
/// trailing-newline contract.
|
|
fn escape_md_headings(body: &str) -> String {
|
|
let mut out = String::with_capacity(body.len());
|
|
for line in body.split_inclusive('\n') {
|
|
let (content, terminator) = match line.strip_suffix('\n') {
|
|
Some(rest) => (rest, "\n"),
|
|
None => (line, ""),
|
|
};
|
|
let trimmed = content.trim_start();
|
|
if is_atx_heading(trimmed) {
|
|
let lead = &content[..content.len() - trimmed.len()];
|
|
out.push_str(lead);
|
|
out.push('\\');
|
|
out.push_str(trimmed);
|
|
} else {
|
|
out.push_str(content);
|
|
}
|
|
out.push_str(terminator);
|
|
}
|
|
out
|
|
}
|
|
|
|
/// Strict `CommonMark` ATX-heading detector: 1-6 leading `#`s followed
|
|
/// by either a space, tab, or end-of-line. Anything tighter (`#tag`,
|
|
/// `#9`) is a non-heading line that the renderer will not promote.
|
|
fn is_atx_heading(line: &str) -> bool {
|
|
let hashes = line.bytes().take_while(|&b| b == b'#').count();
|
|
if !(1..=6).contains(&hashes) {
|
|
return false;
|
|
}
|
|
// Bare `#` / `##` / ... on its own line, or proper ATX with a
|
|
// space/tab after the run of `#`s; anything else (`#tag` / `#9`)
|
|
// is not a heading.
|
|
matches!(line.as_bytes().get(hashes), None | Some(b' ' | b'\t'))
|
|
}
|
|
|
|
fn truncate(s: &str, max: usize) -> String {
|
|
if s.len() <= max {
|
|
return s.to_owned();
|
|
}
|
|
let end = s
|
|
.char_indices()
|
|
.map(|(i, _)| i)
|
|
.take_while(|&i| i <= max.saturating_sub(3))
|
|
.last()
|
|
.unwrap_or(0);
|
|
format!("{}…", &s[..end])
|
|
}
|
|
|
|
/// Detect `@username` mentions on a line. A mention is `@` followed by
|
|
/// at least one username char (alphanumeric / `_` / `-`) where the `@`
|
|
/// is at line start or follows a non-username char — so email-style
|
|
/// `foo@bar.com` does NOT count as a mention.
|
|
fn contains_mention(line: &str) -> bool {
|
|
let bytes = line.as_bytes();
|
|
for (i, &b) in bytes.iter().enumerate() {
|
|
if b != b'@' {
|
|
continue;
|
|
}
|
|
// Boundary: preceding byte must NOT be a username char.
|
|
let boundary_ok = match i.checked_sub(1).map(|j| bytes[j]) {
|
|
None => true,
|
|
Some(prev) => !is_username_byte(prev),
|
|
};
|
|
if !boundary_ok {
|
|
continue;
|
|
}
|
|
// Following byte must be at least one username char.
|
|
if bytes.get(i + 1).is_some_and(|&c| is_username_byte(c)) {
|
|
return true;
|
|
}
|
|
}
|
|
false
|
|
}
|
|
|
|
fn is_username_byte(b: u8) -> bool {
|
|
b.is_ascii_alphanumeric() || b == b'_' || b == b'-'
|
|
}
|
|
|
|
/// Walk `full_body` line-by-line; return lines that contain an
|
|
/// `@username` mention AND aren't already present (as a substring) in
|
|
/// `included_excerpt`. Surfaces tags that fell outside the truncation
|
|
/// window so addressed agents never silently miss a mention on a long
|
|
/// body. See `docs/forge.md::Body excerpt + truncation + heading
|
|
/// escape` for the truncate-before-escape ordering rule.
|
|
fn extract_truncated_mention_lines<'a>(full_body: &'a str, included_excerpt: &str) -> Vec<&'a str> {
|
|
full_body
|
|
.lines()
|
|
.filter(|line| {
|
|
let trimmed = line.trim();
|
|
!trimmed.is_empty() && contains_mention(trimmed)
|
|
})
|
|
.filter(|line| !included_excerpt.contains(line.trim()))
|
|
.collect()
|
|
}
|
|
|
|
/// Build the trailing `mentions (truncated from body):\n > …` block.
|
|
/// Empty string when there's nothing to surface. Caller embeds it
|
|
/// directly before the meta suffix.
|
|
fn render_truncated_mentions(lines: &[&str]) -> String {
|
|
if lines.is_empty() {
|
|
return String::new();
|
|
}
|
|
let mut out = String::from("\n\nmentions (truncated from body):");
|
|
for line in lines {
|
|
write!(out, "\n > {}", line.trim()).ok();
|
|
}
|
|
out
|
|
}
|
|
|
|
/// The escaped body excerpt + overflowed-mention suffix for one issue / PR /
|
|
/// comment body, computed in the one required order: truncate → diff the
|
|
/// overflowed `@mention` lines against the *unescaped* excerpt → heading-escape
|
|
/// the excerpt. Both notification formatters build their body block from this
|
|
/// pair so the ordering contract lives in exactly one place. Returns
|
|
/// `(escaped_excerpt, mentions_suffix)`; `mentions_suffix` is empty when the
|
|
/// body fit (nothing was truncated away).
|
|
fn render_body_excerpt(raw: &str) -> (String, String) {
|
|
let raw_excerpt = truncate(raw, BODY_TRUNCATE);
|
|
let mentions = render_truncated_mentions(&extract_truncated_mention_lines(raw, &raw_excerpt));
|
|
let excerpt = escape_md_headings(&raw_excerpt);
|
|
(excerpt, mentions)
|
|
}
|
|
|
|
/// Map a Forgejo review state to a readable action label.
|
|
/// Returns `None` for non-review states (regular comments have no `state` field;
|
|
/// `PENDING` means the review was saved but not submitted yet).
|
|
/// Forgejo review states: "APPROVED", "`REQUEST_CHANGES`", "COMMENT", "PENDING".
|
|
fn review_state_label(state: &str) -> Option<&str> {
|
|
match state {
|
|
"APPROVED" => Some("approved"),
|
|
"REQUEST_CHANGES" => Some("changes requested"),
|
|
"COMMENT" => Some("review comment"),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
/// Build a human-readable wake message for one Forgejo notification,
|
|
/// or `None` for a self-echo the caller should mark-read without
|
|
/// delivery. Wrapper format table + meta-suffix shape + number/repo
|
|
/// extraction live in `docs/forge.md::Wrapper format` +
|
|
/// `::Meta suffix`.
|
|
async fn format_notification(
|
|
client: &reqwest::Client,
|
|
token: &str,
|
|
notif: &PolledNotification,
|
|
own_login: &str,
|
|
) -> Option<String> {
|
|
let subj = notif.thread.subject.as_ref();
|
|
let title = subj.and_then(|s| s.title.as_deref()).unwrap_or("?");
|
|
let subject_type = subj.and_then(|s| s.r#type);
|
|
// forgejo-api maps a blank `html_url` (Go marshals empty strings) to
|
|
// `None`, so the API-url fallback also covers present-but-empty —
|
|
// deliberate: a fetchable API link beats the raw-HTTP predecessor's
|
|
// empty `url:` line.
|
|
let html_url = subj
|
|
.and_then(|s| s.html_url.as_ref().or(s.url.as_ref()))
|
|
.map_or("", url::Url::as_str);
|
|
|
|
// Extract issue/PR number from the html_url. URL ends with /issues/N or
|
|
// /pulls/N (possibly followed by #anchor for comments). Best-effort.
|
|
let num = html_url
|
|
.split('#')
|
|
.next()
|
|
.and_then(|u| u.rsplit('/').next())
|
|
.and_then(|s| s.parse::<u64>().ok())
|
|
.map(|n| format!(" #{n}"))
|
|
.unwrap_or_default();
|
|
|
|
// Repo slug for multi-repo disambiguation. Falls back gracefully when absent.
|
|
let repo = notif
|
|
.thread
|
|
.repository
|
|
.as_ref()
|
|
.and_then(|r| r.full_name.as_deref())
|
|
.map(|r| format!(" {r}"))
|
|
.unwrap_or_default();
|
|
|
|
// API URLs for fetching content
|
|
let subject_api_url = subj
|
|
.and_then(|s| s.url.as_ref())
|
|
.map_or("", url::Url::as_str);
|
|
let comment_api_url = subj
|
|
.and_then(|s| s.latest_comment_url.as_ref())
|
|
.map_or("", url::Url::as_str);
|
|
let comment_html_url = subj
|
|
.and_then(|s| s.latest_comment_html_url.as_ref())
|
|
.map_or("", url::Url::as_str);
|
|
|
|
// Always fetch subject detail for assignee/reviewer metadata so
|
|
// the meta suffix can show current ownership without a follow-up
|
|
// fetch (see `docs/forge.md::Meta suffix`).
|
|
let subject = if subject_api_url.is_empty() {
|
|
None
|
|
} else {
|
|
fetch_json(client, subject_api_url, token).await
|
|
};
|
|
|
|
// Forgejo's notification `subject.type` is "Pull" / "Issue", never
|
|
// "Pull Request".
|
|
let is_pr = subject_type == Some(NotifySubjectType::Pull);
|
|
let meta_suffix = build_meta_suffix(subject.as_ref(), is_pr);
|
|
|
|
// Determine whether this notification was triggered by a comment/review or
|
|
// by creation/state-change of the subject itself.
|
|
let has_comment = !comment_api_url.is_empty() && comment_api_url != subject_api_url;
|
|
|
|
let meta = NotifMeta {
|
|
title,
|
|
subject_type,
|
|
html_url,
|
|
num,
|
|
repo,
|
|
meta_suffix,
|
|
subject,
|
|
is_pr,
|
|
};
|
|
// A merged/closed subject keeps its `latest_comment_url` set, so a
|
|
// just-merged PR that had any discussion would otherwise route to the
|
|
// comment path and render as `[comment on PR]` instead of `[PR merged]`.
|
|
// When this notification IS the merge/close transition, prefer
|
|
// the state-change path even with a comment url present; a genuine later
|
|
// comment stays on the comment path (see `state_change_is_current`).
|
|
let is_fresh_state_change =
|
|
state_change_is_current(¬if.state, notif.thread.updated_at, meta.subject.as_ref());
|
|
if has_comment && !is_fresh_state_change {
|
|
format_comment_notification(
|
|
client,
|
|
token,
|
|
&meta,
|
|
comment_api_url,
|
|
comment_html_url,
|
|
own_login,
|
|
)
|
|
.await
|
|
} else {
|
|
// State-change path (merge/close/new/activity). A just-merged or
|
|
// closed subject keeps its pre-merge last comment on
|
|
// `latest_comment_url`, so `has_comment` is often true here. When
|
|
// that comment was genuinely posted AFTER the close — a comment
|
|
// racing the merge inside the state-change tolerance window — append
|
|
// it so it isn't lost (best of both worlds; see
|
|
// `docs/forge.md::Merge racing a comment`). The ordinary pre-merge
|
|
// last comment (created before `closed_at`) is left off.
|
|
let comment_tail = if has_comment {
|
|
fresh_post_close_comment_tail(
|
|
client,
|
|
token,
|
|
comment_api_url,
|
|
meta.subject.as_ref(),
|
|
own_login,
|
|
)
|
|
.await
|
|
} else {
|
|
None
|
|
};
|
|
format_state_change_notification(
|
|
notif.thread.updated_at,
|
|
¬if.state,
|
|
&meta,
|
|
own_login,
|
|
comment_tail,
|
|
)
|
|
}
|
|
}
|
|
|
|
/// Shared notification metadata extracted from the polled notification.
|
|
struct NotifMeta<'a> {
|
|
title: &'a str,
|
|
subject_type: Option<NotifySubjectType>,
|
|
html_url: &'a str,
|
|
num: String,
|
|
repo: String,
|
|
meta_suffix: String,
|
|
/// Fetched subject detail (issue/PR JSON); used for review-request detection.
|
|
subject: Option<serde_json::Value>,
|
|
is_pr: bool,
|
|
}
|
|
|
|
/// Build the `\nassignee: ...` (and optionally `\nreviewer: ...`)
|
|
/// suffix appended to every wrapper. Shape + presence rules live in
|
|
/// `docs/forge.md::Meta suffix`.
|
|
fn build_meta_suffix(subject: Option<&serde_json::Value>, is_pr: bool) -> String {
|
|
let assignees: Vec<&str> = subject
|
|
.and_then(|s| s["assignees"].as_array())
|
|
.map(|arr| arr.iter().filter_map(|a| a["login"].as_str()).collect())
|
|
.unwrap_or_default();
|
|
let assignee_line = if assignees.is_empty() {
|
|
"assignee: unassigned".to_owned()
|
|
} else {
|
|
format!("assignee: {}", assignees.join(", "))
|
|
};
|
|
// For PRs, include requested_reviewers when present.
|
|
let reviewer_line = if is_pr {
|
|
let reviewers: Vec<&str> = subject
|
|
.and_then(|s| s["requested_reviewers"].as_array())
|
|
.map(|arr| arr.iter().filter_map(|r| r["login"].as_str()).collect())
|
|
.unwrap_or_default();
|
|
if reviewers.is_empty() {
|
|
None
|
|
} else {
|
|
Some(format!("reviewer: {}", reviewers.join(", ")))
|
|
}
|
|
} else {
|
|
None
|
|
};
|
|
let mut out = format!("\n{assignee_line}");
|
|
if let Some(r) = reviewer_line {
|
|
write!(out, "\n{r}").ok();
|
|
}
|
|
out
|
|
}
|
|
|
|
/// Format a notification triggered by a new comment or review submission.
|
|
async fn format_comment_notification(
|
|
client: &reqwest::Client,
|
|
token: &str,
|
|
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 actor_login = payload
|
|
.as_ref()
|
|
.and_then(|c| c["user"]["login"].as_str())
|
|
.unwrap_or("");
|
|
|
|
// Self-notification filter: skip if we authored the comment/review.
|
|
if !own_login.is_empty() && actor_login == own_login {
|
|
debug!(%own_login, "forge_notify: skipping self-authored comment/review");
|
|
return None;
|
|
}
|
|
|
|
let body_text = payload
|
|
.as_ref()
|
|
.and_then(|c| c["body"].as_str())
|
|
.unwrap_or("")
|
|
.trim();
|
|
|
|
// PR review detection: Forgejo review objects carry a `state`
|
|
// field with values like "APPROVED" / "REQUEST_CHANGES" /
|
|
// "COMMENT". Regular issue/PR comments have no such field. Format
|
|
// reviews distinctly so the agent knows the outcome at a glance.
|
|
let review_state = payload
|
|
.as_ref()
|
|
.and_then(|c| c["state"].as_str())
|
|
.and_then(review_state_label);
|
|
|
|
let url = if comment_html_url.is_empty() {
|
|
meta.html_url
|
|
} else {
|
|
comment_html_url
|
|
};
|
|
let author = if actor_login.is_empty() {
|
|
"?"
|
|
} else {
|
|
actor_login
|
|
};
|
|
let NotifMeta {
|
|
title,
|
|
subject_type,
|
|
num,
|
|
repo,
|
|
meta_suffix,
|
|
..
|
|
} = meta;
|
|
|
|
// Truncate → mention-overflow → escape (see `render_body_excerpt`).
|
|
let (body_for_embed, truncated_mentions) = render_body_excerpt(body_text);
|
|
if let Some(review_label) = review_state {
|
|
// Review submission on a PR.
|
|
let kind = format!("PR {review_label}{num}{repo}");
|
|
let mut out = format!("[{kind}] {title}\nurl: {url}");
|
|
if body_text.is_empty() {
|
|
// Bodiless review: name who reviewed. `reviewed by:` (not
|
|
// `reviewer:`) to avoid colliding with the `reviewer:` line the
|
|
// meta suffix carries for a PR's *requested* reviewers.
|
|
write!(out, "\n\nreviewed by: {author}").ok();
|
|
} else {
|
|
write!(out, "\n\n{author}: {body_for_embed}{truncated_mentions}").ok();
|
|
}
|
|
out.push_str(meta_suffix);
|
|
Some(out)
|
|
} else {
|
|
// Regular comment.
|
|
let kind = format!("comment on {}{num}{repo}", notif_type_label(*subject_type));
|
|
let mut out = format!(
|
|
"[{kind}] {title}\nurl: {url}\n\n{author}: {body_for_embed}{truncated_mentions}"
|
|
);
|
|
out.push_str(meta_suffix);
|
|
Some(out)
|
|
}
|
|
}
|
|
|
|
/// Format a notification triggered by creation or state change of the subject.
|
|
///
|
|
/// Returns `None` for an agent's own *creation* (it opened the issue/PR) —
|
|
/// the same "don't loop claude on its own writes" rule the comment/review
|
|
/// path applies. Only creations are dropped: a later state change on the
|
|
/// agent's own subject is driven by someone else and stays a wake.
|
|
fn format_state_change_notification(
|
|
event_time: Option<OffsetDateTime>,
|
|
notif_state: &str,
|
|
meta: &NotifMeta<'_>,
|
|
own_login: &str,
|
|
comment_tail: Option<String>,
|
|
) -> Option<String> {
|
|
// Classification uses the raw `subject.state` string extracted in
|
|
// `parse_notification` — Forgejo returns "open" / "closed" / "merged"
|
|
// there. We do NOT rely on fetching the PR/issue detail for `merged`:
|
|
// - `subject.url` points to the *issues* endpoint, which returns
|
|
// `pull_request.merged`, not top-level `merged`.
|
|
// - Forgejo API type is "Pull" / "Issue", never "Pull Request".
|
|
|
|
// "New" = the subject is open (or state is absent). Used below for
|
|
// the review-request override.
|
|
let is_new = notif_state == "open" || notif_state.is_empty();
|
|
|
|
let NotifMeta {
|
|
title,
|
|
subject_type,
|
|
html_url,
|
|
num,
|
|
repo,
|
|
meta_suffix,
|
|
subject,
|
|
is_pr,
|
|
} = meta;
|
|
let label = notif_type_label(*subject_type);
|
|
// Only claim "new" when the notification actually fired at creation
|
|
// time. A review submitted with no body carries no
|
|
// `latest_comment_url`, so it lands here instead of on the comment
|
|
// path — and its event time is well after `created_at`. Labeling
|
|
// that "new PR" is misleading (see docs/forge.md, "new vs activity
|
|
// on"): agents dismiss it as a
|
|
// duplicate of the original open notification. When we can't confirm
|
|
// creation, fall back to a neutral "activity on" label.
|
|
let looks_new = notification_is_creation(event_time, subject.as_ref());
|
|
|
|
// Self-authored creation filter: skip an agent being woken by its own
|
|
// freshly-opened issue/PR. The subject payload is already fetched (for
|
|
// assignees / reviewers / body), so its poster `user.login` costs no
|
|
// extra request. Mirrors the self-authored comment/review drop above.
|
|
if looks_new && !own_login.is_empty() {
|
|
let author = subject
|
|
.as_ref()
|
|
.and_then(|s| s["user"]["login"].as_str())
|
|
.unwrap_or("");
|
|
if author == own_login {
|
|
debug!(%own_login, "forge_notify: skipping self-authored creation");
|
|
return None;
|
|
}
|
|
}
|
|
|
|
let kind = match notif_state {
|
|
"merged" => format!("{label} merged{num}{repo}"),
|
|
"closed" => format!("{label} closed{num}{repo}"),
|
|
"open" | "" if looks_new => format!("new {label}{num}{repo}"),
|
|
"open" | "" => format!("activity on {label}{num}{repo}"),
|
|
other => format!("{label}{num}{repo}: {other}"),
|
|
};
|
|
|
|
// Review-request override: Forgejo doesn't reliably set
|
|
// `reason == "review_requested"` (often null), so we check the
|
|
// subject payload's `requested_reviewers` list directly. See
|
|
// `docs/forge.md::Review-request override`.
|
|
let is_review_request = is_new
|
|
&& *is_pr
|
|
&& !own_login.is_empty()
|
|
&& subject
|
|
.as_ref()
|
|
.and_then(|s| s["requested_reviewers"].as_array())
|
|
.is_some_and(|arr| arr.iter().any(|r| r["login"].as_str() == Some(own_login)));
|
|
let kind = if is_review_request {
|
|
format!("review requested{num}{repo}")
|
|
} else {
|
|
kind
|
|
};
|
|
|
|
// Include the start of the issue/PR description so the agent gets context
|
|
// without a follow-up fetch. Same body pipeline as comment bodies.
|
|
let body_block = subject
|
|
.as_ref()
|
|
.and_then(|s| s["body"].as_str())
|
|
.map(str::trim)
|
|
.filter(|s| !s.is_empty())
|
|
.map(|raw| {
|
|
let (excerpt, mentions) = render_body_excerpt(raw);
|
|
format!("\n\n{excerpt}{mentions}")
|
|
})
|
|
.unwrap_or_default();
|
|
|
|
let mut out = format!("[{kind}] {title}\nurl: {html_url}{body_block}");
|
|
// Append a comment that raced the merge/close (best of both worlds), when
|
|
// the caller found one genuinely newer than `closed_at`.
|
|
if let Some(tail) = comment_tail {
|
|
out.push_str(&tail);
|
|
}
|
|
out.push_str(meta_suffix);
|
|
Some(out)
|
|
}
|
|
|
|
/// Decide whether a state-change notification represents the subject's
|
|
/// *creation* (so a `new <kind>` label is truthful) versus later
|
|
/// activity that merely lacked a `latest_comment_url`. Compares the
|
|
/// notification's event time (`updated_at`) against the subject's
|
|
/// `created_at`: within `NEW_ITEM_TOLERANCE_SECS` ⇒ creation. When
|
|
/// either timestamp is missing or unparseable we default to `true`,
|
|
/// preserving the prior "new" behavior rather than masking a genuine
|
|
/// new item behind the neutral fallback. See docs/forge.md, "new vs
|
|
/// activity on".
|
|
fn notification_is_creation(
|
|
event: Option<OffsetDateTime>,
|
|
subject: Option<&serde_json::Value>,
|
|
) -> bool {
|
|
let created = subject
|
|
.and_then(|s| s["created_at"].as_str())
|
|
.and_then(parse_rfc3339);
|
|
match (created, event) {
|
|
(Some(c), Some(e)) => (e - c).whole_seconds().abs() <= NEW_ITEM_TOLERANCE_SECS,
|
|
_ => true,
|
|
}
|
|
}
|
|
|
|
/// Whether this notification represents the subject's *own* merge/close
|
|
/// transition, as opposed to a later comment on an already-merged/closed
|
|
/// subject. A merged/closed PR keeps its `latest_comment_url` set, so
|
|
/// without this a just-merged PR would route to the comment path and
|
|
/// render as `[comment on PR]` instead of `[PR merged]`. We treat
|
|
/// the transition as current when the notification's event time
|
|
/// (`updated_at`) is within `NEW_ITEM_TOLERANCE_SECS` of the subject's
|
|
/// `closed_at` (set for both `merged` and `closed`). A genuine later
|
|
/// comment bumps `updated_at` well past `closed_at`, so it stays on the
|
|
/// comment path (and keeps its comment body). Missing/unparseable
|
|
/// timestamps default to `true` so a merge is never silently hidden
|
|
/// behind a stale comment — mirrors `notification_is_creation`'s
|
|
/// preserve-the-signal fallback.
|
|
fn state_change_is_current(
|
|
state: &str,
|
|
event: Option<OffsetDateTime>,
|
|
subject: Option<&serde_json::Value>,
|
|
) -> bool {
|
|
if !matches!(state, "merged" | "closed") {
|
|
return false;
|
|
}
|
|
let closed = subject
|
|
.and_then(|s| s["closed_at"].as_str())
|
|
.and_then(parse_rfc3339);
|
|
match (closed, event) {
|
|
(Some(c), Some(e)) => (e - c).whole_seconds().abs() <= NEW_ITEM_TOLERANCE_SECS,
|
|
_ => true,
|
|
}
|
|
}
|
|
|
|
/// True when the (already-fetched) comment payload was created strictly
|
|
/// after the subject's `closed_at` — i.e. a comment racing the merge/close,
|
|
/// not the pre-merge last comment a merged/closed subject keeps on
|
|
/// `latest_comment_url`. Unlike `state_change_is_current`, a missing or
|
|
/// unparseable timestamp defaults to `false`: we only append a comment we
|
|
/// can positively place after the close, so an unplaceable one is never
|
|
/// bolted onto a merge message where it might be stale.
|
|
fn comment_is_after_close(
|
|
comment: &serde_json::Value,
|
|
subject: Option<&serde_json::Value>,
|
|
) -> bool {
|
|
let closed = subject
|
|
.and_then(|s| s["closed_at"].as_str())
|
|
.and_then(parse_rfc3339);
|
|
let created = comment["created_at"].as_str().and_then(parse_rfc3339);
|
|
match (created, closed) {
|
|
(Some(cr), Some(cl)) => cr > cl,
|
|
_ => false,
|
|
}
|
|
}
|
|
|
|
/// Render the trailing comment block to append to a merge/close notification
|
|
/// when a comment raced the merge (posted after `closed_at`). Fetches the
|
|
/// comment via `latest_comment_url` and returns `None` when it predates the
|
|
/// close (the ordinary kept last comment), is self-authored (don't echo the
|
|
/// 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(
|
|
client: &reqwest::Client,
|
|
token: &str,
|
|
comment_api_url: &str,
|
|
subject: Option<&serde_json::Value>,
|
|
own_login: &str,
|
|
) -> Option<String> {
|
|
let payload = fetch_json(client, comment_api_url, token).await?;
|
|
if !comment_is_after_close(&payload, subject) {
|
|
return None;
|
|
}
|
|
let author = payload["user"]["login"].as_str().unwrap_or("");
|
|
// Don't surface the agent's own racing comment back to it — mirrors the
|
|
// self-authored filter on the comment/review path.
|
|
if !own_login.is_empty() && author == own_login {
|
|
return None;
|
|
}
|
|
let body = payload["body"].as_str().unwrap_or("").trim();
|
|
if body.is_empty() {
|
|
return None;
|
|
}
|
|
let author = if author.is_empty() { "?" } else { author };
|
|
let (excerpt, mentions) = render_body_excerpt(body);
|
|
Some(format!("\n\ncomment by {author}: {excerpt}{mentions}"))
|
|
}
|
|
|
|
/// Parse an RFC 3339 timestamp as Forgejo emits them
|
|
/// (`2026-06-13T11:18:42+02:00` or `...Z`, optionally with fractional
|
|
/// seconds). Returns `None` on any shape `time` doesn't recognise so
|
|
/// callers can fall back gracefully.
|
|
fn parse_rfc3339(s: &str) -> Option<OffsetDateTime> {
|
|
OffsetDateTime::parse(s, &Rfc3339).ok()
|
|
}
|
|
|
|
/// Minimal, drift-tolerant view of a Forgejo notification thread —
|
|
/// deliberately NOT `forgejo_api::structs::NotificationThread`. That crate
|
|
/// (0.11.0) lags the running `pkgs.forgejo` release, and one field
|
|
/// type/shape drift in the upstream struct fails the WHOLE parse, so every
|
|
/// notification is dropped, never dedup'd / marked-read, and redelivered
|
|
/// forever. This local struct carries only the fields `forge_notify` reads,
|
|
/// each optional + lenient, so unknown or reshaped upstream fields can't
|
|
/// break notification read-state again.
|
|
#[derive(Debug, Deserialize)]
|
|
struct NotificationThread {
|
|
#[serde(default)]
|
|
id: Option<i64>,
|
|
#[serde(default, deserialize_with = "de_opt_rfc3339")]
|
|
updated_at: Option<OffsetDateTime>,
|
|
#[serde(default)]
|
|
subject: Option<NotificationSubject>,
|
|
#[serde(default)]
|
|
repository: Option<NotificationRepo>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
struct NotificationSubject {
|
|
#[serde(default)]
|
|
title: Option<String>,
|
|
#[serde(rename = "type", default, deserialize_with = "de_opt_subject_type")]
|
|
r#type: Option<NotifySubjectType>,
|
|
#[serde(default, deserialize_with = "de_opt_url")]
|
|
html_url: Option<url::Url>,
|
|
#[serde(default, deserialize_with = "de_opt_url")]
|
|
url: Option<url::Url>,
|
|
#[serde(default, deserialize_with = "de_opt_url")]
|
|
latest_comment_url: Option<url::Url>,
|
|
#[serde(default, deserialize_with = "de_opt_url")]
|
|
latest_comment_html_url: Option<url::Url>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
struct NotificationRepo {
|
|
#[serde(default)]
|
|
full_name: Option<String>,
|
|
}
|
|
|
|
/// Deserialize an optional Forgejo subject `type`, mapping any unknown or
|
|
/// missing value to `None` instead of failing the parse — so a new Forgejo
|
|
/// subject type can never break notification read-state.
|
|
fn de_opt_subject_type<'de, D>(d: D) -> Result<Option<NotifySubjectType>, D::Error>
|
|
where
|
|
D: Deserializer<'de>,
|
|
{
|
|
Ok(
|
|
Option::<String>::deserialize(d)?.and_then(|s| match s.as_str() {
|
|
"Pull" => Some(NotifySubjectType::Pull),
|
|
"Issue" => Some(NotifySubjectType::Issue),
|
|
"Commit" => Some(NotifySubjectType::Commit),
|
|
"Repository" => Some(NotifySubjectType::Repository),
|
|
_ => 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).
|
|
fn de_opt_url<'de, D>(d: D) -> Result<Option<url::Url>, D::Error>
|
|
where
|
|
D: Deserializer<'de>,
|
|
{
|
|
Ok(Option::<String>::deserialize(d)?
|
|
.filter(|s| !s.is_empty())
|
|
.and_then(|s| url::Url::parse(&s).ok()))
|
|
}
|
|
|
|
/// Deserialize an optional RFC 3339 timestamp, mapping any unrecognised
|
|
/// shape to `None` via the module's tolerant [`parse_rfc3339`].
|
|
fn de_opt_rfc3339<'de, D>(d: D) -> Result<Option<OffsetDateTime>, D::Error>
|
|
where
|
|
D: Deserializer<'de>,
|
|
{
|
|
Ok(Option::<String>::deserialize(d)?
|
|
.as_deref()
|
|
.and_then(parse_rfc3339))
|
|
}
|
|
|
|
/// One notification from the poll page: the lenient thread plus two raw
|
|
/// fields the typed struct can't carry faithfully.
|
|
struct PolledNotification {
|
|
thread: NotificationThread,
|
|
/// Raw `subject.state`, kept as a string. Forgejo reports `"merged"`
|
|
/// for merged PRs (`services/convert/notification.go`) alongside
|
|
/// open/closed, and the state is matched as a string throughout, so
|
|
/// it's extracted verbatim rather than typed into an enum.
|
|
state: String,
|
|
/// Raw `updated_at` string, byte-identical to what Forgejo sent, so
|
|
/// the persisted delivery-dedupe cursor keeps its exact format
|
|
/// across the typed-client port (no spurious re-deliveries from a
|
|
/// reformatting round-trip).
|
|
updated_at: String,
|
|
}
|
|
|
|
/// Parse one notification JSON object: pull out the raw `subject.state`
|
|
/// and `updated_at` (see [`PolledNotification`]) as strings, then
|
|
/// deserialize the rest into the lenient local [`NotificationThread`].
|
|
/// Returns `None` (with a warn) only for a fundamentally malformed item
|
|
/// (e.g. a non-object) — the rest of the page still delivers.
|
|
fn parse_notification(value: serde_json::Value) -> Option<PolledNotification> {
|
|
let state = value["subject"]["state"].as_str().unwrap_or("").to_owned();
|
|
let updated_at = value["updated_at"].as_str().unwrap_or("").to_owned();
|
|
match serde_json::from_value::<NotificationThread>(value) {
|
|
Ok(thread) => Some(PolledNotification {
|
|
thread,
|
|
state,
|
|
updated_at,
|
|
}),
|
|
Err(e) => {
|
|
warn!("forge_notify: notification parse error: {e}");
|
|
None
|
|
}
|
|
}
|
|
}
|
|
|
|
#[allow(
|
|
clippy::too_many_lines,
|
|
reason = "single-pass notification poll loop — split would obscure the \
|
|
sequential 'fetch / classify / dispatch' rhythm and add helper \
|
|
functions for state shared across all three phases"
|
|
)]
|
|
async fn poll_once(
|
|
forge: &Forgejo,
|
|
client: &reqwest::Client,
|
|
token: &str,
|
|
socket: &Path,
|
|
delivered: &mut HashMap<u64, String>,
|
|
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;
|
|
}
|
|
};
|
|
|
|
if values.is_empty() {
|
|
return;
|
|
}
|
|
|
|
debug!(
|
|
count = values.len(),
|
|
"forge_notify: delivering notifications"
|
|
);
|
|
|
|
let notifications: Vec<PolledNotification> =
|
|
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 {
|
|
continue;
|
|
};
|
|
|
|
// In-process delivery-dedupe: guards against re-firing a wake for a
|
|
// thread whose mark-read (below) transiently failed and so still
|
|
// shows up unread in the next `?all=false` poll. Skip unless its
|
|
// `updated_at` advanced since the version we last delivered (i.e.
|
|
// genuinely new activity). See the `delivered` note in `run`.
|
|
let updated_at = notif.updated_at.clone();
|
|
if !should_deliver(delivered, id, &updated_at) {
|
|
debug!(%id, "forge_notify: skipping (already delivered this version)");
|
|
continue;
|
|
}
|
|
|
|
let body_opt = format_notification(client, token, notif, own_login).await;
|
|
|
|
// None means self-echo — mark read silently, no delivery.
|
|
let Some(body) = body_opt else {
|
|
mark_read(forge, 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
|
|
// 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.
|
|
let req = hive_agent_sock::Request::UpsertTodo {
|
|
subsystem: "forge".to_owned(),
|
|
key: Some(id.to_string()),
|
|
summary: body,
|
|
source: None,
|
|
};
|
|
let deliver_result = hive_sock_client::request::<_, hive_agent_sock::Response>(
|
|
socket,
|
|
&req,
|
|
crate::TODO_SOCKET_RETRY,
|
|
)
|
|
.await
|
|
.map(|_| ());
|
|
match deliver_result {
|
|
Ok(()) => {
|
|
debug!(%id, "forge_notify: todo upserted");
|
|
// Mark the thread read on forge immediately after upserting
|
|
// the todo. The todo is the durable work item now (it stays
|
|
// in `get_loose_ends` until the agent marks it done), so the
|
|
// forge unread flag no longer needs to track agent
|
|
// processing — clearing it keeps forge's unread set tiny by
|
|
// construction, so a container rebuild re-scan finds nothing
|
|
// stale (and idempotent re-upserts wouldn't re-wake anyway).
|
|
// The in-memory `delivered` entry below is only a
|
|
// within-process guard so a transient mark-read failure
|
|
// 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;
|
|
delivered.insert(id, updated_at);
|
|
}
|
|
Err(e) => {
|
|
warn!(%id, error = ?e, "forge_notify: todo upsert failed — leaving unread");
|
|
}
|
|
}
|
|
}
|
|
|
|
// Prune the in-process dedupe map to the threads still present in this
|
|
// poll's unread set. A delivered thread is marked read above, so it
|
|
// drops out of `?all=false` next poll and its entry becomes dead
|
|
// weight; dropping it bounds the map to the current unread size.
|
|
//
|
|
// `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
|
|
.iter()
|
|
.filter_map(|n| n.thread.id.and_then(|id| u64::try_from(id).ok()))
|
|
.collect();
|
|
delivered.retain(|id, _| current_ids.contains(id));
|
|
debug_assert!(
|
|
delivered.len() <= UNREAD_FETCH_LIMIT,
|
|
"in-process dedupe map exceeded the fetch window ({} > {UNREAD_FETCH_LIMIT})",
|
|
delivered.len(),
|
|
);
|
|
}
|
|
|
|
/// 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}"),
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn should_deliver_when_thread_never_seen() {
|
|
let delivered = HashMap::new();
|
|
assert!(should_deliver(&delivered, 42, "2026-06-22T16:00:00Z"));
|
|
}
|
|
|
|
#[test]
|
|
fn should_not_deliver_same_version_again() {
|
|
// 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"));
|
|
}
|
|
|
|
#[test]
|
|
fn should_deliver_when_updated_at_advanced() {
|
|
// 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"));
|
|
}
|
|
|
|
#[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"));
|
|
}
|
|
|
|
#[test]
|
|
fn escape_md_headings_escapes_top_level_atx() {
|
|
// Argus reviews start with `## argus review`, which would
|
|
// otherwise become an h2 inside the wrapper message.
|
|
assert_eq!(
|
|
escape_md_headings("## argus review\n\nlgtm."),
|
|
"\\## argus review\n\nlgtm.",
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn escape_md_headings_escapes_all_heading_depths() {
|
|
let body = "# h1\n## h2\n### h3\n###### h6\nbody";
|
|
assert_eq!(
|
|
escape_md_headings(body),
|
|
"\\# h1\n\\## h2\n\\### h3\n\\###### h6\nbody",
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn escape_md_headings_preserves_indent() {
|
|
// Indented "headings" inside lists / nested quotes keep
|
|
// their leading whitespace so structure isn't visually
|
|
// collapsed by the escape.
|
|
assert_eq!(
|
|
escape_md_headings(" ## indented\nbody"),
|
|
" \\## indented\nbody",
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn escape_md_headings_passes_non_heading_lines_through() {
|
|
let body = "plain text\nwith a #hashtag in middle\n```\n# in fenced code\n```";
|
|
let escaped = escape_md_headings(body);
|
|
// Lines without leading `#` are untouched. The `# in fenced
|
|
// code` line still gets escaped (we don't track fenced-code
|
|
// state) — acceptable: inside a fenced block the escape is
|
|
// visually inert anyway because the renderer treats the
|
|
// content as literal.
|
|
assert!(escaped.contains("plain text"));
|
|
assert!(escaped.contains("with a #hashtag in middle"));
|
|
assert!(escaped.contains("\\# in fenced code"));
|
|
}
|
|
|
|
#[test]
|
|
fn escape_md_headings_handles_empty_and_whitespace_only() {
|
|
assert_eq!(escape_md_headings(""), "");
|
|
assert_eq!(escape_md_headings(" "), " ");
|
|
assert_eq!(escape_md_headings("\n\n"), "\n\n");
|
|
}
|
|
|
|
#[test]
|
|
fn escape_md_headings_skips_non_atx_hash_lines() {
|
|
// ATX requires a space after the `#`s. Lines like `#tag`, a
|
|
// hash-then-digits run, or `#!/bin/bash` are NOT headings —
|
|
// escaping them would just add cosmetic noise where the
|
|
// renderer wouldn't promote the line in the first place.
|
|
let body = "#tag\n#123\n#!/bin/bash\n####### too many hashes\nbody"; // lint:allow: hash-digit heading test input, not a tracker tag
|
|
let escaped = escape_md_headings(body);
|
|
// All four leading `#` lines pass through untouched: too few
|
|
// (still need space), seven `#`s (over the cap), shebang
|
|
// (no space).
|
|
assert_eq!(escaped, body);
|
|
}
|
|
|
|
#[test]
|
|
fn escape_md_headings_handles_bare_hash_lines() {
|
|
// `#` alone on a line IS a valid ATX (h1 with empty text) per
|
|
// CommonMark; escape it to match the renderer's behaviour.
|
|
assert_eq!(escape_md_headings("#"), "\\#");
|
|
assert_eq!(escape_md_headings("##"), "\\##");
|
|
assert_eq!(escape_md_headings("###"), "\\###");
|
|
}
|
|
|
|
#[test]
|
|
fn escape_md_headings_preserves_trailing_newline() {
|
|
// `split_inclusive('\n')` round-trips a body ending in a
|
|
// newline. Important for embedded forge-notify bodies whose
|
|
// source already terminates with `\n` — the wrapper's spacing
|
|
// otherwise gets eaten.
|
|
assert_eq!(escape_md_headings("## h\n"), "\\## h\n");
|
|
assert_eq!(escape_md_headings("body\n"), "body\n");
|
|
assert_eq!(escape_md_headings("no trailing"), "no trailing");
|
|
}
|
|
|
|
#[test]
|
|
fn contains_mention_matches_at_line_start_and_mid_line() {
|
|
assert!(contains_mention("@damocles take a look"));
|
|
assert!(contains_mention("cc @argus please"));
|
|
assert!(contains_mention("see (@mara) for context"));
|
|
// Hyphens / underscores / digits are valid username chars.
|
|
assert!(contains_mention("ping @h-m1nd-2"));
|
|
}
|
|
|
|
#[test]
|
|
fn contains_mention_rejects_email_and_bare_at() {
|
|
// Email addresses (`foo@bar.com`) and `@` followed by
|
|
// whitespace or punctuation are not mentions — boundary check
|
|
// requires the preceding byte to NOT be a username char.
|
|
assert!(!contains_mention("foo@bar.com"));
|
|
assert!(!contains_mention("send to user@example.org"));
|
|
assert!(!contains_mention("just an @"));
|
|
assert!(!contains_mention("@ space"));
|
|
assert!(!contains_mention("plain text"));
|
|
assert!(!contains_mention(""));
|
|
}
|
|
|
|
#[test]
|
|
fn extract_truncated_keeps_mention_lines_outside_excerpt() {
|
|
// Long body where the @mention sits AFTER the excerpt's cutoff
|
|
// — the truncated extractor must surface it.
|
|
let full = "first line\nsecond line\n@damocles tagged here\n";
|
|
let excerpt = "first line\nsecond line\n…"; // mention not present
|
|
let lines = extract_truncated_mention_lines(full, excerpt);
|
|
assert_eq!(lines, vec!["@damocles tagged here"]);
|
|
}
|
|
|
|
#[test]
|
|
fn extract_truncated_drops_mentions_already_in_excerpt() {
|
|
// Mention is inside the embed window already — no need to
|
|
// re-surface, would be noise.
|
|
let full = "@damocles read this\nmore body\n";
|
|
let excerpt = "@damocles read this\nmore body\n…";
|
|
let lines = extract_truncated_mention_lines(full, excerpt);
|
|
assert!(lines.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn extract_truncated_skips_blank_and_no_mention_lines() {
|
|
// Only lines with an actual mention survive — random body
|
|
// text past the cutoff stays dropped.
|
|
let full = "first\n\nsecond paragraph\n@argus reviewer\nfinal\n";
|
|
let excerpt = "first";
|
|
let lines = extract_truncated_mention_lines(full, excerpt);
|
|
assert_eq!(lines, vec!["@argus reviewer"]);
|
|
}
|
|
|
|
#[test]
|
|
fn extract_truncated_does_not_resurface_heading_mention_inside_window() {
|
|
// Regression for the truncate-before-escape ordering rule:
|
|
// if the diff used the escaped excerpt, a body line
|
|
// `# @argus check this` would survive as-is in the body but
|
|
// become `\# @argus check this` in the excerpt, so the
|
|
// `contains` check would fail and the mention would re-surface
|
|
// as if it fell outside the window. Pass the unescaped excerpt
|
|
// (which the caller does) and the duplicate disappears.
|
|
let full = "# @argus check this\nmore body\n";
|
|
let raw_excerpt = full; // fits entirely
|
|
let lines = extract_truncated_mention_lines(full, raw_excerpt);
|
|
assert!(
|
|
lines.is_empty(),
|
|
"heading+mention inside window must not be re-surfaced, got {lines:?}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn render_truncated_mentions_empty_is_empty_string() {
|
|
// Zero overhead on the healthy short-body path: caller
|
|
// concatenates this directly so an empty input must produce
|
|
// no spacing.
|
|
assert_eq!(render_truncated_mentions(&[]), "");
|
|
}
|
|
|
|
#[test]
|
|
fn render_truncated_mentions_formats_block() {
|
|
let lines = ["cc @damocles", " @argus second mention"];
|
|
let rendered = render_truncated_mentions(&lines);
|
|
assert_eq!(
|
|
rendered,
|
|
"\n\nmentions (truncated from body):\n > cc @damocles\n > @argus second mention",
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn parse_rfc3339_handles_offsets_and_z() {
|
|
// Same instant expressed three ways must parse equal
|
|
// (`OffsetDateTime` comparison is instant-based).
|
|
let utc = parse_rfc3339("2026-06-13T09:18:42Z").unwrap();
|
|
let plus2 = parse_rfc3339("2026-06-13T11:18:42+02:00").unwrap();
|
|
let minus5 = parse_rfc3339("2026-06-13T04:18:42-05:00").unwrap();
|
|
assert_eq!(utc, plus2);
|
|
assert_eq!(utc, minus5);
|
|
// Fractional seconds parse; second-granularity comparison holds.
|
|
assert_eq!(
|
|
parse_rfc3339("2026-06-13T09:18:42.512Z")
|
|
.unwrap()
|
|
.unix_timestamp(),
|
|
utc.unix_timestamp(),
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn parse_rfc3339_rejects_garbage() {
|
|
assert!(parse_rfc3339("").is_none());
|
|
assert!(parse_rfc3339("not-a-date").is_none());
|
|
assert!(parse_rfc3339("2026-06-13").is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn notification_is_creation_flags_fresh_and_later_activity() {
|
|
let subject = serde_json::json!({ "created_at": "2026-06-13T11:18:40+02:00" });
|
|
|
|
// Fresh PR: notification event time == created_at → "new".
|
|
let fresh = parse_rfc3339("2026-06-13T11:18:42+02:00");
|
|
assert!(notification_is_creation(fresh, Some(&subject)));
|
|
|
|
// Review hours later on the same PR → not a creation.
|
|
let later = parse_rfc3339("2026-06-13T14:55:00+02:00");
|
|
assert!(!notification_is_creation(later, Some(&subject)));
|
|
|
|
// Missing timestamps → default to creation (preserve prior behavior).
|
|
assert!(notification_is_creation(None, None));
|
|
assert!(notification_is_creation(None, Some(&subject)));
|
|
assert!(notification_is_creation(
|
|
fresh,
|
|
Some(&serde_json::json!({}))
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn state_change_is_current_distinguishes_merge_from_later_comment() {
|
|
// A merged/closed PR whose notification fired ~when it was closed →
|
|
// the transition itself → prefer the state-change ([PR merged]) path
|
|
// even though a merged PR keeps its latest_comment_url set.
|
|
let merged = serde_json::json!({ "closed_at": "2026-06-13T11:18:40+02:00" });
|
|
let at_merge = parse_rfc3339("2026-06-13T11:18:42+02:00");
|
|
assert!(state_change_is_current("merged", at_merge, Some(&merged)));
|
|
assert!(state_change_is_current("closed", at_merge, Some(&merged)));
|
|
|
|
// A comment hours after the merge → not the transition → stays on the
|
|
// comment path so the comment body survives.
|
|
let later = parse_rfc3339("2026-06-13T14:55:00+02:00");
|
|
assert!(!state_change_is_current("merged", later, Some(&merged)));
|
|
|
|
// Non-terminal states never override the comment path.
|
|
assert!(!state_change_is_current("open", at_merge, Some(&merged)));
|
|
assert!(!state_change_is_current("", at_merge, Some(&merged)));
|
|
|
|
// Missing/unparseable timestamps → default true so a merge is never
|
|
// hidden behind a stale comment.
|
|
assert!(state_change_is_current("merged", None, Some(&merged)));
|
|
assert!(state_change_is_current(
|
|
"merged",
|
|
at_merge,
|
|
Some(&serde_json::json!({}))
|
|
));
|
|
assert!(state_change_is_current("merged", None, None));
|
|
}
|
|
|
|
#[test]
|
|
fn parse_notification_tolerates_merged_state() {
|
|
// forgejo-api's `StateType` has no "merged" variant; the raw
|
|
// extraction must keep the item parseable AND preserve the
|
|
// string for the `PR merged` wrapper.
|
|
let polled = parse_notification(serde_json::json!({
|
|
"id": 7,
|
|
"updated_at": "2026-06-13T11:18:42+02:00",
|
|
"url": "http://forge/api/v1/notifications/threads/7",
|
|
"subject": {
|
|
"title": "t",
|
|
"type": "Pull",
|
|
"state": "merged",
|
|
"html_url": "http://forge/o/r/pulls/5",
|
|
"latest_comment_html_url": "",
|
|
"latest_comment_url": "",
|
|
"url": "http://forge/api/v1/repos/o/r/issues/5",
|
|
},
|
|
}))
|
|
.expect("merged-state notification must parse");
|
|
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.subject.as_ref().and_then(|s| s.r#type),
|
|
Some(NotifySubjectType::Pull),
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn parse_notification_missing_subject_degrades() {
|
|
let polled = parse_notification(serde_json::json!({
|
|
"id": 1,
|
|
"updated_at": "2026-06-22T16:00:00Z",
|
|
"url": "",
|
|
}))
|
|
.expect("subject-less notification must parse");
|
|
assert_eq!(polled.state, "");
|
|
assert!(polled.thread.subject.is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn parse_notification_rejects_unrepresentable_item() {
|
|
// A non-object item can't become a `NotificationThread`; it is
|
|
// dropped alone instead of failing the page.
|
|
assert!(parse_notification(serde_json::json!("nonsense")).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn parse_notification_survives_schema_drift() {
|
|
// The whole point of the local lenient struct: a notification
|
|
// carrying fields the upstream forgejo-api struct didn't expect —
|
|
// an unknown top-level key, an unknown subject `type`, a blank
|
|
// url — must still parse so the item can be dedup'd + marked-read.
|
|
// Otherwise the parse fails, the item is dropped, and it
|
|
// redelivers forever.
|
|
let polled = parse_notification(serde_json::json!({
|
|
"id": 42,
|
|
"updated_at": "2026-07-17T12:00:00Z",
|
|
"url": "",
|
|
"some_new_forgejo_field": { "nested": true },
|
|
"subject": {
|
|
"title": "t",
|
|
"type": "SomeBrandNewType",
|
|
"html_url": "",
|
|
"url": "http://forge/api/v1/repos/o/r/issues/9",
|
|
"another_unknown": 123,
|
|
},
|
|
}))
|
|
.expect("drifted notification must still parse");
|
|
assert_eq!(polled.thread.id, Some(42));
|
|
// Unknown subject type degrades to None instead of failing.
|
|
assert!(
|
|
polled
|
|
.thread
|
|
.subject
|
|
.as_ref()
|
|
.and_then(|s| s.r#type)
|
|
.is_none()
|
|
);
|
|
// Blank html_url is None, not a parse error.
|
|
assert!(
|
|
polled
|
|
.thread
|
|
.subject
|
|
.as_ref()
|
|
.is_some_and(|s| s.html_url.is_none())
|
|
);
|
|
}
|
|
|
|
/// Build a `NotifMeta` for the state-change formatter tests. The `&str`
|
|
/// fields borrow `'static` literals so the value is self-contained.
|
|
fn state_change_meta(subject: serde_json::Value) -> NotifMeta<'static> {
|
|
NotifMeta {
|
|
title: "subject title",
|
|
subject_type: Some(NotifySubjectType::Issue),
|
|
html_url: "http://forge/issues/1",
|
|
num: " #1".to_owned(),
|
|
repo: " [agents/x]".to_owned(),
|
|
meta_suffix: "\nassignee: unassigned".to_owned(),
|
|
subject: Some(subject),
|
|
is_pr: false,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn state_change_drops_self_authored_creation() {
|
|
// No timestamps ⇒ treated as a creation; poster login == own_login.
|
|
let meta = state_change_meta(serde_json::json!({ "user": { "login": "damocles" } }));
|
|
assert!(format_state_change_notification(None, "open", &meta, "damocles", None).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn state_change_keeps_other_authored_creation() {
|
|
let meta = state_change_meta(serde_json::json!({ "user": { "login": "someone-else" } }));
|
|
assert!(format_state_change_notification(None, "open", &meta, "damocles", None).is_some());
|
|
}
|
|
|
|
#[test]
|
|
fn state_change_keeps_self_authored_later_activity() {
|
|
// The agent authored the subject, but this notification fired well
|
|
// after creation (someone else acted on it) ⇒ not a creation ⇒ still
|
|
// surfaces.
|
|
let meta = state_change_meta(serde_json::json!({
|
|
"user": { "login": "damocles" },
|
|
"created_at": "2020-01-01T00:00:00Z",
|
|
}));
|
|
let event = parse_rfc3339("2026-06-22T16:00:00Z");
|
|
assert!(
|
|
format_state_change_notification(event, "closed", &meta, "damocles", None).is_some()
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn comment_is_after_close_distinguishes_racing_from_kept_comment() {
|
|
let subject = serde_json::json!({ "closed_at": "2026-06-13T11:18:40+02:00" });
|
|
|
|
// Comment posted a minute after the merge → a comment racing the
|
|
// merge → surface it.
|
|
let racing = serde_json::json!({ "created_at": "2026-06-13T11:19:40+02:00" });
|
|
assert!(comment_is_after_close(&racing, Some(&subject)));
|
|
|
|
// The pre-merge last comment kept on `latest_comment_url` predates
|
|
// the close → must NOT be appended.
|
|
let kept = serde_json::json!({ "created_at": "2026-06-13T10:00:00+02:00" });
|
|
assert!(!comment_is_after_close(&kept, Some(&subject)));
|
|
|
|
// Missing/unparseable timestamps → false (don't append a comment we
|
|
// can't place after the close), opposite of the state-change default.
|
|
assert!(!comment_is_after_close(&racing, None));
|
|
assert!(!comment_is_after_close(
|
|
&serde_json::json!({}),
|
|
Some(&subject)
|
|
));
|
|
assert!(!comment_is_after_close(
|
|
&racing,
|
|
Some(&serde_json::json!({}))
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn format_state_change_appends_comment_tail() {
|
|
// The racing-comment tail lands between the body block and the meta
|
|
// suffix, so the assignee line stays last.
|
|
let meta = state_change_meta(serde_json::json!({
|
|
"user": { "login": "someone-else" },
|
|
"body": "the PR description",
|
|
}));
|
|
let out = format_state_change_notification(
|
|
None,
|
|
"merged",
|
|
&meta,
|
|
"damocles",
|
|
Some("\n\ncomment by argus: nice, merging".to_owned()),
|
|
)
|
|
.expect("merge notification must render");
|
|
assert!(out.contains("comment by argus: nice, merging"));
|
|
// Tail precedes the meta suffix.
|
|
let tail_at = out.find("comment by argus").unwrap();
|
|
let assignee_at = out.find("assignee:").unwrap();
|
|
assert!(tail_at < assignee_at);
|
|
}
|
|
}
|