security(agent): attribute the shared rate window to its top contributor

check_rate_anomaly's 60s window is shared across all sessions/sources —
when it trips, the alert reported only the anonymous aggregate count,
unlike the separate cumulative max_writes_per_session check, which does
name the offending session. A session's write count can never exceed the
window's aggregate count, so whenever the window trips, name the
top-contributing session and source within it in the same alert instead
of adding a second, redundant per-session threshold check.

INT-07

crates/clawhdf5-agent/src/anomaly.rs

@@ -208,6 +208,13 @@ impl WriteAnomalyDetector {
     /// Returns an alert if the number of writes in the last 60 seconds exceeds
     /// `config.max_writes_per_minute`, or if any session has exceeded
     /// `config.max_writes_per_session`.
+    ///
+    /// The 60-second window is a single shared window across all
+    /// sessions/sources, so when it trips the alert additionally names the
+    /// top-contributing session and source within that window — a session
+    /// can never account for more of the window than the aggregate count, so
+    /// this attributes the same trip to its actual offender rather than
+    /// reporting only the anonymous aggregate total.
     pub fn check_rate_anomaly(&self) -> Option<AnomalyAlert> {
         let recent = self.window.len() as u32;
         if recent > self.config.max_writes_per_minute {
@@ -218,11 +225,31 @@ impl WriteAnomalyDetector {
             } else {
                 Severity::Medium
             };
+
+            let mut per_session: std::collections::HashMap<&str, u32> =
+                std::collections::HashMap::new();
+            // MemorySource isn't Eq/Hash, so key by its Display string instead.
+            let mut per_source: std::collections::HashMap<String, u32> =
+                std::collections::HashMap::new();
+            for e in &self.window {
+                *per_session.entry(e.session_id.as_str()).or_insert(0) += 1;
+                *per_source.entry(e.source.to_string()).or_insert(0) += 1;
+            }
+            let top_session = per_session.iter().max_by_key(|&(_, &c)| c);
+            let top_source = per_source.iter().max_by_key(|&(_, &c)| c);
+
+            let attribution = match (top_session, top_source) {
+                (Some((session, s_count)), Some((source, r_count))) => format!(
+                    "; top contributor: session '{session}' with {s_count} writes, \
+                     source {source} with {r_count} writes"
+                ),
+                _ => String::new(),
+            };
             return Some(AnomalyAlert {
                 severity,
                 message: format!(
-                    "Rate limit exceeded: {} writes in last 60s (max {})",
-                    recent, self.config.max_writes_per_minute
+                    "Rate limit exceeded: {} writes in last 60s (max {}){}",
+                    recent, self.config.max_writes_per_minute, attribution
                 ),
                 timestamp: self.last_timestamp,
             });
@@ -402,6 +429,45 @@ mod tests {
         assert!(alert.unwrap().severity >= Severity::Medium);
     }
 
+    /// A single session dominating the shared 60s window must be named in
+    /// the alert, not just the anonymous aggregate count — this is the case
+    /// the separate cumulative max_writes_per_session check doesn't cover
+    /// (the window can trip before the session's lifetime total does).
+    #[test]
+    fn rate_anomaly_names_offending_session() {
+        let mut det = WriteAnomalyDetector::new(cfg());
+        for i in 0..11 {
+            det.record_write(event(1.0 + i as f64 * 0.1, "flood-session", MemorySource::User));
+        }
+        let alert = det.check_rate_anomaly().unwrap();
+        assert!(
+            alert.message.contains("flood-session"),
+            "expected the offending session to be named, got: {}",
+            alert.message
+        );
+    }
+
+    /// When many distinct sessions jointly trip the shared window, the top
+    /// contributor named must actually be the one with the most writes.
+    #[test]
+    fn rate_anomaly_attributes_top_contributor_among_many_sessions() {
+        let mut det = WriteAnomalyDetector::new(cfg());
+        // 5 sessions with 1 write each (below any per-session limit)...
+        for i in 0..5 {
+            det.record_write(event(1.0 + i as f64 * 0.1, "minor-session", MemorySource::User));
+        }
+        // ...plus one session responsible for the majority of the flood.
+        for i in 0..8 {
+            det.record_write(event(2.0 + i as f64 * 0.1, "major-session", MemorySource::User));
+        }
+        let alert = det.check_rate_anomaly().unwrap();
+        assert!(
+            alert.message.contains("major-session"),
+            "expected the top contributor to be named, got: {}",
+            alert.message
+        );
+    }
+
     #[test]
     fn rate_anomaly_critical_3x() {
         let mut det = WriteAnomalyDetector::new(cfg());