crates/nunciod/src/calendar_sync.rs

@@ -43,14 +43,33 @@ pub async fn sync_with_backend(
     start_window: i64,
     end_window: i64,
 ) -> Result<usize, CalendarSyncError> {
+    tracing::debug!(
+        calendar_id = %calendar_id,
+        start_window,
+        end_window,
+        "calendar sync: fetching events from backend"
+    );
     let events = backend
         .fetch_events(calendar_id, start_window, end_window)
-        .await?;
+        .await
+        .inspect_err(|e| {
+            tracing::warn!(
+                calendar_id = %calendar_id,
+                "calendar sync: backend fetch failed: {e}"
+            );
+        })?;
+    let fetched = events.len();
     let mut synced = 0usize;
     for event in events {
         db.save_calendar_event(&event).await?;
         synced += 1;
     }
+    tracing::info!(
+        calendar_id = %calendar_id,
+        fetched,
+        synced,
+        "calendar sync: persisted fetched events"
+    );
     Ok(synced)
 }
 
@@ -72,6 +91,11 @@ pub async fn sync_caldav_account(
     start_window: i64,
     end_window: i64,
 ) -> Result<usize, CalendarSyncError> {
+    tracing::info!(
+        account_id = %account.id,
+        calendar_id = %calendar_id,
+        "calendar sync: dispatching CalDAV sync for account"
+    );
     let client = CalDavClient::new(CalDavAccountConfig {
         account_id: account.id.clone(),
         caldav_url: account.dav_collection_url().unwrap_or_default().to_string(),
@@ -170,4 +194,50 @@ mod tests {
         assert!(matches!(err, CalendarSyncError::Backend(_)));
         assert!(err.to_string().contains("calendar backend error"));
     }
+
+    /// A calendar sync must emit a followable INFO event carrying the calendar
+    /// id and the persisted count, so an operator can trace what a sync did.
+    #[test]
+    fn sync_with_backend_logs_calendar_id_and_synced_count() {
+        use crate::test_tracing::with_recorder;
+        use tracing::Level;
+
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime");
+
+        let (recorder, ()) = with_recorder(|| {
+            runtime.block_on(async {
+                let (db, _dir) = DatabaseEngine::connect_ephemeral()
+                    .await
+                    .expect("ephemeral db");
+                let mock = MockCalendarBackend::new();
+                mock.add_event(mock_event(
+                    "evt-1",
+                    "cal-work",
+                    1_700_000_000,
+                    1_700_003_600,
+                ));
+                sync_with_backend(&db, &mock, "cal-work", 0, i64::MAX)
+                    .await
+                    .expect("sync succeeds");
+            });
+        });
+
+        let events = recorder.events();
+        let persisted = events
+            .iter()
+            .find(|e| e.message() == "calendar sync: persisted fetched events")
+            .expect("a domain event for the persisted sync must be logged");
+        assert_eq!(persisted.level, Level::INFO);
+        assert_eq!(
+            persisted.fields.get("calendar_id").map(String::as_str),
+            Some("cal-work")
+        );
+        assert_eq!(
+            persisted.fields.get("synced").map(String::as_str),
+            Some("1")
+        );
+    }
 }

crates/nunciod/src/contacts_sync.rs

@@ -43,12 +43,28 @@ pub async fn sync_with_backend(
     backend: &dyn ContactsBackend,
     account_id: &str,
 ) -> Result<usize, ContactsSyncError> {
-    let contacts = backend.fetch_contacts(account_id).await?;
+    tracing::info!(
+        account_id = %account_id,
+        "contacts sync: dispatching sync for account"
+    );
+    let contacts = backend.fetch_contacts(account_id).await.inspect_err(|e| {
+        tracing::warn!(
+            account_id = %account_id,
+            "contacts sync: backend fetch failed: {e}"
+        );
+    })?;
+    let fetched = contacts.len();
     let mut synced = 0usize;
     for contact in contacts {
         db.save_contact(&contact).await?;
         synced += 1;
     }
+    tracing::info!(
+        account_id = %account_id,
+        fetched,
+        synced,
+        "contacts sync: persisted fetched contacts"
+    );
     Ok(synced)
 }
 
@@ -120,4 +136,45 @@ mod tests {
         assert!(matches!(err, ContactsSyncError::Backend(_)));
         assert!(err.to_string().contains("contacts backend error"));
     }
+
+    /// A contacts sync must emit a followable INFO event carrying the account
+    /// id and the persisted count.
+    #[test]
+    fn sync_with_backend_logs_account_id_and_synced_count() {
+        use crate::test_tracing::with_recorder;
+        use tracing::Level;
+
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime");
+
+        let (recorder, ()) = with_recorder(|| {
+            runtime.block_on(async {
+                let (db, _dir) = DatabaseEngine::connect_ephemeral()
+                    .await
+                    .expect("ephemeral db");
+                let mock = MockContactsBackend::new();
+                mock.add_contact(mock_contact("ct-1", "acct-contacts-1", "Alice"));
+                sync_with_backend(&db, &mock, "acct-contacts-1")
+                    .await
+                    .expect("sync succeeds");
+            });
+        });
+
+        let events = recorder.events();
+        let persisted = events
+            .iter()
+            .find(|e| e.message() == "contacts sync: persisted fetched contacts")
+            .expect("a domain event for the persisted sync must be logged");
+        assert_eq!(persisted.level, Level::INFO);
+        assert_eq!(
+            persisted.fields.get("account_id").map(String::as_str),
+            Some("acct-contacts-1")
+        );
+        assert_eq!(
+            persisted.fields.get("synced").map(String::as_str),
+            Some("1")
+        );
+    }
 }

crates/nunciod/src/grpc.rs

@@ -68,6 +68,7 @@ use tokio_stream::{Stream, StreamExt};
 use tonic::service::interceptor::InterceptedService;
 use tonic::transport::Server;
 use tonic::{Request, Response, Status};
+use tracing::Instrument;
 
 // Resource bounds applied to the tonic `Server` builder in
 // `serve_on_listener_with_overrides_and_shutdown`. These exist so a
@@ -298,6 +299,17 @@ fn map_tls_mode_from_proto(mode: TlsModeProto) -> Result<nuncio_core::TlsMode, S
     }
 }
 
+/// Short, non-sensitive label for an account's transport kind, safe to log
+/// (carries no host, credential, or address -- just which protocol family the
+/// account uses).
+fn transport_kind(transport: &nuncio_core::Transport) -> &'static str {
+    match transport {
+        nuncio_core::Transport::ImapSmtp(_) => "imap_smtp",
+        nuncio_core::Transport::Jmap(_) => "jmap",
+        nuncio_core::Transport::Dav(_) => "dav",
+    }
+}
+
 /// Maps a `nuncio_core::AccountConfig` onto its wire-format
 /// `nuncio.v1.AccountConfig` representation for `ListAccounts` responses.
 /// The password credential is NEVER part of `AccountConfig` (it lives only
@@ -662,6 +674,12 @@ impl Accounts for AccountsGrpcService {
             return Err(Status::internal(message));
         }
 
+        tracing::info!(
+            account_id = %config.id,
+            transport = transport_kind(&config.transport),
+            "Accounts: account added"
+        );
+
         Ok(Response::new(AddAccountResponse {
             config: Some(map_account_config_to_proto(config)),
         }))
@@ -671,7 +689,7 @@ impl Accounts for AccountsGrpcService {
         &self,
         _request: Request<ListAccountsRequest>,
     ) -> Result<Response<ListAccountsResponse>, Status> {
-        let accounts = self
+        let accounts: Vec<AccountConfigProto> = self
             .db
             .list_accounts()
             .await
@@ -680,6 +698,8 @@ impl Accounts for AccountsGrpcService {
             .map(map_account_config_to_proto)
             .collect();
 
+        tracing::debug!(count = accounts.len(), "Accounts: listed accounts");
+
         Ok(Response::new(ListAccountsResponse { accounts }))
     }
 
@@ -712,6 +732,7 @@ impl Accounts for AccountsGrpcService {
             )));
         }
 
+        let password_rotated = matches!(&req.password, Some(p) if !p.is_empty());
         match req.password {
             // A password rotation writes the new credential first, then
             // persists. Unlike `add_account` (where no prior secret exists,
@@ -735,6 +756,13 @@ impl Accounts for AccountsGrpcService {
             }
         }
 
+        tracing::info!(
+            account_id = %config.id,
+            transport = transport_kind(&config.transport),
+            password_rotated,
+            "Accounts: account updated"
+        );
+
         Ok(Response::new(UpdateAccountResponse {}))
     }
 
@@ -777,6 +805,8 @@ impl Accounts for AccountsGrpcService {
             );
         }
 
+        tracing::info!(account_id = %req.account_id, "Accounts: account removed");
+
         Ok(Response::new(RemoveAccountResponse {}))
     }
 
@@ -809,6 +839,13 @@ impl Accounts for AccountsGrpcService {
 
         let report = self.connection_tester.probe(&config, &password).await;
 
+        tracing::info!(
+            account_id = %config.id,
+            imap_ok = report.imap.ok,
+            smtp_ok = report.smtp.ok,
+            "Accounts: connection probe finished"
+        );
+
         Ok(Response::new(TestAccountConnectionResponse {
             imap_ok: report.imap.ok,
             smtp_ok: report.smtp.ok,
@@ -1137,6 +1174,12 @@ impl Calendar for CalendarGrpcService {
             }
         };
 
+        tracing::info!(
+            calendar_id = %req.calendar_id,
+            synced,
+            "Calendar: sync finished"
+        );
+
         Ok(Response::new(CalendarSyncResponse {
             synced_count: synced as u64,
         }))
@@ -1306,6 +1349,12 @@ impl Contacts for ContactsGrpcService {
             }
         };
 
+        tracing::info!(
+            account_id = %req.account_id,
+            synced,
+            "Contacts: sync finished"
+        );
+
         Ok(Response::new(ContactsSyncResponse {
             synced_count: synced as u64,
         }))
@@ -1421,6 +1470,12 @@ impl Contacts for ContactsGrpcService {
             .await
             .map_err(|e| Status::internal(format!("failed to save contact: {e}")))?;
 
+        tracing::info!(
+            contact_id = %contact.id,
+            account_id = ?contact.account_id,
+            "Contacts: contact created"
+        );
+
         Ok(Response::new(CreateContactResponse {
             contact: Some(map_contact_to_proto(contact)),
         }))
@@ -1601,6 +1656,12 @@ impl Mail for MailGrpcService {
                 )
             })?;
 
+        tracing::info!(
+            message_id = %req.message_id,
+            read = req.read,
+            "Mail: mark-read applied"
+        );
+
         self.event_bus.process_command(CoreCommand::MarkRead {
             message_id: req.message_id,
             read: req.read,
@@ -1724,6 +1785,11 @@ impl Mail for MailGrpcService {
     async fn sync(&self, request: Request<SyncRequest>) -> Result<Response<SyncResponse>, Status> {
         let account_id = request.into_inner().account_id;
 
+        match &account_id {
+            Some(id) => tracing::info!(account_id = %id, "Mail: dispatching sync for account"),
+            None => tracing::info!("Mail: dispatching sync for all configured accounts"),
+        }
+
         let synced = if let Some(backend) = &self.overrides.mail_backend {
             crate::sync::sync_with_backend(
                 &self.db,
@@ -1743,6 +1809,8 @@ impl Mail for MailGrpcService {
             sync_all_configured(&self.sync_dispatcher, &self.db, &self.event_bus).await
         };
 
+        tracing::info!(synced, "Mail: sync finished");
+
         Ok(Response::new(SyncResponse {
             synced_count: synced as u64,
         }))
@@ -1906,6 +1974,8 @@ impl Filters for FiltersGrpcService {
 
         self.reload_engine_from_store().await;
 
+        tracing::info!(rule_id = %rule.id, "Filters: rule created");
+
         Ok(Response::new(CreateRuleResponse {
             rule: Some(map_filter_rule_to_proto(rule)),
         }))
@@ -1957,6 +2027,8 @@ impl Filters for FiltersGrpcService {
 
         self.reload_engine_from_store().await;
 
+        tracing::info!(rule_id = %req.rule_id, "Filters: rule deleted");
+
         Ok(Response::new(DeleteRuleResponse {}))
     }
 
@@ -2082,6 +2154,8 @@ impl Filters for FiltersGrpcService {
 
         self.reload_engine_from_store().await;
 
+        tracing::info!(rule_id = %rule.id, "Filters: rule updated");
+
         Ok(Response::new(UpdateRuleResponse {
             rule: Some(map_filter_rule_to_proto(rule)),
         }))
@@ -2117,6 +2191,12 @@ impl Filters for FiltersGrpcService {
             }
         };
 
+        tracing::info!(
+            format = ?format,
+            rule_count = rules.len(),
+            "Filters: rules exported"
+        );
+
         Ok(Response::new(ExportRulesResponse { content }))
     }
 
@@ -2156,6 +2236,12 @@ impl Filters for FiltersGrpcService {
 
         self.reload_engine_from_store().await;
 
+        tracing::info!(
+            imported_count,
+            error_count = errors.len(),
+            "Filters: rules imported"
+        );
+
         Ok(Response::new(ImportRulesResponse {
             imported_count,
             errors,
@@ -2214,66 +2300,80 @@ impl Filters for FiltersGrpcService {
             req.chunk_size as usize
         };
 
+        tracing::info!(chunk_size, "Filters: triage rescan started");
+
         let (tx, rx) = tokio::sync::mpsc::channel::<Result<TriageProgress, Status>>(16);
         let db = self.db.clone();
         let engine = self.filter_engine.clone();
+        // Carry the per-RPC span into the detached worker so its completion
+        // event inherits the same request_id as the RPC that started the scan.
+        let triage_span = tracing::Span::current();
 
-        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;
+        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;
+                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;
                     }
-                };
-                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 is_last_page {
+                        tracing::info!(
+                            scanned,
+                            matched,
+                            applied,
+                            "Filters: triage rescan finished"
+                        );
+                        return;
                     }
                 }
-
-                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;
-                }
             }
-        });
+            .instrument(triage_span),
+        );
 
         Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
     }
@@ -2323,6 +2423,13 @@ impl Export for ExportGrpcService {
         }
         let format = map_export_format_from_proto(req.format())?;
 
+        let scope_kind = match &req.scope {
+            Some(export_request::Scope::AccountId(_)) => "account",
+            Some(export_request::Scope::FolderId(_)) => "folder",
+            None => "all",
+        };
+        tracing::info!(format = ?format, scope = scope_kind, "Export: mailbox export started");
+
         let messages = match req.scope {
             Some(export_request::Scope::AccountId(account_id)) => {
                 self.db
@@ -2345,6 +2452,12 @@ impl Export for ExportGrpcService {
             .await
             .map_err(|e| Status::internal(format!("export failed: {e}")))?;
 
+        tracing::info!(
+            message_count = summary.message_count,
+            bytes_written = summary.bytes_written,
+            "Export: mailbox export finished"
+        );
+
         Ok(Response::new(ExportResponse {
             output_path: summary.output_path,
             message_count: summary.message_count as u64,
@@ -2427,6 +2540,12 @@ impl Audit for AuditGrpcService {
             .await
             .map_err(|e| Status::internal(format!("failed to verify audit chain: {e}")))?;
 
+        tracing::info!(
+            valid = report.valid,
+            record_count = report.record_count,
+            "Audit: chain verification finished"
+        );
+
         Ok(Response::new(VerifyChainResponse {
             valid: report.valid,
             record_count: report.record_count as u64,
@@ -2929,6 +3048,73 @@ mod tests {
         request
     }
 
+    /// A state-changing RPC must emit its own domain event, and (because every
+    /// handler event inherits the OBS-2 per-RPC span) that event must carry the
+    /// request-scoped `request_id`. Here the `Filters::create_rule` handler is
+    /// invoked inside a stand-in request span to prove the domain event both
+    /// fires and inherits the correlation id -- without duplicating the layer's
+    /// bare entry/exit lines.
+    #[test]
+    fn create_rule_handler_logs_domain_event_under_request_scope() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime");
+
+        let (recorder, ()) = with_recorder(|| {
+            runtime.block_on(async {
+                let (db, _dir) = DatabaseEngine::connect_ephemeral()
+                    .await
+                    .expect("ephemeral db");
+                let db = Arc::new(db);
+                let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rules"));
+                let service = FiltersGrpcService { db, filter_engine };
+
+                // Stand in for the RpcTraceLayer's per-RPC span so the handler's
+                // domain event inherits a request_id, exactly as in production.
+                let span = tracing::info_span!("grpc.request", request_id = "test-req-42");
+                async {
+                    service
+                        .create_rule(Request::new(CreateRuleRequest {
+                            name: "Domain Event Rule".to_string(),
+                            priority: 1,
+                            nsql: SAMPLE_RULE_NSQL.to_string(),
+                        }))
+                        .await
+                        .expect("rule creation succeeds");
+                }
+                .instrument(span)
+                .await;
+            });
+        });
+
+        let events = recorder.events();
+        let created = events
+            .iter()
+            .find(|e| e.message() == "Filters: rule created")
+            .expect("the create_rule handler must emit its domain event");
+        assert_eq!(created.level, Level::INFO);
+        assert!(
+            created.fields.contains_key("rule_id"),
+            "the domain event must carry the rule id"
+        );
+
+        // The domain event was emitted while the handler ran inside the stand-in
+        // request span, so an operator can correlate it back to that request via
+        // the span's `request_id` (the same mechanism the RpcTraceLayer provides
+        // in production -- see `rpc_trace`'s own test).
+        let request_span = recorder
+            .spans()
+            .into_iter()
+            .find(|s| s.name == "grpc.request")
+            .expect("the request span must be recorded");
+        assert_eq!(
+            request_span.fields.get("request_id").map(String::as_str),
+            Some("test-req-42"),
+            "the request span must carry the request-scoped correlation id"
+        );
+    }
+
     #[test]
     fn auth_accept_logs_debug_without_leaking_the_token() {
         let (recorder, result) = with_recorder(|| {

crates/nunciod/src/lifecycle.rs

@@ -88,7 +88,12 @@ impl ShutdownController {
     /// Initiates shutdown. Idempotent: calling it more than once (e.g. a
     /// second Ctrl+C while already draining) is a no-op beyond the first.
     pub fn trigger(&self) {
+        // Drives the event bus to `ShuttingDown` and publishes the
+        // `CoreEvent::ShuttingDown` event to every subscriber (e.g. clients on
+        // `System/Subscribe`), so a graceful shutdown is observable over the
+        // API and not just in local logs.
         self.event_bus.process_command(CoreCommand::Shutdown);
+        tracing::info!("shutdown: ShuttingDown event published to subscribers");
         let _ = self.tx.send(true);
     }
 }
@@ -166,6 +171,29 @@ mod tests {
         assert!(late_clone.is_requested());
     }
 
+    /// Triggering shutdown must both publish the `ShuttingDown` event (flipping
+    /// bus state) and emit a traceable lifecycle log line, so a clean shutdown
+    /// is observable in the logs and not only over the API.
+    #[test]
+    fn trigger_emits_a_traceable_shutting_down_lifecycle_log() {
+        use crate::test_tracing::with_recorder;
+
+        let (recorder, ()) = with_recorder(|| {
+            let event_bus = Arc::new(EventBus::new());
+            let (controller, _signal) = ShutdownController::new(event_bus.clone());
+            controller.trigger();
+            assert_eq!(event_bus.current_state().status, EngineStatus::ShuttingDown);
+        });
+
+        assert!(
+            recorder
+                .events()
+                .iter()
+                .any(|e| e.message() == "shutdown: ShuttingDown event published to subscribers"),
+            "trigger must emit the ShuttingDown lifecycle log"
+        );
+    }
+
     #[tokio::test]
     async fn trigger_is_idempotent() {
         let event_bus = Arc::new(EventBus::new());

crates/nunciod/src/main.rs

@@ -308,10 +308,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         }
     };
 
+    // From here down is the graceful-shutdown drain: the gRPC server has
+    // stopped accepting new connections, so log each remaining teardown step
+    // to make a clean shutdown fully traceable end to end.
+    tracing::info!(
+        "shutdown: gRPC server stopped accepting connections; draining background workers"
+    );
+
     // Bound how long the background workers get to notice the shutdown
     // signal and exit their loops before the database is closed out from
     // under them.
-    let _ = tokio::time::timeout(SHUTDOWN_GRACE_PERIOD, async {
+    let workers_drained = tokio::time::timeout(SHUTDOWN_GRACE_PERIOD, async {
         if let Some(task) = sync_command_task {
             let _ = task.await;
         }
@@ -321,12 +328,22 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         }
     })
     .await;
+    match workers_drained {
+        Ok(()) => tracing::info!("shutdown: background workers (sync, outbox, update) drained"),
+        Err(_) => tracing::warn!(
+            "shutdown: background workers did not all exit within the {:?} grace period; \
+             proceeding to close the database",
+            SHUTDOWN_GRACE_PERIOD
+        ),
+    }
     signal_task.abort();
 
     // Force the WAL checkpoint before the process exits so no committed
     // page is left for SQLite's own deferred "last connection closes"
     // teardown to race (see `DatabaseEngine::close`'s doc comment).
+    tracing::info!("shutdown: closing database and checkpointing the WAL");
     db.close().await;
+    tracing::info!("shutdown: complete");
 
     grpc_result.map_err(|e| format!("nunciod gRPC server failed: {e}"))?;
 

crates/nunciod/src/outbox.rs

@@ -155,6 +155,16 @@ pub async fn execute_pending_mutations(
 
     let mut summary = ExecutionSummary::default();
     for item in pending {
+        // Log the executor's decision-level attempt before issuing the real
+        // remote op: the mutation identity, its operation, and the surrogate
+        // message id it targets (never the message body/subject/recipients).
+        tracing::info!(
+            mutation_id = %item.id,
+            op = %item.mutation_type,
+            message_id = %item.message_id,
+            retry_count = item.retry_count,
+            "outbox: attempting mutation"
+        );
         let outcome = tokio::select! {
             biased;
             _ = shutdown.wait() => {
@@ -174,7 +184,14 @@ pub async fn execute_pending_mutations(
         };
         let next_retry = item.retry_count + 1;
         let (status, retry, tally): (&str, i32, fn(&mut ExecutionSummary)) = match disposition {
-            Disposition::Completed => ("completed", item.retry_count, |s| s.completed += 1),
+            Disposition::Completed => {
+                tracing::info!(
+                    mutation_id = %item.id,
+                    op = %item.mutation_type,
+                    "outbox: mutation completed"
+                );
+                ("completed", item.retry_count, |s| s.completed += 1)
+            }
             Disposition::Permanent(reason) => {
                 tracing::warn!(
                     mutation_id = %item.id,
@@ -298,6 +315,14 @@ async fn dispatch_mailbox_mutation(
         kind,
     };
 
+    tracing::debug!(
+        op = %mutation.mutation_type,
+        remote_id = %spec.remote_id,
+        folder_id = %spec.folder_id,
+        uid_validity = %spec.uid_validity,
+        "outbox: issuing mailbox mutation against remote backend"
+    );
+
     match backend.apply_mutation(&spec).await {
         Ok(()) => Disposition::Completed,
         Err(e) => Disposition::Retry(format!("remote mutation failed: {e}")),
@@ -505,3 +530,157 @@ impl RemoteExecutionEnv for ProductionExecutionEnv {
             .await
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::lifecycle::{ShutdownController, ShutdownSignal};
+    use crate::test_tracing::with_recorder;
+    use nuncio_core::model::Email;
+    use nuncio_core::EventBus;
+    use nuncio_filter::MutationPayload;
+    use nuncio_mail::MockMailBackend;
+    use tracing::Level;
+
+    /// Minimal test [`RemoteExecutionEnv`] whose mailbox mutations always
+    /// genuinely succeed against a recording [`MockMailBackend`]; the forward
+    /// and webhook paths are unused by these tests.
+    struct SucceedingEnv {
+        backend: MockMailBackend,
+    }
+
+    #[async_trait]
+    impl RemoteExecutionEnv for SucceedingEnv {
+        async fn mail_backend(
+            &self,
+            _account_id: &str,
+        ) -> Result<Box<dyn MailBackend>, OutboxExecuteError> {
+            Ok(Box::new(self.backend.clone()))
+        }
+
+        async fn message_sender(
+            &self,
+            account_id: &str,
+        ) -> Result<Box<dyn MessageSender>, OutboxExecuteError> {
+            // The FLAG-only tests never take the forward path; surface an error
+            // rather than a panic if that assumption ever changes.
+            Err(OutboxExecuteError::NotAMailAccount(account_id.to_string()))
+        }
+
+        async fn dispatch_webhook(
+            &self,
+            _url: &str,
+            _rule_id: &str,
+            _message_id: &str,
+            _subject: &str,
+            _sender: &str,
+        ) -> Result<u16, WebhookError> {
+            // The FLAG-only tests never take the webhook path.
+            Err(WebhookError::SigningError(
+                "webhook path unused in test".into(),
+            ))
+        }
+    }
+
+    fn no_shutdown() -> (ShutdownController, ShutdownSignal) {
+        ShutdownController::new(Arc::new(EventBus::new()))
+    }
+
+    fn sample_email() -> Email {
+        Email {
+            id: "msg-outbox-log".to_string(),
+            account_id: "acct-outbox-log".to_string(),
+            folder_id: "INBOX".to_string(),
+            remote_id: "77".to_string(),
+            uid_validity: "9".to_string(),
+            subject: "Confidential outbox subject".to_string(),
+            sender: "alice@example.com".to_string(),
+            recipient: "owner@nuncio.mx".to_string(),
+            received_at: 1_700_000_000,
+            read: false,
+            body_plain: Some("secret outbox body".to_string()),
+            body_html: None,
+            attachments: Vec::new(),
+        }
+    }
+
+    fn flag_mutation() -> PendingRemoteMutation {
+        PendingRemoteMutation {
+            id: "mut-1".to_string(),
+            rule_id: "rule-1".to_string(),
+            message_id: "msg-outbox-log".to_string(),
+            mutation_type: "FLAG".to_string(),
+            payload: serde_json::to_string(&MutationPayload {
+                action_type: "FLAG".to_string(),
+                target: None,
+            })
+            .expect("serialize payload"),
+            status: "pending".to_string(),
+            retry_count: 0,
+            created_at: 1_700_000_000,
+        }
+    }
+
+    /// An outbox drain must log the executor's decision-level attempt BEFORE
+    /// issuing the remote op and the Completed result AFTER, carrying the
+    /// mutation id and operation -- and never the message body or subject.
+    #[test]
+    fn drain_logs_attempt_and_completion_without_leaking_message_content() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime");
+
+        let (recorder, ()) = with_recorder(|| {
+            runtime.block_on(async {
+                let (db, _dir) = DatabaseEngine::connect_ephemeral()
+                    .await
+                    .expect("ephemeral db");
+                db.save_folder_sync_state("acct-outbox-log", "INBOX", "9:100")
+                    .await
+                    .expect("save folder checkpoint");
+                db.save_email(&sample_email()).await.expect("save email");
+                db.save_pending_mutation(&flag_mutation())
+                    .await
+                    .expect("save pending mutation");
+
+                let env = SucceedingEnv {
+                    backend: MockMailBackend::new(),
+                };
+                let (_ctl, mut shutdown) = no_shutdown();
+                let summary = execute_pending_mutations(&db, &env, 10, &mut shutdown).await;
+                assert_eq!(summary.completed, 1, "the mutation must complete");
+            });
+        });
+
+        let events = recorder.events();
+        let attempt = events
+            .iter()
+            .find(|e| e.message() == "outbox: attempting mutation")
+            .expect("an attempt event must be logged before issuing the op");
+        assert_eq!(attempt.level, Level::INFO);
+        assert_eq!(attempt.fields.get("op").map(String::as_str), Some("FLAG"));
+        assert_eq!(
+            attempt.fields.get("mutation_id").map(String::as_str),
+            Some("mut-1")
+        );
+
+        let completed = events
+            .iter()
+            .find(|e| e.message() == "outbox: mutation completed")
+            .expect("a completion event must be logged after a genuine success");
+        assert_eq!(completed.level, Level::INFO);
+        assert_eq!(completed.fields.get("op").map(String::as_str), Some("FLAG"));
+
+        for value in recorder.all_field_values() {
+            assert!(
+                !value.contains("Confidential outbox subject"),
+                "subject leaked into telemetry: {value}"
+            );
+            assert!(
+                !value.contains("secret outbox body"),
+                "body leaked into telemetry: {value}"
+            );
+        }
+    }
+}

crates/nunciod/src/send.rs

@@ -183,6 +183,15 @@ pub async fn send_message_for_account(
     request: ComposeRequest,
 ) -> Result<String, SendError> {
     let config = resolve_sending_account(db, request.account_id.as_deref()).await?;
+    let account_id = config.id.clone();
+    let attachment_count = request.attachments.len();
+    tracing::info!(
+        account_id = %account_id,
+        transport = "smtp",
+        has_cc = request.cc.is_some(),
+        attachment_count,
+        "mail send: dispatching outbound message over real SMTP transport"
+    );
     let password = secrets.get_secret(&config.keyring_secret_key)?;
     let sender = build_smtp_sender(&config, &password)?;
 
@@ -196,8 +205,24 @@ pub async fn send_message_for_account(
         attachments: request.attachments,
     };
 
-    send_with_sender(&sender, &outbound).await?;
-    Ok(generate_message_id())
+    match send_with_sender(&sender, &outbound).await {
+        Ok(()) => {
+            let message_id = generate_message_id();
+            tracing::info!(
+                account_id = %account_id,
+                message_id = %message_id,
+                "mail send: outbound message accepted by SMTP transport"
+            );
+            Ok(message_id)
+        }
+        Err(e) => {
+            tracing::warn!(
+                account_id = %account_id,
+                "mail send: SMTP transport rejected the outbound message: {e}"
+            );
+            Err(e)
+        }
+    }
 }
 
 /// Test/E2E entry point for the `SendMessage` RPC when a
@@ -215,6 +240,14 @@ pub async fn send_message_with_injected_sender(
     request: ComposeRequest,
 ) -> Result<String, SendError> {
     let config = resolve_sending_account(db, request.account_id.as_deref()).await?;
+    let account_id = config.id.clone();
+    tracing::info!(
+        account_id = %account_id,
+        transport = "injected",
+        has_cc = request.cc.is_some(),
+        attachment_count = request.attachments.len(),
+        "mail send: dispatching outbound message via injected sender"
+    );
 
     let outbound = OutboundMessage {
         from: config.email_address,
@@ -226,8 +259,24 @@ pub async fn send_message_with_injected_sender(
         attachments: request.attachments,
     };
 
-    send_with_sender(sender, &outbound).await?;
-    Ok(generate_message_id())
+    match send_with_sender(sender, &outbound).await {
+        Ok(()) => {
+            let message_id = generate_message_id();
+            tracing::info!(
+                account_id = %account_id,
+                message_id = %message_id,
+                "mail send: outbound message accepted by injected sender"
+            );
+            Ok(message_id)
+        }
+        Err(e) => {
+            tracing::warn!(
+                account_id = %account_id,
+                "mail send: injected sender rejected the outbound message: {e}"
+            );
+            Err(e)
+        }
+    }
 }
 
 #[cfg(test)]
@@ -534,4 +583,77 @@ mod tests {
         assert!(matches!(err, SendError::AccountNotFound(id) if id == "acct-does-not-exist"));
         assert!(mock.sent_messages().is_empty());
     }
+
+    /// A send must be followable in the log: a dispatch event before the send
+    /// and an acceptance event after, both scoped to the account id -- and the
+    /// recipient address, subject, and body must never appear in any logged
+    /// field.
+    #[test]
+    fn send_logs_dispatch_and_acceptance_without_leaking_recipient_or_subject() {
+        use crate::test_tracing::with_recorder;
+        use tracing::Level;
+
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime");
+
+        let (recorder, ()) = with_recorder(|| {
+            runtime.block_on(async {
+                let (db, _dir) = DatabaseEngine::connect_ephemeral()
+                    .await
+                    .expect("ephemeral db");
+                let config = sample_account("acct-log");
+                db.save_account(&config).await.expect("save account");
+
+                let mock = MockMessageSender::new();
+                send_message_with_injected_sender(
+                    &db,
+                    &mock,
+                    ComposeRequest {
+                        to: "secret-recipient@example.com".to_string(),
+                        cc: None,
+                        subject: "Confidential Subject Line".to_string(),
+                        body_text: Some("secret body".to_string()),
+                        body_html: None,
+                        attachments: Vec::new(),
+                        account_id: None,
+                    },
+                )
+                .await
+                .expect("send succeeds");
+            });
+        });
+
+        let events = recorder.events();
+        let dispatch = events
+            .iter()
+            .find(|e| e.message() == "mail send: dispatching outbound message via injected sender")
+            .expect("dispatch event must be logged");
+        assert_eq!(dispatch.level, Level::INFO);
+        assert_eq!(
+            dispatch.fields.get("account_id").map(String::as_str),
+            Some("acct-log")
+        );
+        assert!(events
+            .iter()
+            .any(|e| e.message() == "mail send: outbound message accepted by injected sender"));
+
+        // Secret-safety: no recipient, subject, or body text may appear in any
+        // recorded telemetry field.
+        for value in recorder.all_field_values() {
+            assert!(
+                !value.contains("secret-recipient@example.com"),
+                "recipient leaked into telemetry: {value}"
+            );
+            assert!(
+                !value.contains("Confidential Subject Line"),
+                "subject leaked into telemetry: {value}"
+            );
+            assert!(
+                !value.contains("secret body"),
+                "body leaked into telemetry: {value}"
+            );
+        }
+    }
 }