1502 lines
60 KiB
Rust
1502 lines
60 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.
|
|
//!
|
|
//! **Multi-source**: always the internal Forgejo, plus github.com when the
|
|
//! agent has a PAT. Each source polls independently behind
|
|
//! [`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
|
|
//! 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;
|
|
use std::time::Duration;
|
|
|
|
use forgejo_api::structs::NotifySubjectType;
|
|
use serde::{Deserialize, Deserializer};
|
|
use time::OffsetDateTime;
|
|
use time::format_description::well_known::Rfc3339;
|
|
use tracing::{debug, warn};
|
|
|
|
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).
|
|
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.
|
|
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).
|
|
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.
|
|
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
|
|
/// 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;
|
|
|
|
/// 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),
|
|
source.own_login(client),
|
|
)
|
|
.await
|
|
.unwrap_or_default()
|
|
}
|
|
|
|
/// 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;
|
|
}
|
|
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<S: Source>(
|
|
client: &reqwest::Client,
|
|
source: &S,
|
|
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, source).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,
|
|
source,
|
|
&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,
|
|
source,
|
|
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<S: Source>(
|
|
client: &reqwest::Client,
|
|
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, source).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
|
|
};
|
|
|
|
// Deliberately NOT embedding the issue/PR's own description here — mara,
|
|
// triage: "dont pollute todos with full issue or pr description... just
|
|
// the issue title should be enough." Unlike a comment or review body
|
|
// (the actual new content a notification exists to surface), the
|
|
// subject's `body` is the description the agent already wrote or can
|
|
// read from the URL — repeating it in every "new"/"activity on" wake
|
|
// for the same subject is noise, not context. `title` + `url` is the
|
|
// whole point of this wrapper.
|
|
let mut out = format!("[{kind}] {title}\nurl: {html_url}");
|
|
// 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<S: Source>(
|
|
client: &reqwest::Client,
|
|
source: &S,
|
|
comment_api_url: &str,
|
|
subject: Option<&serde_json::Value>,
|
|
own_login: &str,
|
|
) -> Option<String> {
|
|
let payload = fetch_json(client, comment_api_url, source).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 {
|
|
/// 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)]
|
|
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() {
|
|
// 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),
|
|
_ => None,
|
|
}),
|
|
)
|
|
}
|
|
|
|
/// 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).
|
|
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"
|
|
)]
|
|
pub async fn poll_once<S: Source, H: std::hash::BuildHasher>(
|
|
source: &S,
|
|
client: &reqwest::Client,
|
|
socket: &Path,
|
|
delivered: &mut HashMap<String, String, H>,
|
|
own_login: &str,
|
|
) -> 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 poll_hint;
|
|
}
|
|
|
|
debug!(
|
|
source = source.name(),
|
|
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.clone() 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, source, notif, own_login).await;
|
|
|
|
// None means self-echo — mark read silently, no delivery.
|
|
let Some(body) = body_opt else {
|
|
source.mark_read(client, &id).await;
|
|
continue;
|
|
};
|
|
|
|
// Upsert a *todo* (loose-ends v2) on the harness's in-agent socket,
|
|
// 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(format!("{}{id}", source.key_prefix())),
|
|
summary: body,
|
|
source: None,
|
|
reopen_if_acked: false,
|
|
};
|
|
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.
|
|
source.mark_read(client, &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<&str> = notifications
|
|
.iter()
|
|
.filter_map(|n| n.thread.id.as_deref())
|
|
.collect();
|
|
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<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)]
|
|
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".to_owned(), "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".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".to_owned(), "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.as_deref(), 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.as_deref(), 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 url line 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);
|
|
}
|
|
|
|
#[test]
|
|
fn format_state_change_never_embeds_the_subject_body() {
|
|
// mara, triage: "dont pollute todos with full issue or pr
|
|
// description... just the issue title should be enough." Covers
|
|
// both branches that used to embed it — the "new" path and a
|
|
// later state change — so a regression can't sneak back in via
|
|
// either.
|
|
let meta = state_change_meta(serde_json::json!({
|
|
"user": { "login": "someone-else" },
|
|
"body": "the PR description",
|
|
}));
|
|
let new_out = format_state_change_notification(None, "open", &meta, "damocles", None)
|
|
.expect("open notification must render");
|
|
assert!(!new_out.contains("the PR description"));
|
|
|
|
let merged_out = format_state_change_notification(None, "merged", &meta, "damocles", None)
|
|
.expect("merge notification must render");
|
|
assert!(!merged_out.contains("the PR description"));
|
|
}
|
|
}
|