feat(claw): record every watch's poll health, and say so when a streak starts

A watch whose poll fails was invisible: `detect` turned the failure into
one `eprintln!` per tick and nothing else. A watch that had 401'd every
sixty seconds since the day it was created was indistinguishable — on
`/v1/watches`, in the agent's own `list_watches`, everywhere — from a
repository that had simply been quiet. That is how a dead credential
survives for weeks: the log line scrolls, and nothing else remembers.

`watch_health` is a third table beside `watches` and `account_watches`,
keyed by `(scope, forge, subject)`, holding `last_ok_at`, `last_error`,
`last_error_at` and `consecutive_failures`. Every detect pass records
each poll's outcome — repo watches as `repo`, account watches as
`account` — and the pass that *starts* a streak (the first failure after
a success, or after never having succeeded) pushes one device
notification, naming the watch and quoting the forge's error. The rest of
the streak stays quiet: the row carries the count, so the tick's log line
no longer has to.

Readable where the question gets asked: `/v1/watches` answers a `health`
array alongside the two watch lists, and the `list_watches` tool reports
`last_ok_at`, `last_error` and `consecutive_failures` per repo, so "is
this watch working?" is answerable without the daemon's stderr. A watch
that has never polled has no row and reports empty health, which is what
it is.

crates/rhema-claw/src/forge.rs

@@ -31,7 +31,7 @@ use rhema_forge_integration::{
 use rhema_identity::{IdentityStore, SecretKind};
 use rhema_rsd::{Clock, Dispatcher, Event, KeyFn, SystemClock};
 
-use crate::watch_store::{AccountWatchRecord, ClawStore, WatchRecord};
+use crate::watch_store::{AccountWatchRecord, ClawStore, SCOPE_ACCOUNT, SCOPE_REPO, WatchRecord};
 use rhema_forge_integration::{ForgeKind, resolve_token};
 
 /// The credential a watch polls with: the value the forge takes as a
@@ -315,7 +315,16 @@ impl Source for ForgeSource {
                 }
             };
             let client = self.client_for(&record);
-            match self.watcher.poll(client.as_ref(), &repo).await {
+            let polled = self.watcher.poll(client.as_ref(), &repo).await;
+            // Health first, whatever the outcome: a watch that has been
+            // failing since it was created must be able to say so.
+            self.record_poll(
+                SCOPE_REPO,
+                &record.forge,
+                &record.repo,
+                polled.as_ref().err().map(|err| err.to_string()),
+            );
+            match polled {
                 Ok(count) => fed += count,
                 // One dead watch (a lapsed token, a forge outage) must
                 // not starve the kind's other watches either.
@@ -335,7 +344,14 @@ impl Source for ForgeSource {
             if watch.forge != self.name {
                 continue;
             }
-            match self.poll_account(&watch).await {
+            let polled = self.poll_account(&watch).await;
+            self.record_poll(
+                SCOPE_ACCOUNT,
+                &watch.forge,
+                &watch.account_id,
+                polled.as_ref().err().map(|err| err.to_string()),
+            );
+            match polled {
                 Ok(count) => fed += count,
                 // One dead account (a revoked token, a forge outage)
                 // must not starve the kind's other watches either.
@@ -350,6 +366,47 @@ impl Source for ForgeSource {
 }
 
 impl ForgeSource {
+    /// Records one poll's outcome, and says something the first time a
+    /// streak begins: a watch refused on every tick looks exactly like a
+    /// quiet repository from the outside, and only the failing tick knows
+    /// the difference. One notification per streak — the health row
+    /// carries the count, so the log line does not have to.
+    fn record_poll(&self, scope: &str, forge: &str, subject: &str, failure: Option<String>) {
+        // The table stamps in i64 millis like every other column here;
+        // the clock speaks u64.
+        let at = SystemClock.now_millis() as i64;
+        let recorded = match &failure {
+            Some(error) => self
+                .store
+                .record_poll_failure(scope, forge, subject, error, at),
+            None => self.store.record_poll_ok(scope, forge, subject, at),
+        };
+        let health = match recorded {
+            Ok(health) => health,
+            Err(err) => {
+                // Losing the health row must not stop the poll loop.
+                eprintln!("[claw] warning: watch health for {subject}: {err}");
+                return;
+            }
+        };
+        if !health.opens_a_failure_streak() {
+            return;
+        }
+        let Some(error) = failure else {
+            return;
+        };
+        let title = format!("{forge} watch {subject} is failing");
+        let body = format!(
+            "{error}\n\nFirst failure of the streak: GET /v1/watches has the count and the error."
+        );
+        if let Err(err) =
+            self.notifications
+                .push(crate::notifications::DEFAULT_DEVICE, &title, &body)
+        {
+            eprintln!("[claw] warning: notifying {title} failed: {err}");
+        }
+    }
+
     /// One account-watch poll: fetch the account's notifications as the
     /// account, triage each, and land the verdicts — agent dispatches
     /// ingest as role-tagged thread events, surfaces hit the
@@ -654,6 +711,60 @@ mod tests {
         );
     }
 
+    /// A watch that cannot poll is not a quiet watch: the pass records
+    /// the failure in the health table and tells the device once per
+    /// streak, so a dead credential looks dead instead of quiet.
+    #[tokio::test]
+    async fn a_failing_poll_records_health_and_notifies_once() {
+        let store = Arc::new(ClawStore::in_memory().expect("store opens"));
+        // Nothing listens on port 1: every poll fails on the connect.
+        store
+            .add("gitea", "acme/barn", Some("http://127.0.0.1:1"), "")
+            .expect("add");
+        let (dispatcher, accounts, notifications) = test_parts(store.clone());
+        let source = ForgeSource::new(
+            &dispatcher,
+            "gitea",
+            store.clone(),
+            accounts,
+            notifications.clone(),
+        );
+
+        source
+            .detect()
+            .await
+            .expect("a failed poll is not a pass error");
+        let health = store
+            .health(SCOPE_REPO, "gitea", "acme/barn")
+            .expect("health reads")
+            .expect("the failure is recorded");
+        assert_eq!(health.consecutive_failures, 1);
+        assert_eq!(health.last_ok_at, None, "it has never polled once");
+        assert!(health.last_error.is_some(), "the reason is kept");
+
+        let queued = notifications
+            .pending(crate::notifications::DEFAULT_DEVICE, 0)
+            .expect("the queue reads");
+        assert_eq!(queued.len(), 1, "one notification for the streak");
+        assert!(
+            queued[0].title.contains("acme/barn"),
+            "the title names the watch: {}",
+            queued[0].title
+        );
+
+        // The second failing pass extends the streak and stays quiet.
+        source.detect().await.expect("the second pass runs");
+        let health = store
+            .health(SCOPE_REPO, "gitea", "acme/barn")
+            .expect("health reads")
+            .expect("the failure is still recorded");
+        assert_eq!(health.consecutive_failures, 2);
+        let queued = notifications
+            .pending(crate::notifications::DEFAULT_DEVICE, 0)
+            .expect("the queue reads");
+        assert_eq!(queued.len(), 1, "the streak is not re-announced");
+    }
+
     /// The client cache pays off when two watches share their
     /// coordinates: one client, not two.
     #[tokio::test]

crates/rhema-claw/src/http.rs

@@ -834,9 +834,19 @@ async fn watches_list(State(state): State<ClawState>) -> Result<Json<Value>, (St
         .watches
         .list_account_watches()
         .map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
-    Ok(Json(
+    // The poll health rides along for the same reason the account
-        json!({ "watches": watches, "account_watches": account_watches }),
+    // watches do: "has this ever polled?" is the first question a silent
-    ))
+    // watch raises, and a tick that failed with a 401 answers it, while
+    // an empty table would say nothing either way.
+    let health = state
+        .watches
+        .list_health()
+        .map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
+    Ok(Json(json!({
+        "watches": watches,
+        "account_watches": account_watches,
+        "health": health,
+    })))
 }
 
 /// `{"forge", "repo", "host"?, "token"?}` — the shape every watch entry

crates/rhema-claw/src/watch_store.rs

@@ -8,6 +8,12 @@
 //! config no longer carries watches at all — the table is the one
 //! truth, managed over `/v1/watches` and (from chat) the watch tools.
 //!
+//! Beside it, the poll health of every watch ([`WatchHealth`]): when it
+//! last polled successfully, why it last failed and how many failures
+//! that makes. A watch that has been refused by its forge on every tick
+//! since it was created looks exactly like a quiet repository without
+//! this — the difference has to be readable somewhere.
+//!
 //! Two connections, one file: this store opens its own rusqlite
 //! connection to the same database as the dispatcher's store — WAL and a
 //! busy timeout, the pattern rhema-rsd's SQLite tests use — and the two
@@ -98,6 +104,16 @@ CREATE TABLE IF NOT EXISTS account_watches (
     created_at INTEGER NOT NULL,
     UNIQUE(forge, account_id)
 );
+CREATE TABLE IF NOT EXISTS watch_health (
+    scope                TEXT NOT NULL,
+    forge                TEXT NOT NULL,
+    subject              TEXT NOT NULL,
+    last_ok_at           INTEGER,
+    last_error           TEXT,
+    last_error_at        INTEGER,
+    consecutive_failures INTEGER NOT NULL DEFAULT 0,
+    PRIMARY KEY (scope, forge, subject)
+);
 ";
 
 /// Why a watch-table operation failed.
@@ -192,6 +208,45 @@ pub struct AccountWatchRecord {
     pub created_at: i64,
 }
 
+/// The scope of a health row: a repository watch's thread feed.
+pub const SCOPE_REPO: &str = "repo";
+
+/// The scope of a health row: an account's notification feed.
+pub const SCOPE_ACCOUNT: &str = "account";
+
+/// One watch's poll health, as stored: the last success, the last
+/// failure and its reason, and the streak length. `last_ok_at: None` with
+/// `consecutive_failures: 0` is a watch that has not polled yet; a growing
+/// `consecutive_failures` with no `last_ok_at` is a watch that has never
+/// once succeeded — the shape a wrong or mislabelled credential leaves.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct WatchHealth {
+    /// [`SCOPE_REPO`] or [`SCOPE_ACCOUNT`].
+    pub scope: String,
+    /// The forge's source name: `github`, `gitea`, or `gitlab`.
+    pub forge: String,
+    /// The repo (`owner/name`) for a repo watch, the account id for an
+    /// account watch.
+    pub subject: String,
+    /// When a poll last succeeded, millis since the Unix epoch.
+    pub last_ok_at: Option<i64>,
+    /// The last failure's reason, as the poll reported it.
+    pub last_error: Option<String>,
+    /// When that failure happened, millis since the Unix epoch.
+    pub last_error_at: Option<i64>,
+    /// Failures since the last success — what tells a caller to say
+    /// something once, not on every tick of a long outage.
+    pub consecutive_failures: i64,
+}
+
+impl WatchHealth {
+    /// Whether this is the first failure of a streak: the moment worth a
+    /// notification, before the tick's log line becomes wallpaper.
+    pub fn opens_a_failure_streak(&self) -> bool {
+        self.consecutive_failures == 1
+    }
+}
+
 impl ClawStore {
     /// Opens (creating if needed) the watch table in the database at
     /// `path` — the same file the dispatcher's store holds. WAL keeps the
@@ -364,6 +419,142 @@ impl ClawStore {
             .map_err(WatchStoreError::from)?;
         Ok(rows)
     }
+    /// Records a successful poll: the streak resets, and the row stops
+    /// reading as broken.
+    pub fn record_poll_ok(
+        &self,
+        scope: &str,
+        forge: &str,
+        subject: &str,
+        at: i64,
+    ) -> Result<WatchHealth, WatchStoreError> {
+        let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner);
+        conn.execute(
+            "INSERT INTO watch_health (scope, forge, subject, last_ok_at, consecutive_failures)
+             VALUES (?1, ?2, ?3, ?4, 0)
+             ON CONFLICT(scope, forge, subject) DO UPDATE SET
+                last_ok_at = excluded.last_ok_at,
+                consecutive_failures = 0",
+            params![scope, forge, subject, at],
+        )
+        .map_err(WatchStoreError::from)?;
+        read_health(&conn, scope, forge, subject)
+    }
+
+    /// Records a failed poll and the reason: the streak grows, and the
+    /// last success (if any) stays visible beside it.
+    pub fn record_poll_failure(
+        &self,
+        scope: &str,
+        forge: &str,
+        subject: &str,
+        error: &str,
+        at: i64,
+    ) -> Result<WatchHealth, WatchStoreError> {
+        let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner);
+        conn.execute(
+            "INSERT INTO watch_health
+                (scope, forge, subject, last_error, last_error_at, consecutive_failures)
+             VALUES (?1, ?2, ?3, ?4, ?5, 1)
+             ON CONFLICT(scope, forge, subject) DO UPDATE SET
+                last_error = excluded.last_error,
+                last_error_at = excluded.last_error_at,
+                consecutive_failures = watch_health.consecutive_failures + 1",
+            params![scope, forge, subject, error, at],
+        )
+        .map_err(WatchStoreError::from)?;
+        read_health(&conn, scope, forge, subject)
+    }
+
+    /// Every health row, ordered for a stable answer.
+    pub fn list_health(&self) -> Result<Vec<WatchHealth>, WatchStoreError> {
+        let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner);
+        let mut statement = conn
+            .prepare(
+                "SELECT scope, forge, subject, last_ok_at, last_error, last_error_at,
+                        consecutive_failures
+                 FROM watch_health ORDER BY scope ASC, forge ASC, subject ASC",
+            )
+            .map_err(WatchStoreError::from)?;
+        let rows = statement
+            .query_map([], |row| {
+                Ok(WatchHealth {
+                    scope: row.get(0)?,
+                    forge: row.get(1)?,
+                    subject: row.get(2)?,
+                    last_ok_at: row.get(3)?,
+                    last_error: row.get(4)?,
+                    last_error_at: row.get(5)?,
+                    consecutive_failures: row.get(6)?,
+                })
+            })
+            .map_err(WatchStoreError::from)?
+            .collect::<Result<Vec<_>, rusqlite::Error>>()
+            .map_err(WatchStoreError::from)?;
+        Ok(rows)
+    }
+
+    /// One watch's health, when it has polled at all.
+    pub fn health(
+        &self,
+        scope: &str,
+        forge: &str,
+        subject: &str,
+    ) -> Result<Option<WatchHealth>, WatchStoreError> {
+        let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner);
+        let mut statement = conn
+            .prepare(
+                "SELECT scope, forge, subject, last_ok_at, last_error, last_error_at,
+                        consecutive_failures
+                 FROM watch_health WHERE scope = ?1 AND forge = ?2 AND subject = ?3",
+            )
+            .map_err(WatchStoreError::from)?;
+        let mut rows = statement
+            .query_map(params![scope, forge, subject], |row| {
+                Ok(WatchHealth {
+                    scope: row.get(0)?,
+                    forge: row.get(1)?,
+                    subject: row.get(2)?,
+                    last_ok_at: row.get(3)?,
+                    last_error: row.get(4)?,
+                    last_error_at: row.get(5)?,
+                    consecutive_failures: row.get(6)?,
+                })
+            })
+            .map_err(WatchStoreError::from)?;
+        match rows.next() {
+            Some(row) => Ok(Some(row.map_err(WatchStoreError::from)?)),
+            None => Ok(None),
+        }
+    }
+}
+
+/// One health row by key: the upserts' return value, read back under the
+/// connection they wrote with.
+fn read_health(
+    conn: &rusqlite::Connection,
+    scope: &str,
+    forge: &str,
+    subject: &str,
+) -> Result<WatchHealth, WatchStoreError> {
+    conn.query_row(
+        "SELECT scope, forge, subject, last_ok_at, last_error, last_error_at,
+                consecutive_failures
+         FROM watch_health WHERE scope = ?1 AND forge = ?2 AND subject = ?3",
+        params![scope, forge, subject],
+        |row| {
+            Ok(WatchHealth {
+                scope: row.get(0)?,
+                forge: row.get(1)?,
+                subject: row.get(2)?,
+                last_ok_at: row.get(3)?,
+                last_error: row.get(4)?,
+                last_error_at: row.get(5)?,
+                consecutive_failures: row.get(6)?,
+            })
+        },
+    )
+    .map_err(WatchStoreError::from)
 }
 
 /// Milliseconds since the Unix epoch; stamps `created_at`.
@@ -401,6 +592,51 @@ pub fn validate_watch(
 mod tests {
     use super::*;
 
+    /// A watch's health is a streak, not a flag: the first failure opens
+    /// it, the next extends it, and a success closes it without losing
+    /// the last success's stamp.
+    #[test]
+    fn a_health_row_tracks_the_streak_and_survives_a_recovery() {
+        let store = ClawStore::in_memory().expect("store opens");
+        assert_eq!(
+            store
+                .health(SCOPE_REPO, "gitea", "acme/barn")
+                .expect("read"),
+            None,
+            "a watch that never polled has no row"
+        );
+
+        let first = store
+            .record_poll_failure(SCOPE_REPO, "gitea", "acme/barn", "401 denied", 1_000)
+            .expect("failure lands");
+        assert_eq!(first.consecutive_failures, 1);
+        assert!(first.opens_a_failure_streak());
+        assert_eq!(first.last_ok_at, None);
+        assert_eq!(first.last_error.as_deref(), Some("401 denied"));
+
+        let second = store
+            .record_poll_failure(SCOPE_REPO, "gitea", "acme/barn", "401 denied", 2_000)
+            .expect("failure lands");
+        assert_eq!(second.consecutive_failures, 2);
+        assert!(
+            !second.opens_a_failure_streak(),
+            "only the first failure of a streak is news"
+        );
+        assert_eq!(second.last_error_at, Some(2_000));
+
+        let recovered = store
+            .record_poll_ok(SCOPE_REPO, "gitea", "acme/barn", 3_000)
+            .expect("success lands");
+        assert_eq!(recovered.consecutive_failures, 0);
+        assert_eq!(recovered.last_ok_at, Some(3_000));
+        assert_eq!(
+            recovered.last_error.as_deref(),
+            Some("401 denied"),
+            "the last failure stays on the record after a recovery"
+        );
+        assert_eq!(store.list_health().expect("list reads").len(), 1);
+    }
+
     #[test]
     fn validation_fails_a_watch_loudly_and_specificly() {
         assert_eq!(

crates/rhema-claw/src/watch_tools.rs

@@ -18,7 +18,7 @@ use rhema_autonomous_agent::{ToolError, ToolOp, ToolRegistry, async_trait};
 use serde::Deserialize;
 use serde_json::{Value, json};
 
-use crate::watch_store::{ClawStore, validate_watch};
+use crate::watch_store::{ClawStore, SCOPE_REPO, validate_watch};
 use rhema_forge_integration::ForgeKind;
 
 /// The compiled operations behind the manifests, by `run.builtin` id.
@@ -161,13 +161,25 @@ struct ListWatches {
 impl ToolOp for ListWatches {
     async fn invoke(&self, _args: Value) -> Result<String, ToolError> {
         let watches = self.store.list().map_err(|err| err.to_string())?;
+        let health = self.store.list_health().map_err(|err| err.to_string())?;
         let rows: Vec<Value> = watches
             .iter()
             .map(|watch| {
+                // A watch that has never polled has no health row at all:
+                // it answers with every health field empty and a zero
+                // streak, which is exactly what it is.
+                let health = health.iter().find(|row| {
+                    row.scope == SCOPE_REPO && row.forge == watch.forge && row.subject == watch.repo
+                });
                 json!({
                     "forge": watch.forge,
                     "repo": watch.repo,
                     "host": watch.host,
+                    "last_ok_at": health.and_then(|row| row.last_ok_at),
+                    "last_error": health.and_then(|row| row.last_error.clone()),
+                    "consecutive_failures": health
+                        .map(|row| row.consecutive_failures)
+                        .unwrap_or(0),
                 })
             })
             .collect();
@@ -363,8 +375,22 @@ mod tests {
         assert_eq!(
             serde_json::from_str::<Value>(&listed.content).unwrap(),
             json!({"watches": [
-                {"forge": "github", "repo": "acme/widgets", "host": null},
+                {
-                {"forge": "gitlab", "repo": "group/prop", "host": null},
+                    "forge": "github",
+                    "repo": "acme/widgets",
+                    "host": null,
+                    "last_ok_at": null,
+                    "last_error": null,
+                    "consecutive_failures": 0,
+                },
+                {
+                    "forge": "gitlab",
+                    "repo": "group/prop",
+                    "host": null,
+                    "last_ok_at": null,
+                    "last_error": null,
+                    "consecutive_failures": 0,
+                },
             ]}),
             "{}",
             listed.content

crates/rhema-claw/tools/list_watches/TOOL.yaml

@@ -2,7 +2,11 @@
 # compiled builtin named under `run`.
 name: list_watches
 description: >-
-  List the watched repositories: forge, repo, and host for each.
+  List the watched repositories: forge, repo, and host for each, with its
+  poll health — last_ok_at is when it last polled successfully,
+  consecutive_failures is the streak it is on now, and last_error is why
+  the last attempt failed. A watch with failures and no last_ok_at has
+  never polled once.
 replay_safety: safe
 parameters:
   type: object