crates/nuncio-cli/src/args.rs

@@ -517,6 +517,18 @@ pub enum FilterSubcommand {
         #[arg(short, long, default_value_t = 50, help = "Log limit")]
         limit: usize,
     },
+    /// Retroactively apply the current rule set to every stored message,
+    /// streaming cumulative progress as the scan runs.
+    Triage {
+        /// Restrict the scan's reporting to a single rule ID. Not yet
+        /// honored by the daemon (see `nuncio.v1.Filters.Triage`) -- the
+        /// scan always evaluates every active rule.
+        #[arg(short, long, help = "Rule identifier (not yet honored)")]
+        rule_id: Option<String>,
+        /// Messages fetched per streamed progress update.
+        #[arg(short, long, help = "Messages scanned per chunk")]
+        chunk_size: Option<u32>,
+    },
 }
 
 #[cfg(test)]

crates/nuncio-cli/src/runner.rs

@@ -512,6 +512,13 @@ impl HeadlessRunner {
                 FilterSubcommand::Logs { limit } => {
                     self.handle_filter_logs(*limit, json_mode).await
                 }
+                FilterSubcommand::Triage {
+                    rule_id,
+                    chunk_size,
+                } => {
+                    self.handle_filter_triage(rule_id.clone(), *chunk_size, json_mode)
+                        .await
+                }
             },
             Commands::Update { action } => match action {
                 UpdateSubcommand::Check => match nuncio_core::UpdateEngine::new() {
@@ -1484,6 +1491,93 @@ impl HeadlessRunner {
         }
     }
 
+    /// Drains the server-streaming `Filters.Triage` RPC, printing (or
+    /// accumulating, in `--json` mode) each `TriageProgress` update as it
+    /// arrives so the caller can watch a retroactive rescan of the whole
+    /// message store progress in real time rather than blocking silently
+    /// until it finishes.
+    async fn handle_filter_triage(
+        &self,
+        rule_id: Option<String>,
+        chunk_size: Option<u32>,
+        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),
+        };
+
+        let mut stream = match client
+            .triage(nuncio_proto::v1::TriageRequest {
+                rule_id,
+                chunk_size: chunk_size.unwrap_or(0),
+            })
+            .await
+        {
+            Ok(response) => response.into_inner(),
+            Err(status) => {
+                return Self::render_error(
+                    &format!("nunciod daemon rejected triage: {status}"),
+                    json_mode,
+                )
+            }
+        };
+
+        let mut updates = Vec::new();
+        loop {
+            match stream.message().await {
+                Ok(Some(progress)) => {
+                    if !json_mode {
+                        println!(
+                            "scanned={} matched={} actions_applied={} last_message_id={} done={}",
+                            progress.scanned_count,
+                            progress.matched_count,
+                            progress.actions_applied_count,
+                            progress.last_message_id,
+                            progress.done
+                        );
+                    }
+                    let done = progress.done;
+                    updates.push(progress);
+                    if done {
+                        break;
+                    }
+                }
+                Ok(None) => break,
+                Err(status) => {
+                    return Self::render_error(
+                        &format!("nunciod daemon triage stream failed: {status}"),
+                        json_mode,
+                    )
+                }
+            }
+        }
+
+        if json_mode {
+            let updates_json: Vec<serde_json::Value> = updates
+                .iter()
+                .map(|p| {
+                    json!({
+                        "scanned_count": p.scanned_count,
+                        "matched_count": p.matched_count,
+                        "actions_applied_count": p.actions_applied_count,
+                        "last_message_id": p.last_message_id,
+                        "done": p.done,
+                    })
+                })
+                .collect();
+            format_json(&json!({ "updates": updates_json }))
+        } else {
+            match updates.last() {
+                Some(last) => format!(
+                    "Triage complete: scanned={} matched={} actions_applied={}",
+                    last.scanned_count, last.matched_count, last.actions_applied_count
+                ),
+                None => "Triage produced no progress updates.".to_string(),
+            }
+        }
+    }
+
     /// Resolves the gRPC bearer token from the injected vault and dials the
     /// `nuncio.v1.Filters` service at `self.grpc_addr`, shared by every
     /// `filter` handler above.
@@ -3235,8 +3329,8 @@ mod tests {
             ExportRulesRequest, ExportRulesResponse, FilterExecutionLog as FilterExecutionLogProto,
             FilterRule as FilterRuleProto, GetExecutionLogsRequest, GetExecutionLogsResponse,
             ImportRulesRequest, ImportRulesResponse, ListRulesRequest, ListRulesResponse,
-            PreviewRuleRequest, PreviewRuleResponse, UpdateRuleRequest, UpdateRuleResponse,
-            ValidateRuleRequest, ValidateRuleResponse,
+            PreviewRuleRequest, PreviewRuleResponse, TriageProgress, TriageRequest,
+            UpdateRuleRequest, UpdateRuleResponse, ValidateRuleRequest, ValidateRuleResponse,
         };
         use std::sync::Mutex;
 
@@ -3407,6 +3501,37 @@ mod tests {
                     }],
                 }))
             }
+
+            type TriageStream = std::pin::Pin<
+                Box<
+                    dyn tokio_stream::Stream<Item = Result<TriageProgress, tonic::Status>>
+                        + Send
+                        + 'static,
+                >,
+            >;
+
+            async fn triage(
+                &self,
+                _request: tonic::Request<TriageRequest>,
+            ) -> Result<tonic::Response<Self::TriageStream>, tonic::Status> {
+                let updates = vec![
+                    Ok(TriageProgress {
+                        scanned_count: 2,
+                        matched_count: 1,
+                        actions_applied_count: 1,
+                        last_message_id: "msg-stub-2".to_string(),
+                        done: false,
+                    }),
+                    Ok(TriageProgress {
+                        scanned_count: 3,
+                        matched_count: 2,
+                        actions_applied_count: 2,
+                        last_message_id: "msg-stub-3".to_string(),
+                        done: true,
+                    }),
+                ];
+                Ok(tonic::Response::new(Box::pin(tokio_stream::iter(updates))))
+            }
         }
 
         let stub = StubFilters::default();
@@ -3647,6 +3772,25 @@ mod tests {
             .await;
         assert!(logs_out.contains("msg-stub-1"));
         assert!(logs_out.contains(r#""action_taken":"MARK READ""#));
+
+        // `filter triage`: proves the CLI drains the server-streaming
+        // `Triage` RPC (its first real streaming-RPC consumer) and reports
+        // the final cumulative progress rather than only the first chunk.
+        let triage_out = runner
+            .execute_command(
+                &Commands::Filter {
+                    action: FilterSubcommand::Triage {
+                        rule_id: None,
+                        chunk_size: Some(2),
+                    },
+                },
+                true,
+            )
+            .await;
+        assert!(triage_out.contains(r#""scanned_count":3"#));
+        assert!(triage_out.contains(r#""matched_count":2"#));
+        assert!(triage_out.contains(r#""actions_applied_count":2"#));
+        assert!(triage_out.contains(r#""done":true"#));
     }
 
     #[tokio::test]

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

@@ -479,6 +479,13 @@ service Filters {
   // log entries (hash-chain ledger), newest-eligible-first as stored,
   // capped at `limit`.
   rpc GetExecutionLogs(GetExecutionLogsRequest) returns (GetExecutionLogsResponse);
+  // Triage scans every stored message in bounded chunks, evaluating the
+  // daemon's live `FilterEngine` against each one and applying the same
+  // actions live sync would -- so the currently active rule set can be
+  // applied retroactively to mail that was synced before those rules
+  // existed. Streams cumulative progress as it works, ending with a final
+  // `done: true` message once the whole store has been scanned.
+  rpc Triage(TriageRequest) returns (stream TriageProgress);
 }
 
 // FilterRule mirrors `nuncio_filter::FilterRule`'s persisted/user-facing
@@ -675,6 +682,36 @@ message FilterExecutionLog {
   string hash = 7;
 }
 
+// TriageRequest configures a retroactive rescan of the whole message store.
+message TriageRequest {
+  // Scope the scan to a single rule's actions. Currently NOT honored --
+  // `Triage` always evaluates a message against the full live rule set,
+  // matching what a real (non-retroactive) sync run would do. Reserved for
+  // forward compatibility; a client that sets this today gets a whole-store
+  // scan against all active rules, not an error.
+  optional string rule_id = 1;
+  // Number of messages fetched per keyset page. 0 selects the server
+  // default (100).
+  uint32 chunk_size = 2;
+}
+
+// TriageProgress reports the real, cumulative state of an in-progress (or
+// just-finished) `Triage` scan. Every field is the true running total as of
+// the chunk just processed -- never an estimate or a fabricated count.
+message TriageProgress {
+  // Total messages scanned so far.
+  uint64 scanned_count = 1;
+  // Total messages that had at least one action applied or enqueued.
+  uint64 matched_count = 2;
+  // Total individual actions applied or enqueued across all messages.
+  uint64 actions_applied_count = 3;
+  // ID of the last message scanned in the most recently processed chunk.
+  string last_message_id = 4;
+  // True only on the final message of the stream, once the whole store has
+  // been scanned.
+  bool done = 5;
+}
+
 // Export exposes the daemon's real mailbox export engine
 // (`nuncio_core::export`, via `DatabaseEngine::export_messages_to_file`)
 // over authenticated gRPC. The daemon and its caller run on the SAME host

crates/nunciod/src/grpc.rs

@@ -40,8 +40,9 @@ use nuncio_proto::v1::{
     Message as MessageProto, MessageFlagsChanged, MessageSearchHit, PreviewRuleRequest,
     PreviewRuleResponse, SearchMessagesRequest, SearchMessagesResponse, SendMessageRequest,
     SendMessageResponse, ShuttingDown, SubscribeRequest, SyncCompleted, SyncRequest, SyncResponse,
-    SyncStarted, TlsMode as TlsModeProto, UpdateAvailable, UpdateRuleRequest, UpdateRuleResponse,
-    ValidateRuleRequest, ValidateRuleResponse, VerifyChainRequest, VerifyChainResponse,
+    SyncStarted, TlsMode as TlsModeProto, TriageProgress, TriageRequest, UpdateAvailable,
+    UpdateRuleRequest, UpdateRuleResponse, ValidateRuleRequest, ValidateRuleResponse,
+    VerifyChainRequest, VerifyChainResponse,
 };
 use nuncio_store::db::DatabaseEngine;
 use nuncio_store::search::SearchEngine;
@@ -53,7 +54,7 @@ use subtle::ConstantTimeEq;
 use thiserror::Error;
 use tokio::net::TcpListener;
 use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
-use tokio_stream::wrappers::{BroadcastStream, TcpListenerStream};
+use tokio_stream::wrappers::{BroadcastStream, ReceiverStream, TcpListenerStream};
 use tokio_stream::{Stream, StreamExt};
 use tonic::transport::Server;
 use tonic::{Request, Response, Status};
@@ -1472,6 +1473,102 @@ impl Filters for FiltersGrpcService {
 
         Ok(Response::new(GetExecutionLogsResponse { logs }))
     }
+
+    /// Server-streaming retroactive rescan of the whole message store.
+    /// Unlike `System::subscribe` (which wraps an already-live broadcast),
+    /// this RPC actively drives the scan: a spawned worker task walks
+    /// `db.get_message_chunk` in keyset pages, applies the shared
+    /// [`crate::sync::apply_filter_actions`] evaluate-and-route logic to
+    /// each message (the SAME side effects live sync produces), and pushes
+    /// cumulative progress onto an mpsc channel that this method wraps as
+    /// the returned stream.
+    ///
+    /// `rule_id` is accepted for forward compatibility but NOT currently
+    /// honored: `apply_filter_actions` always evaluates against the whole
+    /// live rule set, matching what a real sync pass would do, rather than
+    /// silently narrowing to one rule.
+    ///
+    /// A `get_message_chunk` failure sends a single `Err(Status::internal)`
+    /// on the stream and stops the scan -- never a fabricated `done: true`
+    /// on error. The receiver disconnecting (the caller dropped the call)
+    /// also stops the worker rather than continuing to scan for no one.
+    type TriageStream =
+        Pin<Box<dyn Stream<Item = Result<TriageProgress, Status>> + Send + 'static>>;
+
+    async fn triage(
+        &self,
+        request: Request<TriageRequest>,
+    ) -> Result<Response<Self::TriageStream>, Status> {
+        let req = request.into_inner();
+        let chunk_size = if req.chunk_size == 0 {
+            100
+        } else {
+            req.chunk_size as usize
+        };
+
+        let (tx, rx) = tokio::sync::mpsc::channel::<Result<TriageProgress, Status>>(16);
+        let db = self.db.clone();
+        let engine = self.filter_engine.clone();
+
+        tokio::spawn(async move {
+            let mut last_id = String::new();
+            let mut scanned: u64 = 0;
+            let mut matched: u64 = 0;
+            let mut applied: u64 = 0;
+
+            loop {
+                let batch = match db.get_message_chunk(&last_id, chunk_size).await {
+                    Ok(batch) => batch,
+                    Err(e) => {
+                        let _ = tx
+                            .send(Err(Status::internal(format!(
+                                "failed to read message chunk during triage: {e}"
+                            ))))
+                            .await;
+                        return;
+                    }
+                };
+                let is_last_page = batch.len() < chunk_size;
+
+                for email in &batch {
+                    let actions_for_email =
+                        crate::sync::apply_filter_actions(&db, &engine, email).await;
+                    scanned += 1;
+                    applied += actions_for_email as u64;
+                    if actions_for_email > 0 {
+                        matched += 1;
+                    }
+                }
+
+                if let Some(last) = batch.last() {
+                    last_id = last.id.clone();
+                }
+
+                let done = is_last_page;
+                if tx
+                    .send(Ok(TriageProgress {
+                        scanned_count: scanned,
+                        matched_count: matched,
+                        actions_applied_count: applied,
+                        last_message_id: last_id.clone(),
+                        done,
+                    }))
+                    .await
+                    .is_err()
+                {
+                    // Receiver dropped: the caller disconnected, so there is
+                    // no one left to report progress to.
+                    return;
+                }
+
+                if is_last_page {
+                    return;
+                }
+            }
+        });
+
+        Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
+    }
 }
 
 /// Maps a wire-format `nuncio.v1.ExportFormat` enum value back onto

crates/nunciod/tests/filters_triage_e2e_test.rs

@@ -0,0 +1,267 @@
+//! Offline end-to-end test for the `nuncio.v1.Filters.Triage` streaming RPC.
+//!
+//! Boots a real `nunciod` daemon gRPC server (`nunciod::grpc::
+//! serve_on_listener`) on an ephemeral loopback port, backed by a real
+//! temp file-backed [`DatabaseEngine`] seeded directly via
+//! [`DatabaseEngine::save_email`] (standing in for messages that were synced
+//! before the active rule set existed) and a live [`FilterEngine`] carrying
+//! a real persisted rule.
+//!
+//! Drives ONLY the authenticated `nuncio-proto` `FiltersClient::triage` call
+//! against the live server, proving:
+//!   a. the final cumulative `matched_count`/`actions_applied_count` are
+//!      correct against a store containing both matching and non-matching
+//!      messages;
+//!   b. matching messages show the REAL side effects `apply_filter_actions`
+//!      produces (the read flag flips for an immediate `MARK READ` action;
+//!      an outbox row is queued for a remote `MOVE TO` action) -- not a
+//!      fabricated count with no backing state change;
+//!   c. a store bigger than one `chunk_size` genuinely streams more than one
+//!      non-final `TriageProgress` message, proving this is a real
+//!      multi-chunk stream rather than one batch dressed up as a stream.
+//!
+//! Fully offline and deterministic: no real network, no real OS keyring
+//! (`SecretManager::mock`), no fixed sleeps.
+
+#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+
+use nuncio_core::model::Email;
+use nuncio_core::EventBus;
+use nuncio_filter::{FilterEngine, NsqlParser};
+use nuncio_proto::v1::filters_client::FiltersClient;
+use nuncio_proto::v1::TriageRequest;
+use nuncio_store::db::DatabaseEngine;
+use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT};
+use nunciod::grpc::serve_on_listener;
+use std::sync::Arc;
+use tokio::net::TcpListener;
+
+const ACCOUNT_ID: &str = "acct-triage-e2e-1";
+
+fn sample_email(id: &str, subject: &str) -> Email {
+    Email {
+        id: id.to_string(),
+        account_id: ACCOUNT_ID.to_string(),
+        folder_id: "inbox".to_string(),
+        subject: subject.to_string(),
+        sender: "alice@nuncio.mx".to_string(),
+        recipient: "triage-e2e@nuncio.mx".to_string(),
+        received_at: 1_700_000_000,
+        read: false,
+        body_plain: Some("body".to_string()),
+        body_html: None,
+        attachments: Vec::new(),
+    }
+}
+
+/// Boots the real daemon gRPC server (bearer-token authenticated) on an
+/// ephemeral loopback port, backed by a real temp file-backed
+/// `DatabaseEngine` and the given live `FilterEngine`. Returns the server
+/// address, the minted token, the `DatabaseEngine` handle (used to seed
+/// messages before triggering `Triage`), and the `TempDir` guard, which the
+/// caller must hold for as long as it talks to the server.
+async fn boot_daemon(
+    secrets: Arc<SecretManager>,
+    filter_engine: Arc<FilterEngine>,
+) -> (
+    std::net::SocketAddr,
+    String,
+    Arc<DatabaseEngine>,
+    tempfile::TempDir,
+) {
+    let dir = tempfile::tempdir().expect("create temp dir for file-backed db");
+    let db_path = dir.path().join("triage_e2e.db");
+    let db = Arc::new(
+        DatabaseEngine::connect_file(&db_path, &secrets)
+            .await
+            .expect("open temp file-backed database"),
+    );
+
+    let token_bytes = secrets
+        .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32)
+        .expect("mint gRPC bearer token from mock vault");
+    let token = hex::encode(token_bytes);
+
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind ephemeral loopback port");
+    let addr = listener.local_addr().expect("listener has local addr");
+
+    let event_bus = Arc::new(EventBus::new());
+    let server_db = db.clone();
+    let server_secrets = secrets.clone();
+    let server_token = token.clone();
+    tokio::spawn(async move {
+        let _ = serve_on_listener(
+            listener,
+            event_bus,
+            server_db,
+            filter_engine,
+            server_secrets,
+            server_token,
+        )
+        .await;
+    });
+
+    (addr, token, db, dir)
+}
+
+fn authed_request<T>(token: &str, message: T) -> tonic::Request<T> {
+    let mut request = tonic::Request::new(message);
+    request.metadata_mut().insert(
+        "authorization",
+        format!("Bearer {token}")
+            .parse()
+            .expect("valid ascii metadata value"),
+    );
+    request
+}
+
+#[tokio::test]
+async fn triage_applies_live_rule_retroactively_and_reports_real_cumulative_counts() {
+    let secrets = Arc::new(SecretManager::mock());
+
+    let mark_read_rule = NsqlParser::parse_rule(
+        "Urgent Auto-Read",
+        1,
+        "WHERE subject CONTAINS 'Urgent' ACTION MARK READ",
+    )
+    .expect("parse mark-read rule");
+    let move_rule = NsqlParser::parse_rule(
+        "Archive Rule",
+        2,
+        "WHERE subject CONTAINS 'Archive Me' ACTION MOVE TO 'Archive'",
+    )
+    .expect("parse move rule");
+    let filter_engine =
+        Arc::new(FilterEngine::new(vec![mark_read_rule, move_rule]).expect("compile rules"));
+
+    let (addr, token, db, _dir) = boot_daemon(secrets, filter_engine).await;
+
+    // Seed messages directly into the store, standing in for mail that was
+    // synced BEFORE these rules existed -- exactly the scenario `Triage` is
+    // for. Two match (one per rule), one does not.
+    db.save_email(&sample_email("m-urgent", "Urgent: server down"))
+        .await
+        .expect("seed urgent message");
+    db.save_email(&sample_email("m-archive", "Please Archive Me"))
+        .await
+        .expect("seed archive message");
+    db.save_email(&sample_email("m-plain", "Just saying hi"))
+        .await
+        .expect("seed non-matching message");
+
+    let mut client = FiltersClient::connect(format!("http://{addr}"))
+        .await
+        .expect("client connects");
+
+    let mut stream = client
+        .triage(authed_request(
+            &token,
+            TriageRequest {
+                rule_id: None,
+                chunk_size: 100,
+            },
+        ))
+        .await
+        .expect("triage call succeeds")
+        .into_inner();
+
+    let mut updates = Vec::new();
+    while let Some(progress) = stream.message().await.expect("stream message") {
+        updates.push(progress);
+    }
+
+    let last = updates.last().expect("at least one progress update");
+    assert!(last.done, "final update must be marked done");
+    assert_eq!(last.scanned_count, 3);
+    assert_eq!(
+        last.matched_count, 2,
+        "exactly the two messages with a rule match count as matched"
+    );
+    assert_eq!(
+        last.actions_applied_count, 2,
+        "one action applied per matching message"
+    );
+
+    // Real side effects, not a fabricated count: the MARK READ action
+    // actually flipped the stored read flag...
+    let urgent = db.get_message("m-urgent").await.expect("get urgent");
+    assert!(urgent.read, "MARK READ must genuinely flip the read flag");
+
+    // ...and the MOVE TO action actually queued a real outbox mutation.
+    let pending = db
+        .list_pending_mutations(10)
+        .await
+        .expect("list pending mutations");
+    assert_eq!(
+        pending.len(),
+        1,
+        "the remote MOVE action must enqueue exactly one outbox mutation"
+    );
+    assert_eq!(pending[0].message_id, "m-archive");
+
+    let plain = db.get_message("m-plain").await.expect("get plain");
+    assert!(!plain.read, "a non-matching message must be left untouched");
+}
+
+#[tokio::test]
+async fn triage_streams_multiple_chunks_for_a_store_bigger_than_one_chunk() {
+    let secrets = Arc::new(SecretManager::mock());
+    let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
+
+    let (addr, token, db, _dir) = boot_daemon(secrets, filter_engine).await;
+
+    // Five messages against a chunk size of two forces three chunks
+    // (2 + 2 + 1), so the stream must carry at least two non-final progress
+    // updates before the final `done: true` -- proof this is a real
+    // multi-chunk stream, not one batch disguised as one.
+    for i in 0..5 {
+        db.save_email(&sample_email(
+            &format!("m-{i:02}"),
+            &format!("Message number {i}"),
+        ))
+        .await
+        .expect("seed message");
+    }
+
+    let mut client = FiltersClient::connect(format!("http://{addr}"))
+        .await
+        .expect("client connects");
+
+    let mut stream = client
+        .triage(authed_request(
+            &token,
+            TriageRequest {
+                rule_id: None,
+                chunk_size: 2,
+            },
+        ))
+        .await
+        .expect("triage call succeeds")
+        .into_inner();
+
+    let mut updates = Vec::new();
+    while let Some(progress) = stream.message().await.expect("stream message") {
+        updates.push(progress);
+    }
+
+    let non_final: Vec<_> = updates.iter().filter(|p| !p.done).collect();
+    assert!(
+        non_final.len() > 1,
+        "expected more than one non-final progress update for a multi-chunk scan, got {}",
+        non_final.len()
+    );
+
+    let last = updates.last().expect("at least one progress update");
+    assert!(last.done);
+    assert_eq!(last.scanned_count, 5);
+
+    // Cumulative counts must monotonically increase across the stream --
+    // never reset or double-count between chunks.
+    let mut prev_scanned = 0u64;
+    for update in &updates {
+        assert!(update.scanned_count >= prev_scanned);
+        prev_scanned = update.scanned_count;
+    }
+}

crates/nunciod/tests/grpc_auth_test.rs

@@ -187,3 +187,33 @@ async fn rejects_calls_to_new_filters_rpcs_without_a_bearer_token() {
         .expect_err("call without any authorization header must be rejected");
     assert_eq!(err.code(), Code::Unauthenticated);
 }
+
+/// Confirms the streaming `Filters.Triage` RPC is mounted behind the SAME
+/// `BearerAuthInterceptor` as every other `Filters` RPC -- an unauthenticated
+/// call must be rejected before the stream is ever established, exactly
+/// like an unauthenticated unary call.
+#[tokio::test]
+async fn rejects_triage_calls_without_a_bearer_token() {
+    let event_bus = Arc::new(EventBus::new());
+    let secrets = SecretManager::mock();
+    let token = hex::encode(
+        secrets
+            .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32)
+            .expect("token provisioned from mock vault"),
+    );
+
+    let (addr, _dir) = start_server(event_bus, token).await;
+    let mut client =
+        nuncio_proto::v1::filters_client::FiltersClient::connect(format!("http://{addr}"))
+            .await
+            .expect("client connects");
+
+    let err = client
+        .triage(nuncio_proto::v1::TriageRequest {
+            rule_id: None,
+            chunk_size: 0,
+        })
+        .await
+        .expect_err("call without any authorization header must be rejected");
+    assert_eq!(err.code(), Code::Unauthenticated);
+}