//! Background Forgejo notification poller. Polls
//! `GET /notifications?all=false` every 30s, formats each unread
//! notification as a broker `Wake { from: "forge" }` message, and
//! delivers it to the agent's inbox. Delivered threads are deliberately
//! left UNREAD in forge — the hive-forge read-before-comment guard keys
//! off forge's own unread-state, and the agent reading the thread via the
//! CLI is what marks it read. A delivery-dedupe cursor (thread id →
//! last-delivered `updated_at`) stops the still-unread notification from
//! re-firing a wake every poll; self-echo notifications (the agent's own
//! writes) are still marked read directly. The cursor is persisted as the
//! `forge_cursor` field of the harness's consolidated `hyperhive-harness.json`
//! (via [`crate::events`]) and reloaded on boot so a container
//! rebuild/restart doesn't re-deliver the whole currently-unread backlog —
//! it is a private dedup mirror, NOT forge's read-state, so the
//! read-before-comment guard is untouched.
//!
//! 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 persisted delivery-dedupe cursor: each poll prunes the
/// cursor to the ids in this window, so the `forge_cursor` field in
/// `hyperhive-harness.json` can never exceed this many entries. Keep
/// the two coupled — bumping the fetch limit grows the cursor'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.
///
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");

    // Delivery-dedupe cursor: notification thread id -> the `updated_at`
    // of the version we last woke the agent for. We no longer mark a
    // thread read on delivery (that would consume the unread signal the
    // hive-forge read-before-comment guard relies on), so this map is what
    // stops the same unread notification from re-firing a wake every poll.
    // A new comment bumps `updated_at`, so the thread re-delivers. This is
    // purely anti-spam, NOT a correctness oracle.
    //
    // It is persisted as the `forge_cursor` field of the harness's
    // consolidated state file and reloaded here on boot so a container
    // rebuild/restart doesn't re-deliver the entire currently-unread
    // backlog. Persisting is safe because we only record a thread AFTER a
    // successful broker delivery, and the broker inbox is durable sqlite —
    // so a persisted "delivered" entry can never swallow a wake the agent
    // never received. The cursor is a private dedup mirror, decoupled from
    // forge's own read-state, so it doesn't reintroduce the
    // read-before-comment coupling that the mark-read-on-delivery approach
    // suffered.
    let mut delivered: HashMap<u64, String> = crate::harness_state::read_forge_cursor();
    if !delivered.is_empty() {
        info!(
            entries = delivered.len(),
            "forge_notify: restored delivery-dedupe cursor"
        );
    }

    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;
    }
}

/// 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(&notif.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,
            &notif.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();

    // Tracks whether the dedupe cursor changed this poll (a new delivery
    // recorded, or the prune below dropped now-read threads) so we only
    // rewrite the on-disk cursor when there's something to persist.
    let mut cursor_dirty = false;

    for notif in &notifications {
        let Some(id) = notif.thread.id.and_then(|id| u64::try_from(id).ok()) else {
            continue;
        };

        // Delivery-dedupe: we no longer mark threads read on delivery, so
        // an unread notification reappears in every `?all=false` poll.
        // Skip it silently unless its `updated_at` advanced since the
        // version we last delivered a wake for (i.e. genuinely new
        // activity). See the `delivered` cursor 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;
        };

        let req = hive_agent_sock::Request::Wake {
            from: "forge".to_owned(),
            body,
        };
        let deliver_result = crate::client::request::<_, hive_agent_sock::Response>(socket, &req)
            .await
            .map(|_| ());
        match deliver_result {
            Ok(()) => {
                debug!(%id, "forge_notify: delivered");
                // Record the delivered version in the dedupe cursor INSTEAD
                // of marking the thread read. Leaving it unread is
                // deliberate: the hive-forge read-before-comment guard keys
                // off forge's own unread-state, and the agent reading the
                // thread via the CLI is what marks it read. Recorded only
                // here in the Ok arm — a failed delivery leaves the cursor
                // untouched, so it re-delivers next tick.
                delivered.insert(id, updated_at);
                cursor_dirty = true;
            }
            Err(e) => {
                warn!(%id, error = ?e, "forge_notify: deliver failed — leaving unread");
            }
        }
    }

    // Prune the dedupe cursor down to the threads still present in this
    // poll's unread set. Once the agent reads a thread (marking it read
    // via the CLI) it drops out of `?all=false`, so its cursor entry is
    // dead weight; dropping it bounds the map to the current unread size.
    // If such a thread later goes unread again it carries a fresh
    // `updated_at` and re-delivers correctly.
    //
    // This retain IS the cursor's size bound: `current_ids` comes from a
    // single `limit=UNREAD_FETCH_LIMIT` page, so the persisted cursor can
    // never exceed that many entries — it tracks the unread *window*, not
    // the all-time notification count. The assert makes the invariant
    // loud in tests/dev if a future pagination change silently breaks it.
    let current_ids: HashSet<u64> = notifications
        .iter()
        .filter_map(|n| n.thread.id.and_then(|id| u64::try_from(id).ok()))
        .collect();
    let before_prune = delivered.len();
    delivered.retain(|id, _| current_ids.contains(id));
    debug_assert!(
        delivered.len() <= UNREAD_FETCH_LIMIT,
        "dedupe cursor exceeded the fetch window ({} > {UNREAD_FETCH_LIMIT})",
        delivered.len(),
    );
    if delivered.len() != before_prune {
        cursor_dirty = true;
    }

    // Flush the cursor to the consolidated state file only when it
    // changed, so a rebuild/restart reloads it instead of re-delivering
    // the whole unread backlog.
    if cursor_dirty {
        crate::harness_state::write_forge_cursor(delivered);
    }
}

/// 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 only on the self-echo path (the
/// agent's own comment/review/creation writes) — delivered threads are
/// deliberately left unread for the read-before-comment guard, and a failed
/// delivery is left unread + out of the dedupe cursor so it resurfaces on the
/// next poll tick.
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");
        }
    }
}

#[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);
    }
}