diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4465da4..90b3075 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,7 @@ All notable changes to this project are documented here, following
 - Speedtest now runs providers sequentially, measures each download over the full window, and reports the fastest clean provider — fixing wildly variable, under-reporting results on fast links.
 
 ### Fixed
+- Pre-M0 detection-quality batch (makes the shadow observation's FP/FN signal trustworthy before the AS214806 detection-only deploy; from a full-codebase review + re-triage). **Detector correctness:** windowing/eviction/hold-down now use a **monotonic clock** (`Instant` baseline) so an NTP/wall-clock step can't freeze or force-evict the window; a new **minimum-sample gate** (`--min-samples`, default 8) stops sampling-variance false positives (2 packets at 1-in-65536 no longer extrapolate to a 131k-pps "attack"); the sampling-rate sanity clamp is now **direction-aware** — a low reported rate is clamped up (anti-suppression) while a legitimately-high adaptive-sampling rate is *trusted* up to `expected × --max-sampling-factor` (default 64, floored at 4) instead of being clamped down and hiding a real flood; sFlow decoding is **per-sample resilient** so one malformed sample no longer discards a whole datagram's valid observations (envelope-framing errors stay fatal); and mitigation selection treats destination **port 0** (fragmented / non-TCP-UDP flows) as "no port data" → RTBH, never emitting a useless `dst_port:0` FlowSpec rule (`blackwall-flow`). **Config/API:** the config lexer now treats `#` as a comment only at line-start or after whitespace, so a value like `md5=sec#ret` is no longer silently truncated (`blackwall-config`); `parse_and_resolve()` validates the policy at load for `blackwalld flow` (the M0 entry point) and `bird-config`, so a semantically-invalid config fails at container start instead of mis-behaving through the observation window; and the read API's `?limit=` is clamped to `[1, 1000]` so a negative or huge value can't 500 or dump a whole table (`blackwall-api`). **New observability** (`/metrics`, `blackwall_flow_*`): `min_sample_suppressed_total`, `sampling_near_ceiling_total{pop}`, `sample_decode_errors_total` (distinct from the per-datagram `decode_errors`), and `detections_opened_total` / `detections_cleared_total` — together they let an operator tell a quiet POP from one silently dropping samples. Non-breaking (new CLI flags default to prior behaviour; `--bps-threshold` is documented as L2/frame-length based).
 - Lab CI reliability (blackwall#88): a stuck in-scenario command could hang a gate silently until the job cap. Root cause was two-fold — the harness ran one-shot commands via an unbounded `Command::output()`, and gate drivers ran `cargo test` *inside* the netns, whose long-lived cargo process (holding the Cargo build lock + a subprocess tree) could wedge in CI where a `sudo`-spawned tree escapes GitHub's step timeout. Fixes: `netns::run` now bounds and kills every command (own process group, drained pipes, reader-join timeouts) so it can never block the caller; daemon teardown reaps with a bounded wait; `spawn_bird` redirects its stdio like the other daemons so a killed lab can't leak a pipe-holder; and gates now run the **pre-built** interop binary directly (`scripts/build-lab-tests.sh` → `target/debug/lab-tests/<pkg>-<test>`) instead of `cargo test`, removing the cargo lock/daemon from the scenario entirely.
 - RTBH hardening ahead of production wiring (sub-project C, C1c Part A) — fixes a code review found in the C1b controller and C1a speaker. **The RTBH controller leaked blackholes permanently:** the flow detector emits `Cleared` once, and if it arrived before the RTBH `hold_down` elapsed the withdraw was dropped with no retry, so the victim `/32` stayed blackholed (self-DoSed) forever. The controller now defers such a clear and completes it from a periodic `tick`, adds a `max_ttl` auto-blackhole backstop against a dropped/missed `Cleared`, tracks per-entry origin so an operator (`Manual`) blackhole is never auto-cleared, and cancels a deferred clear when a re-attack (`Opened`/`Updated`) re-asserts the target (`blackwall-rtbh`). **The BGP speaker could panic on a large config:** `push_attr` truncated attribute lengths to one byte, so ≥64 communities overflowed and killed the session task silently — it now emits RFC 4271 §4.3 extended-length attributes; `encode_nlri` also truncates host bits so a non-host prefix can't emit malformed NLRI (`blackwall-bgp`). The speaker further validates its config as iBGP (`local_asn == peer_asn`) and the peer's OPEN (ASN and hold-time, rejecting with the correct NOTIFICATION codes), waits for the peer's KEEPALIVE before declaring the session Established (bounded by the OpenConfirm hold timer), and surfaces send failures on `BgpHandle` instead of silently dropping routes. Controller logic is unit-tested; the speaker changes are covered by the `bgp-bird` lab gate.
 - sFlow decoder dropped every flow sample from real agents: it handled only regular flow samples (type 1), but hsflowd and many switches emit **expanded flow samples (type 3)**, so `decode_datagram` returned zero observations and the volumetric detector never fired on real-agent traffic. The decoder now handles expanded flow samples (`blackwall-flow`); covered by byte-exact tests over real captured hsflowd datagrams. Found by the new `flow-sflow-live` lab gate.
diff --git a/bin/blackwalld/src/main.rs b/bin/blackwalld/src/main.rs
index bf3e9f1..5c4cc7a 100644
--- a/bin/blackwalld/src/main.rs
+++ b/bin/blackwalld/src/main.rs
@@ -89,7 +89,9 @@ enum Command {
         /// Per-destination packets-per-second threshold.
         #[arg(long)]
         pps_threshold: f64,
-        /// Per-destination bits-per-second threshold.
+        /// Per-destination bits-per-second threshold. NOTE: computed from the sFlow
+        /// L2 frame length (includes the Ethernet header), so calibrate against L2
+        /// bytes, not L3 payload.
         #[arg(long)]
         bps_threshold: f64,
         /// Sliding window in seconds.
@@ -98,6 +100,18 @@ enum Command {
         /// Hold-down in seconds before clearing a detection.
         #[arg(long, default_value_t = 30)]
         hold_down_secs: u64,
+        /// Minimum raw sFlow samples in-window before a detection may open (guards
+        /// sampling-variance false positives). 0 disables the gate.
+        #[arg(long, default_value_t = 8)]
+        min_samples: usize,
+        /// Ceiling multiplier applied to an agent's expected sampling rate when
+        /// its reported rate is high. A reported rate above `expected * 4` is
+        /// trusted (adaptive samplers legitimately raise their rate under load)
+        /// up to `expected * max_sampling_factor`, and only clamped down beyond
+        /// that ceiling — never clamped down to `expected` itself, which would
+        /// mask a real flood as a false negative.
+        #[arg(long, default_value_t = 64)]
+        max_sampling_factor: u32,
     },
     /// Apply the ruleset and start the deception engine (requires CAP_NET_ADMIN).
     Run {
@@ -1298,7 +1312,7 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
             Ok(())
         }
         Command::BirdConfig { config } => {
-            let policy = blackwall_config::parse_file(&config)?;
+            let policy = blackwall_config::parse_and_resolve(&config)?;
             match blackwall_bgp::render_bird_ibgp(&policy) {
                 Ok(s) => {
                     print!("{s}");
@@ -1353,8 +1367,10 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
             bps_threshold,
             window_secs,
             hold_down_secs,
+            min_samples,
+            max_sampling_factor,
         } => {
-            let policy = blackwall_config::parse_file(&config)?;
+            let policy = blackwall_config::parse_and_resolve(&config)?;
             if policy.shadow {
                 tracing::warn!(
                     "SHADOW MODE — mitigations are LOGGED, NOT APPLIED (RTBH/FlowSpec/XDP)"
@@ -1365,12 +1381,17 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
             let store = std::sync::Arc::new(blackwall_state::Store::connect(&database_url).await?);
             store.migrate().await?;
             let agents = blackwall_flow::AgentRegistry::from_entries(&policy.pops);
-            let detector = blackwall_flow::ThresholdDetector::new(
-                policy.prefixes.clone(),
+            let detector_config = blackwall_flow::DetectorConfig {
                 pps_threshold,
                 bps_threshold,
-                window_secs * 1000,
-                hold_down_secs * 1000,
+                window_ms: window_secs * 1000,
+                hold_down_ms: hold_down_secs * 1000,
+                min_samples,
+                max_sampling_factor,
+            };
+            let detector = blackwall_flow::ThresholdDetector::new(
+                policy.prefixes.clone(),
+                detector_config,
                 agents,
             );
 
diff --git a/bin/blackwalld/src/metrics.rs b/bin/blackwalld/src/metrics.rs
index 6c9b509..ba81d2f 100644
--- a/bin/blackwalld/src/metrics.rs
+++ b/bin/blackwalld/src/metrics.rs
@@ -100,6 +100,12 @@ async fn gather(sources: &MetricsSources) -> Vec<Metric> {
             kind: MetricKind::Counter,
             value: u64_to_f64(collector.decode_errors()),
         });
+        m.push(Metric {
+            name: "blackwall_flow_sample_decode_errors_total",
+            help: "sFlow samples that failed to decode within an otherwise-valid datagram",
+            kind: MetricKind::Counter,
+            value: u64_to_f64(collector.sample_decode_errors()),
+        });
     }
     if let Some(inflight) = &sources.inflight {
         m.push(Metric {
@@ -240,8 +246,10 @@ fn xdp_block(sources: &MetricsSources) -> Option<String> {
 }
 
 /// Render the per-POP telemetry blocks (`blackwall_flow_pop_last_seen_seconds`,
-/// `blackwall_flow_agent_sampling_mismatch_total`) plus the
-/// `blackwall_flow_unknown_agent_observations_total` scalar, or `None` when the
+/// `blackwall_flow_agent_sampling_mismatch_total`,
+/// `blackwall_flow_sampling_near_ceiling_total`) plus the
+/// `blackwall_flow_unknown_agent_observations_total` and
+/// `blackwall_flow_min_sample_suppressed_total` scalars, or `None` when the
 /// flow daemon has no per-agent snapshot wired up (`sources.agent_stats` is
 /// `None` — the deception engine, which has no sFlow collector).
 ///
@@ -289,6 +297,21 @@ fn agent_stats_block(sources: &MetricsSources, now_ms: u64) -> Option<String> {
                 s.pop, s.mismatches
             );
         }
+        let _ = writeln!(
+            out,
+            "\n# HELP blackwall_flow_sampling_near_ceiling_total Samples per POP whose trusted sampling rate landed at or above half the max-sampling-factor ceiling"
+        );
+        let _ = writeln!(
+            out,
+            "# TYPE blackwall_flow_sampling_near_ceiling_total counter"
+        );
+        for s in &stats {
+            let _ = writeln!(
+                out,
+                "blackwall_flow_sampling_near_ceiling_total{{pop=\"{}\"}} {}",
+                s.pop, s.near_ceiling
+            );
+        }
     }
 
     // Always emitted (a single scalar, not per-label) so the series exists
@@ -313,6 +336,61 @@ fn agent_stats_block(sources: &MetricsSources, now_ms: u64) -> Option<String> {
         "blackwall_flow_unknown_agent_observations_total {unknown}"
     );
 
+    // Always emitted alongside `unknown_agent_observations_total`, same reasoning.
+    let min_sample_suppressed = sources
+        .collector
+        .as_ref()
+        .map_or(0, |c| c.min_sample_suppressed());
+    out.push('\n');
+    let _ = writeln!(
+        out,
+        "# HELP blackwall_flow_min_sample_suppressed_total Detections suppressed by the minimum-sample gate"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE blackwall_flow_min_sample_suppressed_total counter"
+    );
+    let _ = writeln!(
+        out,
+        "blackwall_flow_min_sample_suppressed_total {min_sample_suppressed}"
+    );
+
+    // Always emitted alongside the other scalars above, same reasoning:
+    // distinguishes "quiet network" (both zero) from "POP silently dropping
+    // samples" (decode/suppression counters moving but detections never open).
+    let detections_opened = sources
+        .collector
+        .as_ref()
+        .map_or(0, |c| c.detections_opened());
+    out.push('\n');
+    let _ = writeln!(
+        out,
+        "# HELP blackwall_flow_detections_opened_total Detections opened by the flow detector"
+    );
+    let _ = writeln!(out, "# TYPE blackwall_flow_detections_opened_total counter");
+    let _ = writeln!(
+        out,
+        "blackwall_flow_detections_opened_total {detections_opened}"
+    );
+
+    let detections_cleared = sources
+        .collector
+        .as_ref()
+        .map_or(0, |c| c.detections_cleared());
+    out.push('\n');
+    let _ = writeln!(
+        out,
+        "# HELP blackwall_flow_detections_cleared_total Detections cleared by the flow detector"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE blackwall_flow_detections_cleared_total counter"
+    );
+    let _ = writeln!(
+        out,
+        "blackwall_flow_detections_cleared_total {detections_cleared}"
+    );
+
     Some(out)
 }
 
diff --git a/crates/blackwall-api/src/handlers.rs b/crates/blackwall-api/src/handlers.rs
index 5c10037..4a02d11 100644
--- a/crates/blackwall-api/src/handlers.rs
+++ b/crates/blackwall-api/src/handlers.rs
@@ -14,6 +14,20 @@ use std::sync::Arc;
 /// Default row cap for the `sessions`/`audit` feeds.
 const DEFAULT_LIMIT: i64 = 100;
 
+/// Upper bound on the `sessions`/`audit` feed `?limit=`, regardless of what
+/// the caller requests.
+const MAX_LIMIT: i64 = 1000;
+
+/// Clamp a requested feed `limit` to `[1, MAX_LIMIT]`, defaulting to
+/// `DEFAULT_LIMIT` when absent.
+///
+/// Guards the `LIMIT $1` bind in the feed queries: a negative or zero value
+/// would otherwise reach Postgres and error out, and an unbounded value
+/// would dump the entire table.
+fn clamp_limit(limit: Option<i64>) -> i64 {
+    limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT)
+}
+
 /// `?limit=` query for the capped feeds.
 #[derive(Debug, Deserialize)]
 pub struct LimitQuery {
@@ -178,7 +192,7 @@ pub async fn list_sessions(
     State(s): St,
     Query(q): Query<LimitQuery>,
 ) -> ApiResult<Json<Vec<SessionDto>>> {
-    let limit = q.limit.unwrap_or(DEFAULT_LIMIT);
+    let limit = clamp_limit(q.limit);
     Ok(Json(
         s.sessions(limit)
             .await?
@@ -200,7 +214,7 @@ pub async fn list_audit(
     State(s): St,
     Query(q): Query<LimitQuery>,
 ) -> ApiResult<Json<Vec<AuditDto>>> {
-    let limit = q.limit.unwrap_or(DEFAULT_LIMIT);
+    let limit = clamp_limit(q.limit);
     Ok(Json(
         s.audit(limit)
             .await?
@@ -209,3 +223,17 @@ pub async fn list_audit(
             .collect(),
     ))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn limit_is_clamped() {
+        assert_eq!(clamp_limit(Some(-1)), 1);
+        assert_eq!(clamp_limit(Some(0)), 1);
+        assert_eq!(clamp_limit(Some(i64::MAX)), MAX_LIMIT);
+        assert_eq!(clamp_limit(None), DEFAULT_LIMIT);
+        assert_eq!(clamp_limit(Some(50)), 50);
+    }
+}
diff --git a/crates/blackwall-config/src/error.rs b/crates/blackwall-config/src/error.rs
index 170e9b6..655ff01 100644
--- a/crates/blackwall-config/src/error.rs
+++ b/crates/blackwall-config/src/error.rs
@@ -34,4 +34,12 @@ pub enum ConfigError {
     /// The config file could not be read.
     #[error("reading config: {0}")]
     Io(#[from] std::io::Error),
+    /// The config parsed successfully but is semantically invalid: an owned
+    /// address falls outside every managed prefix, two tenants claim the
+    /// same address, or the same service is defined more than once. Surfaced
+    /// by [`crate::parse_and_resolve`] so load-time paths (`flow`,
+    /// `bird-config`) fail before ever running, instead of misbehaving
+    /// silently at apply time.
+    #[error("config failed validation: {0}")]
+    Resolve(#[from] blackwall_core::PolicyError),
 }
diff --git a/crates/blackwall-config/src/lexer.rs b/crates/blackwall-config/src/lexer.rs
index 35dc4d5..baeecdd 100644
--- a/crates/blackwall-config/src/lexer.rs
+++ b/crates/blackwall-config/src/lexer.rs
@@ -24,7 +24,11 @@ pub fn lex(input: &str) -> Vec<Line> {
     let mut pending: Option<(usize, Vec<String>)> = None;
 
     for (idx, raw) in input.lines().enumerate() {
-        let without_comment = match raw.find('#') {
+        let comment_start = raw
+            .char_indices()
+            .find(|&(i, c)| c == '#' && (i == 0 || raw[..i].ends_with(|p: char| p.is_whitespace())))
+            .map(|(i, _)| i);
+        let without_comment = match comment_start {
             Some(pos) => &raw[..pos],
             None => raw,
         };
@@ -110,4 +114,23 @@ mod tests {
         assert_eq!(lines.len(), 1);
         assert_eq!(lines[0].words, vec!["interface", "wan", "eth0"]);
     }
+
+    #[test]
+    fn hash_inside_token_is_literal() {
+        let out = lex("rtbh md5=sec#ret\n");
+        assert_eq!(out[0].words, vec!["rtbh", "md5=sec#ret"]);
+    }
+
+    #[test]
+    fn key_equals_hash_value_is_literal() {
+        let out = lex("k key=#val\n");
+        assert_eq!(out[0].words, vec!["k", "key=#val"]);
+    }
+
+    #[test]
+    fn whitespace_preceded_hash_still_comments() {
+        assert_eq!(lex("foo # comment\n")[0].words, vec!["foo"]);
+        assert!(lex("# comment only\n").is_empty());
+        assert!(lex("    # indented comment\n").is_empty());
+    }
 }
diff --git a/crates/blackwall-config/src/lib.rs b/crates/blackwall-config/src/lib.rs
index 8877ae1..42a4894 100644
--- a/crates/blackwall-config/src/lib.rs
+++ b/crates/blackwall-config/src/lib.rs
@@ -20,10 +20,29 @@ pub fn parse_file(path: &Path) -> Result<Policy, ConfigError> {
     parse_str(&text)
 }
 
+/// Parse a config file and validate it (`Policy::resolve()`), so a
+/// semantically invalid config (address outside prefixes, duplicate
+/// ownership, duplicate service) fails at load rather than at apply. Used by
+/// load-time paths (`flow`, `bird-config`).
+pub fn parse_and_resolve(path: &Path) -> Result<Policy, ConfigError> {
+    let policy = parse_file(path)?;
+    policy.resolve().map_err(ConfigError::Resolve)?;
+    Ok(policy)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    /// Test-only sibling of [`parse_and_resolve`] that parses from an
+    /// in-memory string instead of a file, mirroring its parse-then-resolve
+    /// behavior for unit tests that don't want to touch the filesystem.
+    fn parse_and_resolve_str(input: &str) -> Result<Policy, ConfigError> {
+        let policy = parse_str(input)?;
+        policy.resolve().map_err(ConfigError::Resolve)?;
+        Ok(policy)
+    }
+
     #[test]
     fn parse_str_round_trips_through_resolve() {
         let policy = parse_str(
@@ -35,4 +54,17 @@ mod tests {
         assert_eq!(resolved.len(), 1);
         assert_eq!(resolved[0].port, 443);
     }
+
+    #[test]
+    fn parse_and_resolve_rejects_out_of_prefix_ownership() {
+        let cfg = "interface wan eth0\nipv4 203.0.113.0/24\ntenant t {\n owns 198.51.100.7\n}\n";
+        let err = parse_and_resolve_str(cfg).unwrap_err();
+        assert!(matches!(err, ConfigError::Resolve(_)));
+    }
+
+    #[test]
+    fn parse_and_resolve_ok_for_flow_only_config() {
+        let cfg = "interface wan eth0\nipv4 203.0.113.0/24\nshadow\n";
+        assert!(parse_and_resolve_str(cfg).is_ok());
+    }
 }
diff --git a/crates/blackwall-flow/src/collector_net.rs b/crates/blackwall-flow/src/collector_net.rs
index c813d77..095ae83 100644
--- a/crates/blackwall-flow/src/collector_net.rs
+++ b/crates/blackwall-flow/src/collector_net.rs
@@ -7,16 +7,20 @@ use crate::metrics::CollectorMetrics;
 use crate::sflow::decode_datagram;
 use crate::sink::MitigationSink;
 use std::net::SocketAddr;
-use std::sync::{Arc, Mutex};
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::sync::{Arc, Mutex, OnceLock};
+use std::time::Instant;
 use tokio::net::UdpSocket;
 
+/// Process-start baseline for the monotonic detector clock.
+fn clock_base() -> Instant {
+    static BASE: OnceLock<Instant> = OnceLock::new();
+    *BASE.get_or_init(Instant::now)
+}
+
+/// Milliseconds since process start (monotonic). Used for all detector windowing,
+/// eviction, and hold-down math — never affected by NTP/wall-clock steps.
 fn now_ms() -> u64 {
-    SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .ok()
-        .and_then(|d| u64::try_from(d.as_millis()).ok())
-        .unwrap_or(0)
+    u64::try_from(clock_base().elapsed().as_millis()).unwrap_or(u64::MAX)
 }
 
 /// Run the collector until the process ends. Binds `listen`, decodes each
@@ -24,8 +28,11 @@ fn now_ms() -> u64 {
 /// window and forwards events to `sink`. Decode errors are logged and skipped.
 ///
 /// When `metrics` is `Some`, the collector increments `datagrams` per received
-/// datagram and `decode_errors` per decode failure, and (after each tick)
-/// publishes the detector's cumulative unknown-agent observation count, for
+/// datagram, `decode_errors` per envelope-level decode failure (the whole
+/// datagram discarded), and `sample_decode_errors` once per malformed sample
+/// inside an otherwise-decoded datagram (the valid samples in that datagram
+/// are still kept), and (after each tick) publishes the detector's cumulative
+/// unknown-agent observation count and minimum-sample-suppressed count, for
 /// the `/metrics` endpoint. Callers with no metrics endpoint pass `None`.
 ///
 /// When `agent_snapshot` is `Some`, the collector overwrites it with
@@ -53,7 +60,10 @@ pub async fn run_collector(
                     Ok((n, _from)) => {
                         if let Some(m) = &metrics { m.incr_datagrams(); }
                         match decode_datagram(&buf[..n]) {
-                            Ok(observations) => {
+                            Ok((observations, sample_errors)) => {
+                                if let Some(m) = &metrics {
+                                    for _ in 0..sample_errors { m.incr_sample_decode_errors(); }
+                                }
                                 let t = now_ms();
                                 for o in &observations { detector.observe(o, t); }
                             }
@@ -73,6 +83,9 @@ pub async fn run_collector(
                 }
                 if let Some(m) = &metrics {
                     m.set_unknown_agent_observations(detector.unknown_agent_observations());
+                    m.set_min_sample_suppressed(detector.min_sample_suppressed());
+                    m.set_detections_opened(detector.detections_opened());
+                    m.set_detections_cleared(detector.detections_cleared());
                 }
                 for event in events {
                     sink.handle(&event).await;
diff --git a/crates/blackwall-flow/src/detector.rs b/crates/blackwall-flow/src/detector.rs
index 975da22..6d55b91 100644
--- a/crates/blackwall-flow/src/detector.rs
+++ b/crates/blackwall-flow/src/detector.rs
@@ -109,6 +109,24 @@ pub trait Detector {
     fn unknown_agent_observations(&self) -> u64 {
         0
     }
+
+    /// Count of detections suppressed solely by the minimum-sample gate;
+    /// default zero for detectors without such a gate.
+    fn min_sample_suppressed(&self) -> u64 {
+        0
+    }
+
+    /// Count of detections opened (`DetectionEvent::Opened`) since start;
+    /// default zero for detectors without such a notion.
+    fn detections_opened(&self) -> u64 {
+        0
+    }
+
+    /// Count of detections cleared (`DetectionEvent::Cleared`) since start;
+    /// default zero for detectors without such a notion.
+    fn detections_cleared(&self) -> u64 {
+        0
+    }
 }
 
 /// Per-POP telemetry snapshot for the metrics endpoint: one entry per agent
@@ -123,6 +141,12 @@ pub struct AgentStat {
     /// Count of samples from this agent whose reported sampling rate was
     /// clamped because it deviated far from the agent's expected rate.
     pub mismatches: u64,
+    /// Count of samples from this agent, among those whose reported rate
+    /// exceeded `expected * 4`, whose effective (post-clamp) rate landed at
+    /// or above half of the `expected * max_sampling_factor` ceiling — an
+    /// early-warning signal that the agent is close to (or already at) the
+    /// hard ceiling.
+    pub near_ceiling: u64,
 }
 
 // ---------------------------------------------------------------------------
@@ -151,6 +175,45 @@ struct DstState {
     last_over_ms: u64,
 }
 
+// ---------------------------------------------------------------------------
+// DetectorConfig
+// ---------------------------------------------------------------------------
+
+/// Scalar knobs for [`ThresholdDetector::new`], grouped into a struct so
+/// adding future knobs (e.g. D3's `max_sampling_factor`) doesn't push the
+/// constructor's arity past `clippy::too_many_arguments`.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub struct DetectorConfig {
+    /// Packets per second above which an attack is declared.
+    pub pps_threshold: f64,
+    /// Bits per second above which an attack is declared.
+    pub bps_threshold: f64,
+    /// Sliding window size in milliseconds for rate computation.
+    pub window_ms: u64,
+    /// Milliseconds below threshold before a detection is cleared.
+    pub hold_down_ms: u64,
+    /// Minimum raw in-window sample count before a detection may open. Guards
+    /// against sampling-variance false positives, where a tiny number of
+    /// samples at a high configured sampling rate extrapolate to an
+    /// over-threshold estimated rate despite carrying almost no statistical
+    /// weight. `0` disables the gate.
+    pub min_samples: usize,
+    /// Ceiling multiplier applied to an agent's expected sampling rate when
+    /// its reported rate is high: a reported rate above `expected * 4` is
+    /// trusted (adaptive samplers legitimately raise their rate under load)
+    /// up to `expected * max_sampling_factor`, and only clamped down beyond
+    /// that ceiling. Does not affect the low-side clamp (a reported rate
+    /// below `expected / 4` is always clamped up to `expected`).
+    ///
+    /// Values below 4 are treated as 4 (the trust band is incoherent below
+    /// the 4× trigger): [`ThresholdDetector::new`] floors the stored value,
+    /// so a misconfigured `0` (which would clamp every high-rate flood's
+    /// volume to zero) or `1..=3` (which would clamp a legitimately-high
+    /// reported rate back down, reintroducing the under-count this clamp
+    /// exists to prevent) can't mask an attack.
+    pub max_sampling_factor: u32,
+}
+
 // ---------------------------------------------------------------------------
 // ThresholdDetector
 // ---------------------------------------------------------------------------
@@ -163,11 +226,17 @@ pub struct ThresholdDetector {
     bps_threshold: f64,
     window_ms: u64,
     hold_down_ms: u64,
+    min_samples: usize,
+    max_sampling_factor: u32,
     state: HashMap<IpAddr, DstState>,
     agents: crate::agents::AgentRegistry,
     agent_last_seen: HashMap<IpAddr, u64>,
     sampling_mismatches: HashMap<IpAddr, u64>,
+    sampling_near_ceiling: HashMap<IpAddr, u64>,
     unknown_agent_observations: u64,
+    min_sample_suppressed: u64,
+    detections_opened: u64,
+    detections_cleared: u64,
 }
 
 impl ThresholdDetector {
@@ -176,31 +245,38 @@ impl ThresholdDetector {
     /// # Parameters
     ///
     /// - `prefixes` — only destinations within these prefixes are monitored.
-    /// - `pps_threshold` — packets per second above which an attack is declared.
-    /// - `bps_threshold` — bits per second above which an attack is declared.
-    /// - `window_ms` — sliding window size in milliseconds for rate computation.
-    /// - `hold_down_ms` — milliseconds below threshold before a detection is cleared.
+    /// - `config` — the scalar thresholds/timings/minimum-sample gate (see
+    ///   [`DetectorConfig`]).
     /// - `agents` — registry of known sFlow agents and their expected sampling
     ///   rates, used for liveness tracking and the sampling-sanity clamp.
     pub fn new(
         prefixes: Vec<IpNet>,
-        pps_threshold: f64,
-        bps_threshold: f64,
-        window_ms: u64,
-        hold_down_ms: u64,
+        config: DetectorConfig,
         agents: crate::agents::AgentRegistry,
     ) -> Self {
         Self {
             prefixes,
-            pps_threshold,
-            bps_threshold,
-            window_ms,
-            hold_down_ms,
+            pps_threshold: config.pps_threshold,
+            bps_threshold: config.bps_threshold,
+            window_ms: config.window_ms,
+            hold_down_ms: config.hold_down_ms,
+            min_samples: config.min_samples,
+            // Floor at 4: the trust band (see `DetectorConfig::max_sampling_factor`
+            // rustdoc) is only coherent when the ceiling is at least `expected * 4`,
+            // the same threshold that triggers the high-rate branch. A misconfigured
+            // 0 would collapse the ceiling to 0 (masking a real flood's volume); a
+            // misconfigured 1..=3 would clamp a legitimately-high reported rate back
+            // down, reintroducing the under-count this clamp exists to prevent.
+            max_sampling_factor: config.max_sampling_factor.max(4),
             state: HashMap::new(),
             agents,
             agent_last_seen: HashMap::new(),
             sampling_mismatches: HashMap::new(),
+            sampling_near_ceiling: HashMap::new(),
             unknown_agent_observations: 0,
+            min_sample_suppressed: 0,
+            detections_opened: 0,
+            detections_cleared: 0,
         }
     }
 
@@ -214,6 +290,38 @@ impl ThresholdDetector {
     pub fn sampling_mismatches(&self) -> &HashMap<IpAddr, u64> {
         &self.sampling_mismatches
     }
+
+    /// Count of samples per agent, among those whose reported rate exceeded
+    /// `expected * 4`, whose effective (post-clamp) rate landed at or above
+    /// half of the `expected * max_sampling_factor` ceiling — an
+    /// early-warning signal that the agent is close to (or already at) the
+    /// hard ceiling, whether from legitimate adaptive-sampler load or a
+    /// misconfiguration.
+    pub fn sampling_near_ceiling(&self) -> &HashMap<IpAddr, u64> {
+        &self.sampling_near_ceiling
+    }
+
+    /// Count of detections suppressed solely by the minimum-sample gate
+    /// (rate crossed threshold but the raw in-window sample count was below
+    /// `min_samples`). Surfaced as `blackwall_flow_min_sample_suppressed_total`.
+    #[must_use]
+    pub fn min_sample_suppressed(&self) -> u64 {
+        self.min_sample_suppressed
+    }
+
+    /// Count of detections opened (`DetectionEvent::Opened`) since start.
+    /// Surfaced as `blackwall_flow_detections_opened_total`.
+    #[must_use]
+    pub fn detections_opened(&self) -> u64 {
+        self.detections_opened
+    }
+
+    /// Count of detections cleared (`DetectionEvent::Cleared`) since start.
+    /// Surfaced as `blackwall_flow_detections_cleared_total`.
+    #[must_use]
+    pub fn detections_cleared(&self) -> u64 {
+        self.detections_cleared
+    }
 }
 
 impl Detector for ThresholdDetector {
@@ -229,14 +337,27 @@ impl Detector for ThresholdDetector {
             Some(expected) => {
                 self.agent_last_seen.insert(obs.agent, now_ms);
 
-                // Clamp an agent whose reported rate deviates far from its
-                // configured expected rate (guards the volume math + collector
-                // against a misconfigured POP), and count the mismatch.
+                // Clamp is DIRECTION-AWARE: a low reported rate deflates the
+                // volume estimate (suppression risk / misconfigured POP), so
+                // it's a hard floor at `expected`. A high reported rate
+                // inflates the estimate, but adaptive samplers legitimately
+                // raise their rate under load, so it's trusted up to a
+                // ceiling and only clamped down beyond that.
                 let lo = expected / 4;
-                let hi = expected.saturating_mul(4);
-                if obs.sampling_rate < lo || obs.sampling_rate > hi {
+                let ceiling = expected.saturating_mul(self.max_sampling_factor);
+                if obs.sampling_rate < lo {
+                    // Low N deflates volume (suppression / misconfigured POP): clamp UP.
                     *self.sampling_mismatches.entry(obs.agent).or_insert(0) += 1;
                     expected
+                } else if obs.sampling_rate > expected.saturating_mul(4) {
+                    // High N inflates, but adaptive samplers legitimately raise
+                    // N under load: trust it up to the ceiling, only clamp beyond.
+                    *self.sampling_mismatches.entry(obs.agent).or_insert(0) += 1;
+                    let trusted = obs.sampling_rate.min(ceiling);
+                    if trusted >= ceiling / 2 {
+                        *self.sampling_near_ceiling.entry(obs.agent).or_insert(0) += 1;
+                    }
+                    trusted
                 } else {
                     obs.sampling_rate
                 }
@@ -252,6 +373,9 @@ impl Detector for ThresholdDetector {
         }
 
         let est_packets = u64::from(effective_rate);
+        // `obs.frame_len` is the sFlow-reported L2 frame length (includes the
+        // Ethernet header), so `est_bytes` — and the resulting bps used against
+        // `bps_threshold` — is on an L2 basis, not L3 payload bytes.
         let est_bytes = u64::from(effective_rate) * u64::from(obs.frame_len);
 
         let entry = self.state.entry(obs.dst).or_insert_with(|| DstState {
@@ -277,6 +401,7 @@ impl Detector for ThresholdDetector {
         let pps_threshold = self.pps_threshold;
         let bps_threshold = self.bps_threshold;
         let hold_down_ms = self.hold_down_ms;
+        let min_samples = self.min_samples;
 
         #[expect(
             clippy::cast_precision_loss,
@@ -286,6 +411,17 @@ impl Detector for ThresholdDetector {
 
         let mut events = Vec::new();
         let mut to_remove = Vec::new();
+        // Accumulated locally and folded into `self.min_sample_suppressed`
+        // after the loop — `self.min_samples`/`self.min_sample_suppressed`
+        // field access would otherwise conflict with the `&mut self.state`
+        // borrow held by the loop below.
+        let mut suppressed: u64 = 0;
+        // Same reasoning as `suppressed` above: accumulated locally and
+        // folded into `self.detections_opened`/`self.detections_cleared`
+        // after the loop, to avoid conflicting with the `&mut self.state`
+        // borrow.
+        let mut opened: u64 = 0;
+        let mut cleared: u64 = 0;
 
         for (dst, state) in &mut self.state {
             // Evict samples outside the window.
@@ -299,6 +435,7 @@ impl Detector for ThresholdDetector {
                         at_ms: now_ms,
                     });
                     to_remove.push(*dst);
+                    cleared = cleared.saturating_add(1);
                 }
                 continue;
             }
@@ -326,7 +463,11 @@ impl Detector for ThresholdDetector {
             )]
             let bps = (total_bytes as f64) * 8.0 / window_secs;
 
-            let over_threshold = pps > pps_threshold || bps > bps_threshold;
+            let rate_over = pps > pps_threshold || bps > bps_threshold;
+            let over_threshold = rate_over && state.samples.len() >= min_samples;
+            if rate_over && !over_threshold {
+                suppressed = suppressed.saturating_add(1);
+            }
 
             if over_threshold {
                 state.last_over_ms = now_ms;
@@ -361,6 +502,7 @@ impl Detector for ThresholdDetector {
                         agents: &self.agents,
                     });
                     events.push(DetectionEvent::Opened(detection));
+                    opened = opened.saturating_add(1);
                 }
             } else if state.open {
                 // Under threshold — check hold-down.
@@ -370,6 +512,7 @@ impl Detector for ThresholdDetector {
                         at_ms: now_ms,
                     });
                     to_remove.push(*dst);
+                    cleared = cleared.saturating_add(1);
                 }
             }
         }
@@ -378,6 +521,10 @@ impl Detector for ThresholdDetector {
             self.state.remove(&dst);
         }
 
+        self.min_sample_suppressed = self.min_sample_suppressed.saturating_add(suppressed);
+        self.detections_opened = self.detections_opened.saturating_add(opened);
+        self.detections_cleared = self.detections_cleared.saturating_add(cleared);
+
         events
     }
 
@@ -388,6 +535,7 @@ impl Detector for ThresholdDetector {
                 pop: self.agents.name(addr).to_owned(),
                 last_seen_ms,
                 mismatches: self.sampling_mismatches.get(&addr).copied().unwrap_or(0),
+                near_ceiling: self.sampling_near_ceiling.get(&addr).copied().unwrap_or(0),
             })
             .collect()
     }
@@ -395,6 +543,18 @@ impl Detector for ThresholdDetector {
     fn unknown_agent_observations(&self) -> u64 {
         self.unknown_agent_observations
     }
+
+    fn min_sample_suppressed(&self) -> u64 {
+        self.min_sample_suppressed
+    }
+
+    fn detections_opened(&self) -> u64 {
+        self.detections_opened
+    }
+
+    fn detections_cleared(&self) -> u64 {
+        self.detections_cleared
+    }
 }
 
 // ---------------------------------------------------------------------------
@@ -582,10 +742,63 @@ mod tests {
         }
     }
 
+    /// Build a `FlowObservation` destined to a host in `203.0.113.0/24`, for the
+    /// monotonic-clock windowing regression test below. `ms` is accepted (and
+    /// named into the call site) purely for readability, pairing with the
+    /// `now_ms` argument passed separately to `observe`/`tick` — `FlowObservation`
+    /// itself carries no timestamp field.
+    fn obs_at(_ms: u64) -> FlowObservation {
+        obs([203, 0, 113, 7], [198, 51, 100, 9], 1, 100)
+    }
+
     fn agent_ip(o: u8) -> std::net::IpAddr {
         std::net::IpAddr::V4(std::net::Ipv4Addr::new(10, 222, 0, o))
     }
 
+    /// A `DetectorConfig` for tests: fixed `window_ms: 10_000, hold_down_ms:
+    /// 30_000` defaults, with the caller supplying the values that vary
+    /// per-test (thresholds + minimum-sample gate).
+    fn test_cfg(pps: f64, bps: f64, min_samples: usize) -> DetectorConfig {
+        test_cfg_factor(pps, bps, min_samples, 64)
+    }
+
+    /// Like [`test_cfg`], but with an explicit `max_sampling_factor` (the
+    /// ceiling multiplier applied to a high reported sampling rate).
+    fn test_cfg_factor(
+        pps: f64,
+        bps: f64,
+        min_samples: usize,
+        max_sampling_factor: u32,
+    ) -> DetectorConfig {
+        DetectorConfig {
+            pps_threshold: pps,
+            bps_threshold: bps,
+            window_ms: 10_000,
+            hold_down_ms: 30_000,
+            min_samples,
+            max_sampling_factor,
+        }
+    }
+
+    impl DetectorConfig {
+        /// Test-only builder for overriding `window_ms` after construction,
+        /// so call sites can read `test_cfg_factor(...).with_window_ms(...)`
+        /// without repeating every other field.
+        fn with_window_ms(mut self, window_ms: u64) -> Self {
+            self.window_ms = window_ms;
+            self
+        }
+    }
+
+    /// Build a `FlowObservation` destined to a host in `203.0.113.0/24` with
+    /// the given `sampling_rate`. `t_ms` is accepted (and named into the call
+    /// site) purely for readability at the call site, pairing with the
+    /// `now_ms` argument passed separately to `observe` — `FlowObservation`
+    /// itself carries no timestamp field.
+    fn obs_rate_at(rate: u32, _t_ms: u64) -> FlowObservation {
+        obs([203, 0, 113, 7], [198, 51, 100, 9], rate, 100)
+    }
+
     /// Test helper for agent-aware observations (distinct from `obs` above,
     /// which predates agent-awareness and is kept for the existing tests).
     fn agent_obs(
@@ -608,14 +821,37 @@ mod tests {
         }
     }
 
+    /// A one-agent `AgentRegistry` named `"ord"` with the given expected
+    /// `sampling` rate, for the direction-aware clamp tests below.
+    fn registry_with(agent: std::net::IpAddr, sampling: u32) -> AgentRegistry {
+        AgentRegistry::from_entries(&[PopEntry {
+            name: "ord".into(),
+            agent,
+            sampling,
+        }])
+    }
+
+    /// Build a `FlowObservation` from `agent` reporting `rate`, destined to a
+    /// host in `203.0.113.0/24`. `t_ms` is accepted (and named into the call
+    /// site) purely for readability, pairing with the `now_ms` argument
+    /// passed separately to `observe` — `FlowObservation` itself carries no
+    /// timestamp field.
+    fn obs_from(agent: std::net::IpAddr, rate: u32, _t_ms: u64) -> FlowObservation {
+        agent_obs(agent, "198.51.100.5", "203.0.113.9", rate, 100)
+    }
+
     fn detector() -> ThresholdDetector {
         // prefix 203.0.113.0/24; pps threshold 100k; bps very high; window 1s; hold-down 2s
         ThresholdDetector::new(
             vec!["203.0.113.0/24".parse().unwrap()],
-            100_000.0,
-            1e15,
-            1000,
-            2000,
+            DetectorConfig {
+                pps_threshold: 100_000.0,
+                bps_threshold: 1e15,
+                window_ms: 1000,
+                hold_down_ms: 2000,
+                min_samples: 0,
+                max_sampling_factor: 64,
+            },
             AgentRegistry::default(),
         )
     }
@@ -767,10 +1003,14 @@ mod tests {
         // window_ms = 0 must be clamped to 1 ms, not produce inf rates.
         let mut d = ThresholdDetector::new(
             vec!["203.0.113.0/24".parse().unwrap()],
-            1e15, // impossibly high pps threshold
-            1e15, // impossibly high bps threshold
-            0,    // zero window — the fix clamps this to 1ms
-            2000,
+            DetectorConfig {
+                pps_threshold: 1e15, // impossibly high pps threshold
+                bps_threshold: 1e15, // impossibly high bps threshold
+                window_ms: 0,        // zero window — the fix clamps this to 1ms
+                hold_down_ms: 2000,
+                min_samples: 0,
+                max_sampling_factor: 64,
+            },
             AgentRegistry::default(),
         );
         // A modest number of samples that should not cross 1e15 threshold.
@@ -790,10 +1030,14 @@ mod tests {
         // Set very high pps threshold but low bps threshold.
         let mut d = ThresholdDetector::new(
             vec!["203.0.113.0/24".parse().unwrap()],
-            1e12,   // pps impossibly high
-            1000.0, // bps very low
-            1000,
-            2000,
+            DetectorConfig {
+                pps_threshold: 1e12,   // pps impossibly high
+                bps_threshold: 1000.0, // bps very low
+                window_ms: 1000,
+                hold_down_ms: 2000,
+                min_samples: 0,
+                max_sampling_factor: 64,
+            },
             AgentRegistry::default(),
         );
         // 1 sample * rate=1 * frame_len=200 → est_bytes=200 → bps = 200*8/1 = 1600 > 1000
@@ -810,10 +1054,14 @@ mod tests {
         // With u128 saturating sums this must NOT panic and must produce a finite, very large rate.
         let mut d = ThresholdDetector::new(
             vec!["203.0.113.0/24".parse().unwrap()],
-            1.0, // very low pps threshold so a detection opens
-            1.0, // very low bps threshold
-            1000,
-            2000,
+            DetectorConfig {
+                pps_threshold: 1.0, // very low pps threshold so a detection opens
+                bps_threshold: 1.0, // very low bps threshold
+                window_ms: 1000,
+                hold_down_ms: 2000,
+                min_samples: 0,
+                max_sampling_factor: 64,
+            },
             AgentRegistry::default(),
         );
         let max_rate = u32::MAX;
@@ -853,10 +1101,14 @@ mod tests {
         let mk = || {
             ThresholdDetector::new(
                 vec!["203.0.113.0/24".parse().unwrap()],
-                1.0,
-                1.0,
-                window_ms,
-                30_000,
+                DetectorConfig {
+                    pps_threshold: 1.0,
+                    bps_threshold: 1.0,
+                    window_ms,
+                    hold_down_ms: 30_000,
+                    min_samples: 0,
+                    max_sampling_factor: 64,
+                },
                 AgentRegistry::from_entries(&[PopEntry {
                     name: "ord".into(),
                     agent: agent_ip(8),
@@ -898,6 +1150,88 @@ mod tests {
         assert_eq!(rogue_d.observed_pps, honest_d.observed_pps);
     }
 
+    #[test]
+    fn low_sampling_rate_clamps_up_to_expected() {
+        // reported N=100 (< expected/4=250) DEFLATES; clamp UP to expected=1000.
+        // window 1s, 1 sample → est_packets == effective_rate. Threshold 500:
+        //   if it used the reported 100 → 100 pps < 500 → no detection;
+        //   clamped to 1000 → 1000 pps > 500 → detection opens. Assert it opens.
+        let agents = registry_with(agent_ip(1), 1000);
+        let mut det = ThresholdDetector::new(
+            vec!["203.0.113.0/24".parse().unwrap()],
+            test_cfg_factor(500.0, 1e18, 1, 64).with_window_ms(1_000),
+            agents,
+        );
+        det.observe(&obs_from(agent_ip(1), 100, 1_000), 1_000); // rate=100, t=1000
+        assert!(det
+            .tick(1_100)
+            .iter()
+            .any(|e| matches!(e, DetectionEvent::Opened(_))));
+        assert_eq!(det.sampling_mismatches().get(&agent_ip(1)), Some(&1));
+    }
+
+    #[test]
+    fn high_sampling_rate_is_trusted_up_to_ceiling() {
+        // reported N=20_000 in (4000, 64000] → trusted. 1 sample, window 1s → 20_000 pps.
+        // threshold 10_000: trusted(20_000) crosses; a clamp-down to expected(1000) would not.
+        let agents = registry_with(agent_ip(1), 1000);
+        let mut det = ThresholdDetector::new(
+            vec!["203.0.113.0/24".parse().unwrap()],
+            test_cfg_factor(10_000.0, 1e18, 1, 64).with_window_ms(1_000),
+            agents,
+        );
+        det.observe(&obs_from(agent_ip(1), 20_000, 1_000), 1_000);
+        assert!(det
+            .tick(1_100)
+            .iter()
+            .any(|e| matches!(e, DetectionEvent::Opened(_))));
+        assert_eq!(det.sampling_mismatches().get(&agent_ip(1)), Some(&1));
+        assert_eq!(det.sampling_near_ceiling().get(&agent_ip(1)), None); // 20k < ceiling/2=32k
+    }
+
+    #[test]
+    fn very_high_sampling_rate_clamps_to_ceiling_and_flags_near() {
+        // reported N=500_000 > ceiling(64_000) → clamp to 64_000; ≥ ceiling/2 → near flag.
+        let agents = registry_with(agent_ip(1), 1000);
+        let mut det = ThresholdDetector::new(
+            vec!["203.0.113.0/24".parse().unwrap()],
+            test_cfg_factor(1e18, 1e30, 1, 64).with_window_ms(1_000), // thresholds huge: no detection needed
+            agents,
+        );
+        det.observe(&obs_from(agent_ip(1), 500_000, 1_000), 1_000);
+        let _ = det.tick(1_100);
+        assert_eq!(det.sampling_near_ceiling().get(&agent_ip(1)), Some(&1));
+    }
+
+    #[test]
+    fn max_sampling_factor_zero_does_not_mask_flood() {
+        // Misconfigured max_sampling_factor=0 must NOT collapse the ceiling to 0
+        // (which would clamp a real flood's volume down to 0 and mask the attack).
+        // The detector floors the effective factor at 4, so the ceiling is
+        // expected*4, and a reported rate above expected*4 is trusted up to that
+        // floor ceiling rather than being clamped to 0.
+        let agents = registry_with(agent_ip(1), 1000);
+        let mut det = ThresholdDetector::new(
+            vec!["203.0.113.0/24".parse().unwrap()],
+            DetectorConfig {
+                max_sampling_factor: 0,
+                ..test_cfg_factor(3_000.0, 1e18, 1, 0).with_window_ms(1_000)
+            },
+            agents,
+        );
+        // reported N=20_000 > expected*4=4_000 → high-rate branch. Without the
+        // floor, ceiling = expected*0 = 0, clamping volume to 0 pps (no detection,
+        // masking the flood). With the floor, ceiling = expected*4 = 4_000, trusted
+        // rate = min(20_000, 4_000) = 4_000 pps > pps_threshold(3_000) → opens.
+        det.observe(&obs_from(agent_ip(1), 20_000, 1_000), 1_000);
+        assert!(
+            det.tick(1_100)
+                .iter()
+                .any(|e| matches!(e, DetectionEvent::Opened(_))),
+            "max_sampling_factor=0 must not mask a real high-rate flood"
+        );
+    }
+
     #[test]
     fn tracks_agent_last_seen() {
         let reg = AgentRegistry::from_entries(&[PopEntry {
@@ -907,10 +1241,7 @@ mod tests {
         }]);
         let mut det = ThresholdDetector::new(
             vec!["203.0.113.0/24".parse().unwrap()],
-            1.0,
-            1.0,
-            10_000,
-            30_000,
+            test_cfg(1.0, 1.0, 0),
             reg,
         );
         det.observe(
@@ -932,10 +1263,7 @@ mod tests {
         }]);
         let mut det = ThresholdDetector::new(
             vec!["203.0.113.0/24".parse().unwrap()],
-            1.0,
-            1.0,
-            10_000,
-            30_000,
+            test_cfg(1.0, 1.0, 0),
             reg,
         );
 
@@ -967,10 +1295,7 @@ mod tests {
         ]);
         let mut det = ThresholdDetector::new(
             vec!["203.0.113.0/24".parse().unwrap()],
-            1.0,
-            1.0,
-            10_000,
-            30_000,
+            test_cfg(1.0, 1.0, 0),
             reg,
         );
 
@@ -998,10 +1323,7 @@ mod tests {
         }]);
         let mut det = ThresholdDetector::new(
             vec!["203.0.113.0/24".parse().unwrap()],
-            1.0,
-            1.0,
-            10_000,
-            30_000,
+            test_cfg(1.0, 1.0, 0),
             reg,
         );
         // Rogue rate (1) vs expected 1000 -> clamped, one mismatch recorded.
@@ -1026,10 +1348,7 @@ mod tests {
         }]);
         let mut det = ThresholdDetector::new(
             vec!["203.0.113.0/24".parse().unwrap()],
-            1.0,
-            1.0,
-            10_000,
-            30_000,
+            test_cfg(1.0, 1.0, 0),
             reg,
         );
         det.observe(
@@ -1048,10 +1367,7 @@ mod tests {
         }]);
         let mut det = ThresholdDetector::new(
             vec!["203.0.113.0/24".parse().unwrap()],
-            1.0,
-            1.0,
-            10_000,
-            30_000,
+            test_cfg(1.0, 1.0, 0),
             reg,
         );
         assert_eq!(det.unknown_agent_observations(), 0);
@@ -1091,10 +1407,7 @@ mod tests {
         ]);
         let mut det = ThresholdDetector::new(
             vec!["203.0.113.0/24".parse().unwrap()],
-            1.0,
-            1.0,
-            10_000,
-            30_000,
+            test_cfg(1.0, 1.0, 0),
             reg,
         );
         // Two POPs each see traffic to the same victim from the same /24.
@@ -1122,4 +1435,84 @@ mod tests {
             "198.51.100.0/24".parse::<ipnet::IpNet>().unwrap()
         );
     }
+
+    #[test]
+    fn detection_windowing_is_monotonic_not_wall_clock() {
+        // Two ticks 5s apart on a MONOTONIC scale must evict correctly even if the
+        // caller's wall clock jumped backward between them. The detector only sees the
+        // ms values it is handed; this documents that the collector must hand it a
+        // monotonic source (regression guard for the collector wiring).
+        let mut det = ThresholdDetector::new(
+            vec!["203.0.113.0/24".parse().unwrap()],
+            test_cfg(100.0, 1_000_000_000.0, 0),
+            crate::agents::AgentRegistry::from_entries(&[]),
+        );
+        // sample at t=1000 (monotonic), window 10s
+        det.observe(&obs_at(1_000), 1_000); // helper builds a FlowObservation for 203.0.113.7
+                                            // a wall-clock backstep would make a naive now() < 1000; monotonic keeps rising:
+        let events = det.tick(2_000); // still within window; no spurious clear
+        assert!(events
+            .iter()
+            .all(|e| !matches!(e, DetectionEvent::Cleared { .. })));
+    }
+
+    #[test]
+    fn min_sample_gate_blocks_variance_false_positive() {
+        // 2 samples @ 1-in-65536 extrapolate to ~131k pps (2 * 65536 / 1s) > 100k
+        // threshold, but with min_samples=8 the gate suppresses the detection and
+        // counts it. window_ms overridden to 1_000 (from test_cfg's 10_000 default)
+        // so the 1-second rate math matches; tick at 1_100 (not yet past the
+        // window) so both samples are still in-window when evaluated.
+        let mut det = ThresholdDetector::new(
+            vec!["203.0.113.0/24".parse().unwrap()],
+            DetectorConfig {
+                window_ms: 1_000,
+                ..test_cfg(100_000.0, 1e18, 8) // pps, bps, min_samples
+            },
+            crate::agents::AgentRegistry::from_entries(&[]),
+        );
+        det.observe(&obs_rate_at(65_536, 1_000), 1_000); // sampling_rate=65536, t=1000
+        det.observe(&obs_rate_at(65_536, 1_100), 1_100);
+        let events = det.tick(1_100);
+        assert!(events
+            .iter()
+            .all(|e| !matches!(e, DetectionEvent::Opened(_))));
+        assert_eq!(det.min_sample_suppressed(), 1);
+    }
+
+    #[test]
+    fn min_sample_gate_allows_real_flood() {
+        let mut det = ThresholdDetector::new(
+            vec!["203.0.113.0/24".parse().unwrap()],
+            test_cfg(100_000.0, 1e18, 8),
+            crate::agents::AgentRegistry::from_entries(&[]),
+        );
+        for i in 0..20 {
+            det.observe(&obs_rate_at(65_536, 1_000 + i), 1_000 + i); // 20 samples ≥ 8
+        }
+        let events = det.tick(2_000);
+        assert!(events
+            .iter()
+            .any(|e| matches!(e, DetectionEvent::Opened(_))));
+        assert_eq!(det.min_sample_suppressed(), 0);
+    }
+
+    #[test]
+    fn detections_opened_cleared_counters_track_events() {
+        let mut det = ThresholdDetector::new(
+            vec!["203.0.113.0/24".parse().unwrap()],
+            DetectorConfig {
+                hold_down_ms: 1_000,
+                ..test_cfg_factor(1.0, 1e18, 0, 64)
+            },
+            crate::agents::AgentRegistry::from_entries(&[]),
+        );
+        for i in 0..10 {
+            det.observe(&obs_rate_at(1000, 1_000 + i), 1_000 + i);
+        }
+        det.tick(2_000); // opens
+        assert_eq!(det.detections_opened(), 1);
+        det.tick(20_000); // window empty + past hold-down → clears
+        assert_eq!(det.detections_cleared(), 1);
+    }
 }
diff --git a/crates/blackwall-flow/src/lib.rs b/crates/blackwall-flow/src/lib.rs
index 11e7132..204d144 100644
--- a/crates/blackwall-flow/src/lib.rs
+++ b/crates/blackwall-flow/src/lib.rs
@@ -14,7 +14,8 @@ mod sink;
 pub use agents::AgentRegistry;
 pub use collector_net::run_collector;
 pub use detector::{
-    AgentStat, AttackKind, Detection, DetectionEvent, Detector, Severity, ThresholdDetector,
+    AgentStat, AttackKind, Detection, DetectionEvent, Detector, DetectorConfig, Severity,
+    ThresholdDetector,
 };
 pub use error::FlowError;
 pub use metrics::CollectorMetrics;
diff --git a/crates/blackwall-flow/src/metrics.rs b/crates/blackwall-flow/src/metrics.rs
index c5c38ef..56c6792 100644
--- a/crates/blackwall-flow/src/metrics.rs
+++ b/crates/blackwall-flow/src/metrics.rs
@@ -15,7 +15,11 @@ use std::sync::atomic::{AtomicU64, Ordering};
 pub struct CollectorMetrics {
     datagrams: AtomicU64,
     decode_errors: AtomicU64,
+    sample_decode_errors: AtomicU64,
     unknown_agent_observations: AtomicU64,
+    min_sample_suppressed: AtomicU64,
+    detections_opened: AtomicU64,
+    detections_cleared: AtomicU64,
 }
 
 impl CollectorMetrics {
@@ -47,6 +51,28 @@ impl CollectorMetrics {
         self.decode_errors.fetch_add(1, Ordering::Relaxed);
     }
 
+    /// Total individual flow/counter *samples* (within an otherwise-decoded
+    /// datagram) that failed to decode and were skipped.
+    ///
+    /// Distinct from [`Self::decode_errors`]: that counter tracks whole
+    /// datagrams whose envelope framing was unreadable (discarding every
+    /// observation in the datagram), while this one tracks samples inside a
+    /// datagram whose envelope decoded fine — e.g. hsflowd batches many flow
+    /// samples per datagram, and one malformed sample no longer discards the
+    /// rest. Keeping the two separate lets an operator tell "this POP is
+    /// sending malformed datagrams" (`decode_errors`) apart from "this POP is
+    /// dropping/corrupting individual samples" (`sample_decode_errors`).
+    #[must_use]
+    pub fn sample_decode_errors(&self) -> u64 {
+        self.sample_decode_errors.load(Ordering::Relaxed)
+    }
+
+    /// Record one sample that failed to decode within an otherwise-valid
+    /// datagram. See [`Self::sample_decode_errors`].
+    pub fn incr_sample_decode_errors(&self) {
+        self.sample_decode_errors.fetch_add(1, Ordering::Relaxed);
+    }
+
     /// Total sample observations attributed to agents absent from the POP
     /// registry (`Detector::unknown_agent_observations`).
     #[must_use]
@@ -65,6 +91,49 @@ impl CollectorMetrics {
         self.unknown_agent_observations
             .store(value, Ordering::Relaxed);
     }
+
+    /// Total detections suppressed solely by the detector's minimum-sample
+    /// gate (`Detector::min_sample_suppressed`).
+    #[must_use]
+    pub fn min_sample_suppressed(&self) -> u64 {
+        self.min_sample_suppressed.load(Ordering::Relaxed)
+    }
+
+    /// Publish the detector's current cumulative minimum-sample-suppressed
+    /// count. Mirrors [`Self::set_unknown_agent_observations`]: the detector
+    /// already accumulates this total internally
+    /// (`Detector::min_sample_suppressed`), so the collector calls this once
+    /// per tick with that total rather than incrementing per event.
+    pub fn set_min_sample_suppressed(&self, value: u64) {
+        self.min_sample_suppressed.store(value, Ordering::Relaxed);
+    }
+
+    /// Total detections opened (`Detector::detections_opened`).
+    #[must_use]
+    pub fn detections_opened(&self) -> u64 {
+        self.detections_opened.load(Ordering::Relaxed)
+    }
+
+    /// Publish the detector's current cumulative detections-opened count.
+    /// Mirrors [`Self::set_min_sample_suppressed`]: the detector already
+    /// accumulates this total internally (`Detector::detections_opened`), so
+    /// the collector calls this once per tick with that total rather than
+    /// incrementing per event.
+    pub fn set_detections_opened(&self, value: u64) {
+        self.detections_opened.store(value, Ordering::Relaxed);
+    }
+
+    /// Total detections cleared (`Detector::detections_cleared`).
+    #[must_use]
+    pub fn detections_cleared(&self) -> u64 {
+        self.detections_cleared.load(Ordering::Relaxed)
+    }
+
+    /// Publish the detector's current cumulative detections-cleared count.
+    /// Mirrors [`Self::set_detections_opened`].
+    pub fn set_detections_cleared(&self, value: u64) {
+        self.detections_cleared.store(value, Ordering::Relaxed);
+    }
 }
 
 #[cfg(test)]
@@ -76,7 +145,20 @@ mod tests {
         let m = CollectorMetrics::new();
         assert_eq!(m.datagrams(), 0);
         assert_eq!(m.decode_errors(), 0);
+        assert_eq!(m.sample_decode_errors(), 0);
         assert_eq!(m.unknown_agent_observations(), 0);
+        assert_eq!(m.min_sample_suppressed(), 0);
+        assert_eq!(m.detections_opened(), 0);
+        assert_eq!(m.detections_cleared(), 0);
+    }
+
+    #[test]
+    fn set_min_sample_suppressed_overwrites_rather_than_accumulates() {
+        let m = CollectorMetrics::new();
+        m.set_min_sample_suppressed(3);
+        assert_eq!(m.min_sample_suppressed(), 3);
+        m.set_min_sample_suppressed(5);
+        assert_eq!(m.min_sample_suppressed(), 5);
     }
 
     #[test]
@@ -89,6 +171,19 @@ mod tests {
         assert_eq!(m.unknown_agent_observations(), 5);
     }
 
+    #[test]
+    fn set_detections_opened_cleared_overwrite_rather_than_accumulate() {
+        let m = CollectorMetrics::new();
+        m.set_detections_opened(3);
+        m.set_detections_cleared(2);
+        assert_eq!(m.detections_opened(), 3);
+        assert_eq!(m.detections_cleared(), 2);
+        m.set_detections_opened(5);
+        m.set_detections_cleared(4);
+        assert_eq!(m.detections_opened(), 5);
+        assert_eq!(m.detections_cleared(), 4);
+    }
+
     #[test]
     fn increment_bumps_counters() {
         let m = CollectorMetrics::new();
@@ -99,6 +194,17 @@ mod tests {
         assert_eq!(m.decode_errors(), 1);
     }
 
+    #[test]
+    fn incr_sample_decode_errors_is_distinct_from_decode_errors() {
+        let m = CollectorMetrics::new();
+        m.incr_decode_errors();
+        m.incr_sample_decode_errors();
+        m.incr_sample_decode_errors();
+        m.incr_sample_decode_errors();
+        assert_eq!(m.decode_errors(), 1, "per-datagram counter unaffected");
+        assert_eq!(m.sample_decode_errors(), 3);
+    }
+
     #[test]
     fn default_matches_new() {
         let m = CollectorMetrics::default();
diff --git a/crates/blackwall-flow/src/select.rs b/crates/blackwall-flow/src/select.rs
index 9361e93..86e7f7b 100644
--- a/crates/blackwall-flow/src/select.rs
+++ b/crates/blackwall-flow/src/select.rs
@@ -45,10 +45,18 @@ pub struct SelectionConfig {
 /// fully offline. Otherwise (diffuse, no ports, or no protocol) falls back to RTBH.
 #[must_use]
 pub fn select(d: &Detection, cfg: &SelectionConfig) -> Mitigation {
-    if d.proto == 0 || d.top_ports.is_empty() {
+    if d.proto == 0 || d.top_ports.is_empty() || d.top_ports.iter().all(|(p, _)| *p == 0) {
+        return Mitigation::Rtbh;
+    }
+    let mut ports: Vec<(u16, f64)> = d
+        .top_ports
+        .iter()
+        .copied()
+        .filter(|(p, _)| *p != 0)
+        .collect();
+    if ports.is_empty() {
         return Mitigation::Rtbh;
     }
-    let mut ports = d.top_ports.clone();
     ports.sort_by(|a, b| b.1.total_cmp(&a.1)); // weight desc
     let mut cumulative = 0.0_f64;
     let mut chosen: Vec<u16> = Vec::new();
@@ -188,6 +196,32 @@ mod tests {
         assert_eq!(select(&det(0, vec![(53, 0.99)]), &cfg()), Mitigation::Rtbh);
     }
 
+    #[test]
+    fn all_port_zero_selects_rtbh() {
+        let m = select(&det(6, vec![(0, 1.0)]), &cfg());
+        assert!(matches!(m, Mitigation::Rtbh));
+    }
+
+    #[test]
+    fn mixed_port_zero_stripped_real_port_flowspec() {
+        // port 0 = 0.2, port 80 = 0.8 -> strip port 0, port 80 concentrates -> FlowSpec{80} only.
+        let m = select(&det(6, vec![(0, 0.2), (80, 0.8)]), &cfg());
+        match m {
+            Mitigation::FlowSpec(rules) => {
+                assert_eq!(rules.len(), 1);
+                assert_eq!(rules[0].dst_port, 80);
+            }
+            other => panic!("expected FlowSpec{{80}}, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn mixed_port_zero_dominant_falls_back_to_rtbh() {
+        // port 0 = 0.7, port 80 = 0.3 -> strip port 0; port 80 alone doesn't reach 0.8 -> RTBH.
+        let m = select(&det(6, vec![(0, 0.7), (80, 0.3)]), &cfg());
+        assert!(matches!(m, Mitigation::Rtbh));
+    }
+
     #[test]
     fn respects_max_flows_cap() {
         // 5 ports each 0.19 reach 0.8 only at the 5th, but max_flows=4 -> diffuse.
diff --git a/crates/blackwall-flow/src/sflow.rs b/crates/blackwall-flow/src/sflow.rs
index a1dccc1..8e2bb6b 100644
--- a/crates/blackwall-flow/src/sflow.rs
+++ b/crates/blackwall-flow/src/sflow.rs
@@ -50,20 +50,36 @@ impl<'a> Cursor<'a> {
     }
 }
 
-/// Decode an sFlow v5 UDP payload into a list of [`FlowObservation`]s.
+/// Decode an sFlow v5 UDP payload into a list of [`FlowObservation`]s, plus a
+/// count of samples that failed to decode.
 ///
 /// Each raw Ethernet header record inside a flow sample that can be parsed
 /// to an IP packet yields one observation.  Counter samples, non-Ethernet
 /// header protocols, and headers that etherparse cannot decode to an IP
 /// layer are silently skipped.
 ///
+/// hsflowd (and other real agents) batch many flow samples into one
+/// datagram; one structurally malformed sample must not discard the valid
+/// observations already decoded from the same datagram. The outer per-sample
+/// framing (`sample_type`/`sample_length`) is read and validated first — once
+/// that succeeds, the exact bounds of the sample body are known regardless of
+/// what is inside it, so a failure *within* a sample (e.g. a record length
+/// past the sample body) cannot desynchronize parsing of the samples that
+/// follow. Such per-sample/record errors are therefore caught, counted in the
+/// returned `u32`, and skipped — decoding continues with the next sample.
+///
 /// # Errors
 ///
-/// Returns [`FlowError::Decode`] only when the datagram is structurally
-/// truncated (a length field would read past the end of the buffer).
-pub fn decode_datagram(bytes: &[u8]) -> Result<Vec<FlowObservation>, FlowError> {
+/// Returns [`FlowError::Decode`] only for **envelope-level** framing errors:
+/// the fixed datagram header (version, agent address, sample count) or a
+/// per-sample header (`sample_type`/`sample_length`) running past the end of
+/// the buffer. These corrupt the cursor position itself, so — unlike a bad
+/// record inside an already-bounded sample body — decoding cannot safely
+/// continue past them.
+pub fn decode_datagram(bytes: &[u8]) -> Result<(Vec<FlowObservation>, u32), FlowError> {
     let mut cur = Cursor::new(bytes);
     let mut observations = Vec::new();
+    let mut sample_errors: u32 = 0;
 
     // Datagram envelope.
     let _version = cur.read_u32()?;
@@ -98,15 +114,21 @@ pub fn decode_datagram(bytes: &[u8]) -> Result<Vec<FlowObservation>, FlowError>
 
         // Flow samples: regular (format 1) and expanded (format 3, used by
         // real agents such as hsflowd). Other sample types (counters) are
-        // skipped.
-        match sample_type & 0xFFF {
-            1 => decode_flow_sample(sample_body, agent, &mut observations)?,
-            3 => decode_expanded_flow_sample(sample_body, agent, &mut observations)?,
+        // skipped. A decode error inside a sample is caught and counted
+        // rather than propagated: the outer `take` above already bounded
+        // this sample's exact extent, so the cursor is safe to continue with
+        // the next sample regardless of what went wrong inside this one.
+        let result = match sample_type & 0xFFF {
+            1 => decode_flow_sample(sample_body, agent, &mut observations),
+            3 => decode_expanded_flow_sample(sample_body, agent, &mut observations),
             _ => continue,
+        };
+        if result.is_err() {
+            sample_errors += 1;
         }
     }
 
-    Ok(observations)
+    Ok((observations, sample_errors))
 }
 
 /// Parse a flow-sample body and append any decoded observations.
@@ -436,7 +458,8 @@ mod tests {
     fn decodes_flow_sample_ipv4_udp() {
         let header = sample_eth_ipv4_udp();
         let dg = sflow_datagram(&header, 1024);
-        let obs = decode_datagram(&dg).unwrap();
+        let (obs, sample_errors) = decode_datagram(&dg).unwrap();
+        assert_eq!(sample_errors, 0);
         assert_eq!(obs.len(), 1);
         let o = obs[0];
         assert_eq!(o.src, IpAddr::V4(Ipv4Addr::new(198, 51, 100, 9)));
@@ -453,7 +476,8 @@ mod tests {
         let header = sample_eth_ipv4_udp();
         // Expanded (type 3) — must produce one observation identical to type 1.
         let dg_expanded = sflow_datagram_expanded(&header, 512);
-        let obs = decode_datagram(&dg_expanded).unwrap();
+        let (obs, sample_errors) = decode_datagram(&dg_expanded).unwrap();
+        assert_eq!(sample_errors, 0);
         assert_eq!(obs.len(), 1, "expanded flow sample yields one observation");
         let o = obs[0];
         assert_eq!(o.dst, IpAddr::V4(Ipv4Addr::new(203, 0, 113, 7)));
@@ -463,7 +487,8 @@ mod tests {
 
         // Regular (type 1) — regression: must still decode after the dispatch change.
         let dg_regular = sflow_datagram(&header, 256);
-        let obs_reg = decode_datagram(&dg_regular).unwrap();
+        let (obs_reg, sample_errors_reg) = decode_datagram(&dg_regular).unwrap();
+        assert_eq!(sample_errors_reg, 0);
         assert_eq!(
             obs_reg.len(),
             1,
@@ -492,13 +517,20 @@ mod tests {
         dd.extend_from_slice(&1u32.to_be_bytes()); // seq
         dd.extend_from_slice(&1000u32.to_be_bytes()); // uptime
         dd.extend_from_slice(&0u32.to_be_bytes()); // num_samples = 0
-        assert!(decode_datagram(&dd).unwrap().is_empty());
+        let (obs, sample_errors) = decode_datagram(&dd).unwrap();
+        assert!(obs.is_empty());
+        assert_eq!(sample_errors, 0);
     }
 
     #[test]
     fn inner_record_length_past_sample_errors() {
         // Build a datagram whose flow sample has num_records=1, record_length=9999
-        // (far past the sample body end) — must propagate as Err.
+        // (far past the sample body end). The outer per-sample framing
+        // (sample_type/sample_length) is intact — only the record *inside*
+        // the sample is malformed — so this is a per-sample error, not an
+        // envelope-level one: the datagram must still decode (0 valid
+        // observations from the bad sample, 1 sample-error counted), not
+        // propagate as `Err`.
         let mut d = Vec::new();
         d.extend_from_slice(&be(5)); // version
         d.extend_from_slice(&be(1)); // agent type ipv4
@@ -525,7 +557,63 @@ mod tests {
         d.extend_from_slice(&be(u32::try_from(flow.len()).unwrap()));
         d.extend_from_slice(&flow);
 
-        assert!(decode_datagram(&d).is_err());
+        let (obs, sample_errors) = decode_datagram(&d).expect("envelope framing is intact");
+        assert!(obs.is_empty(), "the malformed record yields no observation");
+        assert_eq!(sample_errors, 1, "the malformed sample is counted");
+    }
+
+    #[test]
+    fn one_bad_sample_does_not_discard_the_datagram() {
+        // A datagram with [valid flow sample, malformed flow sample] must yield
+        // the valid observation (not lose the whole datagram) plus a
+        // sample-error count of 1 for the malformed one.
+        let header = sample_eth_ipv4_udp();
+        let valid_flow = flow_sample_body(&header, 1024);
+
+        // Malformed flow sample: num_records=1, record_length=9999 (far past
+        // the sample body) — a per-sample/record structural error.
+        let mut malformed_flow = Vec::new();
+        malformed_flow.extend_from_slice(&be(1)); // flow seq
+        malformed_flow.extend_from_slice(&be(0)); // source_id
+        malformed_flow.extend_from_slice(&be(1)); // sampling_rate
+        malformed_flow.extend_from_slice(&be(0)); // sample_pool
+        malformed_flow.extend_from_slice(&be(0)); // drops
+        malformed_flow.extend_from_slice(&be(0)); // input
+        malformed_flow.extend_from_slice(&be(0)); // output
+        malformed_flow.extend_from_slice(&be(1)); // num_records
+        malformed_flow.extend_from_slice(&be(1)); // record_type = raw header
+        malformed_flow.extend_from_slice(&be(9999)); // record_length — far past body
+
+        let mut d = Vec::new();
+        d.extend_from_slice(&be(5)); // version
+        d.extend_from_slice(&be(1)); // agent type ipv4
+        d.extend_from_slice(&[10, 0, 0, 1]); // agent addr
+        d.extend_from_slice(&be(0)); // sub agent
+        d.extend_from_slice(&be(1)); // seq
+        d.extend_from_slice(&be(1000)); // uptime
+        d.extend_from_slice(&be(2)); // num_samples = 2
+
+        d.extend_from_slice(&be(1)); // sample_type = flow sample (valid)
+        d.extend_from_slice(&be(u32::try_from(valid_flow.len()).unwrap()));
+        d.extend_from_slice(&valid_flow);
+
+        d.extend_from_slice(&be(1)); // sample_type = flow sample (malformed)
+        d.extend_from_slice(&be(u32::try_from(malformed_flow.len()).unwrap()));
+        d.extend_from_slice(&malformed_flow);
+
+        let (obs, sample_errors) = decode_datagram(&d).expect("envelope ok");
+        assert_eq!(obs.len(), 1, "the valid sample's observation survives");
+        assert_eq!(
+            sample_errors, 1,
+            "the malformed sample is counted, not propagated as Err"
+        );
+    }
+
+    #[test]
+    fn envelope_truncation_still_errors() {
+        // A datagram cut off inside the fixed envelope header (before even the
+        // sample count can be read) is a framing error and must still be Err.
+        assert!(decode_datagram(&[0, 0, 0, 5]).is_err());
     }
 
     #[test]
@@ -548,7 +636,8 @@ mod tests {
         // Build a datagram with agent 10.222.3.8 (addr type 1) containing one
         // raw-header flow sample (reuse the existing sample-building helper).
         let dg = build_test_datagram_v4_agent([10, 222, 3, 8]);
-        let obs = decode_datagram(&dg).unwrap();
+        let (obs, sample_errors) = decode_datagram(&dg).unwrap();
+        assert_eq!(sample_errors, 0);
         assert!(!obs.is_empty());
         assert!(obs
             .iter()
@@ -559,7 +648,8 @@ mod tests {
     fn decodes_agent_address_v6() {
         let dg =
             build_test_datagram_v6_agent([0x2a, 0x12, 0x9b, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 8]);
-        let obs = decode_datagram(&dg).unwrap();
+        let (obs, sample_errors) = decode_datagram(&dg).unwrap();
+        assert_eq!(sample_errors, 0);
         assert!(!obs.is_empty());
         assert!(obs.iter().all(|o| matches!(o.agent, IpAddr::V6(_))));
     }
diff --git a/crates/blackwall-flow/tests/decode_real.rs b/crates/blackwall-flow/tests/decode_real.rs
index 13f87b6..e1d7d4b 100644
--- a/crates/blackwall-flow/tests/decode_real.rs
+++ b/crates/blackwall-flow/tests/decode_real.rs
@@ -23,7 +23,8 @@ fn decodes_real_hsflowd_expanded_flow_samples() {
     let victim = IpAddr::V4(Ipv4Addr::new(10, 9, 0, 2));
     let mut total = 0;
     for (bytes, expected) in FIXTURES {
-        let obs = decode_datagram(bytes).expect("decode real hsflowd datagram");
+        let (obs, sample_errors) = decode_datagram(bytes).expect("decode real hsflowd datagram");
+        assert_eq!(sample_errors, 0, "real captures have no malformed samples");
         assert_eq!(obs.len(), *expected, "observation count for a fixture");
         for o in &obs {
             assert_eq!(o.dst, victim, "all flood packets target the victim");
diff --git a/crates/blackwall-flow/tests/interop.rs b/crates/blackwall-flow/tests/interop.rs
index bc6e74f..0603642 100644
--- a/crates/blackwall-flow/tests/interop.rs
+++ b/crates/blackwall-flow/tests/interop.rs
@@ -7,7 +7,7 @@
 
 use async_trait::async_trait;
 use blackwall_flow::{
-    run_collector, AgentRegistry, DetectionEvent, MitigationSink, ThresholdDetector,
+    run_collector, AgentRegistry, DetectionEvent, DetectorConfig, MitigationSink, ThresholdDetector,
 };
 use std::net::SocketAddr;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -102,10 +102,14 @@ async fn detects_volumetric_attack() {
     // prefix 203.0.113.0/24; pps 100k; bps effectively off; window 1s; hold-down 2s.
     let detector = ThresholdDetector::new(
         vec!["203.0.113.0/24".parse().unwrap()],
-        100_000.0,
-        1e15,
-        1000,
-        2000,
+        DetectorConfig {
+            pps_threshold: 100_000.0,
+            bps_threshold: 1e15,
+            window_ms: 1000,
+            hold_down_ms: 2000,
+            min_samples: 0,
+            max_sampling_factor: 64,
+        },
         AgentRegistry::default(),
     );
     let collector = tokio::spawn(run_collector(
@@ -145,10 +149,14 @@ async fn detects_live_sflow_attack() {
     // Monitor the victim's prefix; threshold below the lab flood's estimated pps.
     let detector = Box::new(ThresholdDetector::new(
         vec!["10.0.0.0/30".parse().expect("prefix")],
-        20_000.0,      // pps; pinned in Task 5 validation
-        f64::INFINITY, // bps not gated here
-        1000,          // window_ms
-        2000,          // hold_down_ms
+        DetectorConfig {
+            pps_threshold: 20_000.0,      // pps; pinned in Task 5 validation
+            bps_threshold: f64::INFINITY, // bps not gated here
+            window_ms: 1000,
+            hold_down_ms: 2000,
+            min_samples: 0,
+            max_sampling_factor: 64,
+        },
         AgentRegistry::default(),
     ));
     let sink = Arc::new(CountingSink::default());