feat(grpc): Filters gRPC service (create/list/delete/validate/preview)

Add an authenticated Filters service backed by the NSQL engine + store:
CreateRule/DeleteRule persist and hot-reload the live FilterEngine (ArcSwap);
ValidateRule returns an honest valid/error verdict without failing the RPC;
PreviewRule dry-runs a candidate (unsaved) rule against stored mail. Mounted
with the bearer interceptor like the other services (auth-rejection tested).
Wire CLI filter create/list/delete/validate/test to it; the shared filter_engine
instance is reused by both the gRPC path and the legacy filter.* handler during
the transition. filter edit/export/import/logs stay local for now (tracked).

Refs #171

crates/nuncio-cli/src/args.rs

@@ -369,6 +369,12 @@ pub enum FilterSubcommand {
         #[arg(short, long, help = "Rule identifier")]
         id: String,
     },
+    /// Validate an NSQL query's syntax and semantics without persisting it.
+    Validate {
+        /// NSQL query string to validate.
+        #[arg(short, long, help = "NSQL query string")]
+        sql: String,
+    },
     /// Test / dry-run NSQL query against an email.
     Test {
         /// NSQL query string to test.

crates/nuncio-cli/src/runner.rs

@@ -57,6 +57,23 @@ fn message_proto_to_json(message: &nuncio_proto::v1::Message) -> serde_json::Val
     })
 }
 
+/// Renders a `nuncio.v1.FilterRule` (as returned by the daemon's `Filters`
+/// gRPC service, backlog story 2.A / GH #171) into the JSON shape used by
+/// `filter list`/`filter create`'s `--json` output.
+fn filter_rule_proto_to_json(rule: &nuncio_proto::v1::FilterRule) -> serde_json::Value {
+    json!({
+        "id": rule.id,
+        "name": rule.name,
+        "target_account": rule.target_account,
+        "priority": rule.priority,
+        "enabled": rule.enabled,
+        "nsql_text": rule.nsql_text,
+        "actions": rule.actions,
+        "created_at": rule.created_at,
+        "updated_at": rule.updated_at,
+    })
+}
+
 /// Errors emitted by the CLI headless runner.
 #[derive(Error, Debug)]
 pub enum RunnerError {
@@ -368,60 +385,35 @@ impl HeadlessRunner {
                 }
             },
             Commands::Filter { action } => match action {
-                FilterSubcommand::List => {
-                    let rules = self.db.list_filter_rules().await.unwrap_or_default();
-                    if json_mode {
-                        format_json(&json!(rules))
-                    } else if rules.is_empty() {
-                        "No filter rules configured.".to_string()
-                    } else {
-                        let mut out = String::from(
-                            "ID         PRIORITY ENABLED NAME                  NSQL\n",
-                        );
-                        for r in rules {
-                            out.push_str(&format!(
-                                "{:<10} {:<8} {:<7} {:<20} {}\n",
-                                r.id, r.priority, r.enabled, r.name, r.nsql_text
-                            ));
-                        }
-                        out
-                    }
-                }
+                FilterSubcommand::List => self.handle_filter_list(json_mode).await,
                 FilterSubcommand::Create {
                     name,
                     sql,
                     priority,
-                } => match nuncio_filter::NsqlParser::parse_rule(name, *priority, sql) {
-                    Ok(rule) => {
-                        let opts = nuncio_filter::ValidationOptions::default();
-                        if let Err(e) = nuncio_filter::NsqlValidator::validate(&rule, &opts) {
-                            return if json_mode {
-                                format_json_error(&e.to_string())
-                            } else {
-                                format!("Validation Error: {e}")
-                            };
-                        }
-                        if let Err(e) = self.db.save_filter_rule(&rule).await {
-                            return if json_mode {
-                                format_json_error(&e.to_string())
-                            } else {
-                                format!("Database Error: {e}")
-                            };
-                        }
-                        if json_mode {
-                            format_json(&json!(rule))
-                        } else {
-                            format!("✓ Created filter rule '{}' (ID: {}).", rule.name, rule.id)
-                        }
-                    }
-                    Err(e) => {
-                        if json_mode {
-                            format_json_error(&e.to_string())
-                        } else {
-                            format!("Syntax Error: {e}")
-                        }
-                    }
-                },
+                } => {
+                    self.handle_filter_create(name, sql, *priority, json_mode)
+                        .await
+                }
+                FilterSubcommand::Delete { id } => self.handle_filter_delete(id, json_mode).await,
+                FilterSubcommand::Validate { sql } => {
+                    self.handle_filter_validate(sql, json_mode).await
+                }
+                FilterSubcommand::Test { sql, message_id } => {
+                    self.handle_filter_test(sql, message_id.as_deref(), json_mode)
+                        .await
+                }
+                // `filter edit`/`export`/`import`/`logs` are NOT part of
+                // backlog story 2.A (GH #171)'s required `Filters` gRPC
+                // surface (`CreateRule`/`ListRules`/`DeleteRule`/
+                // `ValidateRule`/`PreviewRule`); they still read/write this
+                // runner's own ephemeral local `db` below, exactly as
+                // before. Because `List`/`Create`/`Delete` above now go
+                // through the daemon's real, persistent store instead, a
+                // rule created via `filter create` will NOT show up in
+                // `filter export`/`filter logs` (which only see this
+                // process's throwaway `db`) until these are migrated too --
+                // a known gap tracked for a later story, not a regression
+                // introduced silently here.
                 FilterSubcommand::Edit {
                     id,
                     name,
@@ -474,80 +466,6 @@ impl HeadlessRunner {
                         format!("Rule '{}' not found.", id)
                     }
                 }
-                FilterSubcommand::Delete { id } => {
-                    if let Err(e) = self.db.delete_filter_rule(id).await {
-                        if json_mode {
-                            format_json_error(&e.to_string())
-                        } else {
-                            format!("Error: {e}")
-                        }
-                    } else if json_mode {
-                        format_json(&json!({ "status": "deleted", "id": id }))
-                    } else {
-                        format!("✓ Filter rule '{}' deleted.", id)
-                    }
-                }
-                FilterSubcommand::Test { sql, message_id } => {
-                    match nuncio_filter::NsqlParser::parse_rule("Test Rule", 0, sql) {
-                        Ok(rule) => {
-                            let engine = match nuncio_filter::FilterEngine::new(vec![rule.clone()])
-                            {
-                                Ok(engine) => engine,
-                                Err(e) => {
-                                    return if json_mode {
-                                        format_json_error(&e)
-                                    } else {
-                                        format!("Filter Engine Error: {e}")
-                                    };
-                                }
-                            };
-                            let sample_email = if let Some(mid) = message_id {
-                                self.db.get_message(mid).await.unwrap_or_else(|_| {
-                                    nuncio_core::model::Email {
-                                        id: mid.clone(),
-                                        account_id: "acct-1".to_string(),
-                                        folder_id: "inbox".to_string(),
-                                        subject: "Test Subject".to_string(),
-                                        sender: "test@nuncio.mx".to_string(),
-                                        recipient: "me@nuncio.mx".to_string(),
-                                        received_at: chrono::Utc::now().timestamp(),
-                                        read: false,
-                                        body_plain: Some("Sample body text".to_string()),
-                                        body_html: None,
-                                        attachments: Vec::new(),
-                                    }
-                                })
-                            } else {
-                                nuncio_core::model::Email {
-                                    id: "msg-test".to_string(),
-                                    account_id: "acct-1".to_string(),
-                                    folder_id: "inbox".to_string(),
-                                    subject: "Test Subject".to_string(),
-                                    sender: "test@nuncio.mx".to_string(),
-                                    recipient: "me@nuncio.mx".to_string(),
-                                    received_at: chrono::Utc::now().timestamp(),
-                                    read: false,
-                                    body_plain: Some("Sample body text".to_string()),
-                                    body_html: None,
-                                    attachments: Vec::new(),
-                                }
-                            };
-                            let preview = engine.preview(&sample_email);
-                            if json_mode {
-                                format_json(&json!(preview))
-                            } else {
-                                format!("Dry-run evaluation result: matched={}, actions={:?}, elapsed={}us", preview.matched, preview.actions_evaluated, preview.execution_time_us)
-                            }
-                        }
-                        Err(e) => {
-                            if json_mode {
-                                format_json_error(&e.to_string())
-                            } else {
-                                format!("Syntax Error: {e}")
-                            }
-                        }
-                    }
-                }
                 FilterSubcommand::Export { format } => {
                     let rules = self.db.list_filter_rules().await.unwrap_or_default();
                     if format == "json" || json_mode {
@@ -1023,6 +941,230 @@ impl HeadlessRunner {
         }
     }
 
+    /// `filter list`: a real thin gRPC client of the running `nunciod`
+    /// daemon's `nuncio.v1.Filters` API (backlog story 2.A, GH #171). Lists
+    /// every persisted filter rule from the daemon's real, persistent
+    /// store -- NOT this runner's own ephemeral local `db`, which is thrown
+    /// away when this CLI process exits.
+    async fn handle_filter_list(&self, json_mode: bool) -> String {
+        let mut client = match self.connect_filters_client().await {
+            Ok(client) => client,
+            Err(e) => return Self::render_error(&e, json_mode),
+        };
+
+        match client
+            .list_rules(nuncio_proto::v1::ListRulesRequest {})
+            .await
+        {
+            Ok(response) => {
+                let rules = response.into_inner().rules;
+                if json_mode {
+                    let rules_json: Vec<serde_json::Value> =
+                        rules.iter().map(filter_rule_proto_to_json).collect();
+                    format_json(&json!({ "rules": rules_json }))
+                } else if rules.is_empty() {
+                    "No filter rules configured.".to_string()
+                } else {
+                    let mut out =
+                        String::from("ID         PRIORITY ENABLED NAME                  NSQL\n");
+                    for r in rules {
+                        out.push_str(&format!(
+                            "{:<10} {:<8} {:<7} {:<20} {}\n",
+                            r.id, r.priority, r.enabled, r.name, r.nsql_text
+                        ));
+                    }
+                    out
+                }
+            }
+            Err(status) => Self::render_error(
+                &format!("nunciod daemon rejected list_rules: {status}"),
+                json_mode,
+            ),
+        }
+    }
+
+    /// `filter create`: a real thin gRPC client of the running `nunciod`
+    /// daemon's `nuncio.v1.Filters` API (backlog story 2.A, GH #171). The
+    /// daemon parses, validates (6-pass `NsqlValidator`), and persists the
+    /// rule, then reloads its own live `FilterEngine` -- this runner's own
+    /// ephemeral local `db` is never touched, so the rule survives this CLI
+    /// process exiting.
+    async fn handle_filter_create(
+        &self,
+        name: &str,
+        sql: &str,
+        priority: i32,
+        json_mode: bool,
+    ) -> String {
+        let mut client = match self.connect_filters_client().await {
+            Ok(client) => client,
+            Err(e) => return Self::render_error(&e, json_mode),
+        };
+
+        match client
+            .create_rule(nuncio_proto::v1::CreateRuleRequest {
+                name: name.to_string(),
+                nsql: sql.to_string(),
+                priority,
+            })
+            .await
+        {
+            Ok(response) => match response.into_inner().rule {
+                Some(rule) => {
+                    if json_mode {
+                        format_json(&json!({ "rule": filter_rule_proto_to_json(&rule) }))
+                    } else {
+                        format!("✓ Created filter rule '{}' (ID: {}).", rule.name, rule.id)
+                    }
+                }
+                None => Self::render_error(
+                    "nunciod daemon accepted create_rule but returned no rule",
+                    json_mode,
+                ),
+            },
+            Err(status) if status.code() == tonic::Code::InvalidArgument => {
+                Self::render_error(status.message(), json_mode)
+            }
+            Err(status) => Self::render_error(
+                &format!("nunciod daemon rejected create_rule: {status}"),
+                json_mode,
+            ),
+        }
+    }
+
+    /// `filter delete`: a real thin gRPC client of the running `nunciod`
+    /// daemon's `nuncio.v1.Filters` API (backlog story 2.A, GH #171). The
+    /// daemon deletes the persisted rule and reloads its own live
+    /// `FilterEngine` so the removal takes effect immediately.
+    async fn handle_filter_delete(&self, id: &str, json_mode: bool) -> String {
+        let mut client = match self.connect_filters_client().await {
+            Ok(client) => client,
+            Err(e) => return Self::render_error(&e, json_mode),
+        };
+
+        match client
+            .delete_rule(nuncio_proto::v1::DeleteRuleRequest { id: id.to_string() })
+            .await
+        {
+            Ok(_) => {
+                if json_mode {
+                    format_json(&json!({ "status": "deleted", "id": id }))
+                } else {
+                    format!("✓ Filter rule '{}' deleted.", id)
+                }
+            }
+            Err(status) => Self::render_error(
+                &format!("nunciod daemon rejected delete_rule: {status}"),
+                json_mode,
+            ),
+        }
+    }
+
+    /// `filter validate`: a real thin gRPC client of the running `nunciod`
+    /// daemon's `nuncio.v1.Filters` API (backlog story 2.A, GH #171). Unlike
+    /// `filter create`, an invalid rule is never a connection/RPC failure
+    /// here -- it is the daemon's honestly reported `valid: false` result,
+    /// which this renders as a validation error without ever suggesting
+    /// the daemon itself was unreachable or misbehaving.
+    async fn handle_filter_validate(&self, sql: &str, json_mode: bool) -> String {
+        let mut client = match self.connect_filters_client().await {
+            Ok(client) => client,
+            Err(e) => return Self::render_error(&e, json_mode),
+        };
+
+        match client
+            .validate_rule(nuncio_proto::v1::ValidateRuleRequest {
+                nsql: sql.to_string(),
+            })
+            .await
+        {
+            Ok(response) => {
+                let response = response.into_inner();
+                if json_mode {
+                    format_json(&json!({ "valid": response.valid, "error": response.error }))
+                } else if response.valid {
+                    "✓ NSQL rule is valid.".to_string()
+                } else {
+                    format!("Validation Error: {}", response.error)
+                }
+            }
+            Err(status) => Self::render_error(
+                &format!("nunciod daemon rejected validate_rule: {status}"),
+                json_mode,
+            ),
+        }
+    }
+
+    /// `filter test`: a real thin gRPC client of the running `nunciod`
+    /// daemon's `nuncio.v1.Filters` API (backlog story 2.A, GH #171).
+    /// Dry-run evaluates the given NSQL rule against a stored message (by
+    /// `message_id`, read from the daemon's real, persistent store) or a
+    /// fixed synthetic sample when none is given/resolvable -- the rule is
+    /// never persisted and no action is ever executed.
+    async fn handle_filter_test(
+        &self,
+        sql: &str,
+        message_id: Option<&str>,
+        json_mode: bool,
+    ) -> String {
+        let mut client = match self.connect_filters_client().await {
+            Ok(client) => client,
+            Err(e) => return Self::render_error(&e, json_mode),
+        };
+
+        match client
+            .preview_rule(nuncio_proto::v1::PreviewRuleRequest {
+                nsql: sql.to_string(),
+                message_id: message_id.map(str::to_string),
+            })
+            .await
+        {
+            Ok(response) => {
+                let preview = response.into_inner();
+                if json_mode {
+                    format_json(&json!({
+                        "message_id": preview.message_id,
+                        "matched": preview.matched,
+                        "matched_rule_id": preview.matched_rule_id,
+                        "matched_rule_name": preview.matched_rule_name,
+                        "actions_evaluated": preview.actions_evaluated,
+                        "execution_time_us": preview.execution_time_us,
+                        "condition_traces": preview.condition_traces,
+                    }))
+                } else {
+                    format!(
+                        "Dry-run evaluation result: matched={}, actions={:?}, elapsed={}us",
+                        preview.matched, preview.actions_evaluated, preview.execution_time_us
+                    )
+                }
+            }
+            Err(status) if status.code() == tonic::Code::InvalidArgument => {
+                Self::render_error(status.message(), json_mode)
+            }
+            Err(status) => Self::render_error(
+                &format!("nunciod daemon rejected preview_rule: {status}"),
+                json_mode,
+            ),
+        }
+    }
+
+    /// Resolves the gRPC bearer token from the injected vault and dials the
+    /// `nuncio.v1.Filters` service at `self.grpc_addr` (backlog story 2.A,
+    /// GH #171), shared by every `filter` handler above.
+    async fn connect_filters_client(
+        &self,
+    ) -> Result<nuncio_proto::client::AuthenticatedFiltersClient, String> {
+        let token_bytes = self
+            .secrets
+            .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32)
+            .map_err(|e| format!("failed to read gRPC bearer token from vault: {e}"))?;
+        let token = hex::encode(token_bytes);
+
+        nuncio_proto::client::connect_filters(&self.grpc_addr, &token)
+            .await
+            .map_err(|e| format!("nunciod daemon unreachable at {}: {e}", self.grpc_addr))
+    }
+
     /// `account add`: a real thin gRPC client of the running `nunciod`
     /// daemon's `nuncio.v1.Accounts` API (backlog story 1.C.1 / 1.C.2,
     /// GH #156 / GH #157).
@@ -2043,6 +2185,268 @@ mod tests {
         assert_eq!(recorded_send.body_text, "Let's discuss the roadmap.");
     }
 
+    /// Reference-client proof for backlog story 2.A (GH #171): boots a stub
+    /// `nuncio.v1.Filters` gRPC server (mirroring the `Accounts`/`Mail` stub
+    /// pattern above) and drives the real `HeadlessRunner`'s
+    /// `filter list`/`filter create`/`filter delete`/`filter validate`/
+    /// `filter test` gRPC client paths against it.
+    ///
+    /// Real persistence, `ArcSwap` engine reload, and honest
+    /// validation/preview semantics are proven by `nunciod`'s own
+    /// `grpc::tests`; this test exists purely to prove the CLI's connect +
+    /// call + JSON-format happy path, and its honest invalid-argument error
+    /// handling.
+    #[tokio::test]
+    async fn filter_rpcs_round_trip_over_grpc_to_a_stub_daemon() {
+        use nuncio_proto::v1::filters_server::{Filters as FiltersService, FiltersServer};
+        use nuncio_proto::v1::{
+            CreateRuleRequest, CreateRuleResponse, DeleteRuleRequest, DeleteRuleResponse,
+            FilterRule as FilterRuleProto, ListRulesRequest, ListRulesResponse, PreviewRuleRequest,
+            PreviewRuleResponse, ValidateRuleRequest, ValidateRuleResponse,
+        };
+        use std::sync::Mutex;
+
+        /// Minimal test-only stub of `nuncio.v1.Filters`: records the last
+        /// `DeleteRuleRequest` it received (so this test can assert on
+        /// exactly what the CLI sent over the wire) and otherwise returns
+        /// fixed responses.
+        #[derive(Default)]
+        struct StubFilters {
+            last_delete: Arc<Mutex<Option<DeleteRuleRequest>>>,
+        }
+
+        #[tonic::async_trait]
+        impl FiltersService for StubFilters {
+            async fn create_rule(
+                &self,
+                request: tonic::Request<CreateRuleRequest>,
+            ) -> Result<tonic::Response<CreateRuleResponse>, tonic::Status> {
+                let req = request.into_inner();
+                if req.nsql.contains("BROKEN") {
+                    return Err(tonic::Status::invalid_argument(
+                        "NSQL syntax error: stub rejects BROKEN rules",
+                    ));
+                }
+                Ok(tonic::Response::new(CreateRuleResponse {
+                    rule: Some(FilterRuleProto {
+                        id: "rule-stub-1".to_string(),
+                        name: req.name,
+                        target_account: "*".to_string(),
+                        priority: req.priority,
+                        enabled: true,
+                        nsql_text: req.nsql,
+                        actions: vec!["MARK READ".to_string()],
+                        created_at: 1_700_000_000,
+                        updated_at: 1_700_000_000,
+                    }),
+                }))
+            }
+
+            async fn list_rules(
+                &self,
+                _request: tonic::Request<ListRulesRequest>,
+            ) -> Result<tonic::Response<ListRulesResponse>, tonic::Status> {
+                Ok(tonic::Response::new(ListRulesResponse {
+                    rules: vec![FilterRuleProto {
+                        id: "rule-stub-1".to_string(),
+                        name: "Stub Rule".to_string(),
+                        target_account: "*".to_string(),
+                        priority: 5,
+                        enabled: true,
+                        nsql_text: "WHERE subject CONTAINS 'Urgent' ACTION MARK READ".to_string(),
+                        actions: vec!["MARK READ".to_string()],
+                        created_at: 1_700_000_000,
+                        updated_at: 1_700_000_000,
+                    }],
+                }))
+            }
+
+            async fn delete_rule(
+                &self,
+                request: tonic::Request<DeleteRuleRequest>,
+            ) -> Result<tonic::Response<DeleteRuleResponse>, tonic::Status> {
+                let req = request.into_inner();
+                *self.last_delete.lock().unwrap_or_else(|e| e.into_inner()) = Some(req);
+                Ok(tonic::Response::new(DeleteRuleResponse {}))
+            }
+
+            async fn validate_rule(
+                &self,
+                request: tonic::Request<ValidateRuleRequest>,
+            ) -> Result<tonic::Response<ValidateRuleResponse>, tonic::Status> {
+                let req = request.into_inner();
+                if req.nsql.contains("BROKEN") {
+                    Ok(tonic::Response::new(ValidateRuleResponse {
+                        valid: false,
+                        error: "NSQL syntax error: stub rejects BROKEN rules".to_string(),
+                    }))
+                } else {
+                    Ok(tonic::Response::new(ValidateRuleResponse {
+                        valid: true,
+                        error: String::new(),
+                    }))
+                }
+            }
+
+            async fn preview_rule(
+                &self,
+                _request: tonic::Request<PreviewRuleRequest>,
+            ) -> Result<tonic::Response<PreviewRuleResponse>, tonic::Status> {
+                Ok(tonic::Response::new(PreviewRuleResponse {
+                    message_id: "msg-stub-1".to_string(),
+                    matched: true,
+                    matched_rule_id: Some("rule-stub-1".to_string()),
+                    matched_rule_name: Some("Stub Rule".to_string()),
+                    actions_evaluated: vec!["MARK READ".to_string()],
+                    execution_time_us: 42,
+                    condition_traces: vec!["Rule 'Stub Rule': MATCH".to_string()],
+                }))
+            }
+        }
+
+        let stub = StubFilters::default();
+        let delete_probe = stub.last_delete.clone();
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind ephemeral loopback port");
+        let addr = listener.local_addr().expect("listener has local addr");
+        tokio::spawn(async move {
+            let _ = tonic::transport::Server::builder()
+                .add_service(FiltersServer::new(stub))
+                .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
+                .await;
+        });
+
+        let runner =
+            HeadlessRunner::ephemeral_with(Arc::new(SecretManager::mock()), addr.to_string())
+                .await
+                .expect("ephemeral runner initializes");
+
+        // `filter create` happy path.
+        let create_out = runner
+            .execute_command(
+                &Commands::Filter {
+                    action: FilterSubcommand::Create {
+                        name: "My Rule".to_string(),
+                        sql: "WHERE subject CONTAINS 'Urgent' ACTION MARK READ".to_string(),
+                        priority: 5,
+                    },
+                },
+                true,
+            )
+            .await;
+        assert!(create_out.contains("rule-stub-1"));
+        assert!(create_out.contains("My Rule"));
+
+        // `filter create` honest invalid-argument error.
+        let create_err = runner
+            .execute_command(
+                &Commands::Filter {
+                    action: FilterSubcommand::Create {
+                        name: "Broken".to_string(),
+                        sql: "BROKEN NSQL".to_string(),
+                        priority: 0,
+                    },
+                },
+                true,
+            )
+            .await;
+        assert!(create_err.contains(r#""status":"error""#));
+        assert!(create_err.contains("stub rejects BROKEN rules"));
+
+        // `filter list`.
+        let list_out = runner
+            .execute_command(
+                &Commands::Filter {
+                    action: FilterSubcommand::List,
+                },
+                true,
+            )
+            .await;
+        assert!(list_out.contains("rule-stub-1"));
+        assert!(list_out.contains("Stub Rule"));
+
+        // `filter delete`.
+        let delete_out = runner
+            .execute_command(
+                &Commands::Filter {
+                    action: FilterSubcommand::Delete {
+                        id: "rule-stub-1".to_string(),
+                    },
+                },
+                true,
+            )
+            .await;
+        assert!(delete_out.contains(r#""status":"deleted""#));
+        let recorded_delete = delete_probe
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .clone()
+            .expect("stub daemon received a delete_rule request");
+        assert_eq!(recorded_delete.id, "rule-stub-1");
+
+        // `filter validate` valid + invalid.
+        let validate_ok = runner
+            .execute_command(
+                &Commands::Filter {
+                    action: FilterSubcommand::Validate {
+                        sql: "WHERE subject CONTAINS 'Urgent' ACTION MARK READ".to_string(),
+                    },
+                },
+                true,
+            )
+            .await;
+        assert!(validate_ok.contains(r#""valid":true"#));
+
+        let validate_err = runner
+            .execute_command(
+                &Commands::Filter {
+                    action: FilterSubcommand::Validate {
+                        sql: "BROKEN NSQL".to_string(),
+                    },
+                },
+                true,
+            )
+            .await;
+        assert!(validate_err.contains(r#""valid":false"#));
+        assert!(validate_err.contains("stub rejects BROKEN rules"));
+
+        // `filter test` (PreviewRule) dry-run.
+        let test_out = runner
+            .execute_command(
+                &Commands::Filter {
+                    action: FilterSubcommand::Test {
+                        sql: "WHERE subject CONTAINS 'Urgent' ACTION MARK READ".to_string(),
+                        message_id: Some("msg-stub-1".to_string()),
+                    },
+                },
+                true,
+            )
+            .await;
+        assert!(test_out.contains(r#""matched":true"#));
+        assert!(test_out.contains("msg-stub-1"));
+    }
+
+    #[tokio::test]
+    async fn filter_commands_report_honest_error_when_daemon_unreachable() {
+        let addr = reserve_unreachable_addr().await;
+        let runner = HeadlessRunner::ephemeral_with(Arc::new(SecretManager::mock()), addr)
+            .await
+            .expect("ephemeral runner initializes");
+
+        let out = runner
+            .execute_command(
+                &Commands::Filter {
+                    action: FilterSubcommand::List,
+                },
+                true,
+            )
+            .await;
+        assert!(out.contains(r#""status":"error""#));
+        assert!(out.contains("unreachable"));
+    }
+
     #[test]
     fn runner_error_display() {
         let err = RunnerError::InitFailed("failed to open database".to_string());

crates/nuncio-proto/proto/nuncio/v1/nuncio.proto

@@ -422,3 +422,162 @@ message SyncRequest {
 message SyncResponse {
   uint64 synced_count = 1;
 }
+
+// Filters exposes the NSQL filter engine's rule CRUD, validation, and
+// dry-run preview operations over gRPC (backlog story 2.A, GH #171) -- the
+// authenticated replacement for the existing hand-rolled `filter.*` JSON-RPC
+// IPC methods (`nunciod::main`'s `CustomRpcHandler`), which remain in place
+// (both transports coexist) until they are retired in a later story (2.C).
+//
+// This faithfully exposes the filter engine as it exists today, including
+// its known correctness gaps (e.g. `target_account`/`ON ACCOUNT` matching
+// against `folder_id` rather than the message's actual account field) --
+// fixing those is out of scope here (Phase 3).
+//
+// SECURITY (GH #165): this service MUST be mounted behind the SAME
+// `BearerAuthInterceptor` that guards `System`/`Accounts`/`Mail` -- every
+// gRPC service `nunciod` exposes MUST be authenticated, with no exceptions
+// carved out for "just one more" unauthenticated service. See
+// `nunciod::grpc::serve_on_listener`.
+service Filters {
+  // CreateRule parses and validates an NSQL rule definition, then persists
+  // it and reloads the daemon's live `FilterEngine` rule set (via its
+  // `ArcSwap`-backed hot-reload) so the new rule is active immediately.
+  // Rejects a syntactically or semantically invalid rule with
+  // `Status::invalid_argument` -- nothing is ever persisted on failure.
+  rpc CreateRule(CreateRuleRequest) returns (CreateRuleResponse);
+  // ListRules returns every persisted filter rule, priority ascending.
+  rpc ListRules(ListRulesRequest) returns (ListRulesResponse);
+  // DeleteRule removes a persisted filter rule by ID and reloads the
+  // daemon's live `FilterEngine` rule set so the removal takes effect
+  // immediately.
+  rpc DeleteRule(DeleteRuleRequest) returns (DeleteRuleResponse);
+  // ValidateRule parses and runs the 6-pass `NsqlValidator` audit against
+  // an NSQL rule definition WITHOUT persisting it. Unlike `CreateRule`,
+  // an invalid rule is NOT an RPC failure: it is the expected, honestly
+  // reported result carried in `ValidateRuleResponse.valid`/`error`.
+  rpc ValidateRule(ValidateRuleRequest) returns (ValidateRuleResponse);
+  // PreviewRule dry-run evaluates an ad hoc, NOT-yet-persisted NSQL rule
+  // definition against a stored message (by `message_id`) -- or, if none is
+  // given, a synthetic sample message -- using `FilterEngine::preview`'s
+  // existing dry-run evaluation path. Never persists the rule and never
+  // executes any of its actions.
+  rpc PreviewRule(PreviewRuleRequest) returns (PreviewRuleResponse);
+}
+
+// FilterRule mirrors `nuncio_filter::FilterRule`'s persisted/user-facing
+// fields. The parsed condition AST is intentionally NOT mirrored here --
+// `nsql_text` is the single source of truth, and re-parsing it via
+// `NsqlParser::parse_rule` deterministically reconstructs the identical AST,
+// so there is no independent state to carry over the wire. `actions` is the
+// NSQL-rendered form (`RuleAction::to_nsql`) of the rule's parsed action
+// list, included for display purposes.
+message FilterRule {
+  // Rule unique identifier.
+  string id = 1;
+  // Rule display name.
+  string name = 2;
+  // Target account pattern ("*" for all accounts, an exact address, or a
+  // wildcard like "%@kof22.com").
+  string target_account = 3;
+  // Rule evaluation order (0 = highest).
+  int32 priority = 4;
+  // Active flag.
+  bool enabled = 5;
+  // Full NSQL source code text.
+  string nsql_text = 6;
+  // Rendered NSQL action clauses (e.g. "MOVE TO 'Archive'"), derived from
+  // the parsed AST.
+  repeated string actions = 7;
+  // Creation timestamp (unix seconds).
+  int64 created_at = 8;
+  // Last update timestamp (unix seconds).
+  int64 updated_at = 9;
+}
+
+// CreateRuleRequest carries a new rule's display name, full NSQL source
+// text, and evaluation priority.
+message CreateRuleRequest {
+  // Display name for the rule.
+  string name = 1;
+  // Full NSQL source code text (may include an `ON ACCOUNT` clause and
+  // `ACTION` clause).
+  string nsql = 2;
+  // Evaluation order (0 = highest).
+  int32 priority = 3;
+}
+
+// CreateRuleResponse carries the persisted rule, including its
+// daemon-assigned `id` and timestamps.
+message CreateRuleResponse {
+  FilterRule rule = 1;
+}
+
+// ListRulesRequest currently carries no parameters.
+message ListRulesRequest {}
+
+// ListRulesResponse.
+message ListRulesResponse {
+  repeated FilterRule rules = 1;
+}
+
+// DeleteRuleRequest.
+message DeleteRuleRequest {
+  // Rule unique identifier.
+  string id = 1;
+}
+
+// DeleteRuleResponse currently carries no parameters; a successful RPC
+// response (as opposed to a `Status` error) is confirmation enough.
+message DeleteRuleResponse {}
+
+// ValidateRuleRequest carries only the raw NSQL source text -- validation
+// never persists anything, so no name/priority/id is needed.
+message ValidateRuleRequest {
+  string nsql = 1;
+}
+
+// ValidateRuleResponse. `valid == false` is an ordinary, successfully
+// returned result (never a `Status` error) carrying the honest parse or
+// `NsqlValidator` failure message in `error`.
+message ValidateRuleResponse {
+  // Whether `nsql` parsed AND passed every `NsqlValidator` pass.
+  bool valid = 1;
+  // Empty when `valid` is true; otherwise the parse or validation error
+  // message.
+  string error = 2;
+}
+
+// PreviewRuleRequest carries an ad hoc, NOT-yet-persisted NSQL rule
+// definition to dry-run evaluate. `message_id` optionally names a stored
+// message to evaluate against (looked up via the same store `Mail/
+// GetMessage` reads); when absent (or when `message_id` does not resolve to
+// a stored message), a fixed synthetic sample message is evaluated instead,
+// so this RPC can always return a result.
+message PreviewRuleRequest {
+  // Full NSQL source code text of the rule to preview.
+  string nsql = 1;
+  // Optional stored message id to evaluate the rule against.
+  optional string message_id = 2;
+}
+
+// PreviewRuleResponse mirrors `nuncio_filter::FilterPreviewResult`
+// field-for-field, except `actions_evaluated` is rendered as NSQL strings
+// (`RuleAction::to_nsql`) rather than a structured action AST.
+message PreviewRuleResponse {
+  // Evaluated message id (either the requested `message_id`, or the
+  // synthetic sample message's fixed id when none was given/resolvable).
+  string message_id = 1;
+  // Whether the previewed rule matched the evaluated message.
+  bool matched = 2;
+  // Present only when `matched` is true.
+  optional string matched_rule_id = 3;
+  // Present only when `matched` is true.
+  optional string matched_rule_name = 4;
+  // NSQL-rendered actions that would be executed on a real match.
+  repeated string actions_evaluated = 5;
+  // Evaluation duration in microseconds.
+  uint64 execution_time_us = 6;
+  // Human-readable step-by-step match traces.
+  repeated string condition_traces = 7;
+}

crates/nuncio-proto/src/client.rs

@@ -10,6 +10,7 @@
 //! server-side interceptor in `nunciod::grpc`.
 
 use crate::v1::accounts_client::AccountsClient;
+use crate::v1::filters_client::FiltersClient;
 use crate::v1::mail_client::MailClient;
 use crate::v1::system_client::SystemClient;
 use crate::v1::{Event, SubscribeRequest};
@@ -94,6 +95,11 @@ pub type AuthenticatedAccountsClient =
 /// [`connect_mail`].
 pub type AuthenticatedMailClient = MailClient<InterceptedService<Channel, BearerTokenInterceptor>>;
 
+/// The authenticated `nuncio.v1.Filters` client type returned by
+/// [`connect_filters`].
+pub type AuthenticatedFiltersClient =
+    FiltersClient<InterceptedService<Channel, BearerTokenInterceptor>>;
+
 /// Dials `addr` (a `host:port` pair, e.g. `127.0.0.1:9420`) over plain HTTP
 /// (the loopback gRPC transport is never TLS-wrapped; auth is via bearer
 /// token instead) and returns the connected [`Channel`], shared by every
@@ -164,6 +170,25 @@ pub async fn connect_mail(
     Ok(MailClient::with_interceptor(channel, interceptor))
 }
 
+/// Dials the `nuncio.v1.Filters` gRPC endpoint at `addr` and returns a
+/// client that injects `authorization: Bearer <token>` metadata on every
+/// call (backlog story 2.A, GH #171).
+///
+/// `Filters` is guarded by the exact same `BearerAuthInterceptor` as
+/// `System`/`Accounts`/`Mail` on the server side (see
+/// `nunciod::grpc::serve_on_listener`, GH #165), so this shares
+/// [`BearerTokenInterceptor`] and [`dial`] with [`connect_system`] /
+/// [`connect_accounts`] / [`connect_mail`] rather than hand-rolling a fourth
+/// auth handshake.
+pub async fn connect_filters(
+    addr: &str,
+    token: &str,
+) -> Result<AuthenticatedFiltersClient, ConnectError> {
+    let interceptor = BearerTokenInterceptor::new(token)?;
+    let channel = dial(addr).await?;
+    Ok(FiltersClient::with_interceptor(channel, interceptor))
+}
+
 /// Opens the `nuncio.v1.System/Subscribe` server-streaming RPC on an already
 /// authenticated client (as returned by [`connect_system`]), reusing the
 /// bearer-token interceptor already attached to `client` rather than
@@ -272,6 +297,23 @@ mod tests {
         assert!(err.to_string().contains("127.0.0.1:1"));
     }
 
+    #[tokio::test]
+    async fn connect_filters_fails_closed_on_invalid_token() {
+        let err = connect_filters("127.0.0.1:0", "tok\ntoken")
+            .await
+            .expect_err("invalid token must fail before dialing");
+        assert!(matches!(err, ConnectError::InvalidToken(_)));
+    }
+
+    #[tokio::test]
+    async fn connect_filters_reports_transport_error_when_daemon_unreachable() {
+        let err = connect_filters("127.0.0.1:1", "abc123")
+            .await
+            .expect_err("connecting to an unreachable daemon must fail");
+        assert!(matches!(err, ConnectError::Transport { .. }));
+        assert!(err.to_string().contains("127.0.0.1:1"));
+    }
+
     #[tokio::test]
     async fn connect_system_reports_invalid_endpoint_for_a_malformed_address() {
         // An embedded space is not a legal URI authority character, so

crates/nunciod/src/grpc.rs

@@ -8,27 +8,33 @@
 //! transport is out of scope here and lands in a later story.
 
 use nuncio_core::{CoreCommand, CoreEvent, EventBus};
+use nuncio_filter::{FilterEngine, NsqlParser, NsqlValidator, ValidationOptions};
 use nuncio_mail::{MailBackend, MessageSender};
 use nuncio_proto::v1::accounts_server::{Accounts, AccountsServer};
 use nuncio_proto::v1::event::Kind;
+use nuncio_proto::v1::filters_server::{Filters, FiltersServer};
 use nuncio_proto::v1::mail_server::{Mail, MailServer};
 use nuncio_proto::v1::system_server::{System, SystemServer};
 use nuncio_proto::v1::{
     AccountConfig as AccountConfigProto, AccountProtocol as AccountProtocolProto,
     AddAccountRequest, AddAccountResponse, Attachment as AttachmentProto, BatchFilterProgress,
-    DatabaseRecovered, Event, EventError, FilterExecuted, Folder as FolderProto, GetMessageRequest,
-    GetMessageResponse, GetStatusRequest, GetStatusResponse, ListAccountsRequest,
-    ListAccountsResponse, ListFoldersRequest, ListFoldersResponse, ListMessagesRequest,
-    ListMessagesResponse, MarkReadRequest, MarkReadResponse, Message as MessageProto,
-    MessageFlagsChanged, MessageSearchHit, SearchMessagesRequest, SearchMessagesResponse,
-    SendMessageRequest, SendMessageResponse, ShuttingDown, SubscribeRequest, SyncCompleted,
-    SyncRequest, SyncResponse, SyncStarted, TlsMode as TlsModeProto, UpdateAvailable,
+    CreateRuleRequest, CreateRuleResponse, DatabaseRecovered, DeleteRuleRequest,
+    DeleteRuleResponse, Event, EventError, FilterExecuted, FilterRule as FilterRuleProto,
+    Folder as FolderProto, GetMessageRequest, GetMessageResponse, GetStatusRequest,
+    GetStatusResponse, ListAccountsRequest, ListAccountsResponse, ListFoldersRequest,
+    ListFoldersResponse, ListMessagesRequest, ListMessagesResponse, ListRulesRequest,
+    ListRulesResponse, MarkReadRequest, MarkReadResponse, Message as MessageProto,
+    MessageFlagsChanged, MessageSearchHit, PreviewRuleRequest, PreviewRuleResponse,
+    SearchMessagesRequest, SearchMessagesResponse, SendMessageRequest, SendMessageResponse,
+    ShuttingDown, SubscribeRequest, SyncCompleted, SyncRequest, SyncResponse, SyncStarted,
+    TlsMode as TlsModeProto, UpdateAvailable, ValidateRuleRequest, ValidateRuleResponse,
 };
 use nuncio_store::db::DatabaseEngine;
 use nuncio_store::search::SearchEngine;
 use nuncio_store::vault::SecretManager;
 use std::pin::Pin;
 use std::sync::Arc;
+use std::time::{SystemTime, UNIX_EPOCH};
 use subtle::ConstantTimeEq;
 use thiserror::Error;
 use tokio::net::TcpListener;
@@ -672,6 +678,234 @@ impl Mail for MailGrpcService {
     }
 }
 
+/// Maps a `nuncio_filter::FilterRule` onto its wire-format
+/// `nuncio.v1.FilterRule` representation (backlog story 2.A, GH #171). The
+/// parsed condition AST is deliberately NOT carried over the wire -- see the
+/// message's doc comment in `proto/nuncio/v1/nuncio.proto` -- `actions` is
+/// rendered as NSQL strings via `RuleAction::to_nsql`.
+fn map_filter_rule_to_proto(rule: nuncio_filter::FilterRule) -> FilterRuleProto {
+    FilterRuleProto {
+        id: rule.id,
+        name: rule.name,
+        target_account: rule.target_account,
+        priority: rule.priority,
+        enabled: rule.enabled,
+        nsql_text: rule.nsql_text,
+        actions: rule.actions.iter().map(|a| a.to_nsql()).collect(),
+        created_at: rule.created_at,
+        updated_at: rule.updated_at,
+    }
+}
+
+/// Maps a `nuncio_filter::FilterPreviewResult` onto its wire-format
+/// `nuncio.v1.PreviewRuleResponse` representation.
+fn map_preview_result_to_proto(preview: nuncio_filter::FilterPreviewResult) -> PreviewRuleResponse {
+    PreviewRuleResponse {
+        message_id: preview.message_id,
+        matched: preview.matched,
+        matched_rule_id: preview.matched_rule_id,
+        matched_rule_name: preview.matched_rule_name,
+        actions_evaluated: preview
+            .actions_evaluated
+            .iter()
+            .map(|a| a.to_nsql())
+            .collect(),
+        execution_time_us: preview.execution_time_us,
+        condition_traces: preview.condition_traces,
+    }
+}
+
+/// Builds a fixed synthetic sample message for `PreviewRule` to evaluate
+/// against when the caller supplies no `message_id` (or one that does not
+/// resolve to a stored message), mirroring the exact fallback sample the
+/// CLI's previous local `filter test` path used -- so this RPC can always
+/// return a preview result rather than requiring pre-existing stored mail.
+fn synthetic_preview_email(id: &str) -> nuncio_core::model::Email {
+    let received_at = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_secs() as i64)
+        .unwrap_or_default();
+    nuncio_core::model::Email {
+        id: id.to_string(),
+        account_id: "acct-1".to_string(),
+        folder_id: "inbox".to_string(),
+        subject: "Test Subject".to_string(),
+        sender: "test@nuncio.mx".to_string(),
+        recipient: "me@nuncio.mx".to_string(),
+        received_at,
+        read: false,
+        body_plain: Some("Sample body text".to_string()),
+        body_html: None,
+        attachments: Vec::new(),
+    }
+}
+
+/// `nuncio.v1.Filters` gRPC service implementation backed by the daemon's
+/// live [`DatabaseEngine`] filter-rule CRUD and the daemon's live
+/// [`FilterEngine`] (backlog story 2.A, GH #171) -- the authenticated gRPC
+/// replacement for the existing hand-rolled `filter.*` JSON-RPC IPC methods
+/// (`nunciod::main`'s `CustomRpcHandler`), which remain in place (both
+/// transports coexist) until retired in a later story (2.C).
+///
+/// `CreateRule`/`DeleteRule` persist through `db` FIRST, then reload
+/// `filter_engine`'s `ArcSwap`-backed rule set from the freshly persisted
+/// state via `db.list_filter_rules()`, so the live engine and the persisted
+/// store never disagree about which rules exist. A reload failure (e.g. a
+/// rule whose regex no longer compiles) is logged but does NOT fail the
+/// RPC: the mutation already succeeded in the store, so failing the RPC
+/// here would misreport a successful write as an error.
+struct FiltersGrpcService {
+    db: Arc<DatabaseEngine>,
+    filter_engine: Arc<FilterEngine>,
+}
+
+impl FiltersGrpcService {
+    /// Reloads `self.filter_engine`'s active rule set from `self.db`'s
+    /// current persisted rules, logging (rather than propagating) a reload
+    /// failure -- shared by `create_rule` and `delete_rule` after each
+    /// persists its own mutation.
+    async fn reload_engine_from_store(&self) {
+        match self.db.list_filter_rules().await {
+            Ok(rules) => {
+                if let Err(e) = self.filter_engine.reload_rules(rules) {
+                    tracing::warn!("Filters: failed to reload live FilterEngine rules: {}", e);
+                }
+            }
+            Err(e) => {
+                tracing::warn!(
+                    "Filters: failed to re-list persisted filter rules for engine reload: {}",
+                    e
+                );
+            }
+        }
+    }
+}
+
+#[tonic::async_trait]
+impl Filters for FiltersGrpcService {
+    async fn create_rule(
+        &self,
+        request: Request<CreateRuleRequest>,
+    ) -> Result<Response<CreateRuleResponse>, Status> {
+        let req = request.into_inner();
+
+        let rule = NsqlParser::parse_rule(req.name, req.priority, &req.nsql)
+            .map_err(|e| Status::invalid_argument(format!("NSQL syntax error: {e}")))?;
+
+        NsqlValidator::validate(&rule, &ValidationOptions::default())
+            .map_err(|e| Status::invalid_argument(format!("NSQL validation error: {e}")))?;
+
+        self.db
+            .save_filter_rule(&rule)
+            .await
+            .map_err(|e| Status::internal(format!("failed to persist filter rule: {e}")))?;
+
+        self.reload_engine_from_store().await;
+
+        Ok(Response::new(CreateRuleResponse {
+            rule: Some(map_filter_rule_to_proto(rule)),
+        }))
+    }
+
+    async fn list_rules(
+        &self,
+        _request: Request<ListRulesRequest>,
+    ) -> Result<Response<ListRulesResponse>, Status> {
+        let rules = self
+            .db
+            .list_filter_rules()
+            .await
+            .map_err(|e| Status::internal(format!("failed to list filter rules: {e}")))?
+            .into_iter()
+            .map(map_filter_rule_to_proto)
+            .collect();
+
+        Ok(Response::new(ListRulesResponse { rules }))
+    }
+
+    async fn delete_rule(
+        &self,
+        request: Request<DeleteRuleRequest>,
+    ) -> Result<Response<DeleteRuleResponse>, Status> {
+        let req = request.into_inner();
+        if req.id.is_empty() {
+            return Err(Status::invalid_argument("id is required"));
+        }
+
+        self.db
+            .delete_filter_rule(&req.id)
+            .await
+            .map_err(|e| Status::internal(format!("failed to delete filter rule: {e}")))?;
+
+        self.reload_engine_from_store().await;
+
+        Ok(Response::new(DeleteRuleResponse {}))
+    }
+
+    /// Unlike `create_rule`, an invalid rule is never an RPC failure here --
+    /// see this RPC's doc comment in `proto/nuncio/v1/nuncio.proto`.
+    async fn validate_rule(
+        &self,
+        request: Request<ValidateRuleRequest>,
+    ) -> Result<Response<ValidateRuleResponse>, Status> {
+        let req = request.into_inner();
+
+        let rule = match NsqlParser::parse_rule("Validation Preview", 0, &req.nsql) {
+            Ok(rule) => rule,
+            Err(e) => {
+                return Ok(Response::new(ValidateRuleResponse {
+                    valid: false,
+                    error: format!("NSQL syntax error: {e}"),
+                }));
+            }
+        };
+
+        match NsqlValidator::validate(&rule, &ValidationOptions::default()) {
+            Ok(()) => Ok(Response::new(ValidateRuleResponse {
+                valid: true,
+                error: String::new(),
+            })),
+            Err(e) => Ok(Response::new(ValidateRuleResponse {
+                valid: false,
+                error: format!("NSQL validation error: {e}"),
+            })),
+        }
+    }
+
+    /// Dry-run evaluates an ad hoc, NOT-yet-persisted NSQL rule against a
+    /// stored message (or a synthetic sample -- see
+    /// [`synthetic_preview_email`]), via a fresh one-off [`FilterEngine`]
+    /// built from just that single rule and its existing `preview` path.
+    /// This deliberately does NOT use `self.filter_engine` (the daemon's
+    /// live, persisted rule set): the whole point of a preview is to try out
+    /// a candidate rule that may not be, and may never be, saved.
+    async fn preview_rule(
+        &self,
+        request: Request<PreviewRuleRequest>,
+    ) -> Result<Response<PreviewRuleResponse>, Status> {
+        let req = request.into_inner();
+
+        let rule = NsqlParser::parse_rule("Preview Rule", 0, &req.nsql)
+            .map_err(|e| Status::invalid_argument(format!("NSQL syntax error: {e}")))?;
+
+        let preview_engine = FilterEngine::new(vec![rule])
+            .map_err(|e| Status::internal(format!("failed to build preview engine: {e}")))?;
+
+        let email = match &req.message_id {
+            Some(message_id) if !message_id.is_empty() => {
+                match self.db.get_message(message_id).await {
+                    Ok(email) => email,
+                    Err(_) => synthetic_preview_email(message_id),
+                }
+            }
+            _ => synthetic_preview_email("msg-test"),
+        };
+
+        let preview = preview_engine.preview(&email);
+        Ok(Response::new(map_preview_result_to_proto(preview)))
+    }
+}
+
 /// Bearer-token authentication interceptor for the loopback `nuncio.v1` gRPC
 /// server.
 ///
@@ -719,8 +953,9 @@ impl tonic::service::Interceptor for BearerAuthInterceptor {
 }
 
 /// Binds a loopback TCP listener at `addr` and serves the `nuncio.v1.System`,
-/// `nuncio.v1.Accounts`, and `nuncio.v1.Mail` gRPC services on it, all
-/// authenticated by `token`, until the transport server errors.
+/// `nuncio.v1.Accounts`, `nuncio.v1.Mail`, and `nuncio.v1.Filters` gRPC
+/// services on it, all authenticated by `token`, until the transport server
+/// errors.
 ///
 /// `addr` MUST be a loopback address (e.g. `127.0.0.1:PORT`); callers are
 /// responsible for passing loopback-only addresses (see
@@ -729,6 +964,7 @@ pub async fn serve(
     addr: &str,
     event_bus: Arc<EventBus>,
     db: Arc<DatabaseEngine>,
+    filter_engine: Arc<FilterEngine>,
     secrets: Arc<SecretManager>,
     token: impl Into<Arc<str>>,
 ) -> Result<(), GrpcServeError> {
@@ -738,11 +974,11 @@ pub async fn serve(
             addr: addr.to_string(),
             source,
         })?;
-    serve_on_listener(listener, event_bus, db, secrets, token).await
+    serve_on_listener(listener, event_bus, db, filter_engine, secrets, token).await
 }
 
-/// Serves the `nuncio.v1.System`, `nuncio.v1.Accounts`, and `nuncio.v1.Mail`
-/// gRPC services on an already-bound [`TcpListener`].
+/// Serves the `nuncio.v1.System`, `nuncio.v1.Accounts`, `nuncio.v1.Mail`, and
+/// `nuncio.v1.Filters` gRPC services on an already-bound [`TcpListener`].
 ///
 /// # Security (GH #165)
 ///
@@ -763,6 +999,7 @@ pub async fn serve_on_listener(
     listener: TcpListener,
     event_bus: Arc<EventBus>,
     db: Arc<DatabaseEngine>,
+    filter_engine: Arc<FilterEngine>,
     secrets: Arc<SecretManager>,
     token: impl Into<Arc<str>>,
 ) -> Result<(), GrpcServeError> {
@@ -770,6 +1007,7 @@ pub async fn serve_on_listener(
         listener,
         event_bus,
         db,
+        filter_engine,
         secrets,
         token,
         MailEngineOverrides::default(),
@@ -788,6 +1026,7 @@ pub async fn serve_on_listener_with_overrides(
     listener: TcpListener,
     event_bus: Arc<EventBus>,
     db: Arc<DatabaseEngine>,
+    filter_engine: Arc<FilterEngine>,
     secrets: Arc<SecretManager>,
     token: impl Into<Arc<str>>,
     overrides: MailEngineOverrides,
@@ -812,18 +1051,27 @@ pub async fn serve_on_listener_with_overrides(
     // exactly like `System` and `Accounts` above -- see the hard invariant
     // documented on this function's doc comment (GH #165).
     let mail_service = MailGrpcService {
-        db,
+        db: db.clone(),
         event_bus,
-        secrets,
+        secrets: secrets.clone(),
         overrides,
     };
-    let mail_interceptor = BearerAuthInterceptor::new(token);
+    let mail_interceptor = BearerAuthInterceptor::new(token.clone());
     let mail_svc = MailServer::with_interceptor(mail_service, mail_interceptor);
 
+    // Filters (backlog story 2.A, GH #171): mounted behind its own
+    // `BearerAuthInterceptor`, exactly like `System`/`Accounts`/`Mail` above
+    // -- see the hard invariant documented on this function's doc comment
+    // (GH #165).
+    let filters_service = FiltersGrpcService { db, filter_engine };
+    let filters_interceptor = BearerAuthInterceptor::new(token);
+    let filters_svc = FiltersServer::with_interceptor(filters_service, filters_interceptor);
+
     Server::builder()
         .add_service(system_svc)
         .add_service(accounts_svc)
         .add_service(mail_svc)
+        .add_service(filters_svc)
         .serve_with_incoming(TcpListenerStream::new(listener))
         .await
         .map_err(GrpcServeError::Transport)
@@ -837,12 +1085,15 @@ mod tests {
     use nuncio_proto::v1::system_client::SystemClient;
     use tonic::Code;
 
-    /// Spawns a test server backed by a fresh ephemeral database and a
-    /// fresh [`SecretManager::mock`] vault -- the right default for every
-    /// test that only exercises `System` or doesn't care about pre-existing
-    /// account/keyring state. Tests that DO care (persistence-across-restart,
-    /// credential-secrecy) build their own `db`/`secrets` and call
-    /// [`spawn_test_server_with`] directly instead.
+    /// Spawns a test server backed by a fresh ephemeral database, a fresh
+    /// empty [`FilterEngine`], and a fresh [`SecretManager::mock`] vault --
+    /// the right default for every test that only exercises
+    /// `System`/`Accounts`/`Mail` or doesn't care about pre-existing
+    /// account/keyring/filter-rule state. Tests that DO care
+    /// (persistence-across-restart, credential-secrecy, the `Filters`
+    /// service itself) build their own `db`/`secrets`/`filter_engine` and
+    /// call [`spawn_test_server_with`] / [`spawn_test_server_with_overrides`]
+    /// directly instead.
     async fn spawn_test_server(
         event_bus: Arc<EventBus>,
         token: &str,
@@ -851,21 +1102,25 @@ mod tests {
             .await
             .expect("connect ephemeral test db");
         let secrets = Arc::new(SecretManager::mock());
-        spawn_test_server_with(event_bus, Arc::new(db), secrets, token).await
+        let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
+        spawn_test_server_with(event_bus, Arc::new(db), filter_engine, secrets, token).await
     }
 
     /// Spawns a test server on an ephemeral loopback port backed by the
-    /// given `db` and `secrets`, so tests can share (and re-open) the same
-    /// database path / mock keyring state across multiple server instances.
+    /// given `db`, `filter_engine`, and `secrets`, so tests can share (and
+    /// re-open) the same database path / engine instance / mock keyring
+    /// state across multiple server instances.
     async fn spawn_test_server_with(
         event_bus: Arc<EventBus>,
         db: Arc<DatabaseEngine>,
+        filter_engine: Arc<FilterEngine>,
         secrets: Arc<SecretManager>,
         token: &str,
     ) -> (std::net::SocketAddr, tokio::task::JoinHandle<()>) {
         spawn_test_server_with_overrides(
             event_bus,
             db,
+            filter_engine,
             secrets,
             token,
             MailEngineOverrides::default(),
@@ -874,14 +1129,15 @@ mod tests {
     }
 
     /// Spawns a test server on an ephemeral loopback port backed by the
-    /// given `db`/`secrets`/`overrides` (backlog story 1.C.6, GH #161): the
-    /// helper every test that injects a [`MailEngineOverrides::mail_backend`]
-    /// / [`MailEngineOverrides::message_sender`] uses to prove `Sync` /
-    /// `SendMessage` drive an injected test double over the real
-    /// authenticated gRPC API.
+    /// given `db`/`filter_engine`/`secrets`/`overrides` (backlog story
+    /// 1.C.6, GH #161): the helper every test that injects a
+    /// [`MailEngineOverrides::mail_backend`] / [`MailEngineOverrides::message_sender`]
+    /// uses to prove `Sync` / `SendMessage` drive an injected test double
+    /// over the real authenticated gRPC API.
     async fn spawn_test_server_with_overrides(
         event_bus: Arc<EventBus>,
         db: Arc<DatabaseEngine>,
+        filter_engine: Arc<FilterEngine>,
         secrets: Arc<SecretManager>,
         token: &str,
         overrides: MailEngineOverrides,
@@ -893,7 +1149,13 @@ mod tests {
         let token = token.to_string();
         let handle = tokio::spawn(async move {
             let _ = serve_on_listener_with_overrides(
-                listener, event_bus, db, secrets, token, overrides,
+                listener,
+                event_bus,
+                db,
+                filter_engine,
+                secrets,
+                token,
+                overrides,
             )
             .await;
         });
@@ -1021,8 +1283,17 @@ mod tests {
             .await
             .expect("connect ephemeral test db");
         let secrets = Arc::new(SecretManager::mock());
+        let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
         let _handle = tokio::spawn(async move {
-            let _ = serve("127.0.0.1:0", event_bus, Arc::new(db), secrets, "token").await;
+            let _ = serve(
+                "127.0.0.1:0",
+                event_bus,
+                Arc::new(db),
+                filter_engine,
+                secrets,
+                "token",
+            )
+            .await;
         });
         // Yield so the spawned task is polled at least once and actually
         // reaches the bind + delegate-to-`serve_on_listener` call before
@@ -1037,10 +1308,12 @@ mod tests {
             .await
             .expect("connect ephemeral test db");
         let secrets = Arc::new(SecretManager::mock());
+        let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
         let err = serve(
             "not-a-valid-addr",
             event_bus,
             Arc::new(db),
+            filter_engine,
             secrets,
             "token",
         )
@@ -1313,6 +1586,7 @@ mod tests {
         let (addr, _handle) = spawn_test_server_with(
             Arc::new(EventBus::new()),
             db_first_run.clone(),
+            Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")),
             secrets.clone(),
             "correct-token",
         )
@@ -1348,6 +1622,7 @@ mod tests {
         let (addr2, _handle2) = spawn_test_server_with(
             Arc::new(EventBus::new()),
             db_second_run,
+            Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")),
             secrets.clone(),
             "correct-token",
         )
@@ -1392,6 +1667,7 @@ mod tests {
         let (addr, _handle) = spawn_test_server_with(
             Arc::new(EventBus::new()),
             db.clone(),
+            Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set")),
             secrets.clone(),
             "correct-token",
         )
@@ -1657,8 +1933,15 @@ mod tests {
         let event_bus = Arc::new(EventBus::new());
         let mut events = event_bus.subscribe_events();
         let secrets = Arc::new(SecretManager::mock());
-        let (addr, _handle) =
-            spawn_test_server_with(event_bus, Arc::new(db), secrets, "correct-token").await;
+        let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
+        let (addr, _handle) = spawn_test_server_with(
+            event_bus,
+            Arc::new(db),
+            filter_engine,
+            secrets,
+            "correct-token",
+        )
+        .await;
 
         let mut client = MailClient::connect(format!("http://{addr}"))
             .await
@@ -1853,8 +2136,15 @@ mod tests {
             .set_secret(&config.keyring_secret_key, "irrelevant-password")
             .expect("store credential in mock vault");
 
-        let (addr, _handle) =
-            spawn_test_server_with(Arc::new(EventBus::new()), db, secrets, "correct-token").await;
+        let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
+        let (addr, _handle) = spawn_test_server_with(
+            Arc::new(EventBus::new()),
+            db,
+            filter_engine,
+            secrets,
+            "correct-token",
+        )
+        .await;
 
         let mut client = MailClient::connect(format!("http://{addr}"))
             .await
@@ -1944,9 +2234,11 @@ mod tests {
             mail_backend: Some(Arc::new(mock_backend)),
             message_sender: None,
         };
+        let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
         let (addr, _handle) = spawn_test_server_with_overrides(
             Arc::new(EventBus::new()),
             db,
+            filter_engine,
             secrets,
             "correct-token",
             overrides,
@@ -2017,9 +2309,11 @@ mod tests {
             mail_backend: None,
             message_sender: Some(Arc::new(mock_sender)),
         };
+        let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
         let (addr, _handle) = spawn_test_server_with_overrides(
             Arc::new(EventBus::new()),
             db,
+            filter_engine,
             secrets,
             "correct-token",
             overrides,
@@ -2071,9 +2365,11 @@ mod tests {
             mail_backend: Some(Arc::new(mock_backend)),
             message_sender: None,
         };
+        let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
         let (addr, _handle) = spawn_test_server_with_overrides(
             Arc::new(EventBus::new()),
             db,
+            filter_engine,
             secrets,
             "correct-token",
             overrides,
@@ -2126,9 +2422,11 @@ mod tests {
             mail_backend: None,
             message_sender: Some(Arc::new(mock_sender)),
         };
+        let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
         let (addr, _handle) = spawn_test_server_with_overrides(
             Arc::new(EventBus::new()),
             db,
+            filter_engine,
             secrets,
             "correct-token",
             overrides,
@@ -2145,4 +2443,406 @@ mod tests {
         assert_eq!(err.code(), Code::Internal);
         assert!(err.message().contains("failed to send message"));
     }
+
+    // ---- Filters (backlog story 2.A, GH #171) ----
+
+    use nuncio_proto::v1::filters_client::FiltersClient;
+    use nuncio_proto::v1::{
+        CreateRuleRequest, DeleteRuleRequest, ListRulesRequest, PreviewRuleRequest,
+        ValidateRuleRequest,
+    };
+
+    const SAMPLE_RULE_NSQL: &str =
+        "WHERE subject CONTAINS 'Urgent' ACTION MARK READ, MOVE TO 'Priority'";
+
+    /// Spawns a test server on an ephemeral loopback port backed by a fresh
+    /// ephemeral database and a fresh empty [`FilterEngine`], returning the
+    /// address AND the shared `filter_engine`/`db` handles so `Filters`
+    /// tests can assert directly on live engine state (proving the
+    /// `ArcSwap` reload actually happened) in addition to driving the gRPC
+    /// API.
+    async fn spawn_filters_test_server(
+        token: &str,
+    ) -> (
+        std::net::SocketAddr,
+        tokio::task::JoinHandle<()>,
+        Arc<DatabaseEngine>,
+        Arc<FilterEngine>,
+    ) {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("connect ephemeral test db");
+        let db = Arc::new(db);
+        let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
+        let secrets = Arc::new(SecretManager::mock());
+        let (addr, handle) = spawn_test_server_with(
+            Arc::new(EventBus::new()),
+            db.clone(),
+            filter_engine.clone(),
+            secrets,
+            token,
+        )
+        .await;
+        (addr, handle, db, filter_engine)
+    }
+
+    #[tokio::test]
+    async fn filters_rpcs_reject_missing_bearer_token() {
+        // Confirms `Filters` is mounted behind its own `BearerAuthInterceptor`
+        // exactly like `System`/`Accounts`/`Mail` (GH #165: no un-intercepted
+        // service).
+        let (addr, _handle, _db, _engine) = spawn_filters_test_server("correct-token").await;
+
+        let mut client = FiltersClient::connect(format!("http://{addr}"))
+            .await
+            .expect("client connects");
+
+        let err = client
+            .create_rule(CreateRuleRequest {
+                name: "Unauthed".to_string(),
+                nsql: SAMPLE_RULE_NSQL.to_string(),
+                priority: 0,
+            })
+            .await
+            .expect_err("missing bearer token must be rejected");
+        assert_eq!(err.code(), Code::Unauthenticated);
+
+        let err = client
+            .list_rules(ListRulesRequest {})
+            .await
+            .expect_err("missing bearer token must be rejected");
+        assert_eq!(err.code(), Code::Unauthenticated);
+
+        let err = client
+            .delete_rule(DeleteRuleRequest {
+                id: "rule-1".to_string(),
+            })
+            .await
+            .expect_err("missing bearer token must be rejected");
+        assert_eq!(err.code(), Code::Unauthenticated);
+
+        let err = client
+            .validate_rule(ValidateRuleRequest {
+                nsql: SAMPLE_RULE_NSQL.to_string(),
+            })
+            .await
+            .expect_err("missing bearer token must be rejected");
+        assert_eq!(err.code(), Code::Unauthenticated);
+
+        let err = client
+            .preview_rule(PreviewRuleRequest {
+                nsql: SAMPLE_RULE_NSQL.to_string(),
+                message_id: None,
+            })
+            .await
+            .expect_err("missing bearer token must be rejected");
+        assert_eq!(err.code(), Code::Unauthenticated);
+    }
+
+    /// End-to-end proof for backlog story 2.A (GH #171): `CreateRule`
+    /// persists AND reloads the live `FilterEngine`'s `ArcSwap` rule set
+    /// (proven by evaluating the SAME `filter_engine` instance directly,
+    /// not just re-reading it back over `ListRules`); `ListRules` reflects
+    /// the persisted rule; `DeleteRule` removes it AND reloads the engine
+    /// back to empty.
+    #[tokio::test]
+    async fn create_list_and_delete_rule_round_trip_and_keep_the_live_engine_in_sync() {
+        let (addr, _handle, _db, filter_engine) = spawn_filters_test_server("correct-token").await;
+
+        let mut client = FiltersClient::connect(format!("http://{addr}"))
+            .await
+            .expect("client connects");
+
+        let sample_email = |subject: &str| nuncio_core::model::Email {
+            id: "msg-filters-1".to_string(),
+            account_id: "acct-1".to_string(),
+            folder_id: "inbox".to_string(),
+            subject: subject.to_string(),
+            sender: "alice@nuncio.mx".to_string(),
+            recipient: "bob@nuncio.mx".to_string(),
+            received_at: 1_700_000_000,
+            read: false,
+            body_plain: Some("body".to_string()),
+            body_html: None,
+            attachments: Vec::new(),
+        };
+
+        // Before creating anything, the live engine matches nothing.
+        assert!(filter_engine
+            .evaluate(&sample_email("Urgent Meeting"))
+            .is_empty());
+
+        let create_response = client
+            .create_rule(authed_bearer_request(CreateRuleRequest {
+                name: "Urgent Rule".to_string(),
+                nsql: SAMPLE_RULE_NSQL.to_string(),
+                priority: 5,
+            }))
+            .await
+            .expect("create_rule succeeds")
+            .into_inner();
+        let created_rule = create_response.rule.expect("response carries the rule");
+        assert_eq!(created_rule.name, "Urgent Rule");
+        assert_eq!(created_rule.priority, 5);
+        assert!(created_rule.enabled);
+        assert_eq!(created_rule.nsql_text, SAMPLE_RULE_NSQL);
+        assert!(created_rule.actions.iter().any(|a| a.contains("MARK READ")));
+        let rule_id = created_rule.id.clone();
+
+        // The live `FilterEngine`'s `ArcSwap` rule set was reloaded --
+        // proven by evaluating the SAME instance the server holds, not a
+        // fresh one.
+        let matches = filter_engine.evaluate(&sample_email("Urgent Meeting"));
+        assert_eq!(matches.len(), 1);
+        assert_eq!(matches[0].0.id, rule_id);
+
+        // ListRules reflects the persisted rule.
+        let list_response = client
+            .list_rules(authed_bearer_request(ListRulesRequest {}))
+            .await
+            .expect("list_rules succeeds")
+            .into_inner();
+        assert_eq!(list_response.rules.len(), 1);
+        assert_eq!(list_response.rules[0].id, rule_id);
+
+        // DeleteRule removes it AND reloads the engine back to empty.
+        client
+            .delete_rule(authed_bearer_request(DeleteRuleRequest {
+                id: rule_id.clone(),
+            }))
+            .await
+            .expect("delete_rule succeeds");
+
+        let list_after_delete = client
+            .list_rules(authed_bearer_request(ListRulesRequest {}))
+            .await
+            .expect("list_rules succeeds")
+            .into_inner();
+        assert!(list_after_delete.rules.is_empty());
+        assert!(filter_engine
+            .evaluate(&sample_email("Urgent Meeting"))
+            .is_empty());
+    }
+
+    #[tokio::test]
+    async fn create_rule_rejects_invalid_nsql_and_persists_nothing() {
+        let (addr, _handle, _db, filter_engine) = spawn_filters_test_server("correct-token").await;
+
+        let mut client = FiltersClient::connect(format!("http://{addr}"))
+            .await
+            .expect("client connects");
+
+        let err = client
+            .create_rule(authed_bearer_request(CreateRuleRequest {
+                name: "Broken Rule".to_string(),
+                nsql: "THIS IS NOT VALID NSQL AT ALL {{{".to_string(),
+                priority: 0,
+            }))
+            .await
+            .expect_err("invalid NSQL must be rejected");
+        assert_eq!(err.code(), Code::InvalidArgument);
+
+        let list_response = client
+            .list_rules(authed_bearer_request(ListRulesRequest {}))
+            .await
+            .expect("list_rules succeeds")
+            .into_inner();
+        assert!(
+            list_response.rules.is_empty(),
+            "an invalid rule must never be persisted"
+        );
+        assert!(filter_engine
+            .evaluate(&nuncio_core::model::Email {
+                id: "msg-1".to_string(),
+                account_id: "acct-1".to_string(),
+                folder_id: "inbox".to_string(),
+                subject: "anything".to_string(),
+                sender: "a@b.com".to_string(),
+                recipient: "c@d.com".to_string(),
+                received_at: 0,
+                read: false,
+                body_plain: None,
+                body_html: None,
+                attachments: Vec::new(),
+            })
+            .is_empty());
+    }
+
+    #[tokio::test]
+    async fn delete_rule_rejects_empty_id() {
+        let (addr, _handle, _db, _engine) = spawn_filters_test_server("correct-token").await;
+
+        let mut client = FiltersClient::connect(format!("http://{addr}"))
+            .await
+            .expect("client connects");
+
+        let err = client
+            .delete_rule(authed_bearer_request(DeleteRuleRequest {
+                id: String::new(),
+            }))
+            .await
+            .expect_err("empty id must be rejected");
+        assert_eq!(err.code(), Code::InvalidArgument);
+    }
+
+    /// `ValidateRule` never fails the RPC itself -- an invalid rule is an
+    /// honest `valid: false` result, not a `Status` error (see this RPC's
+    /// doc comment). Also proves validation never persists anything.
+    #[tokio::test]
+    async fn validate_rule_reports_honest_ok_and_error_verdicts_without_persisting() {
+        let (addr, _handle, _db, _engine) = spawn_filters_test_server("correct-token").await;
+
+        let mut client = FiltersClient::connect(format!("http://{addr}"))
+            .await
+            .expect("client connects");
+
+        let valid_response = client
+            .validate_rule(authed_bearer_request(ValidateRuleRequest {
+                nsql: SAMPLE_RULE_NSQL.to_string(),
+            }))
+            .await
+            .expect("validate_rule call itself succeeds for a valid rule")
+            .into_inner();
+        assert!(valid_response.valid);
+        assert!(valid_response.error.is_empty());
+
+        let invalid_response = client
+            .validate_rule(authed_bearer_request(ValidateRuleRequest {
+                nsql: "THIS IS NOT VALID NSQL AT ALL {{{".to_string(),
+            }))
+            .await
+            .expect("validate_rule call itself succeeds even for an invalid rule")
+            .into_inner();
+        assert!(!invalid_response.valid);
+        assert!(!invalid_response.error.is_empty());
+
+        // Nothing was persisted by either validation call.
+        let list_response = client
+            .list_rules(authed_bearer_request(ListRulesRequest {}))
+            .await
+            .expect("list_rules succeeds")
+            .into_inner();
+        assert!(
+            list_response.rules.is_empty(),
+            "validate_rule must never persist a rule"
+        );
+    }
+
+    /// `PreviewRule` dry-run evaluates an ad hoc rule against a stored
+    /// message (proving it reads REAL persisted mail via the same store
+    /// `Mail/GetMessage` reads) without ever persisting the rule itself.
+    #[tokio::test]
+    async fn preview_rule_dry_runs_against_a_stored_message_without_persisting() {
+        let (addr, _handle, db, _engine) = spawn_filters_test_server("correct-token").await;
+
+        db.save_email(&nuncio_core::model::Email {
+            id: "msg-preview-1".to_string(),
+            account_id: "acct-1".to_string(),
+            folder_id: "inbox".to_string(),
+            subject: "Urgent Board Meeting".to_string(),
+            sender: "alice@nuncio.mx".to_string(),
+            recipient: "bob@nuncio.mx".to_string(),
+            received_at: 1_700_000_000,
+            read: false,
+            body_plain: Some("Please review the attached deck".to_string()),
+            body_html: None,
+            attachments: Vec::new(),
+        })
+        .await
+        .expect("seed message");
+
+        let mut client = FiltersClient::connect(format!("http://{addr}"))
+            .await
+            .expect("client connects");
+
+        let response = client
+            .preview_rule(authed_bearer_request(PreviewRuleRequest {
+                nsql: SAMPLE_RULE_NSQL.to_string(),
+                message_id: Some("msg-preview-1".to_string()),
+            }))
+            .await
+            .expect("preview_rule succeeds")
+            .into_inner();
+        assert_eq!(response.message_id, "msg-preview-1");
+        assert!(response.matched);
+        assert!(response
+            .actions_evaluated
+            .iter()
+            .any(|a| a.contains("MARK READ")));
+        assert!(!response.condition_traces.is_empty());
+
+        // Preview evaluates a NON-matching message honestly too.
+        let non_matching = client
+            .preview_rule(authed_bearer_request(PreviewRuleRequest {
+                nsql: "WHERE subject CONTAINS 'Nonexistent Term' ACTION DELETE".to_string(),
+                message_id: Some("msg-preview-1".to_string()),
+            }))
+            .await
+            .expect("preview_rule succeeds")
+            .into_inner();
+        assert!(!non_matching.matched);
+
+        // Previewing must never persist the rule.
+        let list_response = client
+            .list_rules(authed_bearer_request(ListRulesRequest {}))
+            .await
+            .expect("list_rules succeeds")
+            .into_inner();
+        assert!(
+            list_response.rules.is_empty(),
+            "preview_rule must never persist a rule"
+        );
+    }
+
+    /// When no `message_id` is given (or it does not resolve to a stored
+    /// message), `PreviewRule` falls back to a fixed synthetic sample
+    /// message rather than failing the RPC.
+    #[tokio::test]
+    async fn preview_rule_falls_back_to_a_synthetic_message_when_none_is_stored() {
+        let (addr, _handle, _db, _engine) = spawn_filters_test_server("correct-token").await;
+
+        let mut client = FiltersClient::connect(format!("http://{addr}"))
+            .await
+            .expect("client connects");
+
+        let response = client
+            .preview_rule(authed_bearer_request(PreviewRuleRequest {
+                nsql: "WHERE subject CONTAINS 'Test Subject' ACTION FLAG".to_string(),
+                message_id: None,
+            }))
+            .await
+            .expect("preview_rule succeeds with no message_id")
+            .into_inner();
+        assert!(response.matched);
+        assert_eq!(response.message_id, "msg-test");
+
+        let response_missing_id = client
+            .preview_rule(authed_bearer_request(PreviewRuleRequest {
+                nsql: "WHERE subject CONTAINS 'Test Subject' ACTION FLAG".to_string(),
+                message_id: Some("does-not-exist".to_string()),
+            }))
+            .await
+            .expect("preview_rule succeeds with an unresolvable message_id")
+            .into_inner();
+        assert!(response_missing_id.matched);
+        assert_eq!(response_missing_id.message_id, "does-not-exist");
+    }
+
+    #[tokio::test]
+    async fn preview_rule_and_validate_rule_reject_invalid_nsql_syntax() {
+        let (addr, _handle, _db, _engine) = spawn_filters_test_server("correct-token").await;
+
+        let mut client = FiltersClient::connect(format!("http://{addr}"))
+            .await
+            .expect("client connects");
+
+        let err = client
+            .preview_rule(authed_bearer_request(PreviewRuleRequest {
+                nsql: "THIS IS NOT VALID NSQL AT ALL {{{".to_string(),
+                message_id: None,
+            }))
+            .await
+            .expect_err("invalid NSQL must be rejected");
+        assert_eq!(err.code(), Code::InvalidArgument);
+    }
 }

crates/nunciod/src/main.rs

@@ -380,16 +380,23 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     let grpc_token = hex::encode(grpc_token_bytes);
     let grpc_addr = nunciod::grpc::grpc_addr_from_env();
     tracing::info!(
-        "nunciod gRPC (nuncio.v1.System, nuncio.v1.Accounts) starting on {} (loopback only)",
+        "nunciod gRPC (nuncio.v1.System, nuncio.v1.Accounts, nuncio.v1.Mail, nuncio.v1.Filters) \
+         starting on {} (loopback only)",
         grpc_addr
     );
     let grpc_event_bus = event_bus.clone();
     let grpc_db = db.clone();
+    // The gRPC `Filters` service (backlog story 2.A, GH #171) shares this
+    // SAME `filter_engine` instance with the JSON-RPC `filter.*` handler
+    // above, so a rule created/deleted through EITHER transport reloads the
+    // one live `ArcSwap`-backed rule set both transports evaluate against.
+    let grpc_filter_engine = filter_engine.clone();
     let _grpc_task = tokio::spawn(async move {
         if let Err(e) = nunciod::grpc::serve(
             &grpc_addr,
             grpc_event_bus,
             grpc_db,
+            grpc_filter_engine,
             grpc_secrets,
             grpc_token,
         )

crates/nunciod/tests/daemon_e2e_test.rs

@@ -14,6 +14,7 @@
 use nuncio_cli::{AccountSubcommand, Commands, HeadlessRunner, SystemSubcommand};
 use nuncio_core::ipc::{IpcClient, IpcDaemonServer};
 use nuncio_core::{CoreCommand, EventBus};
+use nuncio_filter::FilterEngine;
 use nuncio_store::db::DatabaseEngine;
 use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT};
 use std::sync::Arc;
@@ -157,11 +158,13 @@ async fn cli_system_status_round_trips_over_grpc_to_live_daemon() {
         .await
         .expect("connect ephemeral test db");
     let server_secrets = secrets.clone();
+    let server_filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
     let _server_handle = tokio::spawn(async move {
         let _ = nunciod::grpc::serve_on_listener(
             listener,
             server_event_bus,
             Arc::new(server_db),
+            server_filter_engine,
             server_secrets,
             token,
         )

crates/nunciod/tests/grpc_auth_test.rs

@@ -14,6 +14,7 @@
 #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
 
 use nuncio_core::{CoreCommand, EngineStatus, EventBus};
+use nuncio_filter::FilterEngine;
 use nuncio_proto::v1::system_client::SystemClient;
 use nuncio_proto::v1::GetStatusRequest;
 use nuncio_store::db::DatabaseEngine;
@@ -40,8 +41,17 @@ async fn start_server(event_bus: Arc<EventBus>, token: String) -> std::net::Sock
         .await
         .expect("connect ephemeral test db");
     let secrets = Arc::new(SecretManager::mock());
+    let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
     tokio::spawn(async move {
-        let _ = serve_on_listener(listener, event_bus, Arc::new(db), secrets, token).await;
+        let _ = serve_on_listener(
+            listener,
+            event_bus,
+            Arc::new(db),
+            filter_engine,
+            secrets,
+            token,
+        )
+        .await;
     });
     addr
 }

crates/nunciod/tests/spine_e2e_test.rs

@@ -34,6 +34,7 @@
 
 use nuncio_core::model::{Email, Folder};
 use nuncio_core::EventBus;
+use nuncio_filter::FilterEngine;
 use nuncio_mail::{MockMailBackend, MockMessageSender};
 use nuncio_proto::v1::event::Kind;
 use nuncio_proto::v1::{
@@ -115,11 +116,13 @@ async fn boot_daemon(
     let event_bus = Arc::new(EventBus::new());
     let server_secrets = secrets.clone();
     let server_token = token.clone();
+    let server_filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
     tokio::spawn(async move {
         let _ = serve_on_listener_with_overrides(
             listener,
             event_bus,
             db,
+            server_filter_engine,
             server_secrets,
             server_token,
             overrides,