coven/sync/

sync_manager.rs

//! High-level sync manager: lifecycle, membership, status.
//!
//! Owns the sync lifecycle — cloud home + sync loop — and starts/stops it when
//! a provider is connected/disconnected, no app restart required. The host
//! supplies the config snapshot, keys, encryption, database, clock, and blob
//! handling; coven drives the rest.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, RwLock};

use tokio::sync::watch;
use tracing::{error, info};

use crate::blob::transition::{self, MakeLocalError, MakeRemoteError};
use crate::blob::BlobTransitionObserver;
use crate::clock::ClockRef;
use crate::config::{Config, HomeStorage};
use crate::coven::StoreOpenGuard;
use crate::database::{Database, DbError};
use crate::encryption::EncryptionService;
use crate::keys::{DeviceIdentityCustody, KeyError, MasterKeyCustody, StoreKeys};
use crate::storage::cloud::setup::{SetupError, StorageSetupError};
use crate::storage::cloud::{CloudHome, CloudHomeError};
#[cfg(any(test, feature = "test-utils"))]
use crate::sync::cloud_storage::BlobPathScheme;
use crate::sync::cloud_storage::CloudCipher;
#[cfg(any(test, feature = "test-utils"))]
use crate::sync::cloud_storage::CloudSyncStorage;
use crate::sync::cycle::{InitSyncError, SyncComponents};
/// `MemberInfo` lives next to `MemberRole` in the membership module; coven's
/// public path reaches it through here (re-exported from `lib.rs`).
pub(crate) use crate::sync::membership::MemberInfo;
use crate::sync::membership::MemberRole;
use crate::sync::storage::SyncStorage;
use crate::sync::sync_loop::{SyncLoopError, SyncLoopHandle, SyncLoopStatus};

/// Supplies the host's current config for building the next connection. Starting
/// a loop captures one snapshot; commands on that running loop use its immutable
/// store identity, representation, provider settings, and directory even if the
/// host's next config changes meanwhile.
pub(crate) type ConfigProvider = Arc<dyn Fn() -> Config + Send + Sync>;

#[derive(Debug, thiserror::Error)]
pub enum SyncError {
    #[error("sync is not configured")]
    NotConfigured,
    #[error("sync loop is not running")]
    LoopNotRunning,
    #[error("sharing requires an encrypted cloud home")]
    NotEncryptedHome,
    #[error("no master key is established for this opaque store (locked, or never initialized)")]
    MasterKeyNotEstablished,
    #[error("failed to build cloud home: {0}")]
    CloudHome(#[from] CloudHomeError),
    #[error("failed to create sync storage: {0}")]
    StorageSetup(StorageSetupError),
    #[error("key error: {0}")]
    Key(#[from] KeyError),
    #[error("sync initialization error: {0}")]
    Init(#[from] InitSyncError),
    #[error("Store protocol state: {0}")]
    Protocol(String),
    #[error("{0}")]
    Setup(#[from] SetupError),
    #[error("membership error: {0}")]
    Membership(Box<crate::sync::membership_ops::MembershipOpsError>),
    #[error("circle operation: {0}")]
    Circle(#[from] crate::sync::circle_ops::CircleOperationError),
    #[error("device join: {0}")]
    DeviceJoin(#[from] crate::DeviceJoinError),
    #[error("{0}")]
    Database(#[from] DbError),
    #[error("blob upload drain failed: {0}")]
    BlobUpload(DbError),
    #[error("sync loop error: {0}")]
    Loop(SyncLoopError),
}

impl From<crate::sync::membership_ops::MembershipOpsError> for SyncError {
    fn from(error: crate::sync::membership_ops::MembershipOpsError) -> Self {
        Self::Membership(Box::new(error))
    }
}

/// High-level sync manager.
///
/// Holds the store's master-key custody. The at-rest cipher is resolved from
/// it per [`start_sync`](Self::start_sync) call for an opaque home; a Merge
/// store with scoped rows also loads generation 1 for stable row routing,
/// independent of the home's storage representation.
pub(crate) struct SyncManager {
    config_provider: ConfigProvider,
    key_service: StoreKeys,
    custody: Arc<dyn MasterKeyCustody>,
    identity_custody: Arc<dyn DeviceIdentityCustody>,
    db: Database,
    clock: ClockRef,
    cloudkit_ops: Option<Arc<dyn crate::storage::cloud::cloudkit::CloudKitOps>>,
    observer: Option<Arc<dyn BlobTransitionObserver>>,

    /// The store-directory lock, cloned into every sync loop this manager
    /// starts so the loop's thread keeps it alive for the whole of its final
    /// cycle. The lock releases when the last writer — a running loop, else the
    /// handle — is gone, never while a detached loop thread is still writing.
    open_guard: Arc<StoreOpenGuard>,

    /// The current status value the [`CovenHandle`](crate::CovenHandle) owns, cloned
    /// into every sync loop this manager starts so a subscription outlives the
    /// loop restarts a reconnect performs.
    status_tx: tokio::sync::watch::Sender<SyncLoopStatus>,

    // Mutable sync state — updated when providers are connected/disconnected
    sync_loop_handle: RwLock<Option<Arc<SyncLoopHandle>>>,
    cloud_home: RwLock<Option<Arc<dyn CloudHome>>>,

    /// Serializes the membership operations that mint or rotate the store key —
    /// invite (wraps the key to a new member) and remove (mints a fresh key and
    /// re-wraps it to everyone remaining). Each clones the live cipher at entry
    /// and builds a new keyring on top of it; without this, two rapid ops on one
    /// device would both clone the SAME base generation and the second would
    /// overwrite the first's wraps before the first's head guard could catch it,
    /// forking the fleet. Held for the whole operation so the second waits and
    /// builds on the first's committed state.
    member_ops_lock: tokio::sync::Mutex<()>,
}

impl SyncManager {
    fn require_configured_coordination(&self, config: &Config) -> Result<(), SyncError> {
        crate::storage::cloud::setup::require_serial_coordination_config(
            config,
            self.db.write_policy(),
        )
        .map_err(SyncError::StorageSetup)
    }

    async fn storage_for_command(
        &self,
        config: &Config,
        active_loop: Option<&Arc<SyncLoopHandle>>,
    ) -> Result<Arc<crate::sync::cloud_storage::CloudSyncStorage>, SyncError> {
        if let Some(active_loop) = active_loop {
            return Ok(active_loop.storage().clone());
        }
        let storage = match self.cloud_home() {
            Some(home) => crate::storage::cloud::setup::create_sync_storage_with_home(
                config,
                self.custody.as_ref(),
                self.identity_custody.as_ref(),
                home,
                None,
            ),
            None => {
                crate::storage::cloud::setup::create_sync_storage_with_cloudkit(
                    config,
                    &self.key_service,
                    self.custody.as_ref(),
                    self.identity_custody.as_ref(),
                    None,
                    self.clock.clone(),
                    self.cloudkit_ops.clone(),
                )
                .await
            }
        }
        .map_err(SyncError::StorageSetup)?;
        Ok(Arc::new(storage))
    }

    /// Build the manager off the owned [`Database`]. Session initialization takes
    /// the database's already-seeded register clock into [`SyncComponents`], and
    /// every connected command reads that captured clock from the installed loop.
    ///
    /// Construction is infallible and synchronous: seeding already happened in
    /// the open path. The manager is built lazily, only once a provider is
    /// connected.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        config_provider: ConfigProvider,
        key_service: StoreKeys,
        custody: Arc<dyn MasterKeyCustody>,
        identity_custody: Arc<dyn DeviceIdentityCustody>,
        db: Database,
        clock: ClockRef,
        cloudkit_ops: Option<Arc<dyn crate::storage::cloud::cloudkit::CloudKitOps>>,
        observer: Option<Arc<dyn BlobTransitionObserver>>,
        open_guard: Arc<StoreOpenGuard>,
        status_tx: tokio::sync::watch::Sender<SyncLoopStatus>,
    ) -> Self {
        Self {
            config_provider,
            key_service,
            custody,
            identity_custody,
            db,
            clock,
            cloudkit_ops,
            observer,
            open_guard,
            status_tx,
            sync_loop_handle: RwLock::new(None),
            cloud_home: RwLock::new(None),
            member_ops_lock: tokio::sync::Mutex::new(()),
        }
    }

    pub(crate) fn cloud_home(&self) -> Option<Arc<dyn CloudHome>> {
        self.cloud_home.read().unwrap().clone()
    }

    pub(crate) fn sync_loop_handle(&self) -> Option<Arc<SyncLoopHandle>> {
        self.sync_loop_handle.read().unwrap().clone()
    }

    pub(crate) async fn prepare_serial_resolution(
        &self,
        branch_base: Option<crate::sync::store_commit::StoreBatchCommitRef>,
        store_dir: &crate::store_dir::StoreDir,
    ) -> Result<crate::sync::store_pull::SerialResolutionPlan, SyncError> {
        let loop_handle = self.sync_loop_handle().ok_or(SyncError::LoopNotRunning)?;
        let storage = loop_handle.storage();
        let coordination = storage
            .serial_coordination()
            .map_err(|error| SyncError::Protocol(error.to_string()))?;
        let store_root_hash = self
            .db
            .local_store_root_ref()
            .await?
            .ok_or_else(|| SyncError::Protocol("exact Store root authority is absent".to_string()))?
            .store_root_hash;
        let identity = crate::keys::require_identity(self.identity_custody.as_ref())?;
        crate::sync::store_pull::prepare_serial_resolution(
            &self.db,
            &**storage,
            coordination,
            store_root_hash,
            store_dir,
            branch_base,
            &identity,
        )
        .await
        .map_err(|error| SyncError::Protocol(error.to_string()))
    }

    pub(crate) async fn cleanup_serial_candidates(
        &self,
        branch_id: coven_core::PendingBranchId,
        plan: &crate::sync::store_pull::SerialResolutionPlan,
    ) -> Result<(), SyncError> {
        let loop_handle = self.sync_loop_handle().ok_or(SyncError::LoopNotRunning)?;
        crate::sync::store_pull::cleanup_serial_candidates(
            &self.db,
            &**loop_handle.storage(),
            branch_id,
            plan,
        )
        .await
        .map_err(|error| SyncError::Protocol(error.to_string()))
    }

    pub(crate) async fn abandon_serial_branch(
        &self,
        branch_id: coven_core::PendingBranchId,
        store_dir: &crate::store_dir::StoreDir,
    ) -> Result<coven_core::sync::store_outbound::SerialBranchAbandonment, SyncError> {
        let loop_handle = self.sync_loop_handle().ok_or(SyncError::LoopNotRunning)?;
        let storage = loop_handle.storage();
        let coordination = storage
            .serial_coordination()
            .map_err(|error| SyncError::Protocol(error.to_string()))?;
        let identity = crate::keys::require_identity(self.identity_custody.as_ref())?;
        let device_id = self
            .db
            .get_protocol_state(coven_core::database::LOCAL_DEVICE_ID_STATE_KEY)
            .await?
            .ok_or_else(|| {
                SyncError::Protocol("local Store device identity is absent".to_string())
            })?;
        crate::sync::store_outbound::abandon_serial_branch(
            &self.db,
            &**storage,
            coordination,
            &device_id,
            &identity,
            store_dir,
            branch_id,
        )
        .await
        .map_err(|error| SyncError::Protocol(error.to_string()))
    }

    pub(crate) async fn abandon_merge_candidate(
        &self,
        write_id: coven_core::WriteId,
    ) -> Result<coven_core::sync::store_outbound::MergeCandidateAbandonment, SyncError> {
        let loop_handle = self.sync_loop_handle().ok_or(SyncError::LoopNotRunning)?;
        let identity = crate::keys::require_identity(self.identity_custody.as_ref())?;
        let device_id = self
            .db
            .get_protocol_state(coven_core::database::LOCAL_DEVICE_ID_STATE_KEY)
            .await?
            .ok_or_else(|| {
                SyncError::Protocol("local Store device identity is absent".to_string())
            })?;
        crate::sync::store_outbound::abandon_merge_candidate(
            &self.db,
            &**loop_handle.storage(),
            &device_id,
            &identity,
            write_id,
        )
        .await
        .map_err(|error| SyncError::Protocol(error.to_string()))
    }

    // =========================================================================
    // Sync lifecycle
    // =========================================================================

    /// Resolve the home's at-rest cipher from `storage` and this manager's
    /// custody. A browsable home never consults custody — it stores in the
    /// clear regardless. An opaque home unlocks the master keyring; no key
    /// established (a locked store, or a browsable/opaque storage mismatch)
    /// is [`SyncError::MasterKeyNotEstablished`], surfaced here rather than
    /// deep inside the home build. The single custody→cipher decision both
    /// [`start_sync`](Self::start_sync) and the test-home connect path share.
    fn resolve_cipher(&self, storage: HomeStorage) -> Result<CloudCipher, SyncError> {
        if storage.is_browsable() {
            Ok(CloudCipher::Plaintext)
        } else {
            let keyring = self.custody.unlock()?;
            CloudCipher::for_storage(storage, keyring.map(Into::into))
                .ok_or(SyncError::MasterKeyNotEstablished)
        }
    }

    fn routing_encryption(&self) -> Result<Option<EncryptionService>, SyncError> {
        (self.db.write_policy() == crate::WritePolicy::MergeConcurrent
            && self.db.gates().has_scoped_graph())
        .then(|| crate::handle::routing_encryption_from_custody(self.custody.as_ref()))
        .transpose()
        .map_err(SyncError::from)
    }

    /// Initialize cloud home and sync loop from current config.
    /// Called at startup (if already configured) and after connecting a provider.
    ///
    /// Two outcomes are success: a configured provider whose home builds and whose
    /// loop starts, and a store with no configured provider that legitimately
    /// starts no loop — the latter is a logged `Ok(())` no-op. A cloud-home build
    /// that *fails* (missing credentials, a bad provider config) is an `Err`, not
    /// "no provider connected": the caller must not install a manager that reports
    /// success with nothing started.
    pub(crate) async fn start_sync(&self) -> Result<(), SyncError> {
        let config = (self.config_provider)();

        if config.cloud_home.provider.is_none() {
            self.stop_sync()?;
            // Not a failure: a store with no configured provider starts no
            // cloud home or sync loop.
            info!("start_sync: sync not configured; no loop started");
            return Ok(());
        }

        self.require_configured_coordination(&config)?;
        crate::storage::cloud::setup::require_exact_slot_capabilities_config(&config)
            .map_err(SyncError::StorageSetup)?;

        let routing_encryption = self.routing_encryption()?;

        self.stop_current_connection()?;

        // The home's at-rest cipher, resolved fresh on every start so a
        // stop/start picks up whatever custody now holds. Built once here so
        // the sync loop and storage share one instance — a member removal
        // rotates the key in place through it.
        let cipher = self.resolve_cipher(config.cloud_home.storage)?;

        // Build the cloud home. A failure here is a real fault — surface it so the
        // caller never installs a manager that started nothing.
        let cloud_home = crate::storage::cloud::create_cloud_home_with_cloudkit(
            &config,
            &self.key_service,
            self.clock.clone(),
            self.cloudkit_ops.clone(),
        )
        .await
        .map_err(SyncError::from)?;
        let cloud_home: Arc<dyn CloudHome> = Arc::from(cloud_home);

        // Initialize sync loop. The synced-table set is owned by the Database, so
        // init_sync reads it from there rather than from a separately-held copy.
        // Sync is enabled here, so `None` means a real startup failure (no synced
        // tables, storage/keypair/auth/membership bootstrap) that init_sync already
        // logged — surface it so the caller never installs a manager whose loop
        // never started.
        //
        // Connect never mints a device identity: a locked agent with no
        // identity established must fail here with `KeyError::NoDeviceIdentity`,
        // not silently forge one.
        let storage = crate::storage::cloud::setup::create_sync_storage_with_home(
            &config,
            self.custody.as_ref(),
            self.identity_custody.as_ref(),
            cloud_home.clone(),
            Some(cipher.clone()),
        )
        .map_err(|error| match error {
            crate::storage::cloud::setup::StorageSetupError::Key(error) => SyncError::Key(error),
            error => SyncError::StorageSetup(error),
        })?;

        let initialization = self.store_initialization().await?;
        let components = crate::sync::cycle::init_sync_over_storage(
            &self.db,
            storage,
            initialization,
            routing_encryption,
        )
        .await
        .map_err(SyncError::from)?;

        let _handle = self.install_sync_loop(components, config)?;
        *self.cloud_home.write().unwrap() = Some(cloud_home);

        Ok(())
    }

    /// Build the sync-loop handle off `components`, start it, and install it. The
    /// shared install tail of [`start_sync`](Self::start_sync) and the test-only
    /// [`start_sync_with_home`](Self::start_sync_with_home): both reach it only
    /// after the bootstrap has produced [`SyncComponents`], so the loop handle is
    /// installed whole, never on a half-built bootstrap.
    fn install_sync_loop(
        &self,
        components: SyncComponents,
        config: Config,
    ) -> Result<Arc<SyncLoopHandle>, SyncError> {
        let handle = Arc::new(SyncLoopHandle::new(
            components,
            self.custody.clone(),
            self.clock.clone(),
            config,
            self.observer.clone(),
            self.open_guard.clone(),
            self.status_tx.clone(),
        ));
        handle.start().map_err(SyncError::Loop)?;

        info!("Sync loop started");
        *self.sync_loop_handle.write().unwrap() = Some(handle.clone());
        Ok(handle)
    }

    /// Test-only: stand the sync loop over an injected `home`/`cipher` instead of
    /// building the cloud home from config via `create_cloud_home`.
    ///
    /// The counterpart of [`start_sync`](Self::start_sync) for a host's
    /// integration tests, which drive coven over a mock [`CloudHome`] no provider
    /// match would ever produce. It skips the config-provider gate — the injected
    /// home IS the enablement, there are no real credentials to check — installs
    /// the home, builds a [`CloudSyncStorage`] over it under the supplied `cipher`
    /// (and the config's blob-path scheme), runs the same bootstrap
    /// [`init_sync`](crate::sync::cycle::init_sync) does via
    /// [`init_sync_over_storage`](crate::sync::cycle::init_sync_over_storage), and
    /// starts the loop. A bootstrap failure is an `Err`, the same fail-loud
    /// discipline `start_sync` keeps — and commit-whole: the home and loop handle
    /// are installed only after the keypair load and bootstrap both succeed, so a
    /// failure leaves nothing installed.
    ///
    /// After this returns, the connected loop's storage is reachable via
    /// [`sync_loop_handle`](Self::sync_loop_handle)`().storage()`, so the handle's
    /// read path serves blobs over the same injected home with no separate hook.
    ///
    /// Like [`start_sync`](Self::start_sync), this never mints a device
    /// identity: the caller must establish one under this manager's identity
    /// custody first, or this fails with `KeyError::NoDeviceIdentity`.
    #[cfg(any(test, feature = "test-utils"))]
    pub(crate) async fn start_sync_with_home(
        &self,
        home: std::sync::Arc<dyn CloudHome>,
        cipher: CloudCipher,
    ) -> Result<(), SyncError> {
        self.start_sync_with_home_parts(home, cipher, None).await
    }

    #[cfg(any(test, feature = "test-utils"))]
    pub(crate) async fn start_sync_with_home_and_coordination(
        &self,
        home: std::sync::Arc<dyn CloudHome>,
        coordination: std::sync::Arc<dyn crate::storage::cloud::CloudHeadStorage>,
        cipher: CloudCipher,
    ) -> Result<(), SyncError> {
        self.start_sync_with_home_parts(home, cipher, Some(coordination))
            .await
    }

    #[cfg(any(test, feature = "test-utils"))]
    async fn start_sync_with_home_parts(
        &self,
        home: std::sync::Arc<dyn CloudHome>,
        cipher: CloudCipher,
        coordination: Option<std::sync::Arc<dyn crate::storage::cloud::CloudHeadStorage>>,
    ) -> Result<(), SyncError> {
        let config = (self.config_provider)();
        crate::storage::cloud::setup::require_exact_slot_capabilities_home(
            home.clone(),
            config.cloud_home.provider.clone(),
        )
        .map_err(SyncError::StorageSetup)?;
        let routing_encryption = self.routing_encryption()?;
        self.stop_current_connection()?;

        let keypair = crate::keys::require_identity(self.identity_custody.as_ref())?;
        let blob_paths = if cipher.is_plaintext() {
            BlobPathScheme::Plain
        } else {
            BlobPathScheme::Hashed
        };
        let mut storage = CloudSyncStorage::new(
            home.clone(),
            cipher.clone(),
            blob_paths,
            config.store_id.clone(),
            keypair,
        )?;
        if let Some(coordination) = coordination {
            storage = storage.with_test_serial_coordination(coordination);
        }

        let initialization = self.store_initialization().await?;
        let components = crate::sync::cycle::init_sync_over_storage(
            &self.db,
            storage,
            initialization,
            routing_encryption,
        )
        .await
        .map_err(SyncError::from)?;

        let _handle = self.install_sync_loop(components, config)?;
        *self.cloud_home.write().unwrap() = Some(home);

        Ok(())
    }

    async fn store_initialization(
        &self,
    ) -> Result<crate::sync::cycle::StoreInitialization, SyncError> {
        let Some(expected_store_root) = self.db.local_store_root_ref().await? else {
            return Ok(crate::sync::cycle::StoreInitialization::CreateStore);
        };
        Ok(crate::sync::cycle::StoreInitialization::OpenStore {
            expected_store_root,
        })
    }

    /// Test-only: stand the sync loop over an injected `home` while resolving the
    /// at-rest cipher from custody exactly as [`start_sync`](Self::start_sync)
    /// does — the counterpart of [`start_sync_with_home`](Self::start_sync_with_home)
    /// for proving the established master key is the one sealing traffic. Unlike
    /// that method, which takes the cipher explicitly and never consults custody,
    /// this drives the real [`resolve_cipher`](Self::resolve_cipher) path: an
    /// opaque home with no key established fails [`SyncError::MasterKeyNotEstablished`]
    /// before the loop starts.
    #[cfg(any(test, feature = "test-utils"))]
    pub(crate) async fn start_sync_with_test_home_custody(
        &self,
        home: std::sync::Arc<dyn CloudHome>,
    ) -> Result<(), SyncError> {
        let storage = (self.config_provider)().cloud_home.storage;
        let cipher = self.resolve_cipher(storage)?;
        self.start_sync_with_home(home, cipher).await
    }

    /// Tear down the sync loop and cloud home.
    pub(crate) fn stop_sync(&self) -> Result<(), SyncError> {
        let stop_result = self.stop_current_loop();
        *self.sync_loop_handle.write().unwrap() = None;
        *self.cloud_home.write().unwrap() = None;

        match stop_result {
            Ok(()) => {
                info!("Sync loop stopped");
                Ok(())
            }
            Err(error) => {
                error!("Sync loop stop failed: {error}");
                Err(error)
            }
        }
    }

    fn stop_current_loop(&self) -> Result<(), SyncError> {
        let handle = self.sync_loop_handle.write().unwrap().take();
        if let Some(handle) = handle {
            handle.stop().map_err(SyncError::Loop)?;
        }
        Ok(())
    }

    fn stop_current_connection(&self) -> Result<(), SyncError> {
        let stop_result = self.stop_current_loop();
        *self.cloud_home.write().unwrap() = None;
        stop_result
    }

    // =========================================================================
    // Status / config queries
    // =========================================================================

    pub(crate) fn is_sync_ready(&self) -> bool {
        self.sync_loop_handle
            .read()
            .unwrap()
            .as_ref()
            .is_some_and(|h| h.is_running())
    }

    pub(crate) fn trigger_sync(&self) {
        if let Some(ref sync_loop) = *self.sync_loop_handle.read().unwrap() {
            sync_loop.trigger();
        }
    }

    // =========================================================================
    // Blob locality transitions (make_remote / make_local / cancel_make_remote)
    // =========================================================================

    /// Make `(root_table, root_id)` Remote (Local → Remote): enqueue an upload per
    /// user-provided blob from its external file and record the make_remote intent,
    /// then return. The drain uploads each and flips the gate true on the last (see
    /// [`crate::blob::transition::make_remote`]); the gate flip re-emits the subtree,
    /// the cycle's inline push uploads the root's host-provided blobs, and
    /// `on_root_made_remote` fires. `pin` keeps the uploaded blobs in coven's cache
    /// as pinned (offline) copies.
    pub(crate) async fn make_remote(
        &self,
        root_table: &str,
        root_id: &str,
        pin: bool,
    ) -> Result<(), MakeRemoteError> {
        if !self.is_sync_ready() {
            return Err(MakeRemoteError::SyncNotReady);
        }
        let sync_loop = self
            .sync_loop_handle()
            .ok_or(MakeRemoteError::SyncNotReady)?;
        transition::make_remote(
            &self.db,
            sync_loop.store_dir(),
            sync_loop.hlc(),
            root_table,
            root_id,
            pin,
        )
        .await?;
        self.trigger_sync();
        Ok(())
    }

    /// Cancel an in-flight make_remote of `(root_table, root_id)`: clear its intent
    /// and pending uploads and tombstone any blob that already landed. The gate never
    /// flips, so the root stays Local.
    pub(crate) async fn cancel_make_remote(
        &self,
        root_table: &str,
        root_id: &str,
    ) -> Result<(), MakeRemoteError> {
        if !self.is_sync_ready() {
            return Err(MakeRemoteError::SyncNotReady);
        }
        transition::cancel_make_remote(&self.db, root_table, root_id).await?;
        self.trigger_sync();
        Ok(())
    }

    /// Make `(root_table, root_id)` Local (Remote → Local): bring each blob back to a
    /// local file durability-first — a user-provided blob to the path named in `dest`
    /// (blob id → destination path), a host-provided blob to coven's local store (no
    /// dest) — then flip the gate false, register the user-provided external refs,
    /// and enqueue the cloud deletes in one atomic commit. Awaitable; `cancel` aborts
    /// before the commit (the root stays Remote). `dest` carries user-provided ids
    /// only. Per-blob materialize progress and the completion event reach the
    /// observer this manager was built with.
    pub(crate) async fn make_local(
        &self,
        root_table: &str,
        root_id: &str,
        dest: &HashMap<String, PathBuf>,
        cancel: &watch::Receiver<bool>,
        routing_encryption: Option<EncryptionService>,
    ) -> Result<(), MakeLocalError> {
        if !self.is_sync_ready() {
            return Err(MakeLocalError::SyncNotReady);
        }
        let sync_loop = self
            .sync_loop_handle()
            .ok_or(MakeLocalError::SyncNotReady)?;
        let storage: &dyn SyncStorage = &**sync_loop.storage();
        transition::make_local(
            &self.db,
            storage,
            sync_loop.store_dir(),
            sync_loop.hlc(),
            routing_encryption,
            self.observer.as_deref(),
            root_table,
            root_id,
            dest,
            cancel,
        )
        .await?;
        self.trigger_sync();
        Ok(())
    }

    // =========================================================================
    // Keys / codes
    // =========================================================================

    // =========================================================================
    // Membership
    // =========================================================================

    pub(crate) async fn get_members(&self) -> Result<Vec<MemberInfo>, SyncError> {
        let active_loop = self.sync_loop_handle();
        let config = active_loop
            .as_ref()
            .map(|handle| handle.config().clone())
            .unwrap_or_else(|| (self.config_provider)());
        if active_loop.is_none() && config.cloud_home.provider.is_none() {
            info!("get_members: sync not configured; returning no members");
            return Ok(Vec::new());
        }
        self.require_configured_coordination(&config)?;
        let storage = self
            .storage_for_command(&config, active_loop.as_ref())
            .await?;

        let user_pubkey = crate::keys::identity_public_key(self.identity_custody.as_ref())?;
        crate::sync::membership_ops::get_members(
            &*storage,
            user_pubkey.as_ref().map(|k| k.as_slice()),
            &self.db,
        )
        .await
        .map_err(SyncError::from)
    }

    /// Build a restore code for this store: fetch the current membership-head
    /// floor from the cloud and mint the code from it, so the restorer can seed
    /// its watermark from mint-time
    /// state rather than accepting any signed head as a fresh device would
    /// otherwise have to. Requires a connected provider — unlike the old,
    /// storage-free `generate_restore_code`, minting a trustworthy floor is a
    /// network read, not a pure function of local config and keyring state.
    pub(crate) async fn generate_restore_code(&self) -> Result<String, SyncError> {
        let active_loop = self.sync_loop_handle();
        let config = active_loop
            .as_ref()
            .map(|handle| handle.config().clone())
            .unwrap_or_else(|| (self.config_provider)());
        if active_loop.is_none() && config.cloud_home.provider.is_none() {
            return Err(SyncError::NotConfigured);
        }
        self.require_configured_coordination(&config)?;
        let storage = self
            .storage_for_command(&config, active_loop.as_ref())
            .await?;

        let pinned_owner = self
            .db
            .get_protocol_state(crate::sync::membership_ops::OWNER_PUBKEY_STATE_KEY)
            .await?;
        let founder_pubkey = pinned_owner.clone().ok_or_else(|| {
            SyncError::from(crate::sync::membership_ops::MembershipOpsError::NoFounderChain)
        })?;
        let store_root =
            self.db.local_store_root_ref().await?.ok_or_else(|| {
                SyncError::Protocol("store protocol root hash is absent".to_string())
            })?;
        let membership_floor = match self.db.write_policy() {
            crate::WritePolicy::MergeConcurrent => {
                crate::join_code::MembershipFloor::MergeConcurrent(
                    crate::sync::membership_ops::current_membership_floor(
                        &*storage,
                        &store_root,
                        pinned_owner.as_deref(),
                        Some(&self.db),
                    )
                    .await
                    .map_err(SyncError::from)?,
                )
            }
            crate::WritePolicy::Serial => {
                let coordination = storage.serial_coordination().map_err(|error| {
                    SyncError::Protocol(format!("Serial coordination: {error}"))
                })?;
                let reference =
                    crate::sync::store_outbound::current_serial_head_ref(&self.db, coordination)
                        .await
                        .map_err(|error| SyncError::Protocol(error.to_string()))?;
                crate::join_code::MembershipFloor::Serial(reference)
            }
        };
        let identity = crate::keys::require_identity(self.identity_custody.as_ref())?;
        let authority = crate::sync::restore_code::RestoreAuthority::ActivatedContinuation(
            self.db
                .export_activated_device_continuation(&identity)
                .await?,
        );

        crate::storage::cloud::setup::generate_restore_code(
            &config,
            &self.key_service,
            self.custody.as_ref(),
            store_root,
            founder_pubkey,
            membership_floor,
            authority,
        )
        .map_err(SyncError::from)
    }

    pub(crate) fn invite_member<'a>(
        &'a self,
        public_key_hex: &'a str,
        invitee_email: Option<&'a str>,
        role: MemberRole,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<String, SyncError>> + Send + 'a>>
    {
        Box::pin(async move {
            // Serialize with any other key-minting/rotating member op on this device.
            let _member_ops = self.member_ops_lock.lock().await;

            let sync_loop = self
                .sync_loop_handle
                .read()
                .unwrap()
                .clone()
                .ok_or(SyncError::LoopNotRunning)?;

            // Inviting a member wraps the store key to them, which only an encrypted
            // home has. Refuse before touching the membership chain.
            if sync_loop.current_encryption().is_none() {
                return Err(SyncError::NotEncryptedHome);
            }
            let store_name = sync_loop.config().store_name.clone();
            let invite_code = sync_loop
                .invite_member(public_key_hex, invitee_email, role, &store_name)
                .await
                .map_err(SyncError::from)?;

            Ok(crate::join_code::encode(&invite_code))
        })
    }

    pub(crate) fn remove_member<'a>(
        &'a self,
        public_key_hex: &'a str,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<String, SyncError>> + Send + 'a>>
    {
        Box::pin(async move {
            // Serialize with any other key-minting/rotating member op on this device,
            // so a second removal builds on this one's committed state rather than
            // cloning the same base cipher and overwriting its wraps.
            let _member_ops = self.member_ops_lock.lock().await;

            let sync_loop = self
                .sync_loop_handle
                .read()
                .unwrap()
                .clone()
                .ok_or(SyncError::LoopNotRunning)?;

            // Removing a member rotates the store key, which only an encrypted home
            // has. Refuse up front so a plaintext home never mutates the membership
            // chain or re-wraps keys before the rotation fails.
            if sync_loop.current_encryption().is_none() {
                return Err(SyncError::NotEncryptedHome);
            }

            // Removing a member commits the cloud key rotation and then adopts the
            // rotated key into this device's keyring and live cipher. The host records
            // the returned fingerprint and that a key is stored in its own config; an
            // adoption failure surfaces as its own membership variant naming the
            // half-applied state and its remedies — and, structurally, this device
            // seals nothing new for the cloud until one of those remedies adopts it
            // (`pending_rotation`, shared with the sync loop this same store runs).
            let outcome = sync_loop.remove_member(public_key_hex).await;

            // Durably record the marker's state whether adoption succeeded or failed:
            // on a failed adoption the rotation is committed on the cloud but this
            // device is still sealing under the superseded generation, so a restart
            // must remember the pause; on success the marker is already cleared and
            // this deletes the durable record.
            sync_loop.persist_pending_rotation().await?;

            let fingerprint = outcome.map_err(SyncError::from)?;
            Ok(fingerprint)
        })
    }

    pub(crate) async fn create_circle(&self, name: &str) -> Result<crate::CircleId, SyncError> {
        let sync_loop = self.sync_loop_handle().ok_or(SyncError::LoopNotRunning)?;
        sync_loop.create_circle(name).await.map_err(SyncError::from)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use crate::clock::SystemClock;
    use crate::config::CloudProvider;
    use crate::coven::StoreOpenGuard;
    use crate::encryption::MasterKeyring;
    use crate::keys::{test_keyring, KeyError, StoreKeys};
    use crate::storage::cloud::test_utils::InMemoryCloudHome;
    use crate::storage::cloud::CloudHomeJoinInfo;
    use crate::store_dir::StoreDir;
    use std::sync::Arc;

    struct NoImmutableCopyHome;

    #[async_trait::async_trait]
    impl CloudHome for NoImmutableCopyHome {
        async fn put_object(&self, _key: &str, _data: Vec<u8>) -> Result<(), CloudHomeError> {
            panic!("incapable home must be rejected before I/O")
        }

        async fn open_multipart<'a>(
            &'a self,
            _key: &str,
            _total_len: u64,
        ) -> Result<crate::storage::cloud::BoxPartSink<'a>, CloudHomeError> {
            panic!("incapable home must be rejected before I/O")
        }

        fn multipart_threshold(&self) -> u64 {
            panic!("incapable home must be rejected before I/O")
        }

        async fn read(&self, _key: &str) -> Result<Vec<u8>, CloudHomeError> {
            panic!("incapable home must be rejected before I/O")
        }

        async fn read_range(
            &self,
            _key: &str,
            _start: u64,
            _end: u64,
        ) -> Result<Vec<u8>, CloudHomeError> {
            panic!("incapable home must be rejected before I/O")
        }

        async fn list(&self, _prefix: &str) -> Result<Vec<String>, CloudHomeError> {
            panic!("incapable home must be rejected before I/O")
        }

        async fn delete(&self, _key: &str) -> Result<(), CloudHomeError> {
            panic!("incapable home must be rejected before I/O")
        }

        async fn exists(&self, _key: &str) -> Result<bool, CloudHomeError> {
            panic!("incapable home must be rejected before I/O")
        }

        async fn set_access(
            &self,
            _desired: crate::storage::cloud::CloudAccessState,
        ) -> Result<crate::storage::cloud::CloudAccessOutcome, CloudHomeError> {
            panic!("incapable home must be rejected before I/O")
        }
    }

    /// A custody that never has a master key established — `unlock` always
    /// returns `None`. For tests exercising a locked/unestablished store, or
    /// a browsable home where custody is never consulted at all.
    struct NoKeyCustody;

    impl MasterKeyCustody for NoKeyCustody {
        fn unlock(&self) -> Result<Option<MasterKeyring>, KeyError> {
            Ok(None)
        }
        fn persist(&self, _keyring: &MasterKeyring) -> Result<(), KeyError> {
            Ok(())
        }
        fn forget(&self) -> Result<(), KeyError> {
            Ok(())
        }
    }

    /// The identity sibling of [`NoKeyCustody`]: `unlock` always returns
    /// `None`, for tests exercising a store with no identity established.
    struct NoIdentityCustody;

    impl DeviceIdentityCustody for NoIdentityCustody {
        fn unlock(&self) -> Result<Option<crate::keys::UserKeypair>, KeyError> {
            Ok(None)
        }
        fn persist(&self, _keypair: &crate::keys::UserKeypair) -> Result<(), KeyError> {
            Ok(())
        }
        fn forget(&self) -> Result<(), KeyError> {
            Ok(())
        }
    }

    /// A ready-to-use, already-established identity custody for tests whose
    /// focus is elsewhere (blob transitions, membership, restore-code
    /// generation) — seeded in-memory so it needs no keyring registration.
    fn established_identity_custody() -> Arc<dyn DeviceIdentityCustody> {
        crate::identity_custody::IdentityCustody::InMemory(crate::keys::UserKeypair::generate())
            .resolve("unused-store-id", &StoreDir::new("unused-store-dir"))
    }

    #[tokio::test]
    async fn get_members_surfaces_malformed_cloud_credentials() {
        test_keyring::install();
        let tmp = tempfile::tempdir().expect("temp dir");
        let store_dir = StoreDir::new(tmp.path());
        let store_id = "sync-enabled-malformed-credentials";
        let key_service = StoreKeys::new(store_id.to_string());
        key_service
            .cloud_home_credentials_entry_for_test()
            .expect("create credentials entry")
            .set_password("{")
            .expect("write malformed credentials");
        let join_info = CloudHomeJoinInfo::S3 {
            bucket: "bucket".to_string(),
            region: "region".to_string(),
            endpoint: None,
            access_key: "access".to_string(),
            secret_key: "secret".to_string(),
            key_prefix: None,
        };
        let config = crate::sync::join::build_config(
            store_id,
            "device",
            &store_dir,
            "store",
            &join_info,
            &CloudCipher::Plaintext,
            None,
        );
        let manager = SyncManager::new(
            Arc::new(move || config.clone()),
            key_service,
            Arc::new(NoKeyCustody),
            established_identity_custody(),
            crate::sync::test_helpers::open_test_db(),
            Arc::new(SystemClock),
            None,
            None,
            StoreOpenGuard::acquire_for_test(&store_dir),
            tokio::sync::watch::channel(SyncLoopStatus::Offline).0,
        );

        let error = manager
            .get_members()
            .await
            .expect_err("malformed stored credentials must fail");
        // The typed CloudHomeError survives up through StorageSetup to the public
        // SyncError surface — not flattened into a string — so its retryability
        // verdict is still readable: malformed credentials are a configuration
        // fault the user must fix, not a transient retry.
        let SyncError::StorageSetup(StorageSetupError::CloudHome(cloud_home_error)) = &error else {
            panic!("expected StorageSetup(CloudHome(_)), got {error:?}");
        };
        assert!(matches!(cloud_home_error, CloudHomeError::Configuration(_)));
        assert!(!cloud_home_error.is_retryable());
        assert!(error
            .to_string()
            .contains("malformed cloud home credentials JSON"));
    }

    #[tokio::test]
    async fn start_sync_rejects_an_opaque_home_without_a_master_key() {
        test_keyring::install();
        let (_tmp, store_dir) = crate::sync::test_helpers::temp_store_dir();
        let open_guard = StoreOpenGuard::acquire_for_test(&store_dir);
        let mut config = Config::with_defaults(
            "lib-opaque-no-encryption".to_string(),
            "test-device".to_string(),
            store_dir,
            "Test Store".to_string(),
        );
        // Opaque storage (the default) with a configured provider but no
        // established master key is a locked-store contradiction — custody's
        // `unlock` returns `None`.
        config.cloud_home.provider = Some(CloudProvider::S3);
        let manager = SyncManager::new(
            Arc::new(move || config.clone()),
            StoreKeys::new("lib-opaque-no-encryption".to_string()),
            Arc::new(NoKeyCustody),
            established_identity_custody(),
            crate::sync::test_helpers::open_test_db(),
            Arc::new(SystemClock),
            None,
            None,
            open_guard,
            tokio::sync::watch::channel(SyncLoopStatus::Offline).0,
        );

        let error = manager
            .start_sync()
            .await
            .expect_err("opaque home without an established master key must fail");
        assert!(
            matches!(error, SyncError::MasterKeyNotEstablished),
            "expected MasterKeyNotEstablished, got {error:?}"
        );
    }

    #[tokio::test]
    async fn start_sync_refuses_a_known_unsupported_serial_provider_before_provider_access() {
        let (_tmp, store_dir) = crate::sync::test_helpers::temp_store_dir();
        let open_guard = StoreOpenGuard::acquire_for_test(&store_dir);
        let mut config = Config::with_defaults(
            "serial-google-drive".to_string(),
            "test-device".to_string(),
            store_dir,
            "Serial Store".to_string(),
        );
        config.cloud_home.provider = Some(CloudProvider::GoogleDrive);
        let manager = SyncManager::new(
            Arc::new(move || config.clone()),
            StoreKeys::new("serial-google-drive".to_string()),
            Arc::new(NoKeyCustody),
            Arc::new(NoIdentityCustody),
            crate::sync::test_helpers::open_serial_test_db(),
            Arc::new(SystemClock),
            None,
            None,
            open_guard,
            tokio::sync::watch::channel(SyncLoopStatus::Offline).0,
        );

        let error = manager
            .start_sync()
            .await
            .expect_err("Google Drive cannot coordinate a Serial Store");
        assert!(matches!(
            error,
            SyncError::StorageSetup(StorageSetupError::SerialCoordinationUnavailable {
                provider: CloudProvider::GoogleDrive,
            })
        ));
        assert!(manager.sync_loop_handle().is_none());
        assert!(manager.cloud_home().is_none());
    }

    #[tokio::test]
    async fn immutable_copy_admission_refuses_before_stopping_the_active_loop() {
        let (_tmp, store_dir) = crate::sync::test_helpers::temp_store_dir();
        let open_guard = StoreOpenGuard::acquire_for_test(&store_dir);
        let config = Arc::new(RwLock::new(Config::with_defaults(
            "immutable-admission-before-stop".to_string(),
            "test-device".to_string(),
            store_dir,
            "Blob Store".to_string(),
        )));
        let db = crate::sync::test_helpers::open_test_db_with_blob(crate::BlobDecl::new(
            "photos",
            crate::Provenance::HostProvided,
            crate::CacheFill::CacheLazy,
        ));
        let manager = SyncManager::new(
            {
                let config = config.clone();
                Arc::new(move || config.read().expect("read config").clone())
            },
            StoreKeys::new("immutable-admission-before-stop".to_string()),
            Arc::new(NoKeyCustody),
            established_identity_custody(),
            db,
            Arc::new(SystemClock),
            None,
            None,
            open_guard,
            tokio::sync::watch::channel(SyncLoopStatus::Offline).0,
        );
        manager
            .start_sync_with_home(Arc::new(InMemoryCloudHome::new()), CloudCipher::Plaintext)
            .await
            .expect("install active loop");
        let active_loop = manager.sync_loop_handle().expect("active loop");

        {
            let mut config = config.write().expect("write config");
            config.cloud_home.provider = Some(CloudProvider::S3);
            config.cloud_home.s3_endpoint = Some("https://objects.example".to_string());
            config.cloud_home.s3_exact_slots = None;
        }
        let error = manager
            .start_sync()
            .await
            .expect_err("unsupported immutable-copy provider is refused");

        assert!(matches!(
            error,
            SyncError::StorageSetup(StorageSetupError::ExactSlotsUnavailable {
                provider: CloudProvider::S3,
            })
        ));
        assert!(active_loop.is_running());
        assert!(manager.cloud_home().is_some());

        let error = manager
            .start_sync_with_home(Arc::new(NoImmutableCopyHome), CloudCipher::Plaintext)
            .await
            .expect_err("injected home without immutable-copy storage is refused");
        assert!(matches!(
            error,
            SyncError::StorageSetup(StorageSetupError::ExactSlotsUnavailable {
                provider: CloudProvider::S3,
            })
        ));
        assert!(active_loop.is_running());
        assert!(manager.cloud_home().is_some());
    }

    #[tokio::test]
    async fn start_sync_with_home_stops_the_previous_loop_before_replacement() {
        test_keyring::install();

        let (_tmp, store_dir) = crate::sync::test_helpers::temp_store_dir();
        let open_guard = StoreOpenGuard::acquire_for_test(&store_dir);
        let config = Config::with_defaults(
            "lib-manager-restart".to_string(),
            "test-device".to_string(),
            store_dir,
            "Test Store".to_string(),
        );
        let manager = SyncManager::new(
            Arc::new(move || config.clone()),
            StoreKeys::new("lib-manager-restart".to_string()),
            Arc::new(NoKeyCustody),
            established_identity_custody(),
            crate::sync::test_helpers::open_test_db(),
            Arc::new(SystemClock),
            None,
            None,
            open_guard,
            tokio::sync::watch::channel(SyncLoopStatus::Offline).0,
        );

        let home = Arc::new(InMemoryCloudHome::new());
        manager
            .start_sync_with_home(home.clone(), CloudCipher::Plaintext)
            .await
            .expect("first test home starts");
        let first_loop = manager
            .sync_loop_handle()
            .expect("first loop handle installed");
        assert!(first_loop.is_running(), "first loop starts running");

        manager
            .start_sync_with_home(home, CloudCipher::Plaintext)
            .await
            .expect("replacement test home starts");
        let replacement_loop = manager
            .sync_loop_handle()
            .expect("replacement loop handle installed");

        assert!(
            !first_loop.is_running(),
            "starting sync again stops the previous loop before replacement",
        );
        assert!(
            replacement_loop.is_running(),
            "replacement loop remains running",
        );
    }

    #[tokio::test]
    async fn failed_restart_leaves_no_stale_cloud_home() {
        test_keyring::install();

        let (_tmp, store_dir) = crate::sync::test_helpers::temp_store_dir();
        let open_guard = StoreOpenGuard::acquire_for_test(&store_dir);
        let config = Arc::new(RwLock::new(Config::with_defaults(
            "lib-manager-failed-restart".to_string(),
            "test-device".to_string(),
            store_dir,
            "Test Store".to_string(),
        )));
        let manager = SyncManager::new(
            {
                let config = config.clone();
                Arc::new(move || config.read().unwrap().clone())
            },
            StoreKeys::new("lib-manager-failed-restart".to_string()),
            // An established master key so the opaque default storage passes
            // the cipher precondition and the restart fails at the home build
            // itself.
            crate::custody::KeyCustody::InMemory(MasterKeyring::generate()).resolve(
                "lib-manager-failed-restart",
                &StoreDir::new("unused-store-dir"),
            ),
            established_identity_custody(),
            crate::sync::test_helpers::open_test_db(),
            Arc::new(SystemClock),
            None,
            None,
            open_guard,
            tokio::sync::watch::channel(SyncLoopStatus::Offline).0,
        );

        manager
            .start_sync_with_home(Arc::new(InMemoryCloudHome::new()), CloudCipher::Plaintext)
            .await
            .expect("injected home starts");
        assert!(manager.cloud_home().is_some(), "injected home is installed");

        config.write().unwrap().cloud_home.provider = Some(CloudProvider::S3);
        let error = manager
            .start_sync()
            .await
            .expect_err("invalid configured provider fails restart");
        assert!(
            error.to_string().contains("failed to build cloud home"),
            "restart failure surfaces the provider setup error: {error}",
        );
        assert!(
            manager.sync_loop_handle().is_none(),
            "failed restart leaves no loop installed",
        );
        assert!(
            manager.cloud_home().is_none(),
            "failed restart must not leave the previous cloud home installed",
        );
    }

    /// The `NoDeviceIdentity` sibling of
    /// `start_sync_rejects_an_opaque_home_without_a_master_key`: connecting
    /// with a configured home but no device identity established must fail
    /// typed, with nothing installed — never silently mint one. Browsable
    /// storage so the master-key precondition is out of the way and this
    /// isolates the identity check.
    #[tokio::test]
    async fn start_sync_rejects_a_connect_with_no_device_identity_established() {
        test_keyring::install();

        let (_tmp, store_dir) = crate::sync::test_helpers::temp_store_dir();
        let open_guard = StoreOpenGuard::acquire_for_test(&store_dir);
        let store_id = "lib-no-device-identity".to_string();
        let key_service = StoreKeys::new(store_id.clone());
        key_service
            .set_cloud_home_credentials(&crate::keys::CloudHomeCredentials::S3 {
                access_key: "ak".to_string(),
                secret_key: "sk".to_string(),
            })
            .expect("seed S3 credentials");

        let mut config = Config::with_defaults(
            store_id.clone(),
            "test-device".to_string(),
            store_dir,
            "Test Store".to_string(),
        );
        config.cloud_home.provider = Some(CloudProvider::S3);
        config.cloud_home.storage = HomeStorage::Browsable;
        config.cloud_home.s3_bucket = Some("bucket".to_string());
        config.cloud_home.s3_region = Some("us-east-1".to_string());

        let manager = SyncManager::new(
            Arc::new(move || config.clone()),
            key_service,
            Arc::new(NoKeyCustody),
            Arc::new(NoIdentityCustody),
            crate::sync::test_helpers::open_test_db(),
            Arc::new(SystemClock),
            None,
            None,
            open_guard,
            tokio::sync::watch::channel(SyncLoopStatus::Offline).0,
        );

        let error = manager
            .start_sync()
            .await
            .expect_err("no device identity established must fail the connect");
        assert!(
            matches!(error, SyncError::Key(KeyError::NoDeviceIdentity)),
            "got {error:?}"
        );
        assert!(
            manager.sync_loop_handle().is_none(),
            "a failed connect installs no loop",
        );
        assert!(
            manager.cloud_home().is_none(),
            "a failed connect installs no cloud home",
        );
    }

    #[tokio::test]
    async fn browsable_test_home_with_a_foreign_founder_installs_nothing() {
        test_keyring::install();

        let (_tmp, store_dir) = crate::sync::test_helpers::temp_store_dir();
        let store_id = "lib-foreign-browsable-founder";
        let mut config = Config::with_defaults(
            store_id.to_string(),
            "test-device".to_string(),
            store_dir.clone(),
            "Test Store".to_string(),
        );
        config.cloud_home.storage = HomeStorage::Browsable;
        let home = Arc::new(InMemoryCloudHome::new());
        let attacker = crate::keys::UserKeypair::generate();
        let attacker_storage = CloudSyncStorage::new(
            home.clone(),
            CloudCipher::Plaintext,
            BlobPathScheme::Plain,
            store_id,
            attacker.clone(),
        )
        .expect("build attacker storage");
        let attacker_db = crate::sync::test_helpers::open_test_db();
        crate::sync::store_protocol_root::create_store(
            &attacker_db,
            &attacker_storage,
            store_id,
            &attacker,
        )
        .await
        .expect("publish attacker Store root");

        let victim = crate::keys::UserKeypair::generate();
        let db = crate::sync::test_helpers::open_test_db();
        let manager = SyncManager::new(
            Arc::new(move || config.clone()),
            StoreKeys::new(store_id.to_string()),
            Arc::new(NoKeyCustody),
            crate::identity_custody::IdentityCustody::InMemory(victim)
                .resolve(store_id, &store_dir),
            db.clone(),
            Arc::new(SystemClock),
            None,
            None,
            StoreOpenGuard::acquire_for_test(&store_dir),
            tokio::sync::watch::channel(SyncLoopStatus::Offline).0,
        );

        let error = manager
            .start_sync_with_home(home, CloudCipher::Plaintext)
            .await
            .expect_err("foreign founder must prevent sync startup");
        assert!(
            matches!(error, SyncError::Init(InitSyncError::StoreProtocolRoot(_))),
            "unexpected startup error: {error:?}",
        );
        assert!(manager.sync_loop_handle().is_none());
        assert!(manager.cloud_home().is_none());
        assert_eq!(
            db.get_protocol_state(crate::sync::membership_ops::OWNER_PUBKEY_STATE_KEY)
                .await
                .unwrap(),
            None,
        );
    }

    /// Key material a connect resolves from custody is never cached across
    /// connects — `start_sync` re-derives the cipher fresh every call via
    /// `resolve_cipher`, this manager's single
    /// custody→cipher decision. Persists key A, resolves, swaps what the SAME
    /// custody instance serves to key B (a rotation outside any manager call
    /// — the way a host's own key-rotation flow would), and resolves again:
    /// the second resolution reflects B, not a value cached from the first.
    ///
    /// This drives `resolve_cipher` directly rather than a full
    /// connect/disconnect/reconnect through an opaque home: an opaque store's
    /// membership chain is founded and pinned to the local device on first
    /// connect, so swapping its master key outright (rather than through the
    /// real in-place rotation `remove_member` performs, which also re-wraps
    /// existing membership content) would desync a live home — an unrelated
    /// concern to what this test pins. `resolve_cipher` is the exact
    /// mechanism `start_sync` and the custody-resolving test-home connect
    /// path share, so calling it twice with custody mutated in between is the
    /// real unit behind "reconnect uses new material," without wading into
    /// membership bootstrap.
    #[test]
    fn resolve_cipher_never_caches_reflects_whatever_custody_now_serves() {
        let (_tmp, store_dir) = crate::sync::test_helpers::temp_store_dir();
        let store_id = "lib-resolve-cipher-fresh";
        let custody = crate::custody::KeyCustody::Keyring.resolve(store_id, &store_dir);
        let key_a = MasterKeyring::generate();
        custody.persist(&key_a).expect("establish key A");

        let config = Config::with_defaults(
            store_id.to_string(),
            "test-device".to_string(),
            store_dir.clone(),
            "Test Store".to_string(),
        );
        let manager = SyncManager::new(
            Arc::new(move || config.clone()),
            StoreKeys::new(store_id.to_string()),
            custody.clone(),
            established_identity_custody(),
            crate::sync::test_helpers::open_test_db(),
            Arc::new(SystemClock),
            None,
            None,
            StoreOpenGuard::acquire_for_test(&store_dir),
            tokio::sync::watch::channel(SyncLoopStatus::Offline).0,
        );

        let fingerprint_a = match manager
            .resolve_cipher(crate::config::HomeStorage::Opaque)
            .expect("resolve the cipher custody serves for key A")
        {
            CloudCipher::Encrypted(enc) => enc.fingerprint(),
            CloudCipher::Plaintext => panic!("opaque storage must resolve an encrypted cipher"),
        };
        assert_eq!(fingerprint_a, key_a.fingerprint());

        // Swap what the SAME custody instance serves — outside any manager
        // call, the way a host's own key-rotation flow would.
        let key_b = MasterKeyring::generate();
        custody
            .persist(&key_b)
            .expect("rotate custody's served key to B");

        let fingerprint_b = match manager
            .resolve_cipher(crate::config::HomeStorage::Opaque)
            .expect("resolve the cipher custody serves for key B")
        {
            CloudCipher::Encrypted(enc) => enc.fingerprint(),
            CloudCipher::Plaintext => panic!("opaque storage must resolve an encrypted cipher"),
        };
        assert_eq!(
            fingerprint_b,
            key_b.fingerprint(),
            "the second resolution must reflect key B, not a value cached from the first call",
        );
        assert_ne!(
            fingerprint_a, fingerprint_b,
            "the two resolutions must differ — custody actually served different material",
        );
    }
}