feat(sync): real IMAP fetch->store loop in the daemon (kill fake sync)

Add nunciod::sync: sync_with_backend(&dyn MailBackend) fetches folders +
messages and persists each via save_email, wrapping the status lifecycle so a
sync ALWAYS completes (Idle + SyncCompleted) and can never stick at Syncing;
failures surface as CoreEvent::Error. Production builds a real ImapEngine from
account config + the keyring password. The daemon now claims the EventBus
command receiver and runs real syncs on SyncAll/SyncAccount (previously those
commands -- including post-recovery resync -- were silently inert). Remove the
outbox worker's 'Simulate remote IMAP/JMAP mutation execution' fake; it no
longer fabricates 'completed' (outbound send is 1.C.5). Tests inject
MockMailBackend and assert persistence + the SyncStarted/SyncCompleted/Error
sequence, offline.

Refs #158

crates/nuncio-core/src/ipc/server.rs

@@ -150,7 +150,23 @@ impl IpcDaemonServer {
                     )
                 }
                 "mail.sync_all" => {
+                    // Flip status to `Syncing` and publish `SyncStarted`
+                    // synchronously so a caller polling `system.state` right
+                    // after this RPC returns observes it deterministically.
                     event_bus.process_command(CoreCommand::SyncAll);
+                    // ALSO enqueue the command onto the async command
+                    // channel so whichever real `CoreCommand` consumer owns
+                    // `take_command_receiver()` (e.g. `nunciod`'s real
+                    // inbound-sync command loop, backlog story 1.C.3 / GH
+                    // #158) performs the actual fetch/persist work and later
+                    // flips status back to `Idle` + publishes
+                    // `SyncCompleted`. Best-effort: if no consumer is
+                    // currently running (some test harnesses only exercise
+                    // this transport in isolation), the command is simply
+                    // queued and never drained, which is harmless -- the
+                    // synchronous status flip above already satisfies this
+                    // RPC's own contract.
+                    let _ = event_bus.send_command(CoreCommand::SyncAll).await;
                     JsonRpcResponse::success(req.id, json!({ "status": "dispatched" }))
                 }
                 "mail.mark_read" => {

crates/nunciod/src/lib.rs

@@ -4,6 +4,7 @@
 
 pub mod grpc;
 pub mod orchestrator;
+pub mod sync;
 pub use orchestrator::SelfHealingSyncOrchestrator;
 
 /// Default PERSISTENT database path for `nunciod`: `~/.nuncio/nuncio.db`

crates/nunciod/src/main.rs

@@ -4,7 +4,7 @@
 
 use nuncio_core::ipc::server::CustomRpcHandler;
 use nuncio_core::ipc::IpcDaemonServer;
-use nuncio_core::{CoreEvent, EventBus};
+use nuncio_core::{CoreCommand, CoreEvent, EventBus};
 use nuncio_filter::{FilterEngine, NsqlParser, NsqlValidator, OutboxManager, ValidationOptions};
 use serde_json::json;
 use std::sync::Arc;
@@ -14,7 +14,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     tracing_subscriber::fmt::init();
     tracing::info!("Starting Nuncio Central Daemon Service (nunciod)...");
 
-    let event_bus = Arc::new(EventBus::new());
+    // The `CoreCommand` receiver must be claimed BEFORE the `EventBus` is
+    // wrapped in an `Arc` (`take_command_receiver` needs `&mut self`, which
+    // an `Arc<EventBus>` shared across every subsystem below cannot offer).
+    let mut event_bus_owned = EventBus::new();
+    let command_rx = event_bus_owned.take_command_receiver();
+    let event_bus = Arc::new(event_bus_owned);
     // PERSISTENT database path (backlog stories 1.C.1 / 1.C.2, GH #156 /
     // GH #157): defaults to `~/.nuncio/nuncio.db` -- NOT a temp/ephemeral
     // path -- so accounts (and everything else) survive a daemon restart.
@@ -27,10 +32,72 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     let orchestrator = nunciod::SelfHealingSyncOrchestrator::new(&db_path, event_bus.clone());
     let (db, _summary) = orchestrator.initialize_and_recover().await?;
 
+    // Shared OS keyring vault (backlog stories 1.C.1 / 1.C.2 / 1.C.3, GH
+    // #156 / #157 / #158). ONE `SecretManager` instance is used for every
+    // subsystem that reads/writes account credentials in this process --
+    // the gRPC `Accounts` service below AND the real inbound-sync command
+    // loop -- rather than provisioning separate wrapper instances.
+    let account_secrets = Arc::new(nuncio_store::vault::SecretManager::production());
+
     // Load active rules from SQLite
     let initial_rules = db.list_filter_rules().await.unwrap_or_default();
     let filter_engine = Arc::new(FilterEngine::new(initial_rules)?);
 
+    // Real inbound-sync `CoreCommand` consumer (backlog story 1.C.3, GH
+    // #158). This is the ONE place that claims `EventBus`'s command
+    // receiver, so it is what finally makes `CoreCommand::SyncAll` /
+    // `CoreCommand::SyncAccount` do REAL work -- connecting a real mail
+    // backend, fetching folders/messages, and persisting them via
+    // `DatabaseEngine::save_email` -- instead of only flipping a status
+    // flag. It also finally gives `SelfHealingSyncOrchestrator::
+    // trigger_background_resync`'s post-recovery `send_command` calls a
+    // consumer: previously nothing read this channel, so that resync
+    // trigger was inert.
+    if let Some(mut command_rx) = command_rx {
+        let db_sync = db.clone();
+        let secrets_sync = account_secrets.clone();
+        let event_bus_sync = event_bus.clone();
+        let _sync_command_task = tokio::spawn(async move {
+            while let Some(cmd) = command_rx.recv().await {
+                match cmd {
+                    CoreCommand::SyncAll => {
+                        let synced = nunciod::sync::run_all_accounts_sync(
+                            &db_sync,
+                            &secrets_sync,
+                            &event_bus_sync,
+                        )
+                        .await;
+                        tracing::info!("SyncAll completed: {} message(s) synced", synced);
+                    }
+                    CoreCommand::SyncAccount { account_id } => {
+                        match nunciod::sync::run_account_sync(
+                            &db_sync,
+                            &secrets_sync,
+                            &event_bus_sync,
+                            &account_id,
+                        )
+                        .await
+                        {
+                            Ok(count) => tracing::info!(
+                                "SyncAccount({}) completed: {} message(s) synced",
+                                account_id,
+                                count
+                            ),
+                            Err(e) => {
+                                tracing::warn!("SyncAccount({}) failed: {}", account_id, e)
+                            }
+                        }
+                    }
+                    _ => {}
+                }
+            }
+        });
+    } else {
+        tracing::error!(
+            "EventBus command receiver was already taken; real sync command processing will not run"
+        );
+    }
+
     // Background Outbox Worker Task (#273)
     let db_outbox = db.clone();
     let _outbox_task = tokio::spawn(async move {
@@ -48,9 +115,19 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
                     }
                     let backoff_ms = OutboxManager::calculate_backoff_ms(item.retry_count);
                     tokio::time::sleep(std::time::Duration::from_millis(backoff_ms)).await;
-                    // Simulate remote IMAP/JMAP mutation execution
+                    // Outbound IMAP/JMAP remote mutation execution (actually
+                    // applying a filter action like move/delete/flag on the
+                    // real mail server) is NOT wired yet -- that is backlog
+                    // story 1.C.5. This worker deliberately does NOT mark
+                    // mutations "completed": doing so would fabricate
+                    // success for work that never happened. It only bumps
+                    // the retry counter (so an item that keeps failing still
+                    // eventually flips to "failed" once retries are
+                    // exhausted) and leaves the item "pending" so it is
+                    // retried on the next poll tick once 1.C.5 wires real
+                    // execution.
                     let _ = db_outbox
-                        .update_mutation_status(&item.id, "completed", next_retry)
+                        .update_mutation_status(&item.id, "pending", next_retry)
                         .await;
                 }
             }
@@ -289,13 +366,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     // Runs ALONGSIDE the existing JSON-RPC IPC server below; the migration
     // off the hand-rolled JSON-RPC transport happens in later stories. The
     // bearer token is minted (or loaded, on subsequent runs) from the real
-    // OS keyring vault via `SecretManager::production()`, fails closed if
-    // the keyring is unavailable, and is never logged. `Accounts` reuses
+    // OS keyring vault via the shared `account_secrets` `SecretManager`
+    // (same instance the real-sync command loop above uses), fails closed
+    // if the keyring is unavailable, and is never logged. `Accounts` reuses
     // this SAME `SecretManager` to write account password credentials to
     // the OS keyring (never to SQLite) -- see `nunciod::grpc`'s security
     // comment (GH #165) on why EVERY mounted service shares this one
     // `BearerAuthInterceptor`.
-    let grpc_secrets = Arc::new(nuncio_store::vault::SecretManager::production());
+    let grpc_secrets = account_secrets.clone();
     let grpc_token_bytes = grpc_secrets
         .get_or_create_key_bytes(nuncio_store::vault::GRPC_TOKEN_ACCOUNT, 32)
         .map_err(|e| format!("failed to provision gRPC bearer token from vault: {e}"))?;

crates/nunciod/src/sync.rs

@@ -0,0 +1,624 @@
+//! Real inbound mail synchronization routine (backlog story 1.C.3, GH #158).
+//!
+//! Prior to this module, "sync" was a status-flag flip: `CoreCommand::SyncAll`
+//! / `CoreCommand::SyncAccount` only moved `EngineStatus` to `Syncing` and
+//! published `SyncStarted`, and nothing ever fetched a real message from a
+//! real server into the store. This module closes that gap: for a configured
+//! account it connects a real [`MailBackend`] (a real [`ImapEngine`] for
+//! [`AccountProtocol::ImapSmtp`], built from [`AccountConfig`] plus the
+//! account's keyring-stored password), fetches folders and messages, and
+//! persists every message through [`DatabaseEngine::save_email`].
+//!
+//! # Testability
+//!
+//! [`sync_with_backend`] takes `&dyn MailBackend` rather than a concrete
+//! engine, so it is exercised directly in tests with [`MockMailBackend`] --
+//! no real network or OS keyring access ever happens inside it. Production
+//! entry points ([`run_account_sync`], [`run_all_accounts_sync`]) are the
+//! only places that resolve a real account's credentials and construct a
+//! real backend; they then delegate the actual fetch/persist/event-emission
+//! work to [`sync_with_backend`].
+use nuncio_core::{AccountConfig, AccountProtocol, CoreCommand, EventBus};
+use nuncio_mail::{ImapEngine, JmapEngine, MailBackend, MailError};
+use nuncio_store::db::DatabaseError;
+use nuncio_store::vault::{SecretManager, VaultError};
+use thiserror::Error;
+
+/// Errors that can occur while synchronizing a single account's inbound mail.
+#[derive(Debug, Error)]
+pub enum SyncError {
+    /// No persisted account configuration matches the requested account ID.
+    #[error("account '{0}' is not configured")]
+    AccountNotFound(String),
+    /// Failed to read the account's credential from the secret vault.
+    #[error("failed to read account credential from the secret vault: {0}")]
+    Vault(#[from] VaultError),
+    /// The mail backend (IMAP/JMAP) reported an error.
+    #[error("mail backend error: {0}")]
+    Mail(#[from] MailError),
+    /// A database persistence error occurred while saving synced messages.
+    #[error("database error: {0}")]
+    Store(#[from] DatabaseError),
+}
+
+/// Fetch every folder and every message in every folder from `backend`,
+/// persisting each message via [`DatabaseEngine::save_email`]. Returns the
+/// total number of messages processed (a message id "processed" more than
+/// once, e.g. by a backend that returns the same id from multiple folders,
+/// is counted once per occurrence -- `save_email` itself is `INSERT OR
+/// REPLACE`, so storage stays deduplicated by message id regardless).
+///
+/// Contains no event-bus or credential logic -- pure fetch-and-persist, so
+/// it is the smallest unit tests can exercise directly with a
+/// [`MockMailBackend`].
+async fn fetch_and_persist(
+    db: &nuncio_store::db::DatabaseEngine,
+    backend: &dyn MailBackend,
+) -> Result<usize, SyncError> {
+    let folders = backend.sync_folders().await?;
+    let mut synced = 0usize;
+    for folder in folders {
+        let (emails, _state) = backend.sync_messages(&folder.id, None).await?;
+        for email in emails {
+            db.save_email(&email).await?;
+            synced += 1;
+        }
+    }
+    Ok(synced)
+}
+
+/// Run a real inbound sync against `backend`, wrapping the work with the
+/// standard `CoreCommand`-driven engine-status lifecycle: flips to
+/// `EngineStatus::Syncing` and publishes `SyncStarted` before, always flips
+/// back to `EngineStatus::Idle` and publishes `SyncCompleted` after --
+/// including when the sync fails, so a failed sync can never leave the
+/// engine stuck at `Syncing`. On failure, a `CoreEvent::Error` is published
+/// via `CoreCommand::ReportError` before the completion event.
+///
+/// `account_id` is `Some(id)` for a single-account sync (mirrors
+/// `CoreCommand::SyncAccount`) or `None` for an all-accounts batch (mirrors
+/// `CoreCommand::SyncAll`).
+///
+/// This is the seam production and tests share: production always passes a
+/// real engine (see [`run_account_sync`]); tests pass a [`MockMailBackend`].
+pub async fn sync_with_backend(
+    db: &nuncio_store::db::DatabaseEngine,
+    event_bus: &EventBus,
+    backend: &dyn MailBackend,
+    account_id: Option<String>,
+) -> Result<usize, SyncError> {
+    match &account_id {
+        Some(id) => event_bus.process_command(CoreCommand::SyncAccount {
+            account_id: id.clone(),
+        }),
+        None => event_bus.process_command(CoreCommand::SyncAll),
+    }
+
+    let result = fetch_and_persist(db, backend).await;
+
+    if let Err(e) = &result {
+        event_bus.process_command(CoreCommand::ReportError {
+            message: format!("sync failed: {e}"),
+        });
+    }
+
+    event_bus.complete_sync(account_id);
+    result
+}
+
+/// Look up `account_id`'s persisted configuration.
+async fn find_account(
+    db: &nuncio_store::db::DatabaseEngine,
+    account_id: &str,
+) -> Result<AccountConfig, SyncError> {
+    let accounts = db.list_accounts().await?;
+    accounts
+        .into_iter()
+        .find(|a| a.id == account_id)
+        .ok_or_else(|| SyncError::AccountNotFound(account_id.to_string()))
+}
+
+/// Construct the real [`MailBackend`] engine appropriate for `config`'s
+/// configured protocol, authenticated with `password` (read from the OS
+/// keyring by the caller). This is the ONLY place a concrete protocol engine
+/// is chosen; every other function in this module only ever sees `&dyn
+/// MailBackend`.
+fn build_mail_backend(config: &AccountConfig, password: &str) -> Box<dyn MailBackend> {
+    match config.protocol {
+        AccountProtocol::ImapSmtp => Box::new(ImapEngine::with_credentials(
+            &config.id,
+            &config.server_host,
+            config.server_port,
+            &config.email_address,
+            password,
+        )),
+        AccountProtocol::Jmap => Box::new(JmapEngine::new(&config.id)),
+    }
+}
+
+/// Production entry point for `CoreCommand::SyncAccount`: resolves
+/// `account_id`'s configuration and keyring password, builds the real mail
+/// backend, and runs a full inbound sync through [`sync_with_backend`].
+///
+/// If the account is unknown or its credential cannot be read, a
+/// `CoreEvent::Error` is published and `Err` is returned WITHOUT flipping
+/// the engine status to `Syncing` -- the sync never actually began, so
+/// there is nothing to "complete".
+pub async fn run_account_sync(
+    db: &nuncio_store::db::DatabaseEngine,
+    secrets: &SecretManager,
+    event_bus: &EventBus,
+    account_id: &str,
+) -> Result<usize, SyncError> {
+    let setup = async {
+        let config = find_account(db, account_id).await?;
+        let password = secrets.get_secret(&config.keyring_secret_key)?;
+        Ok::<_, SyncError>(build_mail_backend(&config, &password))
+    }
+    .await;
+
+    match setup {
+        Ok(backend) => {
+            sync_with_backend(
+                db,
+                event_bus,
+                backend.as_ref(),
+                Some(account_id.to_string()),
+            )
+            .await
+        }
+        Err(e) => {
+            event_bus.process_command(CoreCommand::ReportError {
+                message: format!("cannot sync account '{account_id}': {e}"),
+            });
+            Err(e)
+        }
+    }
+}
+
+/// Production entry point for `CoreCommand::SyncAll`: syncs every persisted
+/// account in turn, wrapping the whole batch in a single
+/// `SyncStarted{None}`/`SyncCompleted{None}` pair (rather than one pair per
+/// account). A single account failing to sync (missing credential, backend
+/// error, etc.) publishes a `CoreEvent::Error` and does not stop the
+/// remaining accounts from being attempted. Returns the total number of
+/// messages synced across all accounts.
+pub async fn run_all_accounts_sync(
+    db: &nuncio_store::db::DatabaseEngine,
+    secrets: &SecretManager,
+    event_bus: &EventBus,
+) -> usize {
+    event_bus.process_command(CoreCommand::SyncAll);
+
+    let mut total_synced = 0usize;
+    match db.list_accounts().await {
+        Ok(accounts) => {
+            for config in accounts {
+                match secrets.get_secret(&config.keyring_secret_key) {
+                    Ok(password) => {
+                        let backend = build_mail_backend(&config, &password);
+                        match fetch_and_persist(db, backend.as_ref()).await {
+                            Ok(count) => total_synced += count,
+                            Err(e) => {
+                                event_bus.process_command(CoreCommand::ReportError {
+                                    message: format!(
+                                        "sync failed for account '{}': {e}",
+                                        config.id
+                                    ),
+                                });
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        event_bus.process_command(CoreCommand::ReportError {
+                            message: format!(
+                                "cannot read credential for account '{}': {e}",
+                                config.id
+                            ),
+                        });
+                    }
+                }
+            }
+        }
+        Err(e) => {
+            event_bus.process_command(CoreCommand::ReportError {
+                message: format!("failed to list accounts for sync: {e}"),
+            });
+        }
+    }
+
+    event_bus.complete_sync(None);
+    total_synced
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nuncio_core::model::{Email, Folder};
+    use nuncio_core::{CoreEvent, EngineStatus, TlsMode};
+    use nuncio_mail::MockMailBackend;
+    use nuncio_store::db::DatabaseEngine;
+
+    fn sample_jmap_account(id: &str) -> AccountConfig {
+        AccountConfig {
+            id: id.to_string(),
+            name: "JMAP Test Account".to_string(),
+            email_address: format!("{id}@nuncio.mx"),
+            protocol: AccountProtocol::Jmap,
+            server_host: "jmap.nuncio.mx".to_string(),
+            server_port: 443,
+            use_tls: true,
+            imap_tls_mode: TlsMode::ImplicitTls,
+            smtp_tls_mode: TlsMode::ImplicitTls,
+            keyring_secret_key: format!("nuncio/{id}"),
+            sync_interval_secs: 60,
+        }
+    }
+
+    fn sample_imap_account(id: &str) -> AccountConfig {
+        AccountConfig {
+            id: id.to_string(),
+            name: "IMAP Test Account".to_string(),
+            email_address: format!("{id}@nuncio.mx"),
+            protocol: AccountProtocol::ImapSmtp,
+            server_host: "imap.nuncio.mx".to_string(),
+            server_port: 993,
+            use_tls: true,
+            imap_tls_mode: TlsMode::ImplicitTls,
+            smtp_tls_mode: TlsMode::ImplicitTls,
+            keyring_secret_key: format!("nuncio/{id}"),
+            sync_interval_secs: 60,
+        }
+    }
+
+    fn mock_email(id: &str, folder_id: &str, subject: &str) -> Email {
+        Email {
+            id: id.to_string(),
+            account_id: "acct-mock-1".to_string(),
+            folder_id: folder_id.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(format!("body for {id}")),
+            body_html: None,
+            attachments: Vec::new(),
+        }
+    }
+
+    #[tokio::test]
+    async fn sync_with_backend_persists_mock_messages_and_emits_events() {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+        let mut events = event_bus.subscribe_events();
+
+        let mock = MockMailBackend::new();
+        mock.add_folder(Folder {
+            id: "inbox".to_string(),
+            name: "Inbox".to_string(),
+            total_messages: 2,
+            unread_messages: 2,
+        });
+        mock.add_message(mock_email("m1", "inbox", "Hello"));
+        mock.add_message(mock_email("m2", "inbox", "World"));
+
+        let synced = sync_with_backend(&db, &event_bus, &mock, Some("acct-mock-1".to_string()))
+            .await
+            .expect("sync succeeds");
+        assert_eq!(synced, 2);
+
+        let persisted = db
+            .list_messages("inbox", 10)
+            .await
+            .expect("list persisted messages");
+        assert_eq!(persisted.len(), 2);
+        let subjects: Vec<String> = persisted.iter().map(|e| e.subject.clone()).collect();
+        assert!(subjects.contains(&"Hello".to_string()));
+        assert!(subjects.contains(&"World".to_string()));
+
+        assert_eq!(
+            events.recv().await.expect("start event"),
+            CoreEvent::SyncStarted {
+                account_id: Some("acct-mock-1".to_string())
+            }
+        );
+        assert_eq!(
+            events.recv().await.expect("complete event"),
+            CoreEvent::SyncCompleted {
+                account_id: Some("acct-mock-1".to_string())
+            }
+        );
+        assert_eq!(event_bus.current_state().status, EngineStatus::Idle);
+    }
+
+    #[tokio::test]
+    async fn sync_with_backend_none_account_id_mirrors_sync_all() {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+        let mut events = event_bus.subscribe_events();
+
+        let mock = MockMailBackend::new();
+        mock.add_folder(Folder {
+            id: "inbox".to_string(),
+            name: "Inbox".to_string(),
+            total_messages: 0,
+            unread_messages: 0,
+        });
+
+        let synced = sync_with_backend(&db, &event_bus, &mock, None)
+            .await
+            .expect("sync succeeds");
+        assert_eq!(synced, 0);
+
+        assert_eq!(
+            events.recv().await.expect("start event"),
+            CoreEvent::SyncStarted { account_id: None }
+        );
+        assert_eq!(
+            events.recv().await.expect("complete event"),
+            CoreEvent::SyncCompleted { account_id: None }
+        );
+    }
+
+    #[tokio::test]
+    async fn sync_with_backend_failure_emits_error_and_returns_to_idle() {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+        let mut events = event_bus.subscribe_events();
+
+        let mock = MockMailBackend::new();
+        mock.set_should_fail(true);
+
+        let err = sync_with_backend(&db, &event_bus, &mock, Some("acct-fail".to_string()))
+            .await
+            .expect_err("sync fails");
+        assert!(err.to_string().contains("mail backend error"));
+
+        assert_eq!(
+            events.recv().await.expect("start event"),
+            CoreEvent::SyncStarted {
+                account_id: Some("acct-fail".to_string())
+            }
+        );
+        match events.recv().await.expect("error event") {
+            CoreEvent::Error { message } => assert!(message.contains("sync failed")),
+            other => panic!("expected CoreEvent::Error, got {other:?}"),
+        }
+        assert_eq!(
+            events.recv().await.expect("complete event"),
+            CoreEvent::SyncCompleted {
+                account_id: Some("acct-fail".to_string())
+            }
+        );
+        // Never stuck at Syncing, even on failure.
+        assert_eq!(event_bus.current_state().status, EngineStatus::Idle);
+    }
+
+    #[tokio::test]
+    async fn find_account_returns_config_when_present() {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let config = sample_jmap_account("acct-find-1");
+        db.save_account(&config).await.expect("save account");
+
+        let found = find_account(&db, "acct-find-1")
+            .await
+            .expect("account found");
+        assert_eq!(found.id, "acct-find-1");
+    }
+
+    #[tokio::test]
+    async fn find_account_errors_when_missing() {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+
+        let err = find_account(&db, "acct-missing")
+            .await
+            .expect_err("account not found");
+        assert!(matches!(err, SyncError::AccountNotFound(id) if id == "acct-missing"));
+    }
+
+    #[test]
+    fn build_mail_backend_constructs_imap_engine_for_imap_smtp_protocol() {
+        let config = sample_imap_account("acct-imap-1");
+        // Only constructs the engine -- never invokes any trait method, so
+        // this never touches the network.
+        let _backend = build_mail_backend(&config, "irrelevant-password");
+    }
+
+    #[test]
+    fn build_mail_backend_constructs_jmap_engine_for_jmap_protocol() {
+        let config = sample_jmap_account("acct-jmap-build-1");
+        let _backend = build_mail_backend(&config, "irrelevant-password");
+    }
+
+    #[tokio::test]
+    async fn run_account_sync_persists_real_backend_messages_and_reports_success() {
+        // Uses a real (non-mock) `JmapEngine` -- its `MailBackend` impl
+        // returns deterministic static data with no network I/O, so this
+        // proves `run_account_sync`'s full production wiring (account
+        // lookup -> keyring password -> real backend -> persist -> events)
+        // end-to-end without violating the "no live network calls in
+        // tests" rule.
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+        let mut events = event_bus.subscribe_events();
+        let secrets = SecretManager::mock();
+
+        let config = sample_jmap_account("acct-jmap-real-1");
+        db.save_account(&config).await.expect("save account");
+        secrets
+            .set_secret(&config.keyring_secret_key, "irrelevant-jmap-token")
+            .expect("store credential in mock vault");
+
+        let synced = run_account_sync(&db, &secrets, &event_bus, "acct-jmap-real-1")
+            .await
+            .expect("sync succeeds");
+        // JmapEngine::sync_folders() returns 3 static folders; sync_messages
+        // returns the same 1 static message per folder regardless of
+        // folder_id, so 3 folders => 3 processed messages (deduplicated to
+        // 1 stored row by `save_email`'s INSERT OR REPLACE on message id).
+        assert_eq!(synced, 3);
+
+        let persisted = db
+            .get_message("jmap-msg-1")
+            .await
+            .expect("jmap message persisted");
+        assert_eq!(persisted.subject, "Welcome to JMAP Sync");
+
+        assert_eq!(
+            events.recv().await.expect("start event"),
+            CoreEvent::SyncStarted {
+                account_id: Some("acct-jmap-real-1".to_string())
+            }
+        );
+        assert_eq!(
+            events.recv().await.expect("complete event"),
+            CoreEvent::SyncCompleted {
+                account_id: Some("acct-jmap-real-1".to_string())
+            }
+        );
+        assert_eq!(event_bus.current_state().status, EngineStatus::Idle);
+    }
+
+    #[tokio::test]
+    async fn run_account_sync_reports_error_for_unknown_account_without_starting_sync() {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+        let mut events = event_bus.subscribe_events();
+        let secrets = SecretManager::mock();
+
+        let err = run_account_sync(&db, &secrets, &event_bus, "acct-does-not-exist")
+            .await
+            .expect_err("unknown account fails");
+        assert!(matches!(err, SyncError::AccountNotFound(_)));
+
+        match events.recv().await.expect("error event") {
+            CoreEvent::Error { message } => {
+                assert!(message.contains("acct-does-not-exist"));
+            }
+            other => panic!("expected CoreEvent::Error, got {other:?}"),
+        }
+        // The sync never began, so status must remain at its default Idle.
+        assert_eq!(event_bus.current_state().status, EngineStatus::Idle);
+    }
+
+    #[tokio::test]
+    async fn run_account_sync_reports_error_when_credential_missing() {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+        let mut events = event_bus.subscribe_events();
+        let secrets = SecretManager::mock();
+
+        let config = sample_jmap_account("acct-no-cred-1");
+        db.save_account(&config).await.expect("save account");
+        // Deliberately never calls `secrets.set_secret(...)`.
+
+        let err = run_account_sync(&db, &secrets, &event_bus, "acct-no-cred-1")
+            .await
+            .expect_err("missing credential fails");
+        assert!(matches!(err, SyncError::Vault(_)));
+
+        match events.recv().await.expect("error event") {
+            CoreEvent::Error { message } => {
+                assert!(message.contains("acct-no-cred-1"));
+            }
+            other => panic!("expected CoreEvent::Error, got {other:?}"),
+        }
+        assert_eq!(event_bus.current_state().status, EngineStatus::Idle);
+    }
+
+    #[tokio::test]
+    async fn run_all_accounts_sync_with_no_accounts_returns_zero_and_emits_events() {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+        let mut events = event_bus.subscribe_events();
+        let secrets = SecretManager::mock();
+
+        let total = run_all_accounts_sync(&db, &secrets, &event_bus).await;
+        assert_eq!(total, 0);
+
+        assert_eq!(
+            events.recv().await.expect("start event"),
+            CoreEvent::SyncStarted { account_id: None }
+        );
+        assert_eq!(
+            events.recv().await.expect("complete event"),
+            CoreEvent::SyncCompleted { account_id: None }
+        );
+    }
+
+    #[tokio::test]
+    async fn run_all_accounts_sync_persists_across_multiple_accounts() {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+        let secrets = SecretManager::mock();
+
+        let config_a = sample_jmap_account("acct-all-a");
+        let config_b = sample_jmap_account("acct-all-b");
+        db.save_account(&config_a).await.expect("save account a");
+        db.save_account(&config_b).await.expect("save account b");
+        secrets
+            .set_secret(&config_a.keyring_secret_key, "token-a")
+            .expect("store credential a");
+        secrets
+            .set_secret(&config_b.keyring_secret_key, "token-b")
+            .expect("store credential b");
+
+        let total = run_all_accounts_sync(&db, &secrets, &event_bus).await;
+        // 2 accounts * 3 folders each = 6 processed messages.
+        assert_eq!(total, 6);
+        assert_eq!(event_bus.current_state().status, EngineStatus::Idle);
+    }
+
+    #[tokio::test]
+    async fn run_all_accounts_sync_reports_error_for_account_missing_credential() {
+        let (db, _dir) = DatabaseEngine::connect_ephemeral()
+            .await
+            .expect("ephemeral db");
+        let event_bus = EventBus::new();
+        let mut events = event_bus.subscribe_events();
+        let secrets = SecretManager::mock();
+
+        let config = sample_jmap_account("acct-all-nocred");
+        db.save_account(&config).await.expect("save account");
+        // No credential stored in the vault for this account.
+
+        let total = run_all_accounts_sync(&db, &secrets, &event_bus).await;
+        assert_eq!(total, 0);
+
+        assert_eq!(
+            events.recv().await.expect("start event"),
+            CoreEvent::SyncStarted { account_id: None }
+        );
+        match events.recv().await.expect("error event") {
+            CoreEvent::Error { message } => {
+                assert!(message.contains("acct-all-nocred"));
+            }
+            other => panic!("expected CoreEvent::Error, got {other:?}"),
+        }
+        assert_eq!(
+            events.recv().await.expect("complete event"),
+            CoreEvent::SyncCompleted { account_id: None }
+        );
+    }
+}