diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7eeca6e..39fcc5e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,6 +15,7 @@ All notable changes to this project are documented here, following
 - nftables rendering bound the `prerouting` filter chain to the managed interface (`type filter hook prerouting … device`), which the kernel rejects — only ingress/egress chains may bind a device — so `blackwall-nft::apply` failed on any real ruleset. The chain is now unbound and classification is scoped per-rule with an `iifname` match, the correct pattern for a prerouting filter chain. Because an unbound chain runs for every interface, the closed posture (`default_state` closed) is now enforced by an explicit interface-scoped terminal `drop` rule instead of a chain-wide drop policy, so it no longer black-holes loopback or other-interface host traffic. Caught by the new deception↔scanner lab gate, the first end-to-end run of `apply` against a real `nft`.
 
 ### Added
+- FlowSpec auto-mitigation wiring (sub-project C, C2b-2): makes the C2b-1 auto-mitigation core durable and operable, completing C2b. Announced FlowSpec rules are persisted to PostgreSQL (`flowspec_rules`) and **re-announced on daemon restart**, mirroring RTBH; `blackwalld flow` now runs the FlowSpec manager alongside the RTBH manager off the same iBGP session, so a concentrated attack is auto-dropped per-flow and a diffuse one still blackholes — the full detection→selection→mitigation path in production, not just the lab. A new `flowspec` config directive (`concentration`, `max-flows`, `rate`, `max-rules`, `hold-down`, `ttl`) enables it and reuses the `rtbh` block's BGP peer + eligibility prefixes (rejected at parse time if no `rtbh` block is present). A `blackwalld flowspec add/remove/list` operator CLI records intent to an append-only `flowspec_requests` log (with `created_by`) that the daemon drains and applies, mirroring `rtbh`. `blackwall-rtbh` stays DB-free via the `FlowSpecJournal` trait it defines and `blackwall-state` implements.
 - FlowSpec auto-mitigation core (sub-project C, C2b-1): a detected attack now automatically becomes the *right* mitigation. A pure concentration-based selection inspects the detection's fingerprint (`proto` + `top_ports`): when a small set of flows carries most of the attack it emits FlowSpec drop rules (*drop UDP dport 53 → victim/32*), leaving the victim's other services up; when the attack is diffuse across many ports it falls back to RTBH's whole-IP blackhole (which FlowSpec can't scope). A `FlowSpecController` + single-owner `FlowSpecManager` mirror the RTBH controller/manager (eligibility, concurrent-rule cap, hold-down + deferred-withdraw + TTL, origin, mirror self-heal) and share the one iBGP session via an extended `BgpExecutor`; a `SelectorSink` routes each detection to the FlowSpec or RTBH manager. A new `flowspec-auto-bird` lab gate proves both paths end to end against real **BIRD2**: a concentrated detection installs the flow rule in `flow4tab`, and a diffuse one installs the `/32` blackhole (community `65535:666`). Persistence, the `flowspec` config directive, the `blackwalld flowspec` operator CLI, and the daemon wiring are C2b-2 follow-ons.
 - FlowSpec codec (sub-project C, C2a): finer-grained BGP mitigation than RTBH — drop (or rate-limit) *just the attack flow* instead of null-routing the whole victim IP. A new byte-exact `blackwall-bgp` FlowSpec codec encodes rules for both families (RFC 8955 v4 + RFC 8956 v6, SAFI 133): the minimal DDoS match set — destination-prefix, IP-protocol, destination-port (with the RFC 8955 numeric operator-value encoding) — carrying a traffic-rate action (rate `0.0` = discard, `N` = rate-limit). The OPEN now advertises the FlowSpec capabilities, and the iBGP session gains a FlowSpec inject path (announce/withdraw, re-announced on reconnect like unicast routes). A new `flowspec-bird` lab gate proves an injected rule (`drop UDP dport 53 → 203.0.113.7/32`) reaches real **BIRD2** and validates + installs into its `flow4` table with the full match (`dst 203.0.113.7/32; proto 17; dport 53`). The controller/policy layer, detector auto-trigger, operator CLI, persistence, and richer match components/actions are C2b follow-ons.
 - RTBH control plane (sub-project C, C1c): makes RTBH durable, production-wired, and operable. Active blackholes are persisted to PostgreSQL (`rtbh_blackholes`) and **re-announced on daemon restart** (a crash never drops protection); `blackwalld flow` now auto-blackholes detected attacks through the live C1a BGP speaker, so the D→C link runs in production, not just the lab. A new `blackwalld rtbh add/remove/list` operator CLI records intent to an append-only `rtbh_requests` log (with `created_by` for attribution) that the daemon drains and applies. Internally, a single-owner `RtbhManager` task owns the controller + BGP session and is the sole reconciler (no shared-lock race; removals are explicit events, never inferred from DB absence), superseding the C1b `RtbhSink`. RTBH peering + policy is configured via a new `rtbh` config directive (BGP peer/ASN/router-id, blackhole community, per-family next-hops, cap, hold-down, TTL). `blackwall-rtbh` stays free of a DB dependency via `BgpExecutor`/`BlackholeJournal` traits it defines and `blackwall-bgp`/`blackwall-state` implement. The `rtbh-bird` lab gate now proves both an auto-detected and an operator-manual `/32` (community `65535:666`) reach real BIRD.
diff --git a/Cargo.lock b/Cargo.lock
index 1163dc7..b3b1c21 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -278,6 +278,7 @@ name = "blackwall-state"
 version = "0.1.0"
 dependencies = [
  "async-trait",
+ "blackwall-bgp",
  "blackwall-core",
  "blackwall-flow",
  "blackwall-rtbh",
@@ -321,6 +322,7 @@ dependencies = [
  "blackwall-speedtest",
  "blackwall-state",
  "clap",
+ "ipnet",
  "notify",
  "serde_json",
  "tokio",
diff --git a/README.md b/README.md
index ebe3151..24a9270 100644
--- a/README.md
+++ b/README.md
@@ -203,7 +203,9 @@ limiting. The DDoS-lab traffic-generation foundation (`blackwall-trafficgen`) is
 - ✅ **C2a — FlowSpec codec:** RFC 8955/8956 (SAFI 133) rule encoding + speaker inject path.
 - ✅ **C2b-1 — Auto-mitigation core:** concentration-based selection routes a detection to a
   flow-scoped FlowSpec drop (concentrated) or an RTBH blackhole (diffuse).
-- ⏳ **C2b-2 — FlowSpec control plane:** persistence, `flowspec` config, operator CLI, daemon wiring.
+- ✅ **C2b-2 — FlowSpec control plane:** persistence (`flowspec_rules`, re-announced on restart),
+  `flowspec` config directive, `blackwalld flowspec` operator CLI, and daemon wiring — the FlowSpec
+  manager runs alongside RTBH off the shared iBGP session, driven by the selector.
 - ⏳ **C3 — Looking-glass, C4 — auto-peering, scrubbing, dn42** *(later).*
 
 **D — Detection & telemetry** *(in progress)*
diff --git a/bin/blackwalld/Cargo.toml b/bin/blackwalld/Cargo.toml
index bec4b7e..f4c2c7f 100644
--- a/bin/blackwalld/Cargo.toml
+++ b/bin/blackwalld/Cargo.toml
@@ -19,6 +19,7 @@ blackwall-speedtest = { path = "../../crates/blackwall-speedtest" }
 blackwall-bgp = { path = "../../crates/blackwall-bgp" }
 blackwall-rtbh = { path = "../../crates/blackwall-rtbh" }
 clap = { workspace = true }
+ipnet = { workspace = true }
 serde_json = { workspace = true }
 tokio = { workspace = true, features = ["sync"] }
 tracing = { workspace = true }
diff --git a/bin/blackwalld/src/main.rs b/bin/blackwalld/src/main.rs
index 769647e..3e272fb 100644
--- a/bin/blackwalld/src/main.rs
+++ b/bin/blackwalld/src/main.rs
@@ -126,6 +126,17 @@ enum Command {
         #[command(subcommand)]
         action: RtbhCmd,
     },
+    /// Queue FlowSpec operator intent, or inspect the mirror / intent queue.
+    ///
+    /// Like `rtbh`, the CLI never talks to BGP directly: it only appends to
+    /// (or reads) the `flowspec_requests`/`flowspec_rules` tables. A running
+    /// `blackwalld flow` daemon (with `rtbh` + `flowspec` config blocks) is the
+    /// sole applier of intent.
+    Flowspec {
+        /// Which FlowSpec action to perform.
+        #[command(subcommand)]
+        action: FlowspecCmd,
+    },
 }
 
 /// Operator actions for the `rtbh` subcommand.
@@ -159,6 +170,47 @@ enum RtbhCmd {
     },
 }
 
+/// Operator actions for the `flowspec` subcommand.
+#[derive(Subcommand)]
+enum FlowspecCmd {
+    /// Queue a FlowSpec-add request for the flow `ip proto port`.
+    ///
+    /// Rejected up front (before any database connection) if `ip` falls
+    /// outside the config's FlowSpec-eligible prefixes.
+    Add {
+        /// The victim address to rate-limit.
+        ip: IpAddr,
+        /// IP protocol number (e.g. 17 = UDP, 6 = TCP).
+        proto: u8,
+        /// Destination port.
+        port: u16,
+        /// Traffic-rate action in bytes/sec; `0.0` = drop.
+        #[arg(long, default_value_t = 0.0)]
+        rate: f32,
+        /// Path to the Blackwall config file (must contain a `flowspec` block).
+        #[arg(long)]
+        config: PathBuf,
+        /// Attribution for the request; defaults to `$USER@<hostname>`.
+        #[arg(long)]
+        operator: Option<String>,
+    },
+    /// Queue a FlowSpec-remove request for the flow `ip proto port`.
+    Remove {
+        /// The victim address to stop rate-limiting.
+        ip: IpAddr,
+        /// IP protocol number.
+        proto: u8,
+        /// Destination port.
+        port: u16,
+    },
+    /// List the announced FlowSpec mirror (and, optionally, the intent queue).
+    List {
+        /// Also print the `flowspec_requests` operator intent queue.
+        #[arg(long)]
+        requests: bool,
+    },
+}
+
 #[tokio::main]
 async fn main() -> ExitCode {
     tracing_subscriber::fmt::init();
@@ -417,6 +469,33 @@ fn rtbh_config_from(
     }
 }
 
+/// Build a [`blackwall_rtbh::FlowSpecConfig`] from the policy's prefixes and the
+/// config's `flowspec` block. Shared by the CLI's eligibility pre-check and the
+/// `flow` daemon's manager wiring, so both agree on what is eligible. FlowSpec
+/// reuses `Policy.prefixes` for eligibility (no separate next-hop/peer fields).
+fn flowspec_config_from(
+    policy: &blackwall_core::Policy,
+    fs: &blackwall_core::FlowSpecPolicy,
+) -> blackwall_rtbh::FlowSpecConfig {
+    blackwall_rtbh::FlowSpecConfig {
+        eligible_prefixes: policy.prefixes.clone(),
+        max_rules: fs.max_rules,
+        hold_down: fs.hold_down,
+        max_ttl: fs.max_ttl,
+    }
+}
+
+/// Construct a host route (`/32` for IPv4, `/128` for IPv6) for `target`.
+///
+/// Local mirror of `blackwall_rtbh`'s crate-private `host_prefix`, used to
+/// rebuild a FlowSpec destination match from a stored victim address.
+fn host_prefix(target: IpAddr) -> ipnet::IpNet {
+    match target {
+        IpAddr::V4(a) => ipnet::IpNet::V4(ipnet::Ipv4Net::new(a, 32).expect("v4 /32")),
+        IpAddr::V6(a) => ipnet::IpNet::V6(ipnet::Ipv6Net::new(a, 128).expect("v6 /128")),
+    }
+}
+
 /// Single-owner RTBH reconcile loop: applies auto detection events as they
 /// arrive on `rx`, and on a 1 s tick, calls [`blackwall_rtbh::RtbhManager::tick`]
 /// (completing deferred clears/TTL expiries — mandatory, see the module docs)
@@ -533,6 +612,127 @@ async fn apply_request(
     }
 }
 
+/// Single-owner FlowSpec reconcile loop: the FlowSpec analogue of
+/// [`rtbh_manager_task`]. Applies auto mitigation events as they arrive on
+/// `rx` (an `Open`/`Update`/`Clear` from the collector's [`SelectorSink`]),
+/// and on a 1 s tick calls [`blackwall_rtbh::FlowSpecManager::tick`] (deferred
+/// clears / TTL expiry — mandatory) then drains every `status = 'pending'` row
+/// from `flowspec_requests` via [`apply_flowspec_request`].
+///
+/// Runs until `rx` is closed (i.e. for the process's lifetime, since the
+/// paired `SelectorSink`'s sender is held by the running collector).
+async fn flowspec_manager_task(
+    mut manager: blackwall_rtbh::FlowSpecManager<blackwall_bgp::BgpHandle, blackwall_state::Store>,
+    mut rx: mpsc::Receiver<blackwall_flow::FlowMitigationEvent>,
+    request_store: std::sync::Arc<blackwall_state::Store>,
+) {
+    let mut ticker = tokio::time::interval(std::time::Duration::from_secs(1));
+    loop {
+        tokio::select! {
+            maybe_ev = rx.recv() => {
+                match maybe_ev {
+                    Some(blackwall_flow::FlowMitigationEvent::Open { target, rules }) => {
+                        manager.apply_open(target, &rules, mono_now(), wall_now()).await;
+                    }
+                    Some(blackwall_flow::FlowMitigationEvent::Update { target }) => {
+                        // `apply_updated` is synchronous (in-memory refresh only).
+                        manager.apply_updated(target, mono_now());
+                    }
+                    Some(blackwall_flow::FlowMitigationEvent::Clear { target }) => {
+                        manager.apply_clear(target, mono_now(), wall_now()).await;
+                    }
+                    None => {
+                        tracing::warn!("FlowSpec: event channel closed; manager task exiting");
+                        return;
+                    }
+                }
+            }
+            _ = ticker.tick() => {
+                // Mandatory: completes deferred clears / TTL expiry.
+                manager.tick(mono_now(), wall_now()).await;
+
+                match request_store.pending_flowspec_requests().await {
+                    Ok(reqs) => {
+                        for req in reqs {
+                            apply_flowspec_request(&mut manager, &request_store, req).await;
+                        }
+                    }
+                    Err(err) => {
+                        tracing::warn!(%err, "FlowSpec: failed to read pending flowspec_requests");
+                    }
+                }
+            }
+        }
+    }
+}
+
+/// Apply one pending `flowspec_requests` row and record its outcome — the
+/// FlowSpec analogue of [`apply_request`].
+///
+/// For `"add"`: `Applied` marks the row `applied`; `Deferred` leaves the row
+/// `pending` (retried on the next tick); `Rejected` marks it `rejected` with
+/// the reason. For `"remove"`: withdraws the flow, supersedes any other
+/// still-pending `add` for the same flow key, then marks this row `applied`.
+async fn apply_flowspec_request(
+    manager: &mut blackwall_rtbh::FlowSpecManager<blackwall_bgp::BgpHandle, blackwall_state::Store>,
+    request_store: &blackwall_state::Store,
+    req: blackwall_state::FlowSpecRequestRow,
+) {
+    let rule = blackwall_bgp::FlowSpecRule {
+        dst: host_prefix(req.dst),
+        protocol: Some(req.proto),
+        dst_port: Some(req.dst_port),
+        action: blackwall_bgp::FlowAction::TrafficRate(req.rate),
+    };
+    match req.action.as_str() {
+        "add" => match manager.apply_add(rule, mono_now(), wall_now()).await {
+            blackwall_rtbh::ApplyOutcome::Applied => {
+                if let Err(err) = request_store
+                    .set_flowspec_request_status(req.id, "applied", None)
+                    .await
+                {
+                    tracing::warn!(%err, id = req.id, "FlowSpec: failed to set request status");
+                }
+            }
+            blackwall_rtbh::ApplyOutcome::Deferred => {
+                // Leave `pending`; picked up again on the next tick.
+            }
+            blackwall_rtbh::ApplyOutcome::Rejected(reason) => {
+                if let Err(err) = request_store
+                    .set_flowspec_request_status(req.id, "rejected", Some(&reason))
+                    .await
+                {
+                    tracing::warn!(%err, id = req.id, "FlowSpec: failed to set request status");
+                }
+            }
+        },
+        "remove" => {
+            manager.apply_remove(rule, wall_now()).await;
+            // The operator's remove is the newer intent: cancel any earlier
+            // still-pending add for the same flow key.
+            if let Err(err) = request_store
+                .supersede_pending_flowspec_adds(req.dst, req.proto, req.dst_port, req.id)
+                .await
+            {
+                tracing::warn!(%err, dst = %req.dst, "FlowSpec: failed to supersede pending adds");
+            }
+            if let Err(err) = request_store
+                .set_flowspec_request_status(req.id, "applied", None)
+                .await
+            {
+                tracing::warn!(%err, id = req.id, "FlowSpec: failed to set request status");
+            }
+        }
+        other => {
+            tracing::warn!(
+                action = other,
+                id = req.id,
+                "FlowSpec: unknown request action; ignoring"
+            );
+        }
+    }
+}
+
 /// Core dispatch logic; returns `Err` on any failure.
 async fn run() -> Result<(), Box<dyn std::error::Error>> {
     let cli = Cli::parse();
@@ -617,11 +817,14 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
                         router_id: rtbh.router_id,
                         hold_time: 90,
                     };
+                    // `BgpHandle` is a cloneable mpsc sender; both the RTBH and
+                    // (optionally) FlowSpec managers share the one iBGP session.
                     let (bgp, _bgp_join) = blackwall_bgp::spawn(peer)?;
                     let controller =
                         blackwall_rtbh::RtbhController::new(rtbh_config_from(&policy, &rtbh));
                     let journal: blackwall_state::Store = (*store).clone();
-                    let mut manager = blackwall_rtbh::RtbhManager::new(controller, bgp, journal);
+                    let mut manager =
+                        blackwall_rtbh::RtbhManager::new(controller, bgp.clone(), journal);
 
                     // Rehydrate the controller from the announced mirror before
                     // this session starts accepting new detections/requests.
@@ -641,12 +844,73 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
 
                     let channel_cap = rtbh.max_blackholes.max(1024);
                     let (tx, rx) = mpsc::channel::<blackwall_flow::DetectionEvent>(channel_cap);
-                    let request_store = store.clone();
-                    tokio::spawn(rtbh_manager_task(manager, rx, request_store));
+                    tokio::spawn(rtbh_manager_task(manager, rx, store.clone()));
+
+                    match policy.flowspec.clone() {
+                        // RTBH-only: today's behaviour, Fanout([Pg, Channel→rtbh]).
+                        None => {
+                            let channel_sink: std::sync::Arc<dyn blackwall_flow::MitigationSink> =
+                                std::sync::Arc::new(blackwall_flow::ChannelSink::new(tx));
+                            std::sync::Arc::new(blackwall_flow::FanoutSink(vec![
+                                pg_sink,
+                                channel_sink,
+                            ]))
+                        }
+                        // RTBH + FlowSpec: build a second single-owner manager off
+                        // the SAME BGP session and route detections through a
+                        // SelectorSink instead of the plain RTBH ChannelSink.
+                        Some(fs) => {
+                            let fs_controller = blackwall_rtbh::FlowSpecController::new(
+                                flowspec_config_from(&policy, &fs),
+                            );
+                            let fs_journal: blackwall_state::Store = (*store).clone();
+                            let mut fs_manager = blackwall_rtbh::FlowSpecManager::new(
+                                fs_controller,
+                                bgp,
+                                fs_journal,
+                            );
+
+                            // Rehydrate FlowSpec rules from the announced mirror.
+                            let fs_mirror = store.list_active_flowspec().await?;
+                            let fs_rehydrate: Vec<(
+                                blackwall_bgp::FlowSpecRule,
+                                u64,
+                                blackwall_rtbh::BlackholeOrigin,
+                            )> = fs_mirror
+                                .into_iter()
+                                .map(|row| {
+                                    let origin = match row.origin.as_str() {
+                                        "manual" => blackwall_rtbh::BlackholeOrigin::Manual,
+                                        _ => blackwall_rtbh::BlackholeOrigin::Auto,
+                                    };
+                                    let rule = blackwall_bgp::FlowSpecRule {
+                                        dst: host_prefix(row.dst),
+                                        protocol: Some(row.proto),
+                                        dst_port: Some(row.dst_port),
+                                        action: blackwall_bgp::FlowAction::TrafficRate(row.rate),
+                                    };
+                                    (rule, row.announced_at_ms, origin)
+                                })
+                                .collect();
+                            fs_manager.rehydrate(fs_rehydrate, mono_now()).await;
+
+                            let fs_cap = fs.max_rules.max(1024);
+                            let (fs_tx, fs_rx) =
+                                mpsc::channel::<blackwall_flow::FlowMitigationEvent>(fs_cap);
+                            tokio::spawn(flowspec_manager_task(fs_manager, fs_rx, store.clone()));
 
-                    let channel_sink: std::sync::Arc<dyn blackwall_flow::MitigationSink> =
-                        std::sync::Arc::new(blackwall_flow::ChannelSink::new(tx));
-                    std::sync::Arc::new(blackwall_flow::FanoutSink(vec![pg_sink, channel_sink]))
+                            let selection = blackwall_flow::SelectionConfig {
+                                concentration: fs.concentration,
+                                max_flows: fs.max_flows,
+                                rate: fs.rate,
+                            };
+                            let selector: std::sync::Arc<dyn blackwall_flow::MitigationSink> =
+                                std::sync::Arc::new(blackwall_flow::SelectorSink::new(
+                                    fs_tx, tx, selection,
+                                ));
+                            std::sync::Arc::new(blackwall_flow::FanoutSink(vec![pg_sink, selector]))
+                        }
+                    }
                 }
             };
 
@@ -674,6 +938,7 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
             Ok(())
         }
         Command::Rtbh { action } => run_rtbh(action).await,
+        Command::Flowspec { action } => run_flowspec(action).await,
         Command::Run {
             config,
             database_url,
@@ -1046,3 +1311,112 @@ async fn rtbh_list(requests: bool) -> Result<(), Box<dyn std::error::Error>> {
     }
     Ok(())
 }
+
+/// Dispatch one `flowspec` subcommand.
+async fn run_flowspec(action: FlowspecCmd) -> Result<(), Box<dyn std::error::Error>> {
+    match action {
+        FlowspecCmd::Add {
+            ip,
+            proto,
+            port,
+            rate,
+            config,
+            operator,
+        } => flowspec_add(ip, proto, port, rate, &config, operator).await,
+        FlowspecCmd::Remove { ip, proto, port } => flowspec_remove(ip, proto, port).await,
+        FlowspecCmd::List { requests } => flowspec_list(requests).await,
+    }
+}
+
+/// `flowspec add`: reject the flow up front (no database connection made yet)
+/// if `ip` falls outside the config's FlowSpec-eligible prefixes; otherwise
+/// queue an `"add"` intent row. Unlike RTBH there is no next-hop check.
+async fn flowspec_add(
+    ip: IpAddr,
+    proto: u8,
+    port: u16,
+    rate: f32,
+    config: &std::path::Path,
+    operator: Option<String>,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let policy = blackwall_config::parse_file(config)?;
+    let Some(fs) = policy.flowspec.clone() else {
+        return Err("config has no `flowspec` block; FlowSpec is not enabled".into());
+    };
+    let controller = blackwall_rtbh::FlowSpecController::new(flowspec_config_from(&policy, &fs));
+    if !controller.is_eligible(ip) {
+        return Err(format!("{ip} is outside the configured FlowSpec-eligible prefixes").into());
+    }
+
+    let database_url = std::env::var("DATABASE_URL")
+        .map_err(|_| "DATABASE_URL must be set to queue a flowspec request")?;
+    let store = blackwall_state::Store::connect(&database_url).await?;
+    store.migrate().await?;
+    let created_by = operator.unwrap_or_else(default_operator);
+    let id = store
+        .enqueue_flowspec_request(ip, proto, port, rate, "add", &created_by)
+        .await?;
+    println!("queued (request {id}); the running daemon will announce it.");
+    Ok(())
+}
+
+/// `flowspec remove`: queue a `"remove"` intent row (rate `0.0`).
+async fn flowspec_remove(
+    ip: IpAddr,
+    proto: u8,
+    port: u16,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let database_url = std::env::var("DATABASE_URL")
+        .map_err(|_| "DATABASE_URL must be set to queue a flowspec request")?;
+    let store = blackwall_state::Store::connect(&database_url).await?;
+    store.migrate().await?;
+    let id = store
+        .enqueue_flowspec_request(ip, proto, port, 0.0, "remove", &default_operator())
+        .await?;
+    println!("queued (request {id}); the running daemon will withdraw it.");
+    Ok(())
+}
+
+/// `flowspec list`: print the announced-FlowSpec mirror, and (with
+/// `--requests`) the operator intent queue.
+async fn flowspec_list(requests: bool) -> Result<(), Box<dyn std::error::Error>> {
+    let database_url = std::env::var("DATABASE_URL")
+        .map_err(|_| "DATABASE_URL must be set to list flowspec state")?;
+    let store = blackwall_state::Store::connect(&database_url).await?;
+    store.migrate().await?;
+
+    let active = store.list_active_flowspec().await?;
+    let now = wall_now();
+    println!(
+        "{:<40} {:<6} {:<6} {:<12} {:<8} AGE",
+        "DST", "PROTO", "PORT", "RATE", "ORIGIN"
+    );
+    for row in &active {
+        let age_secs = now.saturating_sub(row.announced_at_ms) / 1000;
+        println!(
+            "{:<40} {:<6} {:<6} {:<12} {:<8} {age_secs}s",
+            row.dst, row.proto, row.dst_port, row.rate, row.origin
+        );
+    }
+
+    if requests {
+        println!();
+        println!(
+            "{:<6} {:<40} {:<6} {:<6} {:<8} {:<10} NOTE",
+            "ID", "DST", "PROTO", "PORT", "ACTION", "STATUS"
+        );
+        for row in store.list_flowspec_requests(None).await? {
+            println!(
+                "{:<6} {:<40} {:<6} {:<6} {:<8} {:<10} {}",
+                row.id,
+                row.dst,
+                row.proto,
+                row.dst_port,
+                row.action,
+                row.status,
+                row.note.as_deref().unwrap_or("")
+            );
+        }
+    }
+    Ok(())
+}
diff --git a/crates/blackwall-config/src/parser.rs b/crates/blackwall-config/src/parser.rs
index ba99fae..25ed168 100644
--- a/crates/blackwall-config/src/parser.rs
+++ b/crates/blackwall-config/src/parser.rs
@@ -4,8 +4,8 @@
 use crate::error::ConfigError;
 use crate::lexer::Line;
 use blackwall_core::{
-    AllowRule, BannerFluxConfig, DnsFluxConfig, L4Proto, Policy, PortState, RtbhPolicy,
-    ServiceTarget, ShapeBandwidth, ShapeRule, Tenant,
+    AllowRule, BannerFluxConfig, DnsFluxConfig, FlowSpecPolicy, L4Proto, Policy, PortState,
+    RtbhPolicy, ServiceTarget, ShapeBandwidth, ShapeRule, Tenant,
 };
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
 
@@ -19,6 +19,7 @@ pub fn parse(lines: &[Line]) -> Result<Policy, ConfigError> {
     let mut banner_flux: Option<BannerFluxConfig> = None;
     let mut dns_flux: Option<DnsFluxConfig> = None;
     let mut rtbh: Option<RtbhPolicy> = None;
+    let mut flowspec: Option<FlowSpecPolicy> = None;
 
     let mut i = 0;
     while i < lines.len() {
@@ -299,6 +300,90 @@ pub fn parse(lines: &[Line]) -> Result<Policy, ConfigError> {
                     max_ttl,
                 });
             }
+            "flowspec" => {
+                if flowspec.is_some() {
+                    return Err(ConfigError::BadValue {
+                        line: line.number,
+                        what: "flowspec",
+                        value: "duplicate".to_owned(),
+                    });
+                }
+                let mut kv: std::collections::HashMap<&str, &str> =
+                    std::collections::HashMap::new();
+                for tok in &line.words[1..] {
+                    let (k, v) = tok.split_once('=').ok_or_else(|| ConfigError::BadValue {
+                        line: line.number,
+                        what: "flowspec",
+                        value: tok.as_str().to_owned(),
+                    })?;
+                    if !matches!(
+                        k,
+                        "concentration" | "max-flows" | "rate" | "max-rules" | "hold-down" | "ttl"
+                    ) {
+                        return Err(ConfigError::BadValue {
+                            line: line.number,
+                            what: "flowspec key",
+                            value: k.to_owned(),
+                        });
+                    }
+                    kv.insert(k, v);
+                }
+                let bad = |what: &'static str, v: &str| ConfigError::BadValue {
+                    line: line.number,
+                    what,
+                    value: v.to_owned(),
+                };
+                let get = |k: &str| -> Result<&str, ConfigError> {
+                    kv.get(k).copied().ok_or_else(|| ConfigError::BadValue {
+                        line: line.number,
+                        what: "flowspec missing key",
+                        value: k.to_owned(),
+                    })
+                };
+                let concentration: f64 = get("concentration")?
+                    .parse()
+                    .map_err(|_| bad("concentration", get("concentration").unwrap_or("")))?;
+                let max_flows: usize = get("max-flows")?
+                    .parse()
+                    .map_err(|_| bad("max-flows", get("max-flows").unwrap_or("")))?;
+                let rate: f32 = get("rate")?
+                    .parse()
+                    .map_err(|_| bad("rate", get("rate").unwrap_or("")))?;
+                let max_rules: usize = get("max-rules")?
+                    .parse()
+                    .map_err(|_| bad("max-rules", get("max-rules").unwrap_or("")))?;
+                let hold_down = parse_duration(line, get("hold-down")?)?;
+                let max_ttl = match kv.get("ttl") {
+                    Some(t) => Some(parse_duration(line, t)?),
+                    None => None,
+                };
+                if let Some(ttl) = max_ttl {
+                    if ttl < hold_down {
+                        return Err(bad("flowspec ttl", "must be >= hold-down"));
+                    }
+                }
+                // Reject misconfigurations that silently break selection: a NaN
+                // or out-of-range `concentration` (NaN makes `cumulative >= c`
+                // always false → FlowSpec never chosen), `max-flows` of 0 (loop
+                // never selects), or a negative `rate` (nonsensical traffic-rate).
+                if !(0.0..=1.0).contains(&concentration) {
+                    return Err(bad("flowspec concentration", "must be in 0.0..=1.0"));
+                }
+                if max_flows == 0 {
+                    return Err(bad("flowspec max-flows", "must be >= 1"));
+                }
+                if rate.is_nan() || rate < 0.0 {
+                    return Err(bad("flowspec rate", "must be >= 0"));
+                }
+                flowspec = Some(FlowSpecPolicy {
+                    concentration,
+                    max_flows,
+                    rate,
+                    max_rules,
+                    hold_down,
+                    max_ttl,
+                });
+            }
             other => {
                 return Err(ConfigError::UnknownDirective {
                     line: line.number,
@@ -316,6 +401,16 @@ pub fn parse(lines: &[Line]) -> Result<Policy, ConfigError> {
         expected: "an `interface` directive",
     })?;
 
+    // FlowSpec reuses the `rtbh` block's BGP peer (single shared iBGP session),
+    // so a `flowspec` directive is meaningless without an `rtbh` block.
+    if flowspec.is_some() && rtbh.is_none() {
+        return Err(ConfigError::BadValue {
+            line: eof_line,
+            what: "flowspec",
+            value: "requires an rtbh block (shared BGP session)".to_owned(),
+        });
+    }
+
     Ok(Policy {
         interface,
         prefixes,
@@ -325,6 +420,7 @@ pub fn parse(lines: &[Line]) -> Result<Policy, ConfigError> {
         banner_flux,
         dns_flux,
         rtbh,
+        flowspec,
     })
 }
 
@@ -980,6 +1076,80 @@ tenant t {
         assert!(parse_text(dup).is_err());
     }
 
+    #[test]
+    fn parses_flowspec_directive() {
+        let src = "\
+interface wan eth0
+ipv4 203.0.113.0/24
+rtbh peer=10.0.0.2:179 local-as=214806 peer-as=214806 router-id=10.222.255.1 next-hop-v4=10.222.255.99 max=256 hold-down=60s ttl=2h
+flowspec concentration=0.8 max-flows=4 rate=0 max-rules=256 hold-down=60s ttl=2h
+";
+        let policy = parse_text(src).unwrap();
+        let fs = policy.flowspec.expect("flowspec present");
+        assert_eq!(fs.concentration, 0.8);
+        assert_eq!(fs.max_flows, 4);
+        assert_eq!(fs.rate, 0.0);
+        assert_eq!(fs.max_rules, 256);
+        assert_eq!(fs.hold_down, std::time::Duration::from_secs(60));
+        assert_eq!(fs.max_ttl, Some(std::time::Duration::from_secs(7200)));
+    }
+
+    #[test]
+    fn flowspec_without_rtbh_is_rejected() {
+        let src = "\
+interface wan eth0
+ipv4 203.0.113.0/24
+flowspec concentration=0.8 max-flows=4 rate=0 max-rules=256 hold-down=60s ttl=2h
+";
+        let err = parse_text(src).unwrap_err();
+        assert!(format!("{err}").contains("flowspec"));
+    }
+
+    #[test]
+    fn flowspec_rejects_invalid_selection_tunables() {
+        let base = "interface wan eth0\nrtbh peer=10.0.0.2:179 local-as=214806 peer-as=214806 router-id=10.222.255.1 next-hop-v4=10.222.255.99 max=256 hold-down=60s ttl=2h\n";
+        for fs in [
+            "flowspec concentration=1.5 max-flows=4 rate=0 max-rules=256 hold-down=60s",
+            "flowspec concentration=nan max-flows=4 rate=0 max-rules=256 hold-down=60s",
+            "flowspec concentration=0.8 max-flows=0 rate=0 max-rules=256 hold-down=60s",
+            "flowspec concentration=0.8 max-flows=4 rate=-1 max-rules=256 hold-down=60s",
+        ] {
+            let src = format!("{base}{fs}\n");
+            assert!(parse_text(&src).is_err(), "should reject: {fs}");
+        }
+    }
+
+    #[test]
+    fn duplicate_flowspec_is_rejected() {
+        let src = "\
+interface wan eth0
+rtbh peer=10.0.0.2:179 local-as=214806 peer-as=214806 router-id=10.222.255.1 next-hop-v4=10.222.255.99 max=256 hold-down=60s ttl=2h
+flowspec concentration=0.8 max-flows=4 rate=0 max-rules=256 hold-down=60s ttl=2h
+flowspec concentration=0.9 max-flows=2 rate=0 max-rules=8 hold-down=30s ttl=1h
+";
+        assert!(parse_text(src).is_err());
+    }
+
+    #[test]
+    fn flowspec_rejects_ttl_below_hold_down() {
+        let src = "\
+interface wan eth0
+rtbh peer=10.0.0.2:179 local-as=1 peer-as=1 router-id=10.0.0.1 next-hop-v4=10.0.0.9 max=8 hold-down=60s
+flowspec concentration=0.8 max-flows=4 rate=0 max-rules=256 hold-down=60s ttl=30s
+";
+        assert!(parse_text(src).is_err());
+    }
+
+    #[test]
+    fn flowspec_rejects_unknown_key() {
+        let src = "\
+interface wan eth0
+rtbh peer=10.0.0.2:179 local-as=1 peer-as=1 router-id=10.0.0.1 next-hop-v4=10.0.0.9 max=8 hold-down=60s
+flowspec concentration=0.8 max-flows=4 rate=0 max-rules=256 hold-down=60s bogus=1
+";
+        assert!(parse_text(src).is_err());
+    }
+
     #[test]
     fn rejects_wrong_token_count_for_directive() {
         // `interface` expects exactly 3 words
diff --git a/crates/blackwall-core/src/flowspec.rs b/crates/blackwall-core/src/flowspec.rs
new file mode 100644
index 0000000..19f9e45
--- /dev/null
+++ b/crates/blackwall-core/src/flowspec.rs
@@ -0,0 +1,43 @@
+//! FlowSpec auto-mitigation policy (selection + controller tunables). Reuses
+//! the `rtbh` block's BGP peer and `Policy.prefixes` for eligibility.
+
+use serde::{Deserialize, Serialize};
+use std::time::Duration;
+
+/// Policy for FlowSpec auto-mitigation. Reuses the `rtbh` block's BGP peer
+/// (single shared iBGP session) and `Policy.prefixes` for eligibility, so it
+/// carries only the selection + controller tunables.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct FlowSpecPolicy {
+    /// Cumulative top-port weight fraction that counts as "concentrated".
+    pub concentration: f64,
+    /// Max distinct flows a single FlowSpec mitigation may emit.
+    pub max_flows: usize,
+    /// Traffic-rate action for emitted rules (bytes/sec; `0.0` = drop).
+    pub rate: f32,
+    /// Hard cap on concurrently-active FlowSpec rules (summed across targets).
+    pub max_rules: usize,
+    /// Deferred-withdraw hold-down.
+    pub hold_down: Duration,
+    /// Optional absolute TTL from last activity.
+    pub max_ttl: Option<Duration>,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn flowspec_policy_roundtrips_serde() {
+        let p = FlowSpecPolicy {
+            concentration: 0.8,
+            max_flows: 4,
+            rate: 0.0,
+            max_rules: 256,
+            hold_down: std::time::Duration::from_secs(60),
+            max_ttl: Some(std::time::Duration::from_secs(7200)),
+        };
+        let json = serde_json::to_string(&p).unwrap();
+        let back: FlowSpecPolicy = serde_json::from_str(&json).unwrap();
+        assert_eq!(p, back);
+    }
+}
diff --git a/crates/blackwall-core/src/lib.rs b/crates/blackwall-core/src/lib.rs
index d4ea977..85a59fd 100644
--- a/crates/blackwall-core/src/lib.rs
+++ b/crates/blackwall-core/src/lib.rs
@@ -1,6 +1,7 @@
 //! Shared domain types and the policy model for Blackwall.
 
 mod dns;
+mod flowspec;
 mod flux;
 mod policy;
 mod port;
@@ -11,6 +12,7 @@ mod shape;
 mod target;
 
 pub use dns::DnsFluxConfig;
+pub use flowspec::FlowSpecPolicy;
 pub use flux::BannerFluxConfig;
 pub use policy::{AllowRule, Policy, Tenant};
 pub use port::PortState;
diff --git a/crates/blackwall-core/src/policy.rs b/crates/blackwall-core/src/policy.rs
index 3371f06..a22897a 100644
--- a/crates/blackwall-core/src/policy.rs
+++ b/crates/blackwall-core/src/policy.rs
@@ -29,7 +29,7 @@ pub struct Tenant {
 }
 
 /// The complete desired firewall policy parsed from config.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct Policy {
     /// The uplink interface Blackwall manages (e.g. `eth0`).
     pub interface: String,
@@ -47,4 +47,6 @@ pub struct Policy {
     pub dns_flux: Option<crate::DnsFluxConfig>,
     /// RTBH control-plane config (`rtbh` directive); `None` disables RTBH.
     pub rtbh: Option<crate::RtbhPolicy>,
+    /// FlowSpec auto-mitigation policy; `None` disables FlowSpec (RTBH-only).
+    pub flowspec: Option<crate::FlowSpecPolicy>,
 }
diff --git a/crates/blackwall-core/src/resolve.rs b/crates/blackwall-core/src/resolve.rs
index 6af62e2..962601a 100644
--- a/crates/blackwall-core/src/resolve.rs
+++ b/crates/blackwall-core/src/resolve.rs
@@ -127,6 +127,7 @@ mod tests {
             banner_flux: None,
             dns_flux: None,
             rtbh: None,
+            flowspec: None,
         }
     }
 
@@ -262,6 +263,7 @@ mod tests {
             banner_flux: None,
             dns_flux: None,
             rtbh: None,
+            flowspec: None,
         };
         let resolved = policy.resolve().expect("empty policy resolves");
         assert!(resolved.is_empty());
diff --git a/crates/blackwall-deception/tests/interop.rs b/crates/blackwall-deception/tests/interop.rs
index 83f0c2a..9af9a4d 100644
--- a/crates/blackwall-deception/tests/interop.rs
+++ b/crates/blackwall-deception/tests/interop.rs
@@ -85,6 +85,7 @@ async fn serves_deception_banner() {
         banner_flux: None,
         dns_flux: None,
         rtbh: None,
+        flowspec: None,
     };
 
     // Apply the REAL nft ruleset: deception TCP on the prefix -> tproxy :61000
@@ -135,6 +136,7 @@ async fn serves_deception_under_load() {
         banner_flux: None,
         dns_flux: None,
         rtbh: None,
+        flowspec: None,
     };
 
     // Apply the REAL nft ruleset: deception TCP on the prefix -> tproxy :61000
diff --git a/crates/blackwall-discovery/src/reconcile.rs b/crates/blackwall-discovery/src/reconcile.rs
index 327a8d1..db5718d 100644
--- a/crates/blackwall-discovery/src/reconcile.rs
+++ b/crates/blackwall-discovery/src/reconcile.rs
@@ -114,6 +114,7 @@ mod tests {
             banner_flux: None,
             dns_flux: None,
             rtbh: None,
+            flowspec: None,
         }
     }
 
diff --git a/crates/blackwall-nft/src/render.rs b/crates/blackwall-nft/src/render.rs
index 845196d..81d9bc0 100644
--- a/crates/blackwall-nft/src/render.rs
+++ b/crates/blackwall-nft/src/render.rs
@@ -427,6 +427,7 @@ mod tests {
             banner_flux: None,
             dns_flux: None,
             rtbh: None,
+            flowspec: None,
         }
     }
 
@@ -448,6 +449,7 @@ mod tests {
             banner_flux: None,
             dns_flux: None,
             rtbh: None,
+            flowspec: None,
         }
     }
 
@@ -692,6 +694,7 @@ mod tests {
             banner_flux: None,
             dns_flux: None,
             rtbh: None,
+            flowspec: None,
         };
         let ruleset = render(&policy).expect("render empty");
         // No resolved services, so real_v4 and real_v6 sets are empty.
diff --git a/crates/blackwall-nft/tests/apply_netns.rs b/crates/blackwall-nft/tests/apply_netns.rs
index be45c02..8b29c9a 100644
--- a/crates/blackwall-nft/tests/apply_netns.rs
+++ b/crates/blackwall-nft/tests/apply_netns.rs
@@ -26,6 +26,7 @@ fn sample() -> Policy {
         banner_flux: None,
         dns_flux: None,
         rtbh: None,
+        flowspec: None,
     }
 }
 
@@ -75,6 +76,7 @@ fn stale_set_elements_removed_on_second_apply() {
         banner_flux: None,
         dns_flux: None,
         rtbh: None,
+        flowspec: None,
     };
     blackwall_nft::apply(&policy_empty).expect("second apply");
 
diff --git a/crates/blackwall-state/Cargo.toml b/crates/blackwall-state/Cargo.toml
index 45a495c..26ad5d7 100644
--- a/crates/blackwall-state/Cargo.toml
+++ b/crates/blackwall-state/Cargo.toml
@@ -6,6 +6,7 @@ license.workspace = true
 repository.workspace = true
 
 [dependencies]
+blackwall-bgp = { path = "../blackwall-bgp" }
 blackwall-core = { path = "../blackwall-core" }
 blackwall-flow = { path = "../blackwall-flow" }
 blackwall-rtbh = { path = "../blackwall-rtbh" }
diff --git a/crates/blackwall-state/migrations/0005_flowspec.sql b/crates/blackwall-state/migrations/0005_flowspec.sql
new file mode 100644
index 0000000..df50159
--- /dev/null
+++ b/crates/blackwall-state/migrations/0005_flowspec.sql
@@ -0,0 +1,26 @@
+CREATE TABLE flowspec_rules (
+    id BIGSERIAL PRIMARY KEY,
+    dst INET NOT NULL,
+    proto INTEGER NOT NULL,
+    dst_port INTEGER NOT NULL,
+    rate DOUBLE PRECISION NOT NULL,
+    origin TEXT NOT NULL CHECK (origin IN ('auto','manual')),
+    announced_at TIMESTAMPTZ NOT NULL,
+    withdrawn_at TIMESTAMPTZ
+);
+CREATE UNIQUE INDEX flowspec_active_uniq ON flowspec_rules (dst, proto, dst_port)
+    WHERE withdrawn_at IS NULL;
+
+CREATE TABLE flowspec_requests (
+    id BIGSERIAL PRIMARY KEY,
+    dst INET NOT NULL,
+    proto INTEGER NOT NULL,
+    dst_port INTEGER NOT NULL,
+    rate DOUBLE PRECISION NOT NULL,
+    action TEXT NOT NULL CHECK (action IN ('add','remove')),
+    created_by TEXT NOT NULL,
+    requested_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','applied','rejected')),
+    note TEXT,
+    applied_at TIMESTAMPTZ
+);
diff --git a/crates/blackwall-state/src/flowspec.rs b/crates/blackwall-state/src/flowspec.rs
new file mode 100644
index 0000000..b32abb0
--- /dev/null
+++ b/crates/blackwall-state/src/flowspec.rs
@@ -0,0 +1,50 @@
+//! Row types for FlowSpec auto-mitigation persistence: the announced-mirror
+//! table and the append-only operator request queue.
+
+use std::net::IpAddr;
+
+/// One row of the `flowspec_rules` announced mirror.
+///
+/// Derives `PartialEq` but not `Eq` because `rate: f32` is not `Eq`.
+#[derive(Debug, Clone, PartialEq)]
+pub struct FlowSpecRuleRow {
+    /// Victim host address (a `/32` or `/128` route is derived downstream).
+    pub dst: IpAddr,
+    /// IP protocol number (e.g. 17 = UDP).
+    pub proto: u8,
+    /// Destination port.
+    pub dst_port: u16,
+    /// Traffic-rate action in bytes/sec; `0.0` = drop.
+    pub rate: f32,
+    /// `"auto"` or `"manual"`.
+    pub origin: String,
+    /// Announce time in epoch milliseconds.
+    pub announced_at_ms: u64,
+    /// Withdraw time in epoch milliseconds, if withdrawn.
+    pub withdrawn_at_ms: Option<u64>,
+}
+
+/// One row of the `flowspec_requests` append-only operator intent queue.
+///
+/// Derives `PartialEq` but not `Eq` because `rate: f32` is not `Eq`.
+#[derive(Debug, Clone, PartialEq)]
+pub struct FlowSpecRequestRow {
+    /// Row id (used to scope supersession).
+    pub id: i64,
+    /// Victim host address.
+    pub dst: IpAddr,
+    /// IP protocol number.
+    pub proto: u8,
+    /// Destination port.
+    pub dst_port: u16,
+    /// Traffic-rate action in bytes/sec; `0.0` = drop.
+    pub rate: f32,
+    /// `"add"` or `"remove"`.
+    pub action: String,
+    /// Operator identity that queued the request.
+    pub created_by: String,
+    /// `"pending"` | `"applied"` | `"rejected"`.
+    pub status: String,
+    /// Optional free-text note (e.g. rejection reason).
+    pub note: Option<String>,
+}
diff --git a/crates/blackwall-state/src/lib.rs b/crates/blackwall-state/src/lib.rs
index ccddd5a..0886283 100644
--- a/crates/blackwall-state/src/lib.rs
+++ b/crates/blackwall-state/src/lib.rs
@@ -3,12 +3,14 @@
 
 mod audit;
 mod error;
+mod flowspec;
 mod rtbh;
 mod services;
 mod sessions;
 mod tenants;
 
 pub use error::StateError;
+pub use flowspec::{FlowSpecRequestRow, FlowSpecRuleRow};
 pub use rtbh::{RtbhBlackholeRow, RtbhRequestRow};
 pub use services::StoredService;
 pub use sessions::SessionRow;
@@ -27,6 +29,32 @@ pub struct Store {
     pool: PgPool,
 }
 
+/// Raw column tuple decoded from a `flowspec_rules` row:
+/// `(dst, proto, dst_port, rate::real, origin, announced_at_ms, withdrawn_at_ms)`.
+type FlowSpecRuleTuple = (
+    sqlx::types::ipnetwork::IpNetwork,
+    i32,
+    i32,
+    f32,
+    String,
+    i64,
+    Option<i64>,
+);
+
+/// Raw column tuple decoded from a `flowspec_requests` row:
+/// `(id, dst, proto, dst_port, rate::real, action, created_by, status, note)`.
+type FlowSpecRequestTuple = (
+    i64,
+    sqlx::types::ipnetwork::IpNetwork,
+    i32,
+    i32,
+    f32,
+    String,
+    String,
+    String,
+    Option<String>,
+);
+
 impl Store {
     /// Connect to PostgreSQL at `database_url` (e.g.
     /// `postgres://blackwall:blackwall@localhost:5432/blackwall`).
@@ -485,6 +513,308 @@ impl Store {
         };
         Ok(rows_to_requests(rows))
     }
+
+    /// Insert or refresh the active `flowspec_rules` mirror row for the
+    /// `(dst, proto, dst_port)` flow key. If an active row already exists its
+    /// `origin` is upgraded per the announcement but never downgraded from
+    /// `manual` to `auto` (an upsert can race a `manual_add` and must not
+    /// clobber it); the `rate` is always refreshed to the latest announcement.
+    pub async fn record_flowspec(
+        &self,
+        dst: IpAddr,
+        proto: u8,
+        dst_port: u16,
+        rate: f32,
+        origin: &str,
+        at_ms: u64,
+    ) -> Result<(), StateError> {
+        let dst = ipnetwork_addr(dst);
+        let at_ms = i64::try_from(at_ms).unwrap_or(i64::MAX);
+        sqlx::query(
+            "INSERT INTO flowspec_rules (dst, proto, dst_port, rate, origin, announced_at) \
+             VALUES ($1, $2, $3, $4, $5, to_timestamp($6::bigint / 1000.0)) \
+             ON CONFLICT (dst, proto, dst_port) WHERE withdrawn_at IS NULL DO UPDATE SET \
+             origin = CASE WHEN flowspec_rules.origin = 'manual' THEN 'manual' ELSE EXCLUDED.origin END, \
+             rate = EXCLUDED.rate",
+        )
+        .bind(dst)
+        .bind(i32::from(proto))
+        .bind(i32::from(dst_port))
+        .bind(f64::from(rate))
+        .bind(origin)
+        .bind(at_ms)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
+    /// Mark the active `flowspec_rules` row for the `(dst, proto, dst_port)`
+    /// flow key as withdrawn. When `only_auto` is set, only a row with
+    /// `origin = 'auto'` is cleared (the auto-path guard that keeps a Cleared
+    /// event from withdrawing a manual rule).
+    pub async fn clear_flowspec(
+        &self,
+        dst: IpAddr,
+        proto: u8,
+        dst_port: u16,
+        at_ms: u64,
+        only_auto: bool,
+    ) -> Result<(), StateError> {
+        let dst = ipnetwork_addr(dst);
+        let at_ms = i64::try_from(at_ms).unwrap_or(i64::MAX);
+        let query = if only_auto {
+            "UPDATE flowspec_rules SET withdrawn_at = to_timestamp($4::bigint / 1000.0) \
+             WHERE dst = $1 AND proto = $2 AND dst_port = $3 AND withdrawn_at IS NULL AND origin = 'auto'"
+        } else {
+            "UPDATE flowspec_rules SET withdrawn_at = to_timestamp($4::bigint / 1000.0) \
+             WHERE dst = $1 AND proto = $2 AND dst_port = $3 AND withdrawn_at IS NULL"
+        };
+        sqlx::query(query)
+            .bind(dst)
+            .bind(i32::from(proto))
+            .bind(i32::from(dst_port))
+            .bind(at_ms)
+            .execute(&self.pool)
+            .await?;
+        Ok(())
+    }
+
+    /// List all currently-active (not withdrawn) FlowSpec rules.
+    ///
+    /// `rate` is selected as `::real` so SQLx decodes it straight into `f32`
+    /// (avoiding a forbidden `f64 as f32` narrowing cast).
+    pub async fn list_active_flowspec(&self) -> Result<Vec<FlowSpecRuleRow>, StateError> {
+        let rows: Vec<FlowSpecRuleTuple> = sqlx::query_as(
+            "SELECT dst, proto, dst_port, rate::real, origin, \
+                (EXTRACT(EPOCH FROM announced_at) * 1000)::bigint, \
+                (EXTRACT(EPOCH FROM withdrawn_at) * 1000)::bigint \
+             FROM flowspec_rules WHERE withdrawn_at IS NULL ORDER BY dst, proto, dst_port",
+        )
+        .fetch_all(&self.pool)
+        .await?;
+
+        let mut out = Vec::with_capacity(rows.len());
+        for (dst, proto, dst_port, rate, origin, announced_at_ms, withdrawn_at_ms) in rows {
+            let announced_at_ms = u64::try_from(announced_at_ms).map_err(|_| {
+                StateError::Db(sqlx::Error::Decode(
+                    format!("announced_at_ms {announced_at_ms} out of u64 range").into(),
+                ))
+            })?;
+            let withdrawn_at_ms = withdrawn_at_ms
+                .map(u64::try_from)
+                .transpose()
+                .map_err(|_| {
+                    StateError::Db(sqlx::Error::Decode(
+                        "withdrawn_at_ms out of u64 range".into(),
+                    ))
+                })?;
+            out.push(FlowSpecRuleRow {
+                dst: dst.ip(),
+                proto: narrow_proto(proto)?,
+                dst_port: narrow_port(dst_port)?,
+                rate,
+                origin,
+                announced_at_ms,
+                withdrawn_at_ms,
+            });
+        }
+        Ok(out)
+    }
+
+    /// Append an operator intent row to `flowspec_requests` (`status =
+    /// 'pending'` by default). Returns the new row's id.
+    pub async fn enqueue_flowspec_request(
+        &self,
+        dst: IpAddr,
+        proto: u8,
+        dst_port: u16,
+        rate: f32,
+        action: &str,
+        created_by: &str,
+    ) -> Result<i64, StateError> {
+        let dst = ipnetwork_addr(dst);
+        let row: (i64,) = sqlx::query_as(
+            "INSERT INTO flowspec_requests (dst, proto, dst_port, rate, action, created_by) \
+             VALUES ($1, $2, $3, $4, $5, $6) RETURNING id",
+        )
+        .bind(dst)
+        .bind(i32::from(proto))
+        .bind(i32::from(dst_port))
+        .bind(f64::from(rate))
+        .bind(action)
+        .bind(created_by)
+        .fetch_one(&self.pool)
+        .await?;
+        Ok(row.0)
+    }
+
+    /// Fetch all FlowSpec requests with `status = 'pending'`, ordered by id.
+    ///
+    /// Status-driven (not a watermark): every tick re-reads the genuinely
+    /// pending set, so a capacity-deferred add is naturally retried (it's still
+    /// `pending`) and a restart never replays already-`applied`/`rejected`
+    /// history.
+    pub async fn pending_flowspec_requests(&self) -> Result<Vec<FlowSpecRequestRow>, StateError> {
+        let rows: Vec<FlowSpecRequestTuple> = sqlx::query_as(
+            "SELECT id, dst, proto, dst_port, rate::real, action, created_by, status, note \
+             FROM flowspec_requests WHERE status = 'pending' ORDER BY id",
+        )
+        .fetch_all(&self.pool)
+        .await?;
+        rows_to_flowspec_requests(rows)
+    }
+
+    /// Mark any still-`pending` `add` request for the `(dst, proto, dst_port)`
+    /// flow key with `id < before_id` as `applied` (with a `"superseded by
+    /// remove"` note), so a later remove cancels an earlier not-yet-applied add
+    /// for the same flow instead of letting it re-announce once capacity frees.
+    ///
+    /// Scoped to `before_id` (the remove request's own id) so a re-add that
+    /// races in after the remove was snapshotted — i.e. has a higher id than
+    /// the remove — is never superseded by it.
+    pub async fn supersede_pending_flowspec_adds(
+        &self,
+        dst: IpAddr,
+        proto: u8,
+        dst_port: u16,
+        before_id: i64,
+    ) -> Result<(), StateError> {
+        let dst = ipnetwork_addr(dst);
+        sqlx::query(
+            "UPDATE flowspec_requests SET status = 'applied', note = 'superseded by remove', \
+             applied_at = now() WHERE action = 'add' AND dst = $1 AND proto = $2 AND dst_port = $3 \
+             AND status = 'pending' AND id < $4",
+        )
+        .bind(dst)
+        .bind(i32::from(proto))
+        .bind(i32::from(dst_port))
+        .bind(before_id)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
+    /// Update a FlowSpec request's status (and optional note), stamping
+    /// `applied_at`.
+    pub async fn set_flowspec_request_status(
+        &self,
+        id: i64,
+        status: &str,
+        note: Option<&str>,
+    ) -> Result<(), StateError> {
+        sqlx::query(
+            "UPDATE flowspec_requests SET status = $2, note = $3, applied_at = now() WHERE id = $1",
+        )
+        .bind(id)
+        .bind(status)
+        .bind(note)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
+    /// List FlowSpec requests, optionally filtered to a single `status`.
+    pub async fn list_flowspec_requests(
+        &self,
+        status_filter: Option<&str>,
+    ) -> Result<Vec<FlowSpecRequestRow>, StateError> {
+        let rows: Vec<FlowSpecRequestTuple> = match status_filter {
+            Some(status) => sqlx::query_as(
+                "SELECT id, dst, proto, dst_port, rate::real, action, created_by, status, note \
+                     FROM flowspec_requests WHERE status = $1 ORDER BY id",
+            )
+            .bind(status)
+            .fetch_all(&self.pool)
+            .await?,
+            None => sqlx::query_as(
+                "SELECT id, dst, proto, dst_port, rate::real, action, created_by, status, note \
+                     FROM flowspec_requests ORDER BY id",
+            )
+            .fetch_all(&self.pool)
+            .await?,
+        };
+        rows_to_flowspec_requests(rows)
+    }
+}
+
+/// Narrow a Postgres `INTEGER` (`i32`) protocol column to `u8`, mapping an
+/// out-of-range value to a decode error rather than a silent `as` truncation.
+fn narrow_proto(v: i32) -> Result<u8, StateError> {
+    u8::try_from(v).map_err(|e| StateError::Db(sqlx::Error::Decode(Box::new(e))))
+}
+
+/// Narrow a Postgres `INTEGER` (`i32`) port column to `u16`, mapping an
+/// out-of-range value to a decode error rather than a silent `as` truncation.
+fn narrow_port(v: i32) -> Result<u16, StateError> {
+    u16::try_from(v).map_err(|e| StateError::Db(sqlx::Error::Decode(Box::new(e))))
+}
+
+/// Map raw `flowspec_requests` rows into [`FlowSpecRequestRow`]s, narrowing the
+/// `INTEGER` proto/port columns to `u8`/`u16` (`rate` is already decoded as
+/// `f32` via a `::real` cast in the SELECT).
+fn rows_to_flowspec_requests(
+    rows: Vec<FlowSpecRequestTuple>,
+) -> Result<Vec<FlowSpecRequestRow>, StateError> {
+    let mut out = Vec::with_capacity(rows.len());
+    for (id, dst, proto, dst_port, rate, action, created_by, status, note) in rows {
+        out.push(FlowSpecRequestRow {
+            id,
+            dst: dst.ip(),
+            proto: narrow_proto(proto)?,
+            dst_port: narrow_port(dst_port)?,
+            rate,
+            action,
+            created_by,
+            status,
+            note,
+        });
+    }
+    Ok(out)
+}
+
+#[async_trait::async_trait]
+impl blackwall_rtbh::FlowSpecJournal for Store {
+    async fn record_announce(
+        &self,
+        rule: blackwall_bgp::FlowSpecRule,
+        origin: blackwall_rtbh::BlackholeOrigin,
+        at_ms: u64,
+    ) -> Result<(), blackwall_rtbh::JournalError> {
+        let (dst, proto, port, rate) = flatten_flowspec(&rule);
+        let o = match origin {
+            blackwall_rtbh::BlackholeOrigin::Auto => "auto",
+            blackwall_rtbh::BlackholeOrigin::Manual => "manual",
+        };
+        self.record_flowspec(dst, proto, port, rate, o, at_ms)
+            .await
+            .map_err(|e| blackwall_rtbh::JournalError(e.to_string()))
+    }
+
+    async fn record_withdraw(
+        &self,
+        rule: blackwall_bgp::FlowSpecRule,
+        at_ms: u64,
+    ) -> Result<(), blackwall_rtbh::JournalError> {
+        let (dst, proto, port, _rate) = flatten_flowspec(&rule);
+        // `only_auto = false` here is safe: the auto-vs-manual guard (never let
+        // an auto-clear withdraw a manual rule) is enforced upstream in
+        // `FlowSpecController` before this ever runs (mirrors the RTBH journal).
+        self.clear_flowspec(dst, proto, port, at_ms, false)
+            .await
+            .map_err(|e| blackwall_rtbh::JournalError(e.to_string()))
+    }
+}
+
+/// Flatten a [`blackwall_bgp::FlowSpecRule`] to the Store's `(dst, proto,
+/// dst_port, rate)` scalar columns. The auto-mitigation path always sets
+/// `protocol`/`dst_port` (a concentrated flow has both); absent components
+/// default to `0`, matching how the mirror keys rows.
+fn flatten_flowspec(rule: &blackwall_bgp::FlowSpecRule) -> (IpAddr, u8, u16, f32) {
+    let dst = rule.dst.addr();
+    let proto = rule.protocol.unwrap_or(0);
+    let port = rule.dst_port.unwrap_or(0);
+    let blackwall_bgp::FlowAction::TrafficRate(rate) = rule.action;
+    (dst, proto, port, rate)
 }
 
 /// Map raw `rtbh_requests` rows into [`RtbhRequestRow`]s.
@@ -1054,6 +1384,191 @@ mod tests {
             .any(|r| r.target == t));
     }
 
+    #[tokio::test]
+    async fn flowspec_record_list_clear_roundtrip() {
+        let Some(url) = test_url() else {
+            eprintln!("DATABASE_URL not set; skipping");
+            return;
+        };
+        let store = Store::connect(&url).await.unwrap();
+        store.migrate().await.unwrap();
+        let dst: IpAddr = "203.0.113.7".parse().unwrap();
+        store
+            .record_flowspec(dst, 17, 53, 0.0, "auto", 1_000)
+            .await
+            .unwrap();
+        let active = store.list_active_flowspec().await.unwrap();
+        let row = active.iter().find(|r| r.dst == dst).expect("row present");
+        assert_eq!(row.proto, 17);
+        assert_eq!(row.dst_port, 53);
+        assert_eq!(row.rate, 0.0);
+        assert_eq!(row.origin, "auto");
+        store
+            .clear_flowspec(dst, 17, 53, 2_000, false)
+            .await
+            .unwrap();
+        assert!(!store
+            .list_active_flowspec()
+            .await
+            .unwrap()
+            .iter()
+            .any(|r| r.dst == dst));
+    }
+
+    #[tokio::test]
+    async fn flowspec_no_downgrade_manual_to_auto() {
+        let Some(url) = test_url() else {
+            eprintln!("DATABASE_URL not set; skipping");
+            return;
+        };
+        let store = Store::connect(&url).await.unwrap();
+        store.migrate().await.unwrap();
+        let dst: IpAddr = "203.0.113.8".parse().unwrap();
+        store
+            .record_flowspec(dst, 6, 443, 0.0, "manual", 1_000)
+            .await
+            .unwrap();
+        // Re-announcing as auto must NOT downgrade the manual origin.
+        store
+            .record_flowspec(dst, 6, 443, 0.0, "auto", 1_500)
+            .await
+            .unwrap();
+        let active = store.list_active_flowspec().await.unwrap();
+        assert_eq!(
+            active.iter().find(|r| r.dst == dst).unwrap().origin,
+            "manual"
+        );
+    }
+
+    #[tokio::test]
+    async fn flowspec_clear_only_auto_keeps_manual() {
+        let Some(url) = test_url() else {
+            eprintln!("DATABASE_URL not set; skipping");
+            return;
+        };
+        let store = Store::connect(&url).await.unwrap();
+        store.migrate().await.unwrap();
+        let dst: IpAddr = "203.0.113.9".parse().unwrap();
+        store
+            .record_flowspec(dst, 6, 80, 0.0, "manual", 1_000)
+            .await
+            .unwrap();
+        // only_auto=true must not clear a manual entry.
+        store.clear_flowspec(dst, 6, 80, 2_000, true).await.unwrap();
+        assert!(store
+            .list_active_flowspec()
+            .await
+            .unwrap()
+            .iter()
+            .any(|r| r.dst == dst));
+    }
+
+    #[tokio::test]
+    async fn flowspec_request_queue_and_supersede() {
+        let Some(url) = test_url() else {
+            eprintln!("DATABASE_URL not set; skipping");
+            return;
+        };
+        let store = Store::connect(&url).await.unwrap();
+        store.migrate().await.unwrap();
+        let dst: IpAddr = "203.0.113.10".parse().unwrap();
+        let add_id = store
+            .enqueue_flowspec_request(dst, 17, 53, 0.0, "add", "op")
+            .await
+            .unwrap();
+        let rm_id = store
+            .enqueue_flowspec_request(dst, 17, 53, 0.0, "remove", "op")
+            .await
+            .unwrap();
+        let pending = store.pending_flowspec_requests().await.unwrap();
+        assert!(pending.iter().any(|r| r.id == add_id && r.action == "add"));
+        assert!(pending
+            .iter()
+            .any(|r| r.id == rm_id && r.action == "remove"));
+
+        store
+            .supersede_pending_flowspec_adds(dst, 17, 53, rm_id)
+            .await
+            .unwrap();
+        // The earlier add is now 'applied'; only the remove stays pending.
+        let pending = store.pending_flowspec_requests().await.unwrap();
+        assert!(
+            !pending.iter().any(|r| r.id == add_id),
+            "superseded add must no longer be pending"
+        );
+        assert!(pending.iter().any(|r| r.id == rm_id));
+        let superseded = store
+            .list_flowspec_requests(None)
+            .await
+            .unwrap()
+            .into_iter()
+            .find(|r| r.id == add_id)
+            .unwrap();
+        assert_eq!(superseded.status, "applied");
+        assert_eq!(superseded.note.as_deref(), Some("superseded by remove"));
+
+        store
+            .set_flowspec_request_status(rm_id, "applied", None)
+            .await
+            .unwrap();
+        assert!(!store
+            .pending_flowspec_requests()
+            .await
+            .unwrap()
+            .iter()
+            .any(|r| r.id == rm_id));
+    }
+
+    #[tokio::test]
+    async fn flowspec_journal_impl() {
+        let Some(url) = test_url() else {
+            eprintln!("DATABASE_URL not set; skipping");
+            return;
+        };
+        use blackwall_rtbh::FlowSpecJournal;
+        let store = Store::connect(&url).await.unwrap();
+        store.migrate().await.unwrap();
+        let dst: IpAddr = "203.0.113.13".parse().unwrap();
+        let rule = blackwall_bgp::FlowSpecRule {
+            dst: "203.0.113.13/32".parse().unwrap(),
+            protocol: Some(17),
+            dst_port: Some(53),
+            action: blackwall_bgp::FlowAction::TrafficRate(0.0),
+        };
+        store
+            .record_announce(rule.clone(), blackwall_rtbh::BlackholeOrigin::Auto, 1_000)
+            .await
+            .unwrap();
+        assert!(store
+            .list_active_flowspec()
+            .await
+            .unwrap()
+            .iter()
+            .any(|r| r.dst == dst && r.proto == 17 && r.dst_port == 53 && r.origin == "auto"));
+        store
+            .record_announce(rule.clone(), blackwall_rtbh::BlackholeOrigin::Manual, 2_000)
+            .await
+            .unwrap();
+        assert_eq!(
+            store
+                .list_active_flowspec()
+                .await
+                .unwrap()
+                .iter()
+                .find(|r| r.dst == dst)
+                .unwrap()
+                .origin,
+            "manual"
+        );
+        store.record_withdraw(rule, 3_000).await.unwrap();
+        assert!(!store
+            .list_active_flowspec()
+            .await
+            .unwrap()
+            .iter()
+            .any(|r| r.dst == dst));
+    }
+
     #[test]
     fn state_error_display_policy() {
         use blackwall_core::PolicyError;
@@ -1130,6 +1645,7 @@ mod tests {
             banner_flux: None,
             dns_flux: None,
             rtbh: None,
+            flowspec: None,
         }
     }
 }