feat(rhema-code): add the colors setting with four palettes

The stored `colors` key names a palette — rhema (the default,
Jadeite), aquamarine, black-and-white, or dracula — and the
Appearance page of /settings cycles it live, like the theme and
icon toggles beside it. Each palette carries a dark and a light
appearance under the existing `theme` key; dracula's light side is
derived, the official palette ships only dark.

The appearance decision moves off token comparison: Theme now
answers is_dark() from the ground/text luminance, so palettes
beyond the original two get the right per-profile fallback tones.

integrations/rhema-forge-integration/src/error.rs

@@ -0,0 +1,73 @@
+//! Why a forge call failed.
+//!
+//! One error type for the whole crate: every backend maps its transport,
+//! HTTP, and decoding failures onto it, so callers match one shape no
+//! matter which forge answered.
+
+/// Everything that can go wrong between a [`crate::Forge`] call and its
+/// answer.
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+pub enum ForgeError {
+    /// The forge answered with a non-2xx status. `body` is a bounded
+    /// snippet of the response body — usually the forge's error JSON.
+    #[error("forge api error (HTTP {status}): {body}")]
+    Api {
+        /// The HTTP status the forge answered with.
+        status: u16,
+        /// A bounded snippet of the response body.
+        body: String,
+    },
+    /// A response body could not be decoded, or a wire row is missing a
+    /// field the normalized vocabulary requires. Loud by design: a shape
+    /// break must surface, not silently drop events.
+    #[error("malformed forge response: {0}")]
+    Parse(String),
+    /// The server kept offering full pages, or offered a page cursor that
+    /// leads nowhere. Polling stopped rather than looping forever.
+    #[error("pagination failed: {0}")]
+    Pagination(String),
+    /// The request never got an answer (DNS, connect, timeout).
+    #[error("network error: {0}")]
+    Network(String),
+}
+
+/// The longest body snippet an [`ForgeError::Api`] keeps: enough for any
+/// forge's error JSON, small enough that error lines stay bounded.
+const BODY_SNIPPET_CAP: usize = 512;
+
+impl ForgeError {
+    /// Builds [`ForgeError::Api`] from a non-2xx status and the raw body,
+    /// truncating the body to [`BODY_SNIPPET_CAP`] characters.
+    pub(crate) fn api(status: reqwest::StatusCode, raw: &str) -> Self {
+        let body: String = raw.chars().take(BODY_SNIPPET_CAP).collect();
+        ForgeError::Api {
+            status: status.as_u16(),
+            body,
+        }
+    }
+
+    /// Maps a transport failure onto [`ForgeError::Network`]; reqwest's
+    /// error is neither `Clone` nor `Eq`, so only its display text travels.
+    pub(crate) fn network(err: reqwest::Error) -> Self {
+        ForgeError::Network(err.to_string())
+    }
+}
+
+/// A wire field the normalized vocabulary cannot do without. A missing one
+/// is a loud [`ForgeError::Parse`], not a silently skipped row.
+pub(crate) fn required<T>(field: Option<T>, what: &str) -> Result<T, ForgeError> {
+    field.ok_or_else(|| ForgeError::Parse(format!("forge row without {what}")))
+}
+
+/// Passes a 2xx response through; turns anything else into
+/// [`ForgeError::Api`] with a bounded body snippet.
+pub(crate) async fn ensure_success(
+    response: reqwest::Response,
+) -> Result<reqwest::Response, ForgeError> {
+    let status = response.status();
+    if status.is_success() {
+        return Ok(response);
+    }
+    let body = response.text().await.unwrap_or_default();
+    Err(ForgeError::api(status, &body))
+}

integrations/rhema-forge-integration/src/event.rs

@@ -0,0 +1,509 @@
+//! The normalized event vocabulary shared by every forge backend.
+//!
+//! Three REST dialects reduce to one shape here. Constraints the
+//! normalization pins (each backend module documents how it meets them):
+//! - An event is a thread-level fact ([`EventKind::Opened`], `Reopened`,
+//!   `Closed`, `Merged`), a comment ([`EventKind::Comment`]), or a
+//!   submitted review ([`EventKind::Review`]). Whether the thread is an
+//!   issue or a pull/merge request rides on [`ForgeEvent::is_pr`], so the
+//!   kind list stays small and the "issue comment vs PR comment" split is
+//!   `EventKind::Comment` + `is_pr`.
+//! - [`ForgeEvent::id`] is the dedup id the RSD dispatcher keys on: the
+//!   feed's item id plus, for thread-level rows, the row's updated time —
+//!   stable while a row is unchanged, new the moment it changes, exactly
+//!   the "thread's updated_at as dedup id" scheme of RSD §8. Comment and
+//!   review rows are immutable enough that their own ids suffice.
+//! - Thread-level polls read row snapshots, not transitions: the actor on
+//!   such an event is the thread author, never the (unknowable from a
+//!   poll) closer or merger.
+
+use core::cmp::Ordering;
+use core::fmt;
+
+use serde::de::{self, Deserializer};
+use serde::ser::Serializer;
+use serde::{Deserialize, Serialize};
+
+/// What happened to a thread.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum EventKind {
+    /// The thread was created.
+    Opened,
+    /// A closed thread returned to open.
+    Reopened,
+    /// The thread was closed (a pull/merge request: closed unmerged).
+    Closed,
+    /// A pull/merge request was merged. Never an issue.
+    Merged,
+    /// A comment was posted; [`ForgeEvent::is_pr`] separates issue from
+    /// pull-request comments.
+    Comment,
+    /// A formal review was submitted on a pull/merge request.
+    Review,
+}
+
+/// The coarse state of a thread as it stands right now.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ThreadState {
+    /// Open for business.
+    Open,
+    /// Closed; a pull/merge request closed unmerged.
+    Closed,
+    /// A merged pull/merge request.
+    Merged,
+}
+
+/// Repository coordinates. `owner` + `name` + a thread [`ForgeEvent::number`]
+/// is the resource key the RSD dispatcher forms (`host:owner/repo#number`).
+///
+/// GitLab's `owner` is the whole namespace path (`group/subgroup`), so
+/// [`RepoRef::path`] is always the full project path the forge addresses.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub struct RepoRef {
+    /// Owner login, or the namespace path on GitLab.
+    pub owner: String,
+    /// Repository or project slug.
+    pub name: String,
+}
+
+impl RepoRef {
+    /// Coordinates for `owner`/`name`. Neither is validated: callers pass
+    /// forge-correct names (the adapters build these from forge data).
+    pub fn new(owner: impl Into<String>, name: impl Into<String>) -> Self {
+        Self {
+            owner: owner.into(),
+            name: name.into(),
+        }
+    }
+
+    /// `owner/name` — the full project path.
+    pub fn path(&self) -> String {
+        format!("{}/{}", self.owner, self.name)
+    }
+}
+
+/// A thread's current state: everything a handler needs to orient without
+/// a second API round-trip.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct Snapshot {
+    /// The thread title.
+    pub title: String,
+    /// The opening body, when the thread has one.
+    pub body: Option<String>,
+    /// Open, closed, or merged.
+    pub state: ThreadState,
+    /// Whether the thread is a pull/merge request.
+    pub is_pr: bool,
+    /// Who opened the thread.
+    pub author: Option<String>,
+    /// Label names, bare.
+    pub labels: Vec<String>,
+    /// Human-facing URL of the thread.
+    pub url: Option<String>,
+    /// The untouched wire row the snapshot came from.
+    pub raw: serde_json::Value,
+}
+
+/// One normalized piece of forge activity: a thread transition, a comment,
+/// or a submitted review.
+///
+/// The shape the RSD dispatcher consumes: [`ForgeEvent::repo`] plus
+/// [`ForgeEvent::number`] form the resource key, [`ForgeEvent::id`] is the
+/// dedup id, and [`ForgeEvent::raw`] carries the untouched wire row for
+/// handler context.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct ForgeEvent {
+    /// Stable dedup id: the feed's item id plus, for thread-level rows,
+    /// the row's updated time (e.g. `github:pull:1824@2026-10-08T12:00:00Z`,
+    /// `github:comment:29482`). Two polls of the same unchanged row produce
+    /// the same id; a changed row produces a new one.
+    pub id: String,
+    /// What happened.
+    pub kind: EventKind,
+    /// The repository the thread lives in.
+    pub repo: RepoRef,
+    /// The thread's number (GitLab: the iid).
+    pub number: u64,
+    /// Whether the thread is a pull/merge request rather than an issue.
+    pub is_pr: bool,
+    /// Who acted: the thread author on thread-level events (a poll sees row
+    /// snapshots, not transitions, so the closer/merger is unknowable), the
+    /// comment or review author otherwise.
+    pub actor: Option<String>,
+    /// The comment or review text; `None` on thread-level events (the
+    /// thread body lives on the snapshot, not on the event).
+    pub body: Option<String>,
+    /// Human-facing URL of the event, when the wire row carries one.
+    pub url: Option<String>,
+    /// When the row was last touched (thread rows) or created/submitted
+    /// (comments, reviews). The poll cursor advances to the newest of
+    /// these.
+    pub timestamp: Timestamp,
+    /// The untouched wire row, minus the keys the event lifts.
+    pub raw: serde_json::Value,
+}
+
+/// Why the account was notified. The forges' own reason vocabularies,
+/// reduced to the handful a triage can act on; whatever a forge says that
+/// does not map rides as [`NotificationReason::Other`] with the forge's
+/// own spelling.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum NotificationReason {
+    /// The account was asked to review a pull/merge request.
+    ReviewRequested,
+    /// New feedback landed on a thread the account authored or
+    /// participates in (a submitted review or a conversation comment).
+    ReviewFeedback,
+    /// The account was assigned the thread.
+    Assigned,
+    /// The account was mentioned or directly addressed.
+    Mention,
+    /// CI status on a thread the account is involved in.
+    CiActivity,
+    /// Activity on a thread the account authored.
+    Author,
+    /// Anything else, in the forge's own spelling.
+    Other(String),
+}
+
+impl NotificationReason {
+    /// The forge-independent spelling, for triage matching and prompts.
+    pub fn as_str(&self) -> &str {
+        match self {
+            NotificationReason::ReviewRequested => "review_requested",
+            NotificationReason::ReviewFeedback => "review_feedback",
+            NotificationReason::Assigned => "assigned",
+            NotificationReason::Mention => "mention",
+            NotificationReason::CiActivity => "ci_activity",
+            NotificationReason::Author => "author",
+            NotificationReason::Other(raw) => raw,
+        }
+    }
+}
+
+/// One account-scoped notification: something happened on a thread that
+/// concerns the account the client authenticates as. Distinct from a
+/// [`ForgeEvent`], which is a thread-level fact lifted from a
+/// repo-scoped feed — the notification says *why the account cares*.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct Notification {
+    /// Stable dedup id, `{source}:notification:{id}@{updated time}`
+    /// (e.g. `github:notification:42188@2026-10-08T12:00:00Z`): a
+    /// notification row that stays put produces the same id across
+    /// polls; a row that moves produces a new one.
+    pub id: String,
+    /// Why the account was notified.
+    pub reason: NotificationReason,
+    /// The repository the thread lives in.
+    pub repo: RepoRef,
+    /// The thread's number (GitLab: the iid).
+    pub number: u64,
+    /// Whether the thread is a pull/merge request rather than an issue.
+    pub is_pr: bool,
+    /// Who acted, when the forge names it.
+    pub actor: Option<String>,
+    /// The thread's title, when the row carries it.
+    pub title: Option<String>,
+    /// Human-facing URL of the thread, when the row carries one.
+    pub url: Option<String>,
+    /// When the notification row was last touched. The poll cursor
+    /// advances to the newest of these.
+    pub timestamp: Timestamp,
+    /// The untouched wire row, minus the keys the notification lifts.
+    pub raw: serde_json::Value,
+}
+
+/// One notifications poll's worth, with the cursor to continue from.
+/// Same fold rules as [`ForgePage`]: oldest first, id-broken ties, and
+/// the cursor never goes backwards.
+#[derive(Debug, Clone, PartialEq)]
+pub struct NotificationPage {
+    /// The notifications, oldest first, ties broken by id — a
+    /// deterministic order regardless of which feed produced a row.
+    pub notifications: Vec<Notification>,
+    /// The cursor for the next poll: the newest event time seen, never
+    /// earlier than the poll's `since`. A backend with no server-side
+    /// time filter ignores the cursor; repetition is harmless under the
+    /// dedup ids.
+    pub cursor: Option<Timestamp>,
+}
+
+impl NotificationPage {
+    /// Folds collected notifications into a page: sorts them and advances
+    /// the cursor past everything seen.
+    pub(crate) fn finish(mut notifications: Vec<Notification>, since: Option<&Timestamp>) -> Self {
+        notifications.sort_by(|a, b| (&a.timestamp, &a.id).cmp(&(&b.timestamp, &b.id)));
+        let mut cursor = since.cloned();
+        for timestamp in notifications.iter().map(|row| &row.timestamp) {
+            cursor = Some(match cursor {
+                Some(current) if current >= *timestamp => current,
+                _ => timestamp.clone(),
+            });
+        }
+        Self {
+            notifications,
+            cursor,
+        }
+    }
+}
+
+/// An RFC 3339 timestamp, kept in the forge's own spelling.
+///
+/// The raw string is what travels: it round-trips onto cursors and into
+/// events unchanged. Ordering compares parsed instants, never text — the
+/// forges disagree on fractional seconds, and `…:00Z` must sort before
+/// `…:00.081Z`, which string comparison gets backwards.
+#[derive(Debug, Clone)]
+pub struct Timestamp {
+    raw: String,
+    /// Whole seconds since the Unix epoch (negative before 1970).
+    secs: i64,
+    /// Sub-second nanoseconds, `[0, 1_000_000_000)`.
+    nanos: u32,
+}
+
+impl Timestamp {
+    /// Parses an RFC 3339 timestamp: `YYYY-MM-DDThh:mm:ss[.frac]` followed
+    /// by `Z` (either case) or a `±hh:mm` offset. `None` on anything else.
+    ///
+    /// Validation is structural, not calendrical (day-of-month is only
+    /// bounds-checked): the forges are the source of truth and only the
+    /// ordering key matters here.
+    pub fn parse(raw: &str) -> Option<Self> {
+        let bytes = raw.as_bytes();
+        // `YYYY-MM-DDThh:mm:ss` is the 19-char floor; an offset must follow.
+        if bytes.len() < 20 || bytes[4] != b'-' || bytes[7] != b'-' {
+            return None;
+        }
+        if bytes[10] != b'T' && bytes[10] != b't' {
+            return None;
+        }
+        if bytes[13] != b':' || bytes[16] != b':' {
+            return None;
+        }
+        let year = digits(raw, 0..4)?;
+        let month = digits(raw, 5..7)?;
+        let day = digits(raw, 8..10)?;
+        let hour = digits(raw, 11..13)?;
+        let minute = digits(raw, 14..16)?;
+        let second = digits(raw, 17..19)?;
+        if !(1..=12).contains(&month) || !(1..=31).contains(&day) {
+            return None;
+        }
+        // 60 accepts the leap seconds RFC 3339 allows.
+        if hour > 23 || minute > 59 || second > 60 {
+            return None;
+        }
+
+        let mut index = 19;
+        let mut nanos: u32 = 0;
+        if bytes[index] == b'.' {
+            index += 1;
+            let start = index;
+            while bytes.get(index).is_some_and(|b| b.is_ascii_digit()) {
+                index += 1;
+            }
+            if index == start {
+                return None;
+            }
+            // Truncate past nanosecond precision, pad short fractions.
+            let mut fraction = raw[start..index].to_owned();
+            fraction.truncate(9);
+            while fraction.len() < 9 {
+                fraction.push('0');
+            }
+            nanos = fraction.parse().ok()?;
+        }
+
+        let offset: i64 = match bytes.get(index) {
+            Some(b'Z') | Some(b'z') => {
+                index += 1;
+                0
+            }
+            Some(sign @ (b'+' | b'-')) => {
+                index += 1;
+                if bytes.get(index + 2) != Some(&b':') {
+                    return None;
+                }
+                let hours = digits(raw, index..index + 2)?;
+                let minutes = digits(raw, index + 3..index + 5)?;
+                if hours > 23 || minutes > 59 {
+                    return None;
+                }
+                index += 5;
+                let total = hours * 3600 + minutes * 60;
+                if *sign == b'+' { total } else { -total }
+            }
+            _ => return None,
+        };
+        if index != bytes.len() {
+            return None; // trailing garbage
+        }
+
+        let secs = days_from_civil(year, month, day) * 86_400 + hour * 3600 + minute * 60 + second
+            - offset;
+        Some(Self {
+            raw: raw.to_owned(),
+            secs,
+            nanos,
+        })
+    }
+
+    /// The raw string, in the forge's own spelling — the form cursors and
+    /// `since`-style query parameters carry.
+    pub fn raw(&self) -> &str {
+        &self.raw
+    }
+
+    /// Whole seconds since the Unix epoch.
+    pub fn epoch_secs(&self) -> i64 {
+        self.secs
+    }
+}
+
+/// A run of ASCII digits at `range`, as its number.
+fn digits(raw: &str, range: core::ops::Range<usize>) -> Option<i64> {
+    let slice = raw.get(range)?;
+    if slice.is_empty() || !slice.bytes().all(|b| b.is_ascii_digit()) {
+        return None;
+    }
+    slice.parse().ok()
+}
+
+/// Days since 1970-01-01 for a civil date (Howard Hinnant's algorithm).
+fn days_from_civil(y: i64, m: i64, d: i64) -> i64 {
+    let y = if m <= 2 { y - 1 } else { y };
+    let era = if y >= 0 { y } else { y - 399 } / 400;
+    let yoe = y - era * 400;
+    let mp = if m > 2 { m - 3 } else { m + 9 };
+    let doy = (153 * mp + 2) / 5 + d - 1;
+    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
+    era * 146_097 + doe - 719_468
+}
+
+impl PartialEq for Timestamp {
+    fn eq(&self, other: &Self) -> bool {
+        self.secs == other.secs && self.nanos == other.nanos
+    }
+}
+impl Eq for Timestamp {}
+
+impl PartialOrd for Timestamp {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+impl Ord for Timestamp {
+    fn cmp(&self, other: &Self) -> Ordering {
+        (self.secs, self.nanos).cmp(&(other.secs, other.nanos))
+    }
+}
+
+impl fmt::Display for Timestamp {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(&self.raw)
+    }
+}
+
+impl Serialize for Timestamp {
+    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        serializer.serialize_str(&self.raw)
+    }
+}
+
+impl<'de> Deserialize<'de> for Timestamp {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        let raw = String::deserialize(deserializer)?;
+        Timestamp::parse(&raw)
+            .ok_or_else(|| de::Error::custom(format!("not an RFC 3339 timestamp: {raw:?}")))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn parses_the_shapes_the_forges_emit() {
+        // GitHub's shape.
+        let zulu = Timestamp::parse("2026-10-08T00:00:00Z").expect("zulu parses");
+        assert_eq!(zulu.epoch_secs(), 1_791_417_600);
+        // GitLab's shape: fractional seconds.
+        let millis = Timestamp::parse("2026-10-08T00:00:00.081Z").expect("millis parse");
+        assert_eq!(millis.epoch_secs(), zulu.epoch_secs());
+        assert!(millis > zulu);
+        // Offsets compare by instant, not spelling.
+        let shifted = Timestamp::parse("2026-10-08T02:00:00+02:00").expect("offset parse");
+        assert_eq!(shifted, zulu);
+        // RFC 3339 allows the lowercase forms.
+        assert_eq!(
+            Timestamp::parse("1970-01-01t00:00:00z")
+                .expect("lowercase parse")
+                .epoch_secs(),
+            0
+        );
+        assert_eq!(
+            Timestamp::parse("2016-01-04T15:31:51.081Z")
+                .expect("gitlab fixture parse")
+                .epoch_secs(),
+            1_451_921_511
+        );
+    }
+
+    #[test]
+    fn rejects_malformed_timestamps() {
+        for raw in [
+            "",
+            "2026-10-08",
+            "2026-10-08T00:00:00",
+            "2026-10-08 00:00:00Z",
+            "2026-13-01T00:00:00Z",
+            "2026-10-32T00:00:00Z",
+            "2026-10-08T24:00:00Z",
+            "2026-10-08T00:60:00Z",
+            "2026-10-08T00:00:00Z ",
+            "2026-10-08T00:00:00+0200",
+            "2026-10-08T00:00:00.Z",
+            "2026-10-08T00:00:00Q",
+        ] {
+            assert!(Timestamp::parse(raw).is_none(), "for {raw:?}");
+        }
+    }
+
+    #[test]
+    fn serde_keeps_the_raw_string() {
+        let ts = Timestamp::parse("2026-10-08T00:00:00.081Z").expect("parses");
+        let encoded = serde_json::to_value(&ts).expect("serializes");
+        assert_eq!(encoded, json!("2026-10-08T00:00:00.081Z"));
+        assert_eq!(
+            serde_json::from_value::<Timestamp>(encoded).expect("deserializes"),
+            ts
+        );
+        assert!(serde_json::from_value::<Timestamp>(json!("nonsense")).is_err());
+    }
+
+    #[test]
+    fn forge_events_round_trip_through_serde() {
+        let event = ForgeEvent {
+            id: "github:issue:7@2026-10-08T00:00:00Z".into(),
+            kind: EventKind::Closed,
+            repo: RepoRef::new("acme", "widgets"),
+            number: 7,
+            is_pr: false,
+            actor: Some("alice".into()),
+            body: None,
+            url: None,
+            timestamp: Timestamp::parse("2026-10-08T00:00:00Z").expect("parses"),
+            raw: json!({ "number": 7 }),
+        };
+        let round: ForgeEvent =
+            serde_json::from_value(serde_json::to_value(&event).expect("serializes"))
+                .expect("deserializes");
+        assert_eq!(round, event);
+    }
+}

integrations/rhema-forge-integration/src/forge.rs

@@ -0,0 +1,145 @@
+//! The forge client contract: one trait, three calls.
+//!
+//! [`Forge`] is the whole surface a consumer needs: poll a repo's
+//! activity against a cursor, post a comment, and fetch a thread
+//! snapshot so an answering handler can orient without a second API
+//! round-trip. Implementations are cheap to clone and share — the
+//! reqwest client inside is reference-counted — and object-safe so a
+//! backend can be swapped without generics.
+//!
+//! Polls are at-least-once. The forges' `since`-style filters match rows
+//! at-or-after the cursor, so boundary rows repeat across polls, and a
+//! row that changes between a backend's feeds can surface twice under
+//! different ids. Both are harmless under the RSD dedup: consumers must
+//! treat the event `id` as the only dedup key, never `(repo, number)`.
+
+use async_trait::async_trait;
+
+use crate::error::ForgeError;
+use crate::event::{ForgeEvent, NotificationPage, RepoRef, Snapshot, Timestamp};
+
+/// One poll's worth of activity, with the cursor to continue from.
+#[derive(Debug, Clone, PartialEq)]
+pub struct ForgePage {
+    /// The events, oldest first, ties broken by id — a deterministic
+    /// order regardless of which feed produced an event.
+    pub events: Vec<ForgeEvent>,
+    /// The cursor for the next poll: the newest event time seen, never
+    /// earlier than the poll's `since`. `None` only when the poll had no
+    /// anchor and came back empty — the repo is simply quiet.
+    pub cursor: Option<Timestamp>,
+}
+
+impl ForgePage {
+    /// Folds collected events into a page: sorts them and advances the
+    /// cursor past everything seen.
+    pub(crate) fn finish(mut events: Vec<ForgeEvent>, since: Option<&Timestamp>) -> Self {
+        events.sort_by(|a, b| (&a.timestamp, &a.id).cmp(&(&b.timestamp, &b.id)));
+        let mut cursor = since.cloned();
+        for timestamp in events.iter().map(|event| &event.timestamp) {
+            cursor = Some(match cursor {
+                Some(current) if current >= *timestamp => current,
+                _ => timestamp.clone(),
+            });
+        }
+        Self { events, cursor }
+    }
+}
+
+/// A forge backend: the client for one REST dialect.
+#[async_trait]
+pub trait Forge: Send + Sync {
+    /// Recent activity for `repo`, oldest first, all pages collected.
+    ///
+    /// `since` anchors the poll: `None` lists from the beginning (a first
+    /// poll); `Some(cursor)` asks for rows at-or-after the cursor, whose
+    /// boundary rows may repeat — their stable ids make the repetition
+    /// dedupable downstream, never duplicated work.
+    async fn poll(
+        &self,
+        repo: &RepoRef,
+        since: Option<&Timestamp>,
+    ) -> Result<ForgePage, ForgeError>;
+
+    /// Posts `body` as a new comment on thread `number`.
+    async fn comment(&self, repo: &RepoRef, number: u64, body: &str) -> Result<(), ForgeError>;
+
+    /// The thread's current state, so a handler can orient without a
+    /// second API round-trip.
+    async fn snapshot(&self, repo: &RepoRef, number: u64) -> Result<Snapshot, ForgeError>;
+
+    /// The account's notification feed: everything addressed to the
+    /// account this client authenticates as, across every repository it
+    /// can see — the account-scoped counterpart of [`Forge::poll`],
+    /// whose feeds are repo-scoped. `since` anchors the poll with the
+    /// same at-least-once semantics as `poll`: boundary rows repeat, and
+    /// the stable ids make the repetition dedupable downstream.
+    async fn notifications(
+        &self,
+        since: Option<&Timestamp>,
+    ) -> Result<NotificationPage, ForgeError>;
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::event::EventKind;
+    use serde_json::json;
+
+    fn event(id: &str, at: &str) -> ForgeEvent {
+        ForgeEvent {
+            id: id.to_owned(),
+            kind: EventKind::Comment,
+            repo: RepoRef::new("acme", "widgets"),
+            number: 7,
+            is_pr: false,
+            actor: None,
+            body: None,
+            url: None,
+            timestamp: Timestamp::parse(at).expect("timestamp parses"),
+            raw: json!({}),
+        }
+    }
+
+    #[test]
+    fn events_sort_oldest_first_with_id_ties() {
+        let page = ForgePage::finish(
+            vec![
+                event("b", "2026-10-08T00:00:00Z"),
+                event("z", "2026-10-07T00:00:00Z"),
+                event("a", "2026-10-08T00:00:00Z"),
+            ],
+            None,
+        );
+        let ids: Vec<&str> = page.events.iter().map(|event| event.id.as_str()).collect();
+        assert_eq!(ids, ["z", "a", "b"]);
+    }
+
+    #[test]
+    fn the_cursor_never_goes_backwards() {
+        // A since newer than everything seen holds: the next poll repeats
+        // the boundary instead of re-listing old history.
+        let since = Timestamp::parse("2026-10-09T00:00:00Z").expect("parses");
+        let page = ForgePage::finish(vec![event("a", "2026-10-08T00:00:00Z")], Some(&since));
+        assert_eq!(page.cursor, Some(since));
+
+        let page = ForgePage::finish(
+            vec![event("a", "2026-10-10T00:00:00Z")],
+            Some(&Timestamp::parse("2026-10-09T00:00:00Z").expect("parses")),
+        );
+        assert_eq!(
+            page.cursor,
+            Some(Timestamp::parse("2026-10-10T00:00:00Z").expect("parses"))
+        );
+    }
+
+    #[test]
+    fn empty_pages_keep_or_drop_the_cursor_depending_on_the_anchor() {
+        let since = Timestamp::parse("2026-10-09T00:00:00Z").expect("parses");
+        let page = ForgePage::finish(vec![], Some(&since));
+        assert_eq!(page.cursor, Some(since));
+
+        let page = ForgePage::finish(vec![], None);
+        assert_eq!(page.cursor, None);
+    }
+}

integrations/rhema-forge-integration/src/gitea.rs

@@ -0,0 +1,829 @@
+//! The Gitea adapter: self-hosted, so the base URL is required and points
+//! at the instance root (`https://forge.example`); every request path is
+//! prefixed with `/api/v1`. Auth is the token in
+//! `Authorization: token <token>`.
+//!
+//! The endpoints mirror GitHub's shape, with the constraints that differ:
+//! - Issues come from `GET /repos/{owner}/{repo}/issues` with
+//!   `state=all&type=issues&since=<RFC 3339>`; `type=issues` is explicit
+//!   so the feed is issues only, whatever the server's default.
+//! - PRs come from `GET /repos/{owner}/{repo}/pulls` with
+//!   `state=all&sort=leastrecent`. Gitea's pulls list has no time filter,
+//!   so the poll pages the whole list and filters `updated_at` client-side
+//!   (belt and braces: `leastrecent` already orders oldest-updated first,
+//!   and the final page sort is deterministic either way).
+//! - Unlike GitHub, Gitea's pulls rows carry the `merged` flag directly,
+//!   so no per-PR detail fetch is needed; only reviews fan out per
+//!   updated PR (`GET /repos/{owner}/{repo}/pulls/{index}/reviews`, no
+//!   repo-wide reviews feed).
+//! - Comments come from the repo-wide
+//!   `GET /repos/{owner}/{repo}/issues/comments` with `since=`; PR
+//!   conversation comments are issue comments, `pull_request_url` set.
+//! - The `since` filters and the client-side cutoff match rows
+//!   at-or-after the cursor, so boundary rows repeat across polls; stable
+//!   ids make the repetition dedupable.
+//! - An updated-but-still-open thread with no close history yields no
+//!   thread event: the usual cause is a comment, which the comments feed
+//!   reports with its own id.
+//! - Reviews carry `submitted`, falling back to `created` on servers that
+//!   leave it unset; `PENDING` reviews were never submitted and are
+//!   skipped.
+//! - Pagination is `limit`/`page` until a short page, with a hard page
+//!   cap. `limit=50` because Gitea's default response cap is 50 rows.
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+use serde::Deserialize;
+use serde::de::DeserializeOwned;
+use serde_json::Value;
+
+use crate::error::{ForgeError, ensure_success, required};
+use crate::event::{
+    EventKind, ForgeEvent, Notification, NotificationPage, NotificationReason, RepoRef, Snapshot,
+    ThreadState, Timestamp,
+};
+use crate::forge::{Forge, ForgePage};
+
+/// Rows requested per page: Gitea's default response cap.
+const PER_PAGE: usize = 50;
+
+/// Pages fetched before a server that keeps returning full pages is
+/// declared misbehaving rather than polled forever.
+const MAX_PAGES: usize = 1000;
+
+/// Gitea backend: an instance base URL, a token, one reqwest client.
+///
+/// An empty token is accepted but useless for writes; public reads work
+/// anonymously.
+pub struct GiteaForge {
+    base_url: String,
+    token: String,
+    http: reqwest::Client,
+}
+
+impl GiteaForge {
+    /// A forge authenticating with `token` against the instance at
+    /// `base_url` (trailing slashes trimmed). Gitea is self-hosted: there
+    /// is no default base.
+    pub fn new(base_url: impl Into<String>, token: impl Into<String>) -> Self {
+        let http = reqwest::Client::builder()
+            // One connect attempt: a dead forge fails fast instead of
+            // stalling the poll loop.
+            .connect_timeout(Duration::from_secs(10))
+            // Every forge call is a bounded request/response, unlike a
+            // model stream, so one overall timeout per request is the
+            // right guard.
+            .timeout(Duration::from_secs(20))
+            .build()
+            .expect("a client builder with only timeouts cannot fail");
+        Self {
+            base_url: base_url.into().trim_end_matches('/').to_owned(),
+            token: token.into(),
+            http,
+        }
+    }
+
+    /// Overrides the base URL (trailing slashes trimmed) — for tests
+    /// against a mock server.
+    pub fn with_base_url(mut self, base_url: impl Into<String>) -> Self {
+        self.base_url = base_url.into().trim_end_matches('/').to_owned();
+        self
+    }
+
+    /// A GET builder for `path` with auth applied: the token rides only in
+    /// the `Authorization` header, never in a URL.
+    fn get(&self, path: &str) -> reqwest::RequestBuilder {
+        let base = &self.base_url;
+        let request = self.http.get(format!("{base}/api/v1{path}"));
+        self.auth(request)
+    }
+
+    /// A POST builder for `path` with auth applied.
+    fn post(&self, path: &str) -> reqwest::RequestBuilder {
+        let base = &self.base_url;
+        let request = self.http.post(format!("{base}/api/v1{path}"));
+        self.auth(request)
+    }
+
+    /// Gitea's token scheme: `Authorization: token <token>`.
+    fn auth(&self, request: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
+        if self.token.is_empty() {
+            return request;
+        }
+        request.header("Authorization", format!("token {}", self.token))
+    }
+
+    /// Follows `limit`/`page` pagination until a short page.
+    async fn paged<T: DeserializeOwned>(
+        &self,
+        path: &str,
+        query: &[(String, String)],
+    ) -> Result<Vec<T>, ForgeError> {
+        let mut rows = Vec::new();
+        for page in 1..=MAX_PAGES {
+            let mut query = query.to_vec();
+            query.push(("limit".to_owned(), PER_PAGE.to_string()));
+            query.push(("page".to_owned(), page.to_string()));
+            let response = self
+                .get(path)
+                .query(&query)
+                .send()
+                .await
+                .map_err(ForgeError::network)?;
+            let batch: Vec<T> = decode(response).await?;
+            let full = batch.len() == PER_PAGE;
+            rows.extend(batch);
+            if !full {
+                return Ok(rows);
+            }
+        }
+        Err(ForgeError::Pagination(format!(
+            "more than {MAX_PAGES} full pages from {path}"
+        )))
+    }
+}
+
+/// Decodes a response: non-2xx to [`ForgeError::Api`] with the body
+/// snippet, undecodable bodies to [`ForgeError::Parse`].
+async fn decode<T: DeserializeOwned>(response: reqwest::Response) -> Result<T, ForgeError> {
+    let response = ensure_success(response).await?;
+    response
+        .json()
+        .await
+        .map_err(|err| ForgeError::Parse(format!("malformed forge response: {err}")))
+}
+
+// --- Wire rows (private): the subset each event lifts, plus the
+// --- untouched row under `raw`.
+
+/// A Gitea user object.
+#[derive(Debug, Deserialize)]
+struct WireUser {
+    login: Option<String>,
+}
+
+/// A label object.
+#[derive(Debug, Deserialize)]
+struct WireLabel {
+    name: Option<String>,
+}
+
+/// The `pull_request` meta on an issue row: its presence is the only PR
+/// marker there; the fields it carries are not used.
+#[derive(Debug, Deserialize)]
+struct WirePullMarker {}
+
+/// One row of the issues list.
+#[derive(Debug, Deserialize)]
+struct WireIssue {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    number: Option<u64>,
+    #[serde(default)]
+    state: Option<String>,
+    #[serde(default)]
+    user: Option<WireUser>,
+    #[serde(default)]
+    labels: Vec<WireLabel>,
+    #[serde(default)]
+    title: Option<String>,
+    #[serde(default)]
+    body: Option<String>,
+    #[serde(default)]
+    created_at: Option<Timestamp>,
+    #[serde(default)]
+    updated_at: Option<Timestamp>,
+    #[serde(default)]
+    closed_at: Option<Timestamp>,
+    #[serde(default)]
+    html_url: Option<String>,
+    #[serde(default)]
+    pull_request: Option<WirePullMarker>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// One row of the pulls list; carries the `merged` flag directly.
+#[derive(Debug, Deserialize)]
+struct WirePull {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    number: Option<u64>,
+    #[serde(default)]
+    state: Option<String>,
+    #[serde(default)]
+    merged: bool,
+    #[serde(default)]
+    user: Option<WireUser>,
+    #[serde(default)]
+    labels: Vec<WireLabel>,
+    #[serde(default)]
+    title: Option<String>,
+    #[serde(default)]
+    body: Option<String>,
+    #[serde(default)]
+    created_at: Option<Timestamp>,
+    #[serde(default)]
+    updated_at: Option<Timestamp>,
+    #[serde(default)]
+    closed_at: Option<Timestamp>,
+    #[serde(default)]
+    html_url: Option<String>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// One row of a PR's reviews.
+#[derive(Debug, Deserialize)]
+struct WireReview {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    user: Option<WireUser>,
+    #[serde(default)]
+    body: Option<String>,
+    #[serde(default)]
+    state: Option<String>,
+    #[serde(default)]
+    submitted: Option<Timestamp>,
+    #[serde(default)]
+    created: Option<Timestamp>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// One row of the repo-wide comments feed.
+#[derive(Debug, Deserialize)]
+struct WireComment {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    user: Option<WireUser>,
+    #[serde(default)]
+    body: Option<String>,
+    #[serde(default)]
+    created_at: Option<Timestamp>,
+    #[serde(default)]
+    updated_at: Option<Timestamp>,
+    #[serde(default)]
+    html_url: Option<String>,
+    /// HTML URL of the parent thread; its trailing segment is the thread
+    /// number.
+    #[serde(default)]
+    issue_url: Option<String>,
+    /// Set (non-empty) exactly on PR conversation comments.
+    #[serde(default)]
+    pull_request_url: Option<String>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// One row of the account notifications feed.
+#[derive(Debug, Clone, Deserialize)]
+struct WireNotification {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    subject: Option<WireNotificationSubject>,
+    #[serde(default)]
+    repository: Option<WireNotificationRepo>,
+    #[serde(default)]
+    updated_at: Option<Timestamp>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// The subject a notification is about.
+#[derive(Debug, Clone, Deserialize)]
+struct WireNotificationSubject {
+    #[serde(default)]
+    title: Option<String>,
+    /// API URL of the thread; its trailing segment is the thread number.
+    #[serde(default)]
+    url: Option<String>,
+    /// `Pull` or `Issue`.
+    #[serde(rename = "type", default)]
+    kind: Option<String>,
+}
+
+/// The repository a notification's thread lives in.
+#[derive(Debug, Clone, Deserialize)]
+struct WireNotificationRepo {
+    #[serde(default)]
+    full_name: Option<String>,
+}
+
+// --- Mapping: wire rows -> normalized events. Pure, so the async
+// --- transport stays thin and the derivations are unit-testable.
+
+/// The thread event an issues-list row carries, if any.
+fn issue_event(row: &WireIssue, repo: &RepoRef) -> Result<Option<ForgeEvent>, ForgeError> {
+    let id = required(row.id, "issue id")?;
+    let number = required(row.number, "issue number")?;
+    let updated = required(row.updated_at.clone(), "issue updated_at")?;
+    let created = required(row.created_at.clone(), "issue created_at")?;
+    let kind = if row.state.as_deref() == Some("closed") {
+        EventKind::Closed
+    } else if row.closed_at.is_some() {
+        EventKind::Reopened
+    } else if created == updated {
+        // Never touched since creation: opening is the only update.
+        EventKind::Opened
+    } else {
+        return Ok(None);
+    };
+    Ok(Some(ForgeEvent {
+        id: format!("gitea:issue:{id}@{}", updated.raw()),
+        kind,
+        repo: repo.clone(),
+        number,
+        is_pr: false,
+        actor: row.user.as_ref().and_then(|user| user.login.clone()),
+        body: None,
+        url: row.html_url.clone(),
+        timestamp: updated,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+/// The thread event a pulls row carries, if any. `merged` wins over the
+/// closed state: a merge closes the PR.
+fn pull_event(row: &WirePull, repo: &RepoRef) -> Result<Option<ForgeEvent>, ForgeError> {
+    let id = required(row.id, "pull id")?;
+    let number = required(row.number, "pull number")?;
+    let updated = required(row.updated_at.clone(), "pull updated_at")?;
+    let created = required(row.created_at.clone(), "pull created_at")?;
+    let kind = if row.merged {
+        EventKind::Merged
+    } else if row.state.as_deref() == Some("closed") {
+        EventKind::Closed
+    } else if row.closed_at.is_some() {
+        EventKind::Reopened
+    } else if created == updated {
+        EventKind::Opened
+    } else {
+        return Ok(None);
+    };
+    Ok(Some(ForgeEvent {
+        id: format!("gitea:pull:{id}@{}", updated.raw()),
+        kind,
+        repo: repo.clone(),
+        number,
+        is_pr: true,
+        actor: row.user.as_ref().and_then(|user| user.login.clone()),
+        body: None,
+        url: row.html_url.clone(),
+        timestamp: updated,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+/// The review event a reviews row carries, if any: never-submitted
+/// (`PENDING`) rows and rows older than the cursor are skipped. The
+/// timestamp falls back from `submitted` to `created`, which older
+/// servers set alone.
+fn review_event(
+    row: &WireReview,
+    repo: &RepoRef,
+    number: u64,
+    since: Option<&Timestamp>,
+) -> Result<Option<ForgeEvent>, ForgeError> {
+    if row.state.as_deref() == Some("PENDING") {
+        return Ok(None);
+    }
+    let timestamp = match row.submitted.clone().or_else(|| row.created.clone()) {
+        Some(timestamp) => timestamp,
+        None => return Ok(None),
+    };
+    if let Some(since) = since {
+        if timestamp < *since {
+            return Ok(None);
+        }
+    }
+    let id = required(row.id, "review id")?;
+    Ok(Some(ForgeEvent {
+        id: format!("gitea:review:{id}"),
+        kind: EventKind::Review,
+        repo: repo.clone(),
+        number,
+        is_pr: true,
+        actor: row.user.as_ref().and_then(|user| user.login.clone()),
+        body: row.body.clone(),
+        // Gitea review rows carry no URL of their own.
+        url: None,
+        timestamp,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+/// The comment event a repo-wide comments row carries.
+fn comment_event(row: &WireComment, repo: &RepoRef) -> Result<Option<ForgeEvent>, ForgeError> {
+    let id = required(row.id, "comment id")?;
+    let issue_url = required(row.issue_url.clone(), "comment issue_url")?;
+    let number = parent_number(&issue_url).ok_or_else(|| {
+        ForgeError::Parse(format!(
+            "comment issue_url without a thread number: {issue_url}"
+        ))
+    })?;
+    let timestamp = required(
+        row.updated_at.clone().or_else(|| row.created_at.clone()),
+        "comment timestamp",
+    )?;
+    let is_pr = row
+        .pull_request_url
+        .as_deref()
+        .is_some_and(|url| !url.is_empty());
+    Ok(Some(ForgeEvent {
+        id: format!("gitea:comment:{id}"),
+        kind: EventKind::Comment,
+        repo: repo.clone(),
+        number,
+        is_pr,
+        actor: row.user.as_ref().and_then(|user| user.login.clone()),
+        body: row.body.clone(),
+        url: row.html_url.clone(),
+        timestamp,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+/// The parent thread number: a forge thread URL's trailing segment.
+fn parent_number(url: &str) -> Option<u64> {
+    url.trim_end_matches('/').rsplit('/').next()?.parse().ok()
+}
+
+/// The notification a notifications row carries. Gitea's feed names no
+/// per-row reason — the thread subject is all a row carries — so every
+/// notification maps to [`NotificationReason::Other`] with a fixed
+/// spelling, and the consumer's triage treats it as its default role.
+fn notification_event(row: &WireNotification) -> Result<Option<Notification>, ForgeError> {
+    let id = required(row.id, "notification id")?;
+    let updated = required(row.updated_at.clone(), "notification updated_at")?;
+    let subject = required(row.subject.clone(), "notification subject")?;
+    let repo_row = required(row.repository.clone(), "notification repository")?;
+    let full_name = required(repo_row.full_name, "notification repository full_name")?;
+    let (owner, name) = full_name.split_once('/').ok_or_else(|| {
+        ForgeError::Parse(format!(
+            "notification repository full_name without a slash: {full_name}"
+        ))
+    })?;
+    let subject_url = subject.url.clone().unwrap_or_default();
+    let number = parent_number(&subject_url).ok_or_else(|| {
+        ForgeError::Parse(format!(
+            "notification subject url without a thread number: {subject_url}"
+        ))
+    })?;
+    Ok(Some(Notification {
+        id: format!("gitea:notification:{id}@{}", updated.raw()),
+        reason: NotificationReason::Other("gitea".to_owned()),
+        repo: RepoRef::new(owner, name),
+        number,
+        is_pr: subject.kind.as_deref() == Some("Pull"),
+        actor: None,
+        title: subject.title.clone(),
+        url: None,
+        timestamp: updated,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+/// The snapshot an issues-list row yields.
+fn snapshot_from_issue(row: &WireIssue) -> Result<Snapshot, ForgeError> {
+    Ok(Snapshot {
+        title: required(row.title.clone(), "issue title")?,
+        body: row.body.clone(),
+        state: if row.state.as_deref() == Some("closed") {
+            ThreadState::Closed
+        } else {
+            ThreadState::Open
+        },
+        is_pr: false,
+        author: row.user.as_ref().and_then(|user| user.login.clone()),
+        labels: row
+            .labels
+            .iter()
+            .filter_map(|label| label.name.clone())
+            .collect(),
+        url: row.html_url.clone(),
+        raw: Value::Object(row.raw.clone()),
+    })
+}
+
+/// The snapshot a pulls row yields; the `merged` flag is authoritative.
+fn snapshot_from_pull(row: &WirePull) -> Result<Snapshot, ForgeError> {
+    Ok(Snapshot {
+        title: required(row.title.clone(), "pull title")?,
+        body: row.body.clone(),
+        state: if row.merged {
+            ThreadState::Merged
+        } else if row.state.as_deref() == Some("closed") {
+            ThreadState::Closed
+        } else {
+            ThreadState::Open
+        },
+        is_pr: true,
+        author: row.user.as_ref().and_then(|user| user.login.clone()),
+        labels: row
+            .labels
+            .iter()
+            .filter_map(|label| label.name.clone())
+            .collect(),
+        url: row.html_url.clone(),
+        raw: Value::Object(row.raw.clone()),
+    })
+}
+
+#[async_trait]
+impl Forge for GiteaForge {
+    async fn poll(
+        &self,
+        repo: &RepoRef,
+        since: Option<&Timestamp>,
+    ) -> Result<ForgePage, ForgeError> {
+        let mut events = Vec::new();
+
+        // Issue feed: server-side `since`, issues only.
+        let issues_path = format!("/repos/{}/{}/issues", repo.owner, repo.name);
+        let mut query = vec![
+            ("state".to_owned(), "all".to_owned()),
+            ("type".to_owned(), "issues".to_owned()),
+        ];
+        if let Some(since) = since {
+            query.push(("since".to_owned(), since.raw().to_owned()));
+        }
+        for row in self.paged::<WireIssue>(&issues_path, &query).await? {
+            if let Some(event) = issue_event(&row, repo)? {
+                events.push(event);
+            }
+        }
+
+        // Pull feed: no server-side time filter exists, so the poll pages
+        // the whole list and cuts off client-side.
+        let pulls_path = format!("/repos/{}/{}/pulls", repo.owner, repo.name);
+        let query = vec![
+            ("state".to_owned(), "all".to_owned()),
+            ("sort".to_owned(), "leastrecent".to_owned()),
+        ];
+        for row in self.paged::<WirePull>(&pulls_path, &query).await? {
+            let updated = required(row.updated_at.clone(), "pull updated_at")?;
+            if let Some(since) = since {
+                if updated < *since {
+                    continue;
+                }
+            }
+            let number = required(row.number, "pull number")?;
+            if let Some(event) = pull_event(&row, repo)? {
+                events.push(event);
+            }
+            // Reviews fan out per updated PR: no repo-wide feed.
+            let reviews_path =
+                format!("/repos/{}/{}/pulls/{number}/reviews", repo.owner, repo.name);
+            for review in self.paged::<WireReview>(&reviews_path, &[]).await? {
+                if let Some(event) = review_event(&review, repo, number, since)? {
+                    events.push(event);
+                }
+            }
+        }
+
+        // Comment feed, repo-wide: covers issue and PR conversation
+        // comments in one shape.
+        let comments_path = format!("/repos/{}/{}/issues/comments", repo.owner, repo.name);
+        let mut query = Vec::new();
+        if let Some(since) = since {
+            query.push(("since".to_owned(), since.raw().to_owned()));
+        }
+        for row in self.paged::<WireComment>(&comments_path, &query).await? {
+            if let Some(event) = comment_event(&row, repo)? {
+                events.push(event);
+            }
+        }
+
+        Ok(ForgePage::finish(events, since))
+    }
+
+    async fn comment(&self, repo: &RepoRef, number: u64, body: &str) -> Result<(), ForgeError> {
+        // The issues endpoint covers PRs too: every PR is an issue here.
+        let path = format!(
+            "/repos/{}/{}/issues/{number}/comments",
+            repo.owner, repo.name
+        );
+        let response = self
+            .post(&path)
+            .json(&serde_json::json!({ "body": body }))
+            .send()
+            .await
+            .map_err(ForgeError::network)?;
+        ensure_success(response).await?;
+        Ok(())
+    }
+
+    async fn snapshot(&self, repo: &RepoRef, number: u64) -> Result<Snapshot, ForgeError> {
+        let issue_path = format!("/repos/{}/{}/issues/{number}", repo.owner, repo.name);
+        let response = self
+            .get(&issue_path)
+            .send()
+            .await
+            .map_err(ForgeError::network)?;
+        let issue: WireIssue = decode(response).await?;
+        if issue.pull_request.is_some() {
+            // The issue view's PR meta is a marker; the pulls row carries
+            // the authoritative merged flag.
+            let pull_path = format!("/repos/{}/{}/pulls/{number}", repo.owner, repo.name);
+            let response = self
+                .get(&pull_path)
+                .send()
+                .await
+                .map_err(ForgeError::network)?;
+            let pull: WirePull = decode(response).await?;
+            Ok(snapshot_from_pull(&pull)?)
+        } else {
+            Ok(snapshot_from_issue(&issue)?)
+        }
+    }
+
+    async fn notifications(
+        &self,
+        since: Option<&Timestamp>,
+    ) -> Result<NotificationPage, ForgeError> {
+        // The unread-notifications feed: everything addressed to the
+        // account, every repository it can see. `since` filters on the
+        // row's updated time; boundary rows repeat, ids dedup them.
+        let mut query = vec![("status-types".to_owned(), "unread".to_owned())];
+        if let Some(since) = since {
+            query.push(("since".to_owned(), since.raw().to_owned()));
+        }
+        let rows = self
+            .paged::<WireNotification>("/notifications", &query)
+            .await?;
+        let mut notifications = Vec::new();
+        for row in &rows {
+            if let Some(notification) = notification_event(row)? {
+                notifications.push(notification);
+            }
+        }
+        Ok(NotificationPage::finish(notifications, since))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn repo() -> RepoRef {
+        RepoRef::new("acme", "widgets")
+    }
+
+    fn row(json: &str) -> WireIssue {
+        serde_json::from_str(json).expect("issue fixture parses")
+    }
+
+    #[test]
+    fn a_closed_issue_row_maps_to_closed() {
+        let event = issue_event(
+            &row(r#"{"id": 107, "number": 7, "state": "closed",
+                    "user": {"login": "alice"},
+                    "created_at": "2026-10-01T09:00:00Z",
+                    "updated_at": "2026-10-02T10:00:00Z",
+                    "closed_at": "2026-10-02T10:00:00Z",
+                    "html_url": "https://forge.example/acme/widgets/issues/7"}"#),
+            &repo(),
+        )
+        .expect("maps")
+        .expect("closed row carries an event");
+        assert_eq!(event.kind, EventKind::Closed);
+        assert_eq!(event.id, "gitea:issue:107@2026-10-02T10:00:00Z");
+        assert!(!event.is_pr);
+    }
+
+    #[test]
+    fn reopen_is_derived_from_a_close_in_the_past() {
+        let event = issue_event(
+            &row(r#"{"id": 104, "number": 4, "state": "open",
+                    "created_at": "2026-10-01T09:00:00Z",
+                    "updated_at": "2026-10-07T09:00:00Z",
+                    "closed_at": "2026-10-03T00:00:00Z"}"#),
+            &repo(),
+        )
+        .expect("maps")
+        .expect("reopened row carries an event");
+        assert_eq!(event.kind, EventKind::Reopened);
+    }
+
+    #[test]
+    fn an_updated_open_row_without_close_history_emits_nothing() {
+        let event = issue_event(
+            &row(r#"{"id": 106, "number": 6, "state": "open",
+                    "created_at": "2026-10-01T09:00:00Z",
+                    "updated_at": "2026-10-05T09:00:00Z",
+                    "closed_at": null}"#),
+            &repo(),
+        )
+        .expect("maps");
+        assert_eq!(event, None);
+    }
+
+    fn pull(json: &str) -> WirePull {
+        serde_json::from_str(json).expect("pull fixture parses")
+    }
+
+    #[test]
+    fn merged_beats_closed_on_a_pull_row() {
+        let merged = pull_event(
+            &pull(
+                r#"{"id": 203, "number": 3, "state": "closed", "merged": true,
+                    "created_at": "2026-10-01T08:00:00Z",
+                    "updated_at": "2026-10-06T12:00:00Z",
+                    "closed_at": "2026-10-06T12:00:00Z"}"#,
+            ),
+            &repo(),
+        )
+        .expect("maps")
+        .expect("merged pull carries an event");
+        assert_eq!(merged.kind, EventKind::Merged);
+        assert_eq!(merged.id, "gitea:pull:203@2026-10-06T12:00:00Z");
+
+        let closed = pull_event(
+            &pull(
+                r#"{"id": 204, "number": 4, "state": "closed", "merged": false,
+                    "created_at": "2026-10-01T08:00:00Z",
+                    "updated_at": "2026-10-06T12:00:00Z",
+                    "closed_at": "2026-10-06T12:00:00Z"}"#,
+            ),
+            &repo(),
+        )
+        .expect("maps")
+        .expect("closed pull carries an event");
+        assert_eq!(closed.kind, EventKind::Closed);
+    }
+
+    #[test]
+    fn review_timestamp_falls_back_to_created() {
+        let review: WireReview = serde_json::from_str(
+            r#"{"id": 31, "state": "APPROVED", "submitted": "2026-10-06T13:00:00Z",
+                "created": "2026-10-06T13:00:00Z"}"#,
+        )
+        .expect("review fixture parses");
+        let event = review_event(&review, &repo(), 3, None)
+            .expect("maps")
+            .expect("review carries an event");
+        assert_eq!(
+            event.timestamp,
+            Timestamp::parse("2026-10-06T13:00:00Z").expect("parses")
+        );
+
+        // Older servers set `created` alone.
+        let bare: WireReview = serde_json::from_str(
+            r#"{"id": 33, "state": "COMMENT", "created": "2026-10-06T14:00:00Z"}"#,
+        )
+        .expect("review fixture parses");
+        let event = review_event(&bare, &repo(), 3, None)
+            .expect("maps")
+            .expect("review carries an event");
+        assert_eq!(
+            event.timestamp,
+            Timestamp::parse("2026-10-06T14:00:00Z").expect("parses")
+        );
+
+        // Neither timestamp: never treatable as submitted.
+        let timeless: WireReview = serde_json::from_str(r#"{"id": 34, "state": "COMMENT"}"#)
+            .expect("review fixture parses");
+        assert_eq!(
+            review_event(&timeless, &repo(), 3, None).expect("maps"),
+            None
+        );
+    }
+
+    #[test]
+    fn a_pr_comment_is_detected_from_its_urls() {
+        let comment: WireComment = serde_json::from_str(
+            r#"{"id": 3001, "user": {"login": "erin"}, "body": "reviewed",
+                "created_at": "2026-10-06T14:00:00Z",
+                "updated_at": "2026-10-06T14:00:00Z",
+                "issue_url": "https://forge.example/acme/widgets/issues/3",
+                "pull_request_url": "https://forge.example/acme/widgets/pulls/3"}"#,
+        )
+        .expect("comment fixture parses");
+        let event = comment_event(&comment, &repo())
+            .expect("maps")
+            .expect("comment carries an event");
+        assert_eq!(event.kind, EventKind::Comment);
+        assert!(event.is_pr);
+        assert_eq!(event.number, 3);
+
+        // An issue comment carries no pull_request_url.
+        let issue_comment: WireComment = serde_json::from_str(
+            r#"{"id": 7001, "created_at": "2026-10-02T11:00:00Z",
+                "updated_at": "2026-10-02T11:00:00Z",
+                "issue_url": "https://forge.example/acme/widgets/issues/7",
+                "pull_request_url": null}"#,
+        )
+        .expect("comment fixture parses");
+        let event = comment_event(&issue_comment, &repo())
+            .expect("maps")
+            .expect("comment carries an event");
+        assert!(!event.is_pr);
+    }
+}

integrations/rhema-forge-integration/src/github.rs

@@ -0,0 +1,846 @@
+//! The GitHub adapter: REST v3, at [`DEFAULT_BASE_URL`] or a GitHub
+//! Enterprise Server base. Auth is the token in `Authorization: Bearer` —
+//! never a query parameter, so the token never lands in URLs or logs.
+//!
+//! Endpoint choices and the constraints they pin:
+//! - Threads come from `GET /repos/{owner}/{repo}/issues` with
+//!   `state=all&sort=updated&direction=asc&since=<RFC 3339>`: the issues
+//!   list is the only thread feed with a server-side time filter, and PR
+//!   rows ride it behind the `pull_request` key.
+//! - The merged flag and reviews are invisible from the issues list: the
+//!   pulls list has no time filter and reviews have no repo-wide feed, so
+//!   every PR row the issues list returns costs one `GET /pulls/{n}`
+//!   (merged detection) and one `GET /pulls/{n}/reviews` (filtered
+//!   client-side by `submitted_at`).
+//! - Comments come from the repo-wide
+//!   `GET /repos/{owner}/{repo}/issues/comments` with
+//!   `sort=created&direction=asc&since=`; a PR conversation comment is the
+//!   same row shape, its `issue_url` ending in `/pulls/`.
+//! - GitHub's `since` matches rows updated at-or-after the instant, so a
+//!   re-poll on the returned cursor repeats boundary rows; stable ids make
+//!   the repetition dedupable.
+//! - An updated-but-still-open thread with no close history yields no
+//!   thread event: the usual cause is a comment, which the comments feed
+//!   reports with its own id. Title/label edits alone surface as nothing.
+//! - A `PENDING` review was never submitted and is skipped; `DISMISSED` is
+//!   activity and normalizes as [`EventKind::Review`].
+//! - A comment edit changes the row's `updated_at` but not its id: the
+//!   edit reappears at the boundary and dedups away. Edits are not
+//!   re-dispatched.
+//! - Pagination is `per_page`/`page` until a short page, with a hard page
+//!   cap. Page offsets over an updated-ascending sort can shift while a
+//!   poll is running; duplicates are absorbed by ids, misses only happen
+//!   to rows that update twice inside one poll window.
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+use serde::Deserialize;
+use serde::de::DeserializeOwned;
+use serde_json::Value;
+
+use crate::error::{ForgeError, ensure_success, required};
+use crate::event::{
+    EventKind, ForgeEvent, Notification, NotificationPage, NotificationReason, RepoRef, Snapshot,
+    ThreadState, Timestamp,
+};
+use crate::forge::{Forge, ForgePage};
+
+/// The production GitHub API base.
+pub const DEFAULT_BASE_URL: &str = "https://api.github.com";
+
+/// GitHub rejects requests without a `User-Agent` outright.
+const USER_AGENT: &str = concat!("rhema-forge-integration/", env!("CARGO_PKG_VERSION"));
+
+/// Rows requested per page.
+const PER_PAGE: usize = 100;
+
+/// Pages fetched before a server that keeps returning full pages is
+/// declared misbehaving rather than polled forever.
+const MAX_PAGES: usize = 1000;
+
+/// GitHub backend: a base URL, a token, one reqwest client.
+///
+/// An empty token is accepted but useless for writes; public reads work
+/// anonymously.
+pub struct GitHubForge {
+    base_url: String,
+    token: String,
+    http: reqwest::Client,
+}
+
+impl GitHubForge {
+    /// A forge authenticating with `token` against [`DEFAULT_BASE_URL`].
+    pub fn new(token: impl Into<String>) -> Self {
+        let http = reqwest::Client::builder()
+            // One connect attempt: a dead forge fails fast instead of
+            // stalling the poll loop.
+            .connect_timeout(Duration::from_secs(10))
+            // Every forge call is a bounded request/response, unlike a
+            // model stream, so one overall timeout per request is the
+            // right guard.
+            .timeout(Duration::from_secs(20))
+            .user_agent(USER_AGENT)
+            .build()
+            .expect("a client builder with only timeouts cannot fail");
+        Self {
+            base_url: DEFAULT_BASE_URL.to_owned(),
+            token: token.into(),
+            http,
+        }
+    }
+
+    /// Overrides the base URL (trailing slashes trimmed) — for GitHub
+    /// Enterprise Server and for tests against a mock server.
+    pub fn with_base_url(mut self, base_url: impl Into<String>) -> Self {
+        self.base_url = base_url.into().trim_end_matches('/').to_owned();
+        self
+    }
+
+    /// A GET builder for `path` with auth applied: the token rides only in
+    /// the `Authorization: Bearer` header, never in a URL.
+    fn get(&self, path: &str) -> reqwest::RequestBuilder {
+        let base = &self.base_url;
+        let request = self.http.get(format!("{base}{path}"));
+        if self.token.is_empty() {
+            request
+        } else {
+            request.bearer_auth(&self.token)
+        }
+    }
+
+    /// A POST builder for `path` with auth applied.
+    fn post(&self, path: &str) -> reqwest::RequestBuilder {
+        let base = &self.base_url;
+        let request = self.http.post(format!("{base}{path}"));
+        if self.token.is_empty() {
+            request
+        } else {
+            request.bearer_auth(&self.token)
+        }
+    }
+
+    /// Follows `per_page`/`page` pagination until a short page.
+    async fn paged<T: DeserializeOwned>(
+        &self,
+        path: &str,
+        query: &[(String, String)],
+    ) -> Result<Vec<T>, ForgeError> {
+        let mut rows = Vec::new();
+        for page in 1..=MAX_PAGES {
+            let mut query = query.to_vec();
+            query.push(("per_page".to_owned(), PER_PAGE.to_string()));
+            query.push(("page".to_owned(), page.to_string()));
+            let response = self
+                .get(path)
+                .query(&query)
+                .send()
+                .await
+                .map_err(ForgeError::network)?;
+            let batch: Vec<T> = decode(response).await?;
+            let full = batch.len() == PER_PAGE;
+            rows.extend(batch);
+            if !full {
+                return Ok(rows);
+            }
+        }
+        Err(ForgeError::Pagination(format!(
+            "more than {MAX_PAGES} full pages from {path}"
+        )))
+    }
+}
+
+/// Decodes a response: non-2xx to [`ForgeError::Api`] with the body
+/// snippet, undecodable bodies to [`ForgeError::Parse`].
+async fn decode<T: DeserializeOwned>(response: reqwest::Response) -> Result<T, ForgeError> {
+    let response = ensure_success(response).await?;
+    response
+        .json()
+        .await
+        .map_err(|err| ForgeError::Parse(format!("malformed forge response: {err}")))
+}
+
+// --- Wire rows (private): the subset each event lifts, plus the
+// --- untouched row under `raw`.
+
+/// A GitHub user object.
+#[derive(Debug, Deserialize)]
+struct WireUser {
+    login: Option<String>,
+}
+
+/// A label object.
+#[derive(Debug, Deserialize)]
+struct WireLabel {
+    name: Option<String>,
+}
+
+/// The `pull_request` key: its presence is the only PR marker on issue
+/// rows; the fields it carries are not used.
+#[derive(Debug, Deserialize)]
+struct WirePullMarker {}
+
+/// One row of the issues list.
+#[derive(Debug, Deserialize)]
+struct WireIssue {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    number: Option<u64>,
+    #[serde(default)]
+    state: Option<String>,
+    #[serde(default)]
+    user: Option<WireUser>,
+    #[serde(default)]
+    labels: Vec<WireLabel>,
+    #[serde(default)]
+    title: Option<String>,
+    #[serde(default)]
+    body: Option<String>,
+    #[serde(default)]
+    created_at: Option<Timestamp>,
+    #[serde(default)]
+    updated_at: Option<Timestamp>,
+    #[serde(default)]
+    closed_at: Option<Timestamp>,
+    #[serde(default)]
+    html_url: Option<String>,
+    #[serde(default)]
+    pull_request: Option<WirePullMarker>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// One `GET /pulls/{n}` row.
+#[derive(Debug, Deserialize)]
+struct WirePull {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    number: Option<u64>,
+    #[serde(default)]
+    state: Option<String>,
+    #[serde(default)]
+    user: Option<WireUser>,
+    #[serde(default)]
+    labels: Vec<WireLabel>,
+    #[serde(default)]
+    title: Option<String>,
+    #[serde(default)]
+    body: Option<String>,
+    #[serde(default)]
+    created_at: Option<Timestamp>,
+    #[serde(default)]
+    updated_at: Option<Timestamp>,
+    #[serde(default)]
+    closed_at: Option<Timestamp>,
+    #[serde(default)]
+    merged_at: Option<Timestamp>,
+    #[serde(default)]
+    html_url: Option<String>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// One row of a PR's reviews.
+#[derive(Debug, Deserialize)]
+struct WireReview {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    user: Option<WireUser>,
+    #[serde(default)]
+    body: Option<String>,
+    #[serde(default)]
+    state: Option<String>,
+    #[serde(default)]
+    submitted_at: Option<Timestamp>,
+    #[serde(default)]
+    html_url: Option<String>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// One row of the repo-wide comments feed.
+#[derive(Debug, Deserialize)]
+struct WireComment {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    user: Option<WireUser>,
+    #[serde(default)]
+    body: Option<String>,
+    #[serde(default)]
+    created_at: Option<Timestamp>,
+    #[serde(default)]
+    updated_at: Option<Timestamp>,
+    #[serde(default)]
+    html_url: Option<String>,
+    /// API URL of the parent thread; its trailing segment is the thread
+    /// number, and `/pulls/` in it marks a PR comment.
+    #[serde(default)]
+    issue_url: Option<String>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// One row of the account notifications feed.
+#[derive(Debug, Clone, Deserialize)]
+struct WireNotification {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    reason: Option<String>,
+    #[serde(default)]
+    subject: Option<WireNotificationSubject>,
+    #[serde(default)]
+    repository: Option<WireNotificationRepo>,
+    #[serde(default)]
+    updated_at: Option<Timestamp>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// The subject a notification is about.
+#[derive(Debug, Clone, Deserialize)]
+struct WireNotificationSubject {
+    #[serde(default)]
+    title: Option<String>,
+    /// API URL of the thread; its trailing segment is the thread number.
+    #[serde(default)]
+    url: Option<String>,
+    /// `PullRequest`, `Issue`, `Release`, ...
+    #[serde(rename = "type", default)]
+    kind: Option<String>,
+}
+
+/// The repository a notification's thread lives in.
+#[derive(Debug, Clone, Deserialize)]
+struct WireNotificationRepo {
+    #[serde(default)]
+    full_name: Option<String>,
+}
+
+// --- Mapping: wire rows -> normalized events. Pure, so the async
+// --- transport stays thin and the derivations are unit-testable.
+
+/// The thread event an issues-list row carries, if any. State transitions
+/// are derivable from the row; other updates are not.
+fn issue_event(row: &WireIssue, repo: &RepoRef) -> Result<Option<ForgeEvent>, ForgeError> {
+    let id = required(row.id, "issue id")?;
+    let number = required(row.number, "issue number")?;
+    let updated = required(row.updated_at.clone(), "issue updated_at")?;
+    let created = required(row.created_at.clone(), "issue created_at")?;
+    let kind = if row.state.as_deref() == Some("closed") {
+        EventKind::Closed
+    } else if row.closed_at.is_some() {
+        EventKind::Reopened
+    } else if created == updated {
+        // Never touched since creation: opening is the only update.
+        EventKind::Opened
+    } else {
+        return Ok(None);
+    };
+    Ok(Some(ForgeEvent {
+        id: format!("github:issue:{id}@{}", updated.raw()),
+        kind,
+        repo: repo.clone(),
+        number,
+        is_pr: false,
+        actor: row.user.as_ref().and_then(|user| user.login.clone()),
+        body: None,
+        url: row.html_url.clone(),
+        timestamp: updated,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+/// The thread event a pulls row carries, if any. The merged flag wins over
+/// the closed state: a merge closes the PR.
+fn pull_event(row: &WirePull, repo: &RepoRef) -> Result<Option<ForgeEvent>, ForgeError> {
+    let id = required(row.id, "pull id")?;
+    let number = required(row.number, "pull number")?;
+    let updated = required(row.updated_at.clone(), "pull updated_at")?;
+    let created = required(row.created_at.clone(), "pull created_at")?;
+    let kind = if row.merged_at.is_some() {
+        EventKind::Merged
+    } else if row.state.as_deref() == Some("closed") {
+        EventKind::Closed
+    } else if row.closed_at.is_some() {
+        EventKind::Reopened
+    } else if created == updated {
+        EventKind::Opened
+    } else {
+        return Ok(None);
+    };
+    Ok(Some(ForgeEvent {
+        id: format!("github:pull:{id}@{}", updated.raw()),
+        kind,
+        repo: repo.clone(),
+        number,
+        is_pr: true,
+        actor: row.user.as_ref().and_then(|user| user.login.clone()),
+        body: None,
+        url: row.html_url.clone(),
+        timestamp: updated,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+/// The review event a reviews row carries, if any: never-submitted
+/// (`PENDING`) rows and rows older than the cursor are skipped.
+fn review_event(
+    row: &WireReview,
+    repo: &RepoRef,
+    number: u64,
+    since: Option<&Timestamp>,
+) -> Result<Option<ForgeEvent>, ForgeError> {
+    if row.state.as_deref() == Some("PENDING") {
+        return Ok(None);
+    }
+    // A review without a submission time was never submitted.
+    let timestamp = required(row.submitted_at.clone(), "review submitted_at")?;
+    if let Some(since) = since {
+        if timestamp < *since {
+            return Ok(None);
+        }
+    }
+    let id = required(row.id, "review id")?;
+    Ok(Some(ForgeEvent {
+        id: format!("github:review:{id}"),
+        kind: EventKind::Review,
+        repo: repo.clone(),
+        number,
+        is_pr: true,
+        actor: row.user.as_ref().and_then(|user| user.login.clone()),
+        body: row.body.clone(),
+        url: row.html_url.clone(),
+        timestamp,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+/// The comment event a repo-wide comments row carries.
+fn comment_event(row: &WireComment, repo: &RepoRef) -> Result<Option<ForgeEvent>, ForgeError> {
+    let id = required(row.id, "comment id")?;
+    let issue_url = required(row.issue_url.clone(), "comment issue_url")?;
+    let number = parent_number(&issue_url).ok_or_else(|| {
+        ForgeError::Parse(format!(
+            "comment issue_url without a thread number: {issue_url}"
+        ))
+    })?;
+    let timestamp = required(
+        row.updated_at.clone().or_else(|| row.created_at.clone()),
+        "comment timestamp",
+    )?;
+    Ok(Some(ForgeEvent {
+        id: format!("github:comment:{id}"),
+        kind: EventKind::Comment,
+        repo: repo.clone(),
+        number,
+        is_pr: issue_url.contains("/pulls/"),
+        actor: row.user.as_ref().and_then(|user| user.login.clone()),
+        body: row.body.clone(),
+        url: row.html_url.clone(),
+        timestamp,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+/// The parent thread number: a forge thread URL's trailing segment.
+fn parent_number(url: &str) -> Option<u64> {
+    url.trim_end_matches('/').rsplit('/').next()?.parse().ok()
+}
+
+/// The notification a notifications row carries. GitHub's `reason` field
+/// is the row's whole why; the subject's API URL yields the thread.
+fn notification_event(row: &WireNotification) -> Result<Option<Notification>, ForgeError> {
+    let id = required(row.id, "notification id")?;
+    let updated = required(row.updated_at.clone(), "notification updated_at")?;
+    let subject = required(row.subject.clone(), "notification subject")?;
+    let repo_row = required(row.repository.clone(), "notification repository")?;
+    let full_name = required(repo_row.full_name, "notification repository full_name")?;
+    let (owner, name) = full_name.split_once('/').ok_or_else(|| {
+        ForgeError::Parse(format!(
+            "notification repository full_name without a slash: {full_name}"
+        ))
+    })?;
+    let subject_url = subject.url.clone().unwrap_or_default();
+    let number = parent_number(&subject_url).ok_or_else(|| {
+        ForgeError::Parse(format!(
+            "notification subject url without a thread number: {subject_url}"
+        ))
+    })?;
+    Ok(Some(Notification {
+        id: format!("github:notification:{id}@{}", updated.raw()),
+        reason: github_reason(row.reason.as_deref()),
+        repo: RepoRef::new(owner, name),
+        number,
+        is_pr: subject.kind.as_deref() == Some("PullRequest"),
+        actor: None,
+        title: subject.title.clone(),
+        url: None,
+        timestamp: updated,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+/// GitHub's reason strings, reduced to the shared vocabulary. `comment`
+/// means "activity on a thread you participate in" — for the account
+/// this client authenticates as, that is overwhelmingly feedback on its
+/// own threads, so it maps to [`NotificationReason::ReviewFeedback`].
+fn github_reason(reason: Option<&str>) -> NotificationReason {
+    match reason {
+        Some("review_requested") => NotificationReason::ReviewRequested,
+        Some("assign") => NotificationReason::Assigned,
+        Some("mention") | Some("team_mention") => NotificationReason::Mention,
+        Some("ci_activity") => NotificationReason::CiActivity,
+        Some("author") => NotificationReason::Author,
+        Some("comment") => NotificationReason::ReviewFeedback,
+        Some(other) => NotificationReason::Other(other.to_owned()),
+        None => NotificationReason::Other("unreasoned".to_owned()),
+    }
+}
+
+/// The snapshot an issues-list row yields.
+fn snapshot_from_issue(row: &WireIssue) -> Result<Snapshot, ForgeError> {
+    Ok(Snapshot {
+        title: required(row.title.clone(), "issue title")?,
+        body: row.body.clone(),
+        state: if row.state.as_deref() == Some("closed") {
+            ThreadState::Closed
+        } else {
+            ThreadState::Open
+        },
+        is_pr: false,
+        author: row.user.as_ref().and_then(|user| user.login.clone()),
+        labels: row
+            .labels
+            .iter()
+            .filter_map(|label| label.name.clone())
+            .collect(),
+        url: row.html_url.clone(),
+        raw: Value::Object(row.raw.clone()),
+    })
+}
+
+/// The snapshot a pulls row yields; `merged_at` is the authoritative
+/// merged flag the issue view of a PR cannot prove.
+fn snapshot_from_pull(row: &WirePull) -> Result<Snapshot, ForgeError> {
+    Ok(Snapshot {
+        title: required(row.title.clone(), "pull title")?,
+        body: row.body.clone(),
+        state: if row.merged_at.is_some() {
+            ThreadState::Merged
+        } else if row.state.as_deref() == Some("closed") {
+            ThreadState::Closed
+        } else {
+            ThreadState::Open
+        },
+        is_pr: true,
+        author: row.user.as_ref().and_then(|user| user.login.clone()),
+        labels: row
+            .labels
+            .iter()
+            .filter_map(|label| label.name.clone())
+            .collect(),
+        url: row.html_url.clone(),
+        raw: Value::Object(row.raw.clone()),
+    })
+}
+
+#[async_trait]
+impl Forge for GitHubForge {
+    async fn poll(
+        &self,
+        repo: &RepoRef,
+        since: Option<&Timestamp>,
+    ) -> Result<ForgePage, ForgeError> {
+        let mut events = Vec::new();
+
+        // Thread feed: the issues list, the only feed with a server-side
+        // time filter. PR rows ride behind the `pull_request` key.
+        let threads_path = format!("/repos/{}/{}/issues", repo.owner, repo.name);
+        let mut query = vec![
+            ("state".to_owned(), "all".to_owned()),
+            ("sort".to_owned(), "updated".to_owned()),
+            ("direction".to_owned(), "asc".to_owned()),
+        ];
+        if let Some(since) = since {
+            query.push(("since".to_owned(), since.raw().to_owned()));
+        }
+        let mut pull_numbers = Vec::new();
+        for row in self.paged::<WireIssue>(&threads_path, &query).await? {
+            if row.pull_request.is_some() {
+                pull_numbers.push(required(row.number, "issue number")?);
+            } else if let Some(event) = issue_event(&row, repo)? {
+                events.push(event);
+            }
+        }
+
+        // Per updated PR: the merged flag and the reviews are invisible
+        // from the issues list.
+        for number in pull_numbers {
+            let detail_path = format!("/repos/{}/{}/pulls/{number}", repo.owner, repo.name);
+            let response = self
+                .get(&detail_path)
+                .send()
+                .await
+                .map_err(ForgeError::network)?;
+            let pull: WirePull = decode(response).await?;
+            if let Some(event) = pull_event(&pull, repo)? {
+                events.push(event);
+            }
+            let reviews = self
+                .paged::<WireReview>(&format!("{detail_path}/reviews"), &[])
+                .await?;
+            for review in &reviews {
+                if let Some(event) = review_event(review, repo, number, since)? {
+                    events.push(event);
+                }
+            }
+        }
+
+        // Comment feed, repo-wide: covers issue and PR conversation
+        // comments in one shape.
+        let comments_path = format!("/repos/{}/{}/issues/comments", repo.owner, repo.name);
+        let mut query = vec![
+            ("sort".to_owned(), "created".to_owned()),
+            ("direction".to_owned(), "asc".to_owned()),
+        ];
+        if let Some(since) = since {
+            query.push(("since".to_owned(), since.raw().to_owned()));
+        }
+        for row in self.paged::<WireComment>(&comments_path, &query).await? {
+            if let Some(event) = comment_event(&row, repo)? {
+                events.push(event);
+            }
+        }
+
+        Ok(ForgePage::finish(events, since))
+    }
+
+    async fn comment(&self, repo: &RepoRef, number: u64, body: &str) -> Result<(), ForgeError> {
+        // The issues endpoint covers PRs too: every PR is an issue here.
+        let path = format!(
+            "/repos/{}/{}/issues/{number}/comments",
+            repo.owner, repo.name
+        );
+        let response = self
+            .post(&path)
+            .json(&serde_json::json!({ "body": body }))
+            .send()
+            .await
+            .map_err(ForgeError::network)?;
+        ensure_success(response).await?;
+        Ok(())
+    }
+
+    async fn snapshot(&self, repo: &RepoRef, number: u64) -> Result<Snapshot, ForgeError> {
+        let issue_path = format!("/repos/{}/{}/issues/{number}", repo.owner, repo.name);
+        let response = self
+            .get(&issue_path)
+            .send()
+            .await
+            .map_err(ForgeError::network)?;
+        let issue: WireIssue = decode(response).await?;
+        if issue.pull_request.is_some() {
+            // The issue view of a PR lacks the authoritative merged flag.
+            let pull_path = format!("/repos/{}/{}/pulls/{number}", repo.owner, repo.name);
+            let response = self
+                .get(&pull_path)
+                .send()
+                .await
+                .map_err(ForgeError::network)?;
+            let pull: WirePull = decode(response).await?;
+            Ok(snapshot_from_pull(&pull)?)
+        } else {
+            Ok(snapshot_from_issue(&issue)?)
+        }
+    }
+
+    async fn notifications(
+        &self,
+        since: Option<&Timestamp>,
+    ) -> Result<NotificationPage, ForgeError> {
+        // The unread-notifications feed: everything addressed to the
+        // account, every repository it can see. `since` filters on the
+        // row's updated time; boundary rows repeat, ids dedup them.
+        let mut query = Vec::new();
+        if let Some(since) = since {
+            query.push(("since".to_owned(), since.raw().to_owned()));
+        }
+        let rows = self
+            .paged::<WireNotification>("/notifications", &query)
+            .await?;
+        let mut notifications = Vec::new();
+        for row in &rows {
+            if let Some(notification) = notification_event(row)? {
+                notifications.push(notification);
+            }
+        }
+        Ok(NotificationPage::finish(notifications, since))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn repo() -> RepoRef {
+        RepoRef::new("acme", "widgets")
+    }
+
+    fn row(json: &str) -> WireIssue {
+        serde_json::from_str(json).expect("issue fixture parses")
+    }
+
+    #[test]
+    fn a_closed_issue_row_maps_to_closed() {
+        let event = issue_event(
+            &row(r#"{"id": 1007, "number": 7, "state": "closed",
+                    "user": {"login": "alice"}, "labels": [],
+                    "created_at": "2026-10-01T09:00:00Z",
+                    "updated_at": "2026-10-02T10:00:00Z",
+                    "closed_at": "2026-10-02T10:00:00Z",
+                    "html_url": "https://github.com/acme/widgets/issues/7"}"#),
+            &repo(),
+        )
+        .expect("maps")
+        .expect("closed row carries an event");
+        assert_eq!(event.kind, EventKind::Closed);
+        assert_eq!(event.id, "github:issue:1007@2026-10-02T10:00:00Z");
+        assert_eq!(event.number, 7);
+        assert!(!event.is_pr);
+        assert_eq!(event.actor.as_deref(), Some("alice"));
+        assert_eq!(event.raw["title"], serde_json::Value::Null);
+    }
+
+    #[test]
+    fn reopen_is_derived_from_a_close_in_the_past() {
+        let event = issue_event(
+            &row(r#"{"id": 1004, "number": 4, "state": "open",
+                    "created_at": "2026-10-01T09:00:00Z",
+                    "updated_at": "2026-10-07T09:00:00Z",
+                    "closed_at": "2026-10-03T00:00:00Z"}"#),
+            &repo(),
+        )
+        .expect("maps")
+        .expect("reopened row carries an event");
+        assert_eq!(event.kind, EventKind::Reopened);
+    }
+
+    #[test]
+    fn an_updated_open_row_without_close_history_emits_nothing() {
+        let event = issue_event(
+            &row(r#"{"id": 1006, "number": 6, "state": "open",
+                    "created_at": "2026-10-01T09:00:00Z",
+                    "updated_at": "2026-10-05T09:00:00Z",
+                    "closed_at": null}"#),
+            &repo(),
+        )
+        .expect("maps");
+        assert_eq!(event, None);
+    }
+
+    #[test]
+    fn a_row_missing_a_required_field_is_a_parse_error() {
+        let err = issue_event(&row(r#"{"number": 7, "state": "closed"}"#), &repo())
+            .expect_err("missing id is loud");
+        assert!(matches!(err, ForgeError::Parse(_)), "got: {err:?}");
+    }
+
+    fn pull(json: &str) -> WirePull {
+        serde_json::from_str(json).expect("pull fixture parses")
+    }
+
+    #[test]
+    fn merged_beats_closed_on_a_pull_row() {
+        let merged = pull_event(
+            &pull(
+                r#"{"id": 9009, "number": 9, "state": "closed",
+                    "created_at": "2026-10-01T09:00:00Z",
+                    "updated_at": "2026-10-06T12:00:00Z",
+                    "closed_at": "2026-10-06T12:00:00Z",
+                    "merged_at": "2026-10-06T12:00:00Z"}"#,
+            ),
+            &repo(),
+        )
+        .expect("maps")
+        .expect("merged pull carries an event");
+        assert_eq!(merged.kind, EventKind::Merged);
+        assert!(merged.is_pr);
+        assert_eq!(merged.id, "github:pull:9009@2026-10-06T12:00:00Z");
+
+        let closed = pull_event(
+            &pull(
+                r#"{"id": 9010, "number": 10, "state": "closed",
+                    "created_at": "2026-10-01T09:00:00Z",
+                    "updated_at": "2026-10-06T12:00:00Z",
+                    "closed_at": "2026-10-06T12:00:00Z",
+                    "merged_at": null}"#,
+            ),
+            &repo(),
+        )
+        .expect("maps")
+        .expect("closed pull carries an event");
+        assert_eq!(closed.kind, EventKind::Closed);
+    }
+
+    #[test]
+    fn pending_reviews_without_submission_are_skipped() {
+        let review: WireReview =
+            serde_json::from_str(r#"{"id": 82, "state": "PENDING", "submitted_at": null}"#)
+                .expect("review fixture parses");
+        let event = review_event(&review, &repo(), 8, None).expect("maps");
+        assert_eq!(event, None);
+    }
+
+    #[test]
+    fn a_pr_comment_is_detected_from_its_issue_url() {
+        let comment: WireComment = serde_json::from_str(
+            r#"{"id": 8001, "user": {"login": "dave"}, "body": "looks",
+                "created_at": "2026-10-05T10:30:00Z",
+                "updated_at": "2026-10-05T10:30:00Z",
+                "issue_url": "https://api.github.com/repos/acme/widgets/pulls/8"}"#,
+        )
+        .expect("comment fixture parses");
+        let event = comment_event(&comment, &repo())
+            .expect("maps")
+            .expect("comment carries an event");
+        assert_eq!(event.kind, EventKind::Comment);
+        assert!(event.is_pr);
+        assert_eq!(event.number, 8);
+        assert_eq!(event.id, "github:comment:8001");
+    }
+
+    #[test]
+    fn a_comment_url_without_a_number_is_a_parse_error() {
+        let comment: WireComment = serde_json::from_str(
+            r#"{"id": 8002, "issue_url": "https://api.github.com/repos/acme/widgets/issues/"}"#,
+        )
+        .expect("comment fixture parses");
+        let err = comment_event(&comment, &repo()).expect_err("loud failure");
+        assert!(matches!(err, ForgeError::Parse(_)), "got: {err:?}");
+    }
+
+    #[test]
+    fn the_raw_payload_keeps_the_fields_the_event_does_not_lift() {
+        let event = issue_event(
+            &row(
+                r#"{"id": 1007, "number": 7, "state": "closed", "title": "crash",
+                    "created_at": "2026-10-01T09:00:00Z",
+                    "updated_at": "2026-10-02T10:00:00Z",
+                    "closed_at": "2026-10-02T10:00:00Z",
+                    "node_id": "MDU6SXNzdWUxMDA3"}"#,
+            ),
+            &repo(),
+        )
+        .expect("maps")
+        .expect("closed row carries an event");
+        // Undeclared keys survive; the lifted ones (id, number, state,
+        // title, ...) do not.
+        assert_eq!(event.raw["node_id"], "MDU6SXNzdWUxMDA3");
+        assert!(event.raw.get("title").is_none());
+    }
+}

integrations/rhema-forge-integration/src/gitlab.rs

@@ -0,0 +1,714 @@
+//! The GitLab adapter: gitlab.com or self-hosted, at
+//! [`DEFAULT_BASE_URL`] — every request path is prefixed with `/api/v4`,
+//! and the project is addressed by its URL-encoded full path. Auth is the
+//! token in the `PRIVATE-TOKEN` header.
+//!
+//! Endpoint choices and the constraints that differ:
+//! - Issues and merge requests come from
+//!   `GET /projects/{encoded path}/{issues,merge_requests}` with
+//!   `updated_after=<RFC 3339>&order_by=updated_at&sort=asc`: two separate
+//!   feeds, since GitLab has no combined thread feed.
+//! - Issue iids and merge-request iids are separate sequences: a number
+//!   can exist as both. Wherever a bare number must resolve, the issue
+//!   feed is tried first and the merge-request feed is the 404 fallback;
+//!   an iid that is neither surfaces the merge-request 404.
+//! - GitLab has no repo-wide notes feed and the per-thread notes endpoint
+//!   has no updated filter, so every updated thread gets one paginated
+//!   `GET .../{issues,merge_requests}/{iid}/notes?sort=asc&order_by=created_at`
+//!   call and the `created_at` cutoff is client-side. Note edits do not
+//!   re-fire (notes are keyed by creation time).
+//! - System notes (state changes, label flips, ...) are skipped: the
+//!   thread-level transitions they record already surface from the thread
+//!   rows.
+//! - `updated_after` matches rows at-or-after the instant, so a re-poll on
+//!   the returned cursor repeats boundary rows; stable ids make the
+//!   repetition dedupable.
+//! - Merge-request state `merged` maps to [`EventKind::Merged`], `closed`
+//!   (unmerged) to [`EventKind::Closed`]; issue state `closed` to
+//!   [`EventKind::Closed`].
+//! - Pagination is offset-based (`per_page`/`page`) with the next page in
+//!   the `X-Next-Page` response header. Offset pages over an
+//!   updated-ascending sort can shift while a poll runs; duplicates are
+//!   absorbed by ids, misses only happen to rows that update twice inside
+//!   one poll window.
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+use serde::Deserialize;
+use serde::de::DeserializeOwned;
+use serde_json::Value;
+
+use crate::error::{ForgeError, ensure_success, required};
+use crate::event::{
+    EventKind, ForgeEvent, Notification, NotificationPage, NotificationReason, RepoRef, Snapshot,
+    ThreadState, Timestamp,
+};
+use crate::forge::{Forge, ForgePage};
+
+/// The production GitLab base.
+pub const DEFAULT_BASE_URL: &str = "https://gitlab.com";
+
+/// Rows requested per page.
+const PER_PAGE: usize = 100;
+
+/// Pages fetched before a server that keeps offering pages is declared
+/// misbehaving rather than polled forever.
+const MAX_PAGES: usize = 1000;
+
+/// GitLab backend: a base URL, a token, one reqwest client.
+///
+/// An empty token is accepted but useless for writes; public reads work
+/// anonymously.
+pub struct GitLabForge {
+    base_url: String,
+    token: String,
+    http: reqwest::Client,
+}
+
+impl GitLabForge {
+    /// A forge authenticating with `token` against [`DEFAULT_BASE_URL`].
+    pub fn new(token: impl Into<String>) -> Self {
+        let http = reqwest::Client::builder()
+            // One connect attempt: a dead forge fails fast instead of
+            // stalling the poll loop.
+            .connect_timeout(Duration::from_secs(10))
+            // Every forge call is a bounded request/response, unlike a
+            // model stream, so one overall timeout per request is the
+            // right guard.
+            .timeout(Duration::from_secs(20))
+            .build()
+            .expect("a client builder with only timeouts cannot fail");
+        Self {
+            base_url: DEFAULT_BASE_URL.to_owned(),
+            token: token.into(),
+            http,
+        }
+    }
+
+    /// Overrides the base URL (trailing slashes trimmed) — for
+    /// self-hosted instances and for tests against a mock server.
+    pub fn with_base_url(mut self, base_url: impl Into<String>) -> Self {
+        self.base_url = base_url.into().trim_end_matches('/').to_owned();
+        self
+    }
+
+    /// A GET builder for `path` with auth applied: the token rides only in
+    /// the `PRIVATE-TOKEN` header, never in a URL.
+    fn get(&self, path: &str) -> reqwest::RequestBuilder {
+        let base = &self.base_url;
+        let request = self.http.get(format!("{base}/api/v4{path}"));
+        self.auth(request)
+    }
+
+    /// A POST builder for `path` with auth applied.
+    fn post(&self, path: &str) -> reqwest::RequestBuilder {
+        let base = &self.base_url;
+        let request = self.http.post(format!("{base}/api/v4{path}"));
+        self.auth(request)
+    }
+
+    /// GitLab's token scheme: the `PRIVATE-TOKEN` header.
+    fn auth(&self, request: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
+        if self.token.is_empty() {
+            return request;
+        }
+        request.header("PRIVATE-TOKEN", &self.token)
+    }
+
+    /// Follows offset pagination via the `X-Next-Page` response header:
+    /// absent or empty ends the feed.
+    async fn paged<T: DeserializeOwned>(
+        &self,
+        path: &str,
+        query: &[(String, String)],
+    ) -> Result<Vec<T>, ForgeError> {
+        let mut rows = Vec::new();
+        let mut page: u32 = 1;
+        loop {
+            let mut query = query.to_vec();
+            query.push(("per_page".to_owned(), PER_PAGE.to_string()));
+            query.push(("page".to_owned(), page.to_string()));
+            let response = self
+                .get(path)
+                .query(&query)
+                .send()
+                .await
+                .map_err(ForgeError::network)?;
+            // The next-page cursor is a header: read it before the body
+            // moves.
+            let next = response
+                .headers()
+                .get("x-next-page")
+                .and_then(|value| value.to_str().ok())
+                .unwrap_or_default()
+                .to_owned();
+            rows.extend(decode::<Vec<T>>(response).await?);
+            if next.is_empty() {
+                return Ok(rows);
+            }
+            page = next
+                .parse()
+                .map_err(|_| ForgeError::Pagination(format!("non-numeric x-next-page: {next}")))?;
+            if page > MAX_PAGES as u32 {
+                return Err(ForgeError::Pagination(format!(
+                    "more than {MAX_PAGES} pages from {path}"
+                )));
+            }
+        }
+    }
+}
+
+/// Decodes a response: non-2xx to [`ForgeError::Api`] with the body
+/// snippet, undecodable bodies to [`ForgeError::Parse`].
+async fn decode<T: DeserializeOwned>(response: reqwest::Response) -> Result<T, ForgeError> {
+    let response = ensure_success(response).await?;
+    response
+        .json()
+        .await
+        .map_err(|err| ForgeError::Parse(format!("malformed forge response: {err}")))
+}
+
+/// The project path (`owner/name`, possibly with subgroups) as GitLab's
+/// API needs it: the whole path percent-encoded, slashes included.
+fn encode_project_path(path: &str) -> String {
+    let mut encoded = String::with_capacity(path.len());
+    for byte in path.bytes() {
+        match byte {
+            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
+                encoded.push(byte as char)
+            }
+            other => encoded.push_str(&format!("%{other:02X}")),
+        }
+    }
+    encoded
+}
+
+// --- Wire rows (private): the subset each event lifts, plus the
+// --- untouched row under `raw`.
+
+/// A GitLab author object; the login field is spelled `username` here.
+#[derive(Debug, Clone, Deserialize)]
+struct WireAuthor {
+    username: Option<String>,
+}
+
+/// One issue or merge-request row: the same subset lifts from both.
+#[derive(Debug, Deserialize)]
+struct WireThread {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    iid: Option<u64>,
+    #[serde(default)]
+    state: Option<String>,
+    #[serde(default)]
+    author: Option<WireAuthor>,
+    #[serde(default)]
+    labels: Vec<String>,
+    #[serde(default)]
+    title: Option<String>,
+    /// The opening body: GitLab spells it `description`.
+    #[serde(default)]
+    description: Option<String>,
+    #[serde(default)]
+    created_at: Option<Timestamp>,
+    #[serde(default)]
+    updated_at: Option<Timestamp>,
+    #[serde(default)]
+    closed_at: Option<Timestamp>,
+    #[serde(default)]
+    web_url: Option<String>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// One note (comment) row of a thread's notes feed.
+#[derive(Debug, Deserialize)]
+struct WireNote {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    body: Option<String>,
+    #[serde(default)]
+    system: bool,
+    #[serde(default)]
+    author: Option<WireAuthor>,
+    #[serde(default)]
+    created_at: Option<Timestamp>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// One row of the account's todo list — GitLab's notification feed.
+#[derive(Debug, Clone, Deserialize)]
+struct WireTodo {
+    #[serde(default)]
+    id: Option<i64>,
+    #[serde(default)]
+    project: Option<WireTodoProject>,
+    #[serde(default)]
+    target_type: Option<String>,
+    #[serde(default)]
+    target: Option<WireTodoTarget>,
+    #[serde(default)]
+    action_name: Option<String>,
+    /// Human-facing URL of the target thread.
+    #[serde(default)]
+    target_url: Option<String>,
+    #[serde(default)]
+    updated_at: Option<Timestamp>,
+    #[serde(flatten)]
+    raw: serde_json::Map<String, Value>,
+}
+
+/// The project a todo's thread lives in.
+#[derive(Debug, Clone, Deserialize)]
+struct WireTodoProject {
+    #[serde(default)]
+    path_with_namespace: Option<String>,
+}
+
+/// The thread a todo points at.
+#[derive(Debug, Clone, Deserialize)]
+struct WireTodoTarget {
+    #[serde(default)]
+    iid: Option<u64>,
+    #[serde(default)]
+    title: Option<String>,
+    #[serde(default)]
+    author: Option<WireAuthor>,
+}
+
+/// GitLab's `action_name` strings, reduced to the shared vocabulary.
+fn gitlab_reason(action: Option<&str>) -> NotificationReason {
+    match action {
+        Some("review_requested") | Some("approval_required") => NotificationReason::ReviewRequested,
+        Some("assigned") => NotificationReason::Assigned,
+        Some("mentioned") | Some("directly_addressed") => NotificationReason::Mention,
+        Some("build_failed") | Some("build_success") => NotificationReason::CiActivity,
+        Some(other) => NotificationReason::Other(other.to_owned()),
+        None => NotificationReason::Other("unactioned".to_owned()),
+    }
+}
+
+/// The notification a todo row carries. The todo list has no server-side
+/// time filter, so the poll refetches every pending todo and the dedup
+/// ids absorb the repetition; the cursor is advisory only here.
+fn todo_notification(row: &WireTodo) -> Result<Option<Notification>, ForgeError> {
+    let id = required(row.id, "todo id")?;
+    let updated = required(row.updated_at.clone(), "todo updated_at")?;
+    let project = required(row.project.clone(), "todo project")?;
+    let path = required(project.path_with_namespace, "todo project path")?;
+    let (owner, name) = path
+        .rsplit_once('/')
+        .ok_or_else(|| ForgeError::Parse(format!("todo project path without a slash: {path}")))?;
+    let target = required(row.target.clone(), "todo target")?;
+    let iid = required(target.iid, "todo target iid")?;
+    let is_pr = row.target_type.as_deref() == Some("MergeRequest");
+    Ok(Some(Notification {
+        id: format!("gitlab:notification:{id}@{}", updated.raw()),
+        reason: gitlab_reason(row.action_name.as_deref()),
+        repo: RepoRef::new(owner, name),
+        number: iid,
+        is_pr,
+        actor: target
+            .author
+            .as_ref()
+            .and_then(|author| author.username.clone()),
+        title: target.title.clone(),
+        url: row.target_url.clone(),
+        timestamp: updated,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+// --- Mapping: wire rows -> normalized events. Pure, so the async
+// --- transport stays thin and the derivations are unit-testable.
+
+/// The thread event a thread row carries, if any. `merged` only exists on
+/// merge requests; a still-open row with a close in its past reopened.
+fn thread_event(
+    row: &WireThread,
+    repo: &RepoRef,
+    is_pr: bool,
+) -> Result<Option<ForgeEvent>, ForgeError> {
+    let id = required(row.id, "thread id")?;
+    let iid = required(row.iid, "thread iid")?;
+    let updated = required(row.updated_at.clone(), "thread updated_at")?;
+    let created = required(row.created_at.clone(), "thread created_at")?;
+    let state = row.state.as_deref().unwrap_or_default();
+    let kind = if is_pr && state == "merged" {
+        EventKind::Merged
+    } else if state == "closed" {
+        EventKind::Closed
+    } else if row.closed_at.is_some() {
+        EventKind::Reopened
+    } else if created == updated {
+        // Never touched since creation: opening is the only update.
+        EventKind::Opened
+    } else {
+        return Ok(None);
+    };
+    let item = if is_pr { "mr" } else { "issue" };
+    Ok(Some(ForgeEvent {
+        id: format!("gitlab:{item}:{id}@{}", updated.raw()),
+        kind,
+        repo: repo.clone(),
+        number: iid,
+        is_pr,
+        actor: row
+            .author
+            .as_ref()
+            .and_then(|author| author.username.clone()),
+        body: None,
+        url: row.web_url.clone(),
+        timestamp: updated,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+/// The comment event a note row carries, if any: system notes record
+/// transitions the thread rows already surface, and rows older than the
+/// cursor are skipped.
+fn note_event(
+    row: &WireNote,
+    repo: &RepoRef,
+    iid: u64,
+    is_pr: bool,
+    since: Option<&Timestamp>,
+) -> Result<Option<ForgeEvent>, ForgeError> {
+    if row.system {
+        return Ok(None);
+    }
+    let timestamp = required(row.created_at.clone(), "note created_at")?;
+    if let Some(since) = since {
+        if timestamp < *since {
+            return Ok(None);
+        }
+    }
+    let id = required(row.id, "note id")?;
+    Ok(Some(ForgeEvent {
+        id: format!("gitlab:note:{id}"),
+        kind: EventKind::Comment,
+        repo: repo.clone(),
+        number: iid,
+        is_pr,
+        actor: row
+            .author
+            .as_ref()
+            .and_then(|author| author.username.clone()),
+        body: row.body.clone(),
+        // Note rows carry no URL of their own; the thread's `web_url` is
+        // on the thread row.
+        url: None,
+        timestamp,
+        raw: Value::Object(row.raw.clone()),
+    }))
+}
+
+/// The snapshot a thread row yields.
+fn snapshot_from_row(row: &WireThread, is_pr: bool) -> Result<Snapshot, ForgeError> {
+    Ok(Snapshot {
+        title: required(row.title.clone(), "thread title")?,
+        body: row.description.clone(),
+        state: match row.state.as_deref() {
+            Some("merged") => ThreadState::Merged,
+            Some("closed") => ThreadState::Closed,
+            _ => ThreadState::Open,
+        },
+        is_pr,
+        author: row
+            .author
+            .as_ref()
+            .and_then(|author| author.username.clone()),
+        labels: row.labels.clone(),
+        url: row.web_url.clone(),
+        raw: Value::Object(row.raw.clone()),
+    })
+}
+
+#[async_trait]
+impl Forge for GitLabForge {
+    async fn poll(
+        &self,
+        repo: &RepoRef,
+        since: Option<&Timestamp>,
+    ) -> Result<ForgePage, ForgeError> {
+        let project = encode_project_path(&repo.path());
+        let base = format!("/projects/{project}");
+        let mut events = Vec::new();
+
+        // Thread feeds: issues and merge requests, updated-ascending.
+        let mut query = vec![
+            ("order_by".to_owned(), "updated_at".to_owned()),
+            ("sort".to_owned(), "asc".to_owned()),
+        ];
+        if let Some(since) = since {
+            query.push(("updated_after".to_owned(), since.raw().to_owned()));
+        }
+        let mut threads: Vec<(WireThread, bool)> = Vec::new();
+        for row in self
+            .paged::<WireThread>(&format!("{base}/issues"), &query)
+            .await?
+        {
+            if let Some(event) = thread_event(&row, repo, false)? {
+                events.push(event);
+            }
+            threads.push((row, false));
+        }
+        for row in self
+            .paged::<WireThread>(&format!("{base}/merge_requests"), &query)
+            .await?
+        {
+            if let Some(event) = thread_event(&row, repo, true)? {
+                events.push(event);
+            }
+            threads.push((row, true));
+        }
+
+        // Notes, per updated thread: no repo-wide feed, no updated filter
+        // on the endpoint, so each thread's notes are paged and cut off
+        // client-side.
+        for (row, is_pr) in &threads {
+            let iid = required(row.iid, "thread iid")?;
+            let segment = if *is_pr { "merge_requests" } else { "issues" };
+            let notes_query = vec![
+                ("order_by".to_owned(), "created_at".to_owned()),
+                ("sort".to_owned(), "asc".to_owned()),
+            ];
+            let notes_path = format!("{base}/{segment}/{iid}/notes");
+            for note in self.paged::<WireNote>(&notes_path, &notes_query).await? {
+                if let Some(event) = note_event(&note, repo, iid, *is_pr, since)? {
+                    events.push(event);
+                }
+            }
+        }
+
+        Ok(ForgePage::finish(events, since))
+    }
+
+    async fn comment(&self, repo: &RepoRef, number: u64, body: &str) -> Result<(), ForgeError> {
+        let project = encode_project_path(&repo.path());
+        // Issue iids and MR iids are separate sequences: try the issue
+        // feed first, fall back on its 404. An iid that is neither
+        // surfaces the merge-request 404.
+        let notes_path = |segment: &str| format!("/projects/{project}/{segment}/{number}/notes");
+        let response = self
+            .post(&notes_path("issues"))
+            .json(&serde_json::json!({ "body": body }))
+            .send()
+            .await
+            .map_err(ForgeError::network)?;
+        let response = if response.status() == reqwest::StatusCode::NOT_FOUND {
+            self.post(&notes_path("merge_requests"))
+                .json(&serde_json::json!({ "body": body }))
+                .send()
+                .await
+                .map_err(ForgeError::network)?
+        } else {
+            response
+        };
+        ensure_success(response).await?;
+        Ok(())
+    }
+
+    async fn snapshot(&self, repo: &RepoRef, number: u64) -> Result<Snapshot, ForgeError> {
+        let project = encode_project_path(&repo.path());
+        let resource = |segment: &str| format!("/projects/{project}/{segment}/{number}");
+        let response = self
+            .get(&resource("issues"))
+            .send()
+            .await
+            .map_err(ForgeError::network)?;
+        // Same iid sequences as `comment`: issues first, 404 fallback.
+        let (row, is_pr) = match decode::<WireThread>(response).await {
+            Ok(row) => (row, false),
+            Err(ForgeError::Api { status: 404, .. }) => {
+                let response = self
+                    .get(&resource("merge_requests"))
+                    .send()
+                    .await
+                    .map_err(ForgeError::network)?;
+                let row: WireThread = decode(response).await?;
+                (row, true)
+            }
+            Err(other) => return Err(other),
+        };
+        Ok(snapshot_from_row(&row, is_pr)?)
+    }
+
+    async fn notifications(
+        &self,
+        since: Option<&Timestamp>,
+    ) -> Result<NotificationPage, ForgeError> {
+        // The pending-todo feed: everything addressed to the account,
+        // every project it can see. The endpoint has no time filter, so
+        // `since` only shapes the returned cursor; the refetch is
+        // absorbed by the dedup ids.
+        let rows = self.paged::<WireTodo>("/todos", &[]).await?;
+        let mut notifications = Vec::new();
+        for row in &rows {
+            if let Some(notification) = todo_notification(row)? {
+                notifications.push(notification);
+            }
+        }
+        Ok(NotificationPage::finish(notifications, since))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn repo() -> RepoRef {
+        RepoRef::new("acme", "widgets")
+    }
+
+    #[test]
+    fn project_paths_encode_their_slashes() {
+        assert_eq!(encode_project_path("acme/widgets"), "acme%2Fwidgets");
+        assert_eq!(encode_project_path("group/sub/proj"), "group%2Fsub%2Fproj");
+        // Unreserved characters pass through byte for byte.
+        assert_eq!(encode_project_path("safe-name_1.0~x"), "safe-name_1.0~x");
+        assert_eq!(encode_project_path("spa ce"), "spa%20ce");
+    }
+
+    fn thread(json: &str) -> WireThread {
+        serde_json::from_str(json).expect("thread fixture parses")
+    }
+
+    #[test]
+    fn a_closed_issue_row_maps_to_closed() {
+        let event = thread_event(
+            &thread(
+                r#"{"id": 407, "iid": 7, "state": "closed",
+                    "author": {"username": "alice"},
+                    "created_at": "2026-10-01T09:00:00Z",
+                    "updated_at": "2026-10-02T10:00:00.081Z",
+                    "closed_at": "2026-10-02T10:00:00Z",
+                    "web_url": "https://gitlab.com/acme/widgets/-/issues/7"}"#,
+            ),
+            &repo(),
+            false,
+        )
+        .expect("maps")
+        .expect("closed row carries an event");
+        assert_eq!(event.kind, EventKind::Closed);
+        // GitLab's fractional seconds survive verbatim in the id.
+        assert_eq!(event.id, "gitlab:issue:407@2026-10-02T10:00:00.081Z");
+        assert_eq!(event.number, 7);
+        assert!(!event.is_pr);
+    }
+
+    #[test]
+    fn reopened_and_opened_derive_from_the_close_history() {
+        let reopened = thread_event(
+            &thread(
+                r#"{"id": 404, "iid": 4, "state": "opened",
+                    "created_at": "2026-10-01T09:00:00Z",
+                    "updated_at": "2026-10-07T09:00:00Z",
+                    "closed_at": "2026-10-03T00:00:00Z"}"#,
+            ),
+            &repo(),
+            false,
+        )
+        .expect("maps")
+        .expect("reopened row carries an event");
+        assert_eq!(reopened.kind, EventKind::Reopened);
+
+        let opened = thread_event(
+            &thread(
+                r#"{"id": 405, "iid": 5, "state": "opened",
+                    "created_at": "2026-10-03T08:00:00Z",
+                    "updated_at": "2026-10-03T08:00:00Z",
+                    "closed_at": null}"#,
+            ),
+            &repo(),
+            false,
+        )
+        .expect("maps")
+        .expect("fresh row carries an event");
+        assert_eq!(opened.kind, EventKind::Opened);
+    }
+
+    #[test]
+    fn merged_merge_requests_map_to_merged() {
+        let event = thread_event(
+            &thread(
+                r#"{"id": 503, "iid": 3, "state": "merged",
+                    "created_at": "2026-10-01T08:00:00Z",
+                    "updated_at": "2026-10-06T12:00:00Z",
+                    "closed_at": "2026-10-06T12:00:00Z"}"#,
+            ),
+            &repo(),
+            true,
+        )
+        .expect("maps")
+        .expect("merged row carries an event");
+        assert_eq!(event.kind, EventKind::Merged);
+        assert!(event.is_pr);
+        assert_eq!(event.id, "gitlab:mr:503@2026-10-06T12:00:00Z");
+
+        // Unmerged close on an MR stays Closed.
+        let closed = thread_event(
+            &thread(
+                r#"{"id": 504, "iid": 3, "state": "closed",
+                    "created_at": "2026-10-01T08:00:00Z",
+                    "updated_at": "2026-10-06T12:00:00Z",
+                    "closed_at": "2026-10-06T12:00:00Z"}"#,
+            ),
+            &repo(),
+            true,
+        )
+        .expect("maps")
+        .expect("closed row carries an event");
+        assert_eq!(closed.kind, EventKind::Closed);
+    }
+
+    #[test]
+    fn system_notes_and_stale_notes_are_skipped() {
+        let system: WireNote = serde_json::from_str(
+            r#"{"id": 301, "body": "changed the description", "system": true,
+                "created_at": "2026-10-02T11:00:00Z"}"#,
+        )
+        .expect("note fixture parses");
+        assert_eq!(
+            note_event(&system, &repo(), 7, false, None).expect("maps"),
+            None
+        );
+
+        let stale: WireNote = serde_json::from_str(
+            r#"{"id": 303, "body": "old", "system": false,
+                "created_at": "2025-01-01T00:00:00Z"}"#,
+        )
+        .expect("note fixture parses");
+        let since = Timestamp::parse("2026-10-01T00:00:00Z").expect("parses");
+        assert_eq!(
+            note_event(&stale, &repo(), 7, false, Some(&since)).expect("maps"),
+            None
+        );
+    }
+
+    #[test]
+    fn snapshot_states_map_from_the_row_state() {
+        let snapshot = snapshot_from_row(
+            &thread(
+                r#"{"id": 503, "iid": 3, "state": "merged", "title": "fix crash",
+                    "description": "the fix", "author": {"username": "bob"},
+                    "labels": ["bug", "critical"],
+                    "web_url": "https://gitlab.com/acme/widgets/-/merge_requests/3"}"#,
+            ),
+            true,
+        )
+        .expect("maps");
+        assert_eq!(snapshot.state, ThreadState::Merged);
+        assert!(snapshot.is_pr);
+        assert_eq!(
+            snapshot.labels,
+            vec!["bug".to_owned(), "critical".to_owned()]
+        );
+        assert_eq!(snapshot.author.as_deref(), Some("bob"));
+        assert_eq!(snapshot.body.as_deref(), Some("the fix"));
+    }
+}

integrations/rhema-forge-integration/src/lib.rs

@@ -0,0 +1,70 @@
+//! `rhema-forge-integration` — the forge integration: one normalized
+//! event vocabulary and one client contract for GitHub, Gitea, and
+//! GitLab, and the ingest path that turns forge activity into RSD
+//! dispatch.
+//!
+//! An integration, not an agent: no driver, no model, no loop lives
+//! here. A consumer composes the crate with its own agent tier — the
+//! [`crate::Watcher`] polls a [`crate::Forge`] for activity and feeds
+//! the events to a [`rhema_rsd::Dispatcher`](rhema_rsd) (the forge's
+//! stable dedup id becomes the event id, and the resource key is spec
+//! §8's `host:owner/repo#number` shape, so re-polling unchanged history
+//! dedups away and a changed row re-dispatches), and a dispatched
+//! thread is answered through the same [`crate::Forge`] capabilities,
+//! with [`crate::decode`] recovering the event behind an inbox item.
+//! Polling cadence, handler policy, and the loop itself are the
+//! consumer's — the claw daemon today, any other agent consumer of the
+//! framework tomorrow; the dispatch guarantees are the dispatcher's.
+//!
+//! ```
+//! use rhema_forge_integration::{Forge, GitHubForge, RepoRef};
+//!
+//! # async fn demo() -> Result<(), rhema_forge_integration::ForgeError> {
+//! let forge = GitHubForge::new(std::env::var("GITHUB_TOKEN").unwrap());
+//! // First poll: everything. Later polls: pass `page.cursor` back in.
+//! let page = forge.poll(&RepoRef::new("acme", "widgets"), None).await?;
+//! for event in &page.events {
+//!     println!("{:?} on {}#{}", event.kind, event.repo.path(), event.number);
+//! }
+//! # Ok(())
+//! # }
+//! ```
+//!
+//! Layout:
+//! - [`event`] — the normalized vocabulary: `ForgeEvent`, `EventKind`,
+//!   `RepoRef`, `Snapshot`, `Timestamp`
+//! - [`error`] — the crate's single forge error type
+//! - [`forge`] — the `Forge` client contract and `ForgePage`
+//! - [`github`] — the GitHub adapter (api.github.com)
+//! - [`gitea`] — the Gitea adapter (self-hosted, `/api/v1`)
+//! - [`gitlab`] — the GitLab adapter (gitlab.com or self-hosted, `/api/v4`)
+//! - [`watcher`] — the dispatcher seam: polls a forge into an RSD
+//!   `Dispatcher` (keys, dedup, cursors) and decodes delivered inbox
+//!   items back into forge events
+
+#![forbid(unsafe_code)]
+#![deny(missing_docs)]
+
+pub mod error;
+pub mod event;
+pub mod forge;
+pub mod gitea;
+pub mod github;
+pub mod gitlab;
+pub mod watcher;
+
+// Re-exported so implementors of the async seam can apply #[async_trait]
+// without adding the crate themselves.
+pub use async_trait::async_trait;
+pub use error::ForgeError;
+pub use event::{
+    EventKind, ForgeEvent, Notification, NotificationPage, NotificationReason, RepoRef, Snapshot,
+    ThreadState, Timestamp,
+};
+pub use forge::{Forge, ForgePage};
+pub use gitea::GiteaForge;
+pub use github::GitHubForge;
+pub use gitlab::GitLabForge;
+pub use watcher::{
+    WatchError, Watcher, decode, forge_key_fn, parse_key, register_forge, resource_key,
+};

integrations/rhema-forge-integration/src/watcher.rs

@@ -0,0 +1,186 @@
+//! The dispatcher seam: poll a forge feed into the RSD dispatcher, and
+//! decode a delivered inbox item back into the forge event it carried.
+//!
+//! The [`Watcher`] owns one cursor per repo (the newest event time fed to
+//! the dispatcher) and turns forge events into
+//! [`rhema_rsd::Event`]s: the forge's stable dedup id stays the event id,
+//! the serialized [`ForgeEvent`](crate::event::ForgeEvent) is the payload,
+//! and the resource key is the spec §8 shape —
+//! `{source}:{owner}/{name}#{number}` (see [`resource_key`]). Re-polling
+//! unchanged history therefore dedups away in the dispatcher, and a
+//! changed row re-dispatches.
+//!
+//! The watcher is loop-free: one `poll` per call. Cadence belongs to the
+//! embedder's loop.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+use rhema_rsd::{Clock, Dispatcher, Event, InboxItem, KeyFn, ResourceKey, SystemClock};
+
+use crate::error::ForgeError;
+use crate::event::{ForgeEvent, RepoRef, Timestamp};
+use crate::forge::Forge;
+
+/// Why a watch step failed: the forge call, or the dispatcher seam behind
+/// it.
+#[derive(Debug, thiserror::Error)]
+pub enum WatchError {
+    /// The poll itself failed.
+    #[error(transparent)]
+    Forge(#[from] ForgeError),
+    /// The dispatcher rejected or could not persist an event.
+    #[error(transparent)]
+    Dispatch(#[from] rhema_rsd::DispatchError),
+}
+
+/// The resource key of a thread on `source`: `{source}:{path}#{number}` —
+/// spec §8's `github:acme/widgets#142` shape.
+pub fn resource_key(source: &str, repo: &RepoRef, number: u64) -> ResourceKey {
+    ResourceKey::from(format!("{source}:{}#{number}", repo.path()))
+}
+
+/// Recovers the thread coordinates from [`resource_key`]'s shape. The
+/// inverse of the key builder, so a dispatch never needs the original
+/// event to reach the forge.
+pub fn parse_key(key: &ResourceKey) -> Option<(RepoRef, u64)> {
+    let (head, number) = key.as_str().rsplit_once('#')?;
+    let path = head.split_once(':')?.1;
+    let (owner, name) = path.rsplit_once('/')?;
+    Some((RepoRef::new(owner, name), number.parse().ok()?))
+}
+
+/// The payload a watcher builds for every event: just the coordinates the
+/// canonical key function reads, before the full [`ForgeEvent`].
+#[derive(serde::Deserialize)]
+struct KeyParts {
+    repo: RepoRef,
+    number: u64,
+}
+
+/// Registers `source` on `dispatcher` with the canonical key function —
+/// the one [`resource_key`] implements. Do this once before the first
+/// ingest.
+pub fn register_forge(dispatcher: &Dispatcher, source: &str) {
+    let prefix = format!("{source}:");
+    dispatcher.register_source(
+        source.to_owned(),
+        Arc::new(move |event: &Event| {
+            match serde_json::from_value::<KeyParts>(event.payload.clone()) {
+                Ok(parts) => {
+                    ResourceKey::from(format!("{prefix}{}#{}", parts.repo.path(), parts.number))
+                }
+                // Infallible seam: a payload the watcher itself did not
+                // build still gets a unique key, so it can neither clobber
+                // another thread nor wedge the dispatcher.
+                Err(_) => ResourceKey::from(format!("{prefix}unparsable#{}", event.id)),
+            }
+        }),
+    );
+}
+
+/// The `KeyFn` [`register_forge`] installs, for embedders wiring the
+/// source themselves.
+pub fn forge_key_fn(source: &str) -> Arc<KeyFn> {
+    let prefix = format!("{source}:");
+    Arc::new(
+        move |event: &Event| match serde_json::from_value::<KeyParts>(event.payload.clone()) {
+            Ok(parts) => {
+                ResourceKey::from(format!("{prefix}{}#{}", parts.repo.path(), parts.number))
+            }
+            Err(_) => ResourceKey::from(format!("{prefix}unparsable#{}", event.id)),
+        },
+    )
+}
+
+/// Polls forge feeds into a [`Dispatcher`], one repo at a time, keeping
+/// each repo's cursor.
+pub struct Watcher {
+    dispatcher: Arc<Dispatcher>,
+    source: String,
+    cursors: Mutex<HashMap<RepoRef, Timestamp>>,
+}
+
+impl Watcher {
+    /// A watcher feeding `dispatcher` as `source` (e.g. `"github"`).
+    /// [`register_forge`] must have installed the source's key function.
+    pub fn new(dispatcher: Arc<Dispatcher>, source: impl Into<String>) -> Self {
+        Self {
+            dispatcher,
+            source: source.into(),
+            cursors: Mutex::new(HashMap::new()),
+        }
+    }
+
+    /// The cursor `repo` polls from next: `None` before the first poll
+    /// (everything the forge returns is fed), then the newest event time
+    /// seen.
+    pub fn cursor(&self, repo: &RepoRef) -> Option<Timestamp> {
+        self.cursors
+            .lock()
+            .expect("cursor map lock")
+            .get(repo)
+            .cloned()
+    }
+
+    /// One poll of `repo`: the page's events are ingested in feed order
+    /// and the cursor advances to the page's. Returns how many events were
+    /// fed to the dispatcher — replayed ids included, they dedup there.
+    pub async fn poll(&self, forge: &dyn Forge, repo: &RepoRef) -> Result<usize, WatchError> {
+        let since = self.cursor(repo);
+        let page = forge.poll(repo, since.as_ref()).await?;
+        let now = SystemClock.now_millis();
+        let mut fed = 0;
+        for event in &page.events {
+            let payload = serde_json::to_value(event)
+                .map_err(|err| ForgeError::Parse(format!("event does not serialize: {err}")))?;
+            self.dispatcher.ingest(Event {
+                id: event.id.clone(),
+                source: self.source.clone(),
+                payload,
+                received_at: now,
+            })?;
+            fed += 1;
+        }
+        if let Some(cursor) = page.cursor {
+            self.cursors
+                .lock()
+                .expect("cursor map lock")
+                .insert(repo.clone(), cursor);
+        }
+        Ok(fed)
+    }
+}
+
+/// Decodes a delivered inbox item back into the forge event the watcher
+/// fed. A payload this crate did not build is a loud parse error, never a
+/// silently skipped event.
+pub fn decode(item: &InboxItem) -> Result<ForgeEvent, ForgeError> {
+    serde_json::from_value(item.event.payload.clone())
+        .map_err(|err| ForgeError::Parse(format!("inbox payload is not a forge event: {err}")))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn the_key_shape_round_trips_through_parse() {
+        let repo = RepoRef::new("acme", "widgets");
+        let key = resource_key("github", &repo, 142);
+        assert_eq!(key.as_str(), "github:acme/widgets#142");
+        assert_eq!(parse_key(&key), Some((repo, 142)));
+    }
+
+    #[test]
+    fn a_gitlab_subgroup_path_survives_the_round_trip() {
+        let repo = RepoRef::new("group/subgroup", "widgets");
+        let key = resource_key("gitlab", &repo, 7);
+        assert_eq!(parse_key(&key), Some((repo, 7)));
+    }
+
+    #[test]
+    fn a_key_without_a_number_does_not_parse() {
+        assert_eq!(parse_key(&ResourceKey::from("github:acme/widgets")), None);
+    }
+}