A25: surface a clock-skew warning instead of failing silently

A validly-signed gossip payload rejected only by the 120s replay
freshness window (GossipReject::OutOfWindow) now drives a room-level
"clocks out of sync" warning banner, instead of silently dropping the
peer so the room shows "1 in room" with no error.

Observe-only: verify_gossip's accept/reject decision and
GOSSIP_FRESHNESS_MS are unchanged; the payload is still dropped exactly
as before. The warning is gated strictly on OutOfWindow (which, because
the signature is verified first, implies a genuine authenticated peer
whose clock is skewed), never on BadSignature.

Policy lives in a pure, unit-tested ClockSkewMonitor seam with injected
now_ms: >=3 OutOfWindow drops from the same author within 60s warn once,
5-min per-author cooldown, bounded/pruned author map. The warning rides
the existing in-process RoomEvent -> UiEvent -> transient-banner path
(no wire/serialization or dependency change).

Implemented by Codex (gpt-5.5), senior-audited against the 5-point
checklist and independently verified (452 lib tests, clippy
--all-targets clean, release build green). Tests-green only; a 2-machine
deliberate-skew field test is still owed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/network/gossip.rs

@@ -152,6 +152,100 @@ fn prune_stale_mutations(
     seen.retain(|_, last_ts| *last_ts >= floor);
 }
 
+/// Three signed, out-of-window payloads inside one minute is enough to distinguish
+/// a persistently skewed clock from a single delayed gossip frame without making
+/// the user wait long. Repeats are suppressed for five minutes per author.
+const CLOCK_SKEW_OBSERVATION_WINDOW_MS: u64 = 60_000;
+const CLOCK_SKEW_WARNING_THRESHOLD: usize = 3;
+const CLOCK_SKEW_COOLDOWN_MS: u64 = 5 * 60_000;
+const CLOCK_SKEW_AUTHORS_SOFT_CAP: usize = 256;
+const CLOCK_SKEW_AUTHORS_HARD_CAP: usize = 512;
+const CLOCK_SKEW_AUTHOR_TTL_MS: u64 = CLOCK_SKEW_COOLDOWN_MS;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct ClockSkewWarning {
+    author: EndpointId,
+    /// Positive means the peer's sender-stamped clock is ahead of ours.
+    skew_ms: i64,
+}
+
+#[derive(Debug, Default)]
+struct ClockSkewMonitor {
+    authors: HashMap<EndpointId, ClockSkewAuthorState>,
+}
+
+#[derive(Debug, Default)]
+struct ClockSkewAuthorState {
+    observed_at: Vec<u64>,
+    last_seen_ms: u64,
+    last_warned_ms: Option<u64>,
+}
+
+impl ClockSkewMonitor {
+    fn observe(
+        &mut self,
+        author: EndpointId,
+        skew_ms: i64,
+        now_ms: u64,
+    ) -> Option<ClockSkewWarning> {
+        if self.authors.len() > CLOCK_SKEW_AUTHORS_SOFT_CAP {
+            self.prune_stale_authors(now_ms);
+        }
+
+        let warning = {
+            let state = self.authors.entry(author).or_default();
+            state.last_seen_ms = now_ms;
+            let floor = now_ms.saturating_sub(CLOCK_SKEW_OBSERVATION_WINDOW_MS);
+            state.observed_at.retain(|ts| *ts >= floor);
+            state.observed_at.push(now_ms);
+            if state.observed_at.len() > CLOCK_SKEW_WARNING_THRESHOLD {
+                let excess = state.observed_at.len() - CLOCK_SKEW_WARNING_THRESHOLD;
+                state.observed_at.drain(0..excess);
+            }
+
+            let threshold_met = state.observed_at.len() >= CLOCK_SKEW_WARNING_THRESHOLD;
+            let in_cooldown = state
+                .last_warned_ms
+                .is_some_and(|last| now_ms.saturating_sub(last) < CLOCK_SKEW_COOLDOWN_MS);
+            if threshold_met && !in_cooldown {
+                state.last_warned_ms = Some(now_ms);
+                Some(ClockSkewWarning { author, skew_ms })
+            } else {
+                None
+            }
+        };
+
+        if self.authors.len() > CLOCK_SKEW_AUTHORS_HARD_CAP {
+            self.drop_oldest_authors();
+        }
+
+        warning
+    }
+
+    fn prune_stale_authors(&mut self, now_ms: u64) {
+        let stale_before = now_ms.saturating_sub(CLOCK_SKEW_AUTHOR_TTL_MS);
+        self.authors.retain(|_, state| {
+            let last_warning_live = state
+                .last_warned_ms
+                .is_some_and(|last| now_ms.saturating_sub(last) < CLOCK_SKEW_COOLDOWN_MS);
+            last_warning_live || state.last_seen_ms >= stale_before
+        });
+    }
+
+    fn drop_oldest_authors(&mut self) {
+        let remove_count = self.authors.len().saturating_sub(CLOCK_SKEW_AUTHORS_SOFT_CAP);
+        let mut by_age: Vec<_> = self
+            .authors
+            .iter()
+            .map(|(author, state)| (*author, state.last_seen_ms))
+            .collect();
+        by_age.sort_by_key(|(_, last_seen_ms)| *last_seen_ms);
+        for (author, _) in by_age.into_iter().take(remove_count) {
+            self.authors.remove(&author);
+        }
+    }
+}
+
 /// Maximum number of distinct peers we hold in a room roster at once.
 ///
 /// Everyone with the room ticket is an authenticated *insider*: a signature only
@@ -413,6 +507,7 @@ impl RoomState for IrohGossipState {
         let handle = tokio::spawn(async move {
             crate::log_msg(&format!("Spawned gossip topic loop for self_id={:?}", self_id));
             let mut state_mutations_seen = HashMap::new();
+            let mut clock_skew_monitor = ClockSkewMonitor::default();
 
             // Broadcast initial state
             let initial_payload = {
@@ -453,9 +548,28 @@ impl RoomState for IrohGossipState {
                                 // action: a forged/stale payload is dropped here
                                 // so it can't impersonate, evict, or poison
                                 // presence/address-book (security S2).
-                                if let Err(reason) =
-                                    verify_gossip(&payload, &topic_bytes, now_millis(), GOSSIP_FRESHNESS_MS)
-                                {
+                                let received_now_ms = now_millis();
+                                if let Err(reason) = verify_gossip(
+                                    &payload,
+                                    &topic_bytes,
+                                    received_now_ms,
+                                    GOSSIP_FRESHNESS_MS,
+                                ) {
+                                    if reason == GossipReject::OutOfWindow {
+                                        let skew_ms = payload.ts as i64 - received_now_ms as i64;
+                                        if let Some(warning) = clock_skew_monitor.observe(
+                                            payload.author,
+                                            skew_ms,
+                                            received_now_ms,
+                                        ) {
+                                            let _ = event_tx
+                                                .send(RoomEvent::ClockSkewSuspected {
+                                                    author: warning.author,
+                                                    skew_ms: warning.skew_ms,
+                                                })
+                                                .await;
+                                        }
+                                    }
                                     crate::log_msg(&format!(
                                         "Gossip dropped unauthenticated/stale payload claiming author={:?}: {:?}",
                                         payload.author, reason
@@ -950,6 +1064,93 @@ mod tests {
         assert!(!seen.contains_key(&(a, StateMutationKind::Announce)));
     }
 
+    #[test]
+    fn clock_skew_monitor_single_drop_does_not_warn() {
+        let author = fresh_id();
+        let mut monitor = ClockSkewMonitor::default();
+
+        assert_eq!(monitor.observe(author, -121_000, 10_000), None);
+    }
+
+    #[test]
+    fn clock_skew_monitor_three_drops_in_window_warn_once() {
+        let author = fresh_id();
+        let mut monitor = ClockSkewMonitor::default();
+
+        assert_eq!(monitor.observe(author, -121_000, 10_000), None);
+        assert_eq!(monitor.observe(author, -122_000, 40_000), None);
+        assert_eq!(
+            monitor.observe(author, -123_000, 69_999),
+            Some(ClockSkewWarning { author, skew_ms: -123_000 })
+        );
+        assert_eq!(monitor.observe(author, -124_000, 70_000), None);
+    }
+
+    #[test]
+    fn clock_skew_monitor_cooldown_suppresses_repeats() {
+        let author = fresh_id();
+        let mut monitor = ClockSkewMonitor::default();
+
+        assert_eq!(monitor.observe(author, 121_000, 0), None);
+        assert_eq!(monitor.observe(author, 122_000, 10_000), None);
+        assert!(monitor.observe(author, 123_000, 20_000).is_some());
+
+        assert_eq!(monitor.observe(author, 124_000, 30_000), None);
+        assert_eq!(monitor.observe(author, 125_000, 310_000), None);
+        assert_eq!(monitor.observe(author, 126_000, 319_000), None);
+        assert_eq!(monitor.observe(author, 127_000, 319_999), None);
+        assert_eq!(
+            monitor.observe(author, 128_000, 320_000),
+            Some(ClockSkewWarning { author, skew_ms: 128_000 })
+        );
+    }
+
+    #[test]
+    fn clock_skew_monitor_tracks_distinct_authors_independently() {
+        let a = fresh_id();
+        let b = fresh_id();
+        let mut monitor = ClockSkewMonitor::default();
+
+        assert_eq!(monitor.observe(a, -121_000, 0), None);
+        assert_eq!(monitor.observe(a, -121_000, 1_000), None);
+        assert_eq!(monitor.observe(b, 121_000, 0), None);
+        assert_eq!(monitor.observe(b, 121_000, 1_000), None);
+        assert_eq!(
+            monitor.observe(b, 121_000, 2_000),
+            Some(ClockSkewWarning { author: b, skew_ms: 121_000 })
+        );
+        assert_eq!(
+            monitor.observe(a, -121_000, 2_000),
+            Some(ClockSkewWarning { author: a, skew_ms: -121_000 })
+        );
+    }
+
+    #[test]
+    fn clock_skew_monitor_prunes_stale_authors_when_over_cap() {
+        let mut monitor = ClockSkewMonitor::default();
+        for _ in 0..=CLOCK_SKEW_AUTHORS_SOFT_CAP {
+            assert_eq!(monitor.observe(fresh_id(), -121_000, 1), None);
+        }
+        assert!(monitor.authors.len() > CLOCK_SKEW_AUTHORS_SOFT_CAP);
+
+        let current = fresh_id();
+        assert_eq!(
+            monitor.observe(current, -121_000, CLOCK_SKEW_AUTHOR_TTL_MS + 2),
+            None
+        );
+        assert_eq!(monitor.authors.len(), 1);
+        assert!(monitor.authors.contains_key(&current));
+    }
+
+    #[test]
+    fn clock_skew_monitor_hard_cap_bounds_fresh_author_growth() {
+        let mut monitor = ClockSkewMonitor::default();
+        for now_ms in 0..(CLOCK_SKEW_AUTHORS_HARD_CAP as u64 + 10) {
+            let _ = monitor.observe(fresh_id(), -121_000, now_ms);
+            assert!(monitor.authors.len() <= CLOCK_SKEW_AUTHORS_HARD_CAP);
+        }
+    }
+
     #[test]
     fn sanitize_endpoint_addr_caps_address_count() {
         use std::net::SocketAddr;