refactor(proposal): put DB access behind a mockable ProposalStore trait

.gitattributes

@@ -0,0 +1 @@
+* text=auto eol=lf

server/Cargo.lock

@@ -674,6 +674,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "anstyle"
+version = "1.0.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000"
+
 [[package]]
 name = "anyhow"
 version = "1.0.102"
@@ -767,6 +773,7 @@ dependencies = [
  "kameo",
  "kameo_actors",
  "ml-dsa",
+ "mockall",
  "mutants",
  "pem",
  "proptest",
@@ -1983,6 +1990,12 @@ dependencies = [
  "syn 2.0.117",
 ]
 
+[[package]]
+name = "downcast"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1435fa1053d8b2fbbe9be7e97eca7f33d37b28409959813daefc1446a14247f1"
+
 [[package]]
 name = "downcast-rs"
 version = "2.0.2"
@@ -2235,6 +2248,15 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "fragile"
+version = "2.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8878864ba14bb86e818a412bfd6f18f9eabd4ec0f008a28e8f7eb61db532fcf9"
+dependencies = [
+ "futures-core",
+]
+
 [[package]]
 name = "fs_extra"
 version = "1.3.0"
@@ -3352,6 +3374,32 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "mockall"
+version = "0.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1a6ceddfe3ce334925e96bf420fdb2dcee5bed6c632a168ece622676dadeaf8a"
+dependencies = [
+ "cfg-if",
+ "downcast",
+ "fragile",
+ "mockall_derive",
+ "predicates",
+ "predicates-tree",
+]
+
+[[package]]
+name = "mockall_derive"
+version = "0.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9cfe16fbe8a314aeec0b861ac24e60b1e123e97634bab045475b9d6a18416fd8"
+dependencies = [
+ "cfg-if",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.117",
+]
+
 [[package]]
 name = "module-lattice"
 version = "0.2.2"
@@ -3785,6 +3833,32 @@ dependencies = [
  "zerocopy",
 ]
 
+[[package]]
+name = "predicates"
+version = "3.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ada8f2932f28a27ee7b70dd6c1c39ea0675c55a36879ab92f3a715eaa1e63cfe"
+dependencies = [
+ "anstyle",
+ "predicates-core",
+]
+
+[[package]]
+name = "predicates-core"
+version = "1.0.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cad38746f3166b4031b1a0d39ad9f954dd291e7854fcc0eed52ee41a0b50d144"
+
+[[package]]
+name = "predicates-tree"
+version = "1.0.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d0de1b847b39c8131db0467e9df1ff60e6d0562ab8e9a16e568ad0fdb372e2f2"
+dependencies = [
+ "predicates-core",
+ "termtree",
+]
+
 [[package]]
 name = "prettyplease"
 version = "0.2.37"
@@ -5171,6 +5245,12 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "termtree"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f50febec83f5ee1df3015341d8bd429f2d1cc62bcba7ea2076759d315084683"
+
 [[package]]
 name = "test-log"
 version = "0.2.20"

server/crates/arbiter-server/Cargo.toml

@@ -59,6 +59,7 @@ proptest = "1.11.0"
 rstest.workspace = true
 test-log = { version = "0.2", default-features = false, features = ["trace"] }
 ml-dsa.workspace = true
+mockall = "0.15.0"
 
 [lib]
 doctest = false

server/crates/arbiter-server/src/actors/evm/mod.rs

@@ -1,9 +1,13 @@
 use crate::{
-    actors::vault::{CreateNew, Decrypt, Vault},
+    actors::{
+        proposal_manager::events::ProposalApproved,
+        vault::{CreateNew, Decrypt, Vault},
+    },
     crypto::integrity,
     db::{
         DatabaseError, DatabasePool,
-        models::{self, EvmWalletId},
+        models::{self, EvmWalletId, ProposalId},
+        proposal::{ProposalKind, grant_wallet_access, one_off_transaction, persistent_grant},
         schema,
     },
     evm::{
@@ -23,8 +27,9 @@ use diesel::{
     ExpressionMethods, OptionalExtension as _, QueryDsl, SelectableHelper as _, dsl::insert_into,
 };
 use diesel_async::RunQueryDsl;
-use kameo::{Actor, actor::ActorRef, messages};
+use kameo::{Actor, actor::ActorRef, messages, prelude::Message};
 use rand::{SeedableRng, rng, rngs::StdRng};
+use tracing::error;
 
 pub use crate::evm::safe_signer;
 
@@ -62,6 +67,9 @@ pub enum Error {
 
     #[error("Integrity violation: {0}")]
     Integrity(#[from] integrity::Error),
+
+    #[error("Signing error: {0}")]
+    Sign(#[from] SignTransactionError),
 }
 
 #[derive(Actor)]
@@ -267,3 +275,142 @@ impl EvmActor {
         Ok(signer.sign_transaction_sync(&mut transaction)?)
     }
 }
+
+impl Message<ProposalApproved> for EvmActor {
+    type Reply = ();
+
+    /// Every subscriber sees every approval and acts only on the kinds it owns.
+    async fn handle(
+        &mut self,
+        msg: ProposalApproved,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        let result = match msg.kind {
+            ProposalKind::GrantWalletAccess(settings) => self.grant_wallet_access(&settings).await,
+            ProposalKind::ApprovePersistentGrant(settings) => {
+                self.create_persistent_grant(*settings).await
+            }
+            ProposalKind::ApproveOneOffTransaction(settings) => {
+                self.sign_one_off_transaction(msg.id, *settings).await
+            }
+            _ => return,
+        };
+
+        if let Err(error) = result {
+            error!(
+                ?error,
+                proposal_id = msg.id.to_raw(),
+                "Failed to execute an approved proposal"
+            );
+        }
+    }
+}
+
+impl EvmActor {
+    async fn grant_wallet_access(
+        &mut self,
+        settings: &grant_wallet_access::Settings,
+    ) -> Result<(), Error> {
+        let mut conn = self.db.get().await.map_err(DatabaseError::from)?;
+
+        insert_into(schema::evm_wallet_access::table)
+            .values((
+                schema::evm_wallet_access::wallet_id.eq(EvmWalletId::from_raw(settings.wallet_id)),
+                schema::evm_wallet_access::client_id.eq(settings.client_id),
+            ))
+            .execute(&mut conn)
+            .await
+            .map_err(DatabaseError::from)?;
+
+        Ok(())
+    }
+
+    async fn create_persistent_grant(
+        &mut self,
+        grant: persistent_grant::Settings,
+    ) -> Result<(), Error> {
+        use crate::evm::policies::{
+            TransactionRateLimit, VolumeRateLimit, ether_transfer, token_transfers,
+        };
+        use alloy::primitives::U256;
+        use chrono::Duration;
+
+        let volume = |limit: persistent_grant::VolumeLimit| VolumeRateLimit {
+            max_volume: U256::from_be_bytes(limit.max_volume),
+            window: Duration::seconds(limit.window_secs),
+        };
+
+        let basic = SharedGrantSettings {
+            wallet_access_id: grant.wallet_access_id,
+            chain: grant.chain_id,
+            valid_from: grant
+                .valid_from_secs
+                .and_then(|s| chrono::DateTime::from_timestamp(s, 0)),
+            valid_until: grant
+                .valid_until_secs
+                .and_then(|s| chrono::DateTime::from_timestamp(s, 0)),
+            max_gas_fee_per_gas: grant.max_gas_fee_per_gas.map(U256::from_be_bytes),
+            max_priority_fee_per_gas: grant.max_priority_fee_per_gas.map(U256::from_be_bytes),
+            rate_limit: grant.rate_limit.map(|r| TransactionRateLimit {
+                count: r.count,
+                window: Duration::seconds(r.window_secs),
+            }),
+        };
+
+        let specific = match grant.specific {
+            persistent_grant::Specific::EtherTransfer { targets, limit } => {
+                SpecificGrant::EtherTransfer(ether_transfer::Settings {
+                    target: targets.into_iter().map(Address::from).collect(),
+                    limit: volume(limit),
+                })
+            }
+            persistent_grant::Specific::TokenTransfer {
+                token_contract,
+                receiver,
+                volume_limits,
+            } => SpecificGrant::TokenTransfer(token_transfers::Settings {
+                token_contract: Address::from(token_contract),
+                target: receiver.map(Address::from),
+                volume_limits: volume_limits.into_iter().map(volume).collect(),
+            }),
+        };
+
+        self.operator_create_grant(basic, specific).await?;
+
+        Ok(())
+    }
+
+    async fn sign_one_off_transaction(
+        &mut self,
+        proposal_id: ProposalId,
+        tx: one_off_transaction::Settings,
+    ) -> Result<(), Error> {
+        use alloy::{
+            eips::eip2930::AccessList,
+            primitives::{Bytes, TxKind, U256},
+        };
+
+        let transaction = TxEip1559 {
+            chain_id: tx.chain_id,
+            nonce: tx.nonce,
+            gas_limit: tx.gas_limit,
+            max_fee_per_gas: tx.max_fee_per_gas,
+            max_priority_fee_per_gas: tx.max_priority_fee_per_gas,
+            to: TxKind::Call(Address::from(tx.to)),
+            value: U256::from_be_bytes(tx.value),
+            input: Bytes::from(tx.input),
+            access_list: AccessList::default(),
+        };
+
+        let signature = self
+            .client_sign_transaction(tx.client_id, Address::from(tx.wallet_address), transaction)
+            .await?;
+
+        let mut conn = self.db.get().await.map_err(DatabaseError::from)?;
+        one_off_transaction::store_signature(proposal_id, &signature, &mut conn)
+            .await
+            .map_err(DatabaseError::from)?;
+
+        Ok(())
+    }
+}

server/crates/arbiter-server/src/actors/mod.rs

@@ -1,14 +1,21 @@
 use crate::{
     actors::{
-        bootstrap::Bootstrapper, evm::EvmActor, flow_coordinator::FlowCoordinator,
+        bootstrap::Bootstrapper,
-        operator_registry::OperatorRegistry, proposal_manager::ProposalManager, vault::Vault,
+        evm::EvmActor,
+        flow_coordinator::FlowCoordinator,
+        operator_registry::OperatorRegistry,
+        proposal_manager::{ProposalManager, events::ProposalApproved},
+        vault::Vault,
         vault_coordinator::VaultCoordinator,
     },
     db,
 };
 
 use kameo::actor::{ActorRef, Spawn};
-use kameo_actors::{DeliveryStrategy, message_bus::MessageBus};
+use kameo_actors::{
+    DeliveryStrategy,
+    message_bus::{MessageBus, Register},
+};
 use thiserror::Error;
 
 pub mod bootstrap;
@@ -55,14 +62,18 @@ impl GlobalActors {
             db.clone(),
             key_holder.clone(),
         ));
+        // Approved proposals are executed by whoever owns the kind, not by ProposalManager.
+        for recipient in [
+            evm.clone().recipient::<ProposalApproved>(),
+            vault_coordinator.clone().recipient::<ProposalApproved>(),
+            key_holder.clone().recipient::<ProposalApproved>(),
+        ] {
+            let _ = message_bus.tell(Register(recipient)).await;
+        }
+
         Ok(Self {
             bootstrapper: Bootstrapper::spawn(Bootstrapper::new(&db).await?),
-            proposal_manager: ProposalManager::spawn(ProposalManager::new(
+            proposal_manager: ProposalManager::spawn(ProposalManager::new(db, message_bus.clone())),
-                db,
-                key_holder.clone(),
-                evm.clone(),
-                vault_coordinator.clone(),
-            )),
             vault: key_holder,
             vault_coordinator,
             flow_coordinator: FlowCoordinator::spawn(FlowCoordinator::new(

server/crates/arbiter-server/src/actors/proposal_manager.rs

@@ -1,36 +1,34 @@
 use crate::{
-    actors::{
+    actors::proposal_manager::{
-        evm::EvmActor,
+        events::ProposalApproved,
-        vault::Vault,
+        store::{DieselProposalStore, ProposalStore, Tally},
-        vault_coordinator::{StartRekey, VaultCoordinator},
     },
     crypto::governance,
     db::{
         self,
-        functions::unixepoch,
         models::{
-            NewProposal, NewProposalVote, NewRecoveryProposalVote, NewRecoveryWakeupRequest,
+            NewProposalVote, NewRecoveryProposalVote, OperatorIdentityId, Proposal, ProposalId,
-            OperatorIdentityId, Proposal, ProposalId, ProposalStatus, RecoveryOperatorIdentityId,
+            ProposalStatus, RecoveryOperatorIdentityId, SqliteTimestamp,
-            SqliteTimestamp,
         },
-        proposal::{ProposalKind, ProposalKindTag, one_off_transaction, persistent_grant},
+        proposal::{ProposalKind, ProposalKindTag},
-        schema,
     },
 };
 use chrono::Utc;
-use diesel::{
-    ExpressionMethods as _, QueryDsl,
-    dsl::{exists, select},
-};
-use diesel_async::{AsyncConnection as _, RunQueryDsl};
 use kameo::{Actor, actor::ActorRef, messages};
-use std::collections::HashMap;
+use kameo_actors::message_bus::{MessageBus, Publish};
-use strum::IntoDiscriminant as _;
+use std::sync::Arc;
-use tracing::{error, warn};
+use tracing::warn;
+
+pub mod events;
+pub mod store;
 
 pub const DEFAULT_TTL_SECS: u32 = 7 * 24 * 60 * 60; // 7 days
 pub const MAX_TTL_SECS: u32 = DEFAULT_TTL_SECS;
 
+/// Recovery operators stay asleep for this long after a wake-up is requested, so the other
+/// operators have time to dispute it (§3.6).
+const WAKEUP_DELAY_SECS: i32 = 14 * 24 * 60 * 60; // 14 days
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum VoteOutcome {
     Pending,
@@ -58,8 +56,8 @@ pub enum Error {
     DatabaseConnection(#[from] db::PoolError),
     #[error("Database query error: {0}")]
     DatabaseQuery(#[from] diesel::result::Error),
-    #[error("Execution failed: {0}")]
+    #[error("Proposal manager is unavailable")]
-    ExecutionFailed(String),
+    Unavailable,
     #[error("Recovery operators are sleeping")]
     RecoveryNotActive,
     #[error("Recovery operators may only vote on operator replacement")]
@@ -82,25 +80,21 @@ pub struct ProposalSummary {
 
 #[derive(Actor)]
 pub struct ProposalManager {
-    pub(crate) db: db::DatabasePool,
+    pub(crate) store: Arc<dyn ProposalStore>,
-    pub(crate) vault: ActorRef<Vault>,
+    pub(crate) events: ActorRef<MessageBus>,
-    pub(crate) evm: ActorRef<EvmActor>,
-    pub(crate) vault_coordinator: ActorRef<VaultCoordinator>,
 }
 
 impl ProposalManager {
-    pub const fn new(
+    pub fn new(db: db::DatabasePool, events: ActorRef<MessageBus>) -> Self {
-        db: db::DatabasePool,
+        Self::with_store(Arc::new(DieselProposalStore::new(db)), events)
-        vault: ActorRef<Vault>,
+    }
-        evm: ActorRef<EvmActor>,
+
-        vault_coordinator: ActorRef<VaultCoordinator>,
+    /// Builds the actor over an arbitrary store, so tests can supply a mock.
+    pub(crate) const fn with_store(
+        store: Arc<dyn ProposalStore>,
+        events: ActorRef<MessageBus>,
     ) -> Self {
-        Self {
+        Self { store, events }
-            db,
-            vault,
-            evm,
-            vault_coordinator,
-        }
     }
 }
 
@@ -120,98 +114,18 @@ impl ProposalManager {
         let expires_at =
             SqliteTimestamp::from(Utc::now() + chrono::Duration::seconds(i64::from(ttl)));
 
-        let id: ProposalId = self
+        self.store.create(kind, initiator_id, expires_at).await
-            .db
-            .get()
-            .await?
-            .transaction(async |conn| {
-                let id: ProposalId = diesel::insert_into(schema::proposal::table)
-                    .values(&NewProposal {
-                        kind: kind.discriminant(),
-                        initiator_id,
-                        expires_at,
-                    })
-                    .returning(schema::proposal::id)
-                    .get_result(conn)
-                    .await?;
-                db::proposal::insert_kind(conn, id, &kind).await?;
-                Ok::<_, diesel::result::Error>(id)
-            })
-            .await?;
-
-        Ok(id)
     }
 
     #[message]
     pub async fn query_pending(&mut self, operator_id: OperatorIdentityId) -> Vec<ProposalSummary> {
-        #[expect(
+        self.store
-            clippy::cast_possible_truncation,
+            .pending_for(operator_id)
-            clippy::as_conversions,
-            reason = "fixme! #84; this will break in 2038"
-        )]
-        let now_ts = Utc::now().timestamp() as i32;
-
-        let Ok(mut conn) = self.db.get().await else {
-            warn!("query_pending: failed to acquire DB connection");
-            return vec![];
-        };
-
-        let voted_ids: Vec<ProposalId> = schema::proposal_vote::table
-            .filter(schema::proposal_vote::operator_id.eq(operator_id))
-            .select(schema::proposal_vote::proposal_id)
-            .load(&mut conn)
             .await
-            .unwrap_or_default();
+            .unwrap_or_else(|e| {
-
+                warn!(?e, "query_pending failed");
-        let proposals: Vec<Proposal> = schema::proposal::table
+                vec![]
-            .filter(schema::proposal::status.eq(ProposalStatus::Pending))
-            .filter(schema::proposal::expires_at.gt(now_ts))
-            .filter(diesel::dsl::not(schema::proposal::id.eq_any(&voted_ids)))
-            .load(&mut conn)
-            .await
-            .unwrap_or_default();
-
-        let ids: Vec<ProposalId> = proposals.iter().map(|p| p.id).collect();
-        let tallies: Vec<(ProposalId, bool, i64)> = schema::proposal_vote::table
-            .filter(schema::proposal_vote::proposal_id.eq_any(&ids))
-            .group_by((
-                schema::proposal_vote::proposal_id,
-                schema::proposal_vote::approve,
-            ))
-            .select((
-                schema::proposal_vote::proposal_id,
-                schema::proposal_vote::approve,
-                diesel::dsl::count_star(),
-            ))
-            .load(&mut conn)
-            .await
-            .unwrap_or_default();
-
-        let mut by_proposal: HashMap<ProposalId, (i64, i64)> = HashMap::new();
-        for (proposal_id, approve, count) in tallies {
-            let entry = by_proposal.entry(proposal_id).or_insert((0, 0));
-            if approve {
-                entry.0 += count;
-            } else {
-                entry.1 += count;
-            }
-        }
-
-        proposals
-            .into_iter()
-            .map(|p| {
-                let (approve_count, reject_count) =
-                    by_proposal.get(&p.id).copied().unwrap_or((0, 0));
-                ProposalSummary {
-                    id: p.id,
-                    kind: p.kind,
-                    initiator_id: p.initiator_id,
-                    expires_at: p.expires_at,
-                    approve_count,
-                    reject_count,
-                }
             })
-            .collect()
     }
 
     #[message]
@@ -222,146 +136,35 @@ impl ProposalManager {
         approve: bool,
         signature: Vec<u8>,
     ) -> Result<VoteOutcome, Error> {
-        let mut conn = self.db.get().await?;
+        let proposal = self.store.load(proposal_id).await?;
 
-        // Load proposal — must exist
+        // Checked before the status check so AlreadyVoted takes priority.
-        let proposal: Proposal = schema::proposal::table
+        if self.store.has_voted(proposal_id, operator_id).await? {
-            .find(proposal_id)
-            .first(&mut conn)
-            .await
-            .map_err(|e| match e {
-                diesel::result::Error::NotFound => Error::ProposalNotFound,
-                other => Error::DatabaseQuery(other),
-            })?;
-
-        // Check for duplicate vote before status check so AlreadyVoted takes priority
-        let already_voted: bool = select(exists(
-            schema::proposal_vote::table
-                .filter(schema::proposal_vote::proposal_id.eq(proposal_id))
-                .filter(schema::proposal_vote::operator_id.eq(operator_id)),
-        ))
-        .get_result(&mut conn)
-        .await?;
-        if already_voted {
             return Err(Error::AlreadyVoted);
         }
 
-        if proposal.status != ProposalStatus::Pending {
+        Self::check_votable(&proposal)?;
-            return Err(Error::ProposalNotPending);
-        }
 
-        if proposal.expires_at.0 <= Utc::now() {
+        let public_key = self.store.operator_public_key(operator_id).await?;
-            return Err(Error::ProposalExpired);
+        governance::verify_vote(&public_key, proposal_id, approve, &signature)
-        }
-
-        // Load operator public key from operator_identity
-        let pubkey_bytes: Vec<u8> = schema::operator_identity::table
-            .find(operator_id)
-            .select(schema::operator_identity::public_key)
-            .first(&mut conn)
-            .await
-            .map_err(|e| match e {
-                diesel::result::Error::NotFound => Error::OperatorNotFound,
-                other => Error::DatabaseQuery(other),
-            })?;
-
-        governance::verify_vote(&pubkey_bytes, proposal_id, approve, &signature)
             .map_err(|_| Error::InvalidSignature)?;
 
-        // Insert vote
+        self.store
-        diesel::insert_into(schema::proposal_vote::table)
+            .record_vote(NewProposalVote {
-            .values(&NewProposalVote {
                 proposal_id,
                 operator_id,
                 approve,
                 signature,
             })
-            .execute(&mut conn)
             .await?;
 
-        // Quorum check
+        let mut tally = self.store.tally(proposal_id).await?;
-        let total_operators: i64 = schema::operator_identity::table
+        // §3.5: recovery operators only join the electorate once they are awake.
-            .count()
+        if !self.store.is_recovery_active().await? {
-            .get_result(&mut conn)
+            tally.total_recovery = 0;
-            .await?;
-        let recovery_active = Self::is_recovery_active_conn(&mut conn).await?;
-        let total_recovery: i64 = if recovery_active {
-            schema::recovery_operator_identity::table
-                .count()
-                .get_result(&mut conn)
-                .await?
-        } else {
-            0
-        };
-        #[expect(
-            clippy::cast_possible_truncation,
-            clippy::cast_sign_loss,
-            clippy::as_conversions,
-            reason = "operator count is always a small positive integer"
-        )]
-        let threshold = if proposal.kind.requires_full_quorum() {
-            // §3.3: key-rotation proposals require every eligible voter to approve
-            // §3.5: when recovery is active, recovery operators also vote on replace_operator
-            (total_operators + total_recovery) as usize
-        } else {
-            crate::crypto::shamir::shamir_threshold(total_operators as usize)
-        };
-
-        let ordinary_approve: i64 = schema::proposal_vote::table
-            .filter(schema::proposal_vote::proposal_id.eq(proposal_id))
-            .filter(schema::proposal_vote::approve.eq(true))
-            .count()
-            .get_result(&mut conn)
-            .await?;
-        let recovery_approve: i64 = schema::recovery_proposal_vote::table
-            .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id))
-            .filter(schema::recovery_proposal_vote::approve.eq(true))
-            .count()
-            .get_result(&mut conn)
-            .await?;
-        let approve_count = ordinary_approve + recovery_approve;
-
-        let ordinary_reject: i64 = schema::proposal_vote::table
-            .filter(schema::proposal_vote::proposal_id.eq(proposal_id))
-            .filter(schema::proposal_vote::approve.eq(false))
-            .count()
-            .get_result(&mut conn)
-            .await?;
-        let recovery_reject: i64 = schema::recovery_proposal_vote::table
-            .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id))
-            .filter(schema::recovery_proposal_vote::approve.eq(false))
-            .count()
-            .get_result(&mut conn)
-            .await?;
-        let reject_count = ordinary_reject + recovery_reject;
-
-        #[expect(
-            clippy::cast_possible_wrap,
-            clippy::as_conversions,
-            reason = "threshold is derived from operator count, always fits i64"
-        )]
-        let threshold_i64 = threshold as i64;
-
-        if approve_count >= threshold_i64 {
-            diesel::update(schema::proposal::table.find(proposal_id))
-                .set(schema::proposal::status.eq(ProposalStatus::Approved))
-                .execute(&mut conn)
-                .await?;
-            drop(conn); // release connection before async execution
-            self.execute_proposal(&proposal).await?;
-            return Ok(VoteOutcome::Approved);
         }
 
-        let total_eligible = total_operators + total_recovery;
+        self.settle(&proposal, &tally).await
-        if reject_count > total_eligible - threshold_i64 {
-            diesel::update(schema::proposal::table.find(proposal_id))
-                .set(schema::proposal::status.eq(ProposalStatus::Rejected))
-                .execute(&mut conn)
-                .await?;
-            return Ok(VoteOutcome::Rejected);
-        }
-
-        Ok(VoteOutcome::Pending)
     }
 
     /// §3.6: Any ordinary operator may request recovery wake-up.
@@ -371,17 +174,10 @@ impl ProposalManager {
         &mut self,
         operator_id: OperatorIdentityId,
     ) -> Result<(), Error> {
-        let mut conn = self.db.get().await?;
+        if self.store.has_uncancelled_wakeup().await? {
-        if Self::has_uncancelled_wakeup(&mut conn).await? {
             return Err(Error::WakeupAlreadyPending);
         }
-        diesel::insert_into(schema::recovery_wakeup_request::table)
+        self.store.request_wakeup(operator_id).await
-            .values(&NewRecoveryWakeupRequest {
-                requested_by: operator_id,
-            })
-            .execute(&mut conn)
-            .await?;
-        Ok(())
     }
 
     /// §3.6: Any ordinary operator may cancel a pending wake-up request.
@@ -391,19 +187,11 @@ impl ProposalManager {
         &mut self,
         operator_id: OperatorIdentityId,
     ) -> Result<(), Error> {
-        let mut conn = self.db.get().await?;
+        if self.store.cancel_wakeup(operator_id).await? {
-        let rows_updated = diesel::update(schema::recovery_wakeup_request::table)
+            Ok(())
-            .filter(schema::recovery_wakeup_request::cancelled_at.is_null())
+        } else {
-            .set((
+            Err(Error::NoActiveWakeup)
-                schema::recovery_wakeup_request::cancelled_by.eq(Some(operator_id)),
-                schema::recovery_wakeup_request::cancelled_at.eq(Some(SqliteTimestamp::now())),
-            ))
-            .execute(&mut conn)
-            .await?;
-        if rows_updated == 0 {
-            return Err(Error::NoActiveWakeup);
         }
-        Ok(())
     }
 
     /// §3.5: Recovery operators may only vote on operator replacement proposals.
@@ -416,374 +204,130 @@ impl ProposalManager {
         approve: bool,
         signature: Vec<u8>,
     ) -> Result<VoteOutcome, Error> {
-        let mut conn = self.db.get().await?;
+        let proposal = self.store.load(proposal_id).await?;
-
-        let proposal: Proposal = schema::proposal::table
-            .find(proposal_id)
-            .first(&mut conn)
-            .await
-            .map_err(|e| match e {
-                diesel::result::Error::NotFound => Error::ProposalNotFound,
-                other => Error::DatabaseQuery(other),
-            })?;
 
         if proposal.kind != ProposalKindTag::ReplaceOperator {
             return Err(Error::NotAllowedForRecoveryOperator);
         }
 
-        if !Self::is_recovery_active_conn(&mut conn).await? {
+        if !self.store.is_recovery_active().await? {
             return Err(Error::RecoveryNotActive);
         }
 
-        let already_voted: bool = select(exists(
+        if self
-            schema::recovery_proposal_vote::table
+            .store
-                .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id))
+            .has_recovery_voted(proposal_id, recovery_operator_id)
-                .filter(
+            .await?
-                    schema::recovery_proposal_vote::recovery_operator_id.eq(recovery_operator_id),
+        {
-                ),
-        ))
-        .get_result(&mut conn)
-        .await?;
-        if already_voted {
             return Err(Error::AlreadyVoted);
         }
 
-        if proposal.status != ProposalStatus::Pending {
+        Self::check_votable(&proposal)?;
-            return Err(Error::ProposalNotPending);
-        }
 
-        if proposal.expires_at.0 <= Utc::now() {
+        let public_key = self
-            return Err(Error::ProposalExpired);
+            .store
-        }
+            .recovery_operator_public_key(recovery_operator_id)
-
+            .await?;
-        let pubkey_bytes: Vec<u8> = schema::recovery_operator_identity::table
+        governance::verify_vote(&public_key, proposal_id, approve, &signature)
-            .find(recovery_operator_id)
-            .select(schema::recovery_operator_identity::public_key)
-            .first(&mut conn)
-            .await
-            .map_err(|e| match e {
-                diesel::result::Error::NotFound => Error::OperatorNotFound,
-                other => Error::DatabaseQuery(other),
-            })?;
-
-        governance::verify_vote(&pubkey_bytes, proposal_id, approve, &signature)
             .map_err(|_| Error::InvalidSignature)?;
 
-        diesel::insert_into(schema::recovery_proposal_vote::table)
+        self.store
-            .values(&NewRecoveryProposalVote {
+            .record_recovery_vote(NewRecoveryProposalVote {
                 proposal_id,
                 recovery_operator_id,
                 approve,
                 signature,
             })
-            .execute(&mut conn)
             .await?;
 
-        // Quorum: all ordinary + all recovery operators must approve (§3.3 + §3.5)
+        let tally = self.store.tally(proposal_id).await?;
-        let total_ordinary: i64 = schema::operator_identity::table
+        self.settle(&proposal, &tally).await
-            .count()
-            .get_result(&mut conn)
-            .await?;
-        let total_recovery: i64 = schema::recovery_operator_identity::table
-            .count()
-            .get_result(&mut conn)
-            .await?;
-        let threshold_i64 = total_ordinary + total_recovery;
-
-        let ordinary_approve: i64 = schema::proposal_vote::table
-            .filter(schema::proposal_vote::proposal_id.eq(proposal_id))
-            .filter(schema::proposal_vote::approve.eq(true))
-            .count()
-            .get_result(&mut conn)
-            .await?;
-        let recovery_approve: i64 = schema::recovery_proposal_vote::table
-            .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id))
-            .filter(schema::recovery_proposal_vote::approve.eq(true))
-            .count()
-            .get_result(&mut conn)
-            .await?;
-        let approve_count = ordinary_approve + recovery_approve;
-
-        if approve_count >= threshold_i64 {
-            diesel::update(schema::proposal::table.find(proposal_id))
-                .set(schema::proposal::status.eq(ProposalStatus::Approved))
-                .execute(&mut conn)
-                .await?;
-            drop(conn);
-            self.execute_proposal(&proposal).await?;
-            return Ok(VoteOutcome::Approved);
-        }
-
-        let recovery_reject: i64 = schema::recovery_proposal_vote::table
-            .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id))
-            .filter(schema::recovery_proposal_vote::approve.eq(false))
-            .count()
-            .get_result(&mut conn)
-            .await?;
-        let ordinary_reject: i64 = schema::proposal_vote::table
-            .filter(schema::proposal_vote::proposal_id.eq(proposal_id))
-            .filter(schema::proposal_vote::approve.eq(false))
-            .count()
-            .get_result(&mut conn)
-            .await?;
-        let reject_count = ordinary_reject + recovery_reject;
-
-        if reject_count > threshold_i64 - approve_count - reject_count {
-            diesel::update(schema::proposal::table.find(proposal_id))
-                .set(schema::proposal::status.eq(ProposalStatus::Rejected))
-                .execute(&mut conn)
-                .await?;
-            return Ok(VoteOutcome::Rejected);
-        }
-
-        Ok(VoteOutcome::Pending)
     }
 }
 
 impl ProposalManager {
-    const WAKEUP_DELAY_SECS: i32 = 14 * 24 * 60 * 60;
+    /// A vote only counts while the proposal is still open.
-
+    fn check_votable(proposal: &Proposal) -> Result<(), Error> {
-    /// Returns true when an uncancelled wakeup request has passed the 14-day dispute window.
+        if proposal.status != ProposalStatus::Pending {
-    async fn is_recovery_active_conn(conn: &mut db::DatabaseConnection) -> Result<bool, Error> {
+            return Err(Error::ProposalNotPending);
-        select(exists(
+        }
-            schema::recovery_wakeup_request::table
+        if proposal.expires_at.0 <= Utc::now() {
-                .filter(schema::recovery_wakeup_request::cancelled_at.is_null())
+            return Err(Error::ProposalExpired);
-                .filter(
+        }
-                    schema::recovery_wakeup_request::requested_at
+        Ok(())
-                        .le(unixepoch("now") - Self::WAKEUP_DELAY_SECS),
-                ),
-        ))
-        .get_result(conn)
-        .await
-        .map_err(Error::from)
     }
 
-    /// Returns true when there is any uncancelled wakeup request (pending or active).
+    /// Pure quorum arithmetic — no I/O, so the rules can be tested directly (§3.3).
-    async fn has_uncancelled_wakeup(conn: &mut db::DatabaseConnection) -> Result<bool, Error> {
+    ///
-        select(exists(schema::recovery_wakeup_request::table.filter(
+    /// A proposal is rejected once approval has become unreachable: even if every voter
-            schema::recovery_wakeup_request::cancelled_at.is_null(),
+    /// who has not spoken yet approved, the threshold could not be met.
-        )))
+    #[must_use]
-        .get_result(conn)
+    pub(crate) const fn evaluate_quorum(tally: &Tally, requires_full_quorum: bool) -> VoteOutcome {
-        .await
+        let total_eligible = tally.total_ordinary + tally.total_recovery;
-        .map_err(Error::from)
-    }
 
-    async fn execute_proposal(&self, proposal: &Proposal) -> Result<(), Error> {
+        #[expect(
-        let mut conn = self.db.get().await?;
+            clippy::cast_possible_truncation,
-        let kind = db::proposal::load_kind(&mut conn, proposal.id, proposal.kind).await?;
+            clippy::cast_sign_loss,
-        drop(conn);
+            clippy::cast_possible_wrap,
+            clippy::as_conversions,
+            reason = "operator counts are always small positive integers"
+        )]
+        // §3.3: key-rotation proposals require every eligible voter to approve.
+        // §3.5: when recovery is active, recovery operators are eligible too.
+        let threshold: i64 = if requires_full_quorum {
+            total_eligible
+        } else {
+            crate::crypto::shamir::shamir_threshold(tally.total_ordinary as usize) as i64
+        };
 
-        match kind {
+        if tally.approve >= threshold {
-            ProposalKind::ApproveSdkClient(s) => self.execute_approve_sdk_client(s.client_id).await,
+            VoteOutcome::Approved
-            ProposalKind::GrantWalletAccess(s) => {
+        } else if tally.reject > total_eligible - threshold {
-                self.execute_grant_wallet_access(s.wallet_id, s.client_id)
+            VoteOutcome::Rejected
-                    .await
+        } else {
-            }
+            VoteOutcome::Pending
-            ProposalKind::ReplaceOperator(s) => {
-                self.execute_replace_operator(s.old_operator_id, s.new_pubkey)
-                    .await
-            }
-            ProposalKind::TriggerRekey => self.execute_trigger_rekey().await,
-            ProposalKind::ApprovePersistentGrant(grant) => {
-                self.execute_approve_persistent_grant(*grant).await
-            }
-            ProposalKind::ApproveOneOffTransaction(tx) => {
-                self.execute_approve_one_off_transaction(proposal.id, *tx)
-                    .await
-            }
         }
     }
 
-    async fn execute_grant_wallet_access(
+    /// Applies the quorum rules to a fresh tally and records whatever they decide.
-        &self,
+    async fn settle(&self, proposal: &Proposal, tally: &Tally) -> Result<VoteOutcome, Error> {
-        wallet_id: i32,
+        let outcome = Self::evaluate_quorum(tally, proposal.kind.requires_full_quorum());
-        client_id: i32,
-    ) -> Result<(), Error> {
-        use crate::db::models::EvmWalletId;
 
-        let mut conn = self.db.get().await.map_err(Error::DatabaseConnection)?;
+        match outcome {
-
+            VoteOutcome::Approved => self.announce_approval(proposal).await?,
-        diesel::insert_into(schema::evm_wallet_access::table)
+            VoteOutcome::Rejected => {
-            .values((
+                self.store
-                schema::evm_wallet_access::wallet_id.eq(EvmWalletId::from_raw(wallet_id)),
+                    .set_status(proposal.id, ProposalStatus::Rejected)
-                schema::evm_wallet_access::client_id.eq(client_id),
+                    .await?;
-            ))
-            .execute(&mut conn)
-            .await
-            .map_err(|e| Error::ExecutionFailed(format!("grant wallet access: {e}")))?;
-
-        Ok(())
-    }
-
-    /// Updates the old operator's public key in-place (preserving their DB id and history),
-    /// removes their old Shamir share, then begins a coordinated re-key (§3.3).
-    async fn execute_replace_operator(
-        &self,
-        old_operator_id: OperatorIdentityId,
-        new_pubkey: Vec<u8>,
-    ) -> Result<(), Error> {
-        let mut conn = self.db.get().await.map_err(Error::DatabaseConnection)?;
-
-        diesel::update(schema::operator_identity::table)
-            .filter(schema::operator_identity::id.eq(old_operator_id))
-            .set(schema::operator_identity::public_key.eq(&new_pubkey))
-            .execute(&mut conn)
-            .await
-            .map_err(|e| Error::ExecutionFailed(format!("update operator pubkey: {e}")))?;
-
-        // Remove the old Shamir share; finalize_rekey will store a fresh one.
-        diesel::delete(schema::operator::table)
-            .filter(schema::operator::id.eq(Some(old_operator_id)))
-            .execute(&mut conn)
-            .await
-            .map_err(|e| Error::ExecutionFailed(format!("remove old operator share: {e}")))?;
-
-        drop(conn);
-
-        self.vault_coordinator
-            .ask(StartRekey {})
-            .await
-            .map_err(|e| Error::ExecutionFailed(format!("start rekey: {e}")))?;
-
-        Ok(())
-    }
-
-    /// Triggers a Shamir re-key with the current operator set (§3.3).
-    async fn execute_trigger_rekey(&self) -> Result<(), Error> {
-        self.vault_coordinator
-            .ask(StartRekey {})
-            .await
-            .map_err(|e| Error::ExecutionFailed(format!("start rekey: {e}")))?;
-        Ok(())
-    }
-
-    async fn execute_approve_one_off_transaction(
-        &self,
-        proposal_id: ProposalId,
-        tx: one_off_transaction::Settings,
-    ) -> Result<(), Error> {
-        use crate::actors::evm::ClientSignTransaction;
-        use alloy::{
-            consensus::TxEip1559,
-            eips::eip2930::AccessList,
-            primitives::{Address, Bytes, TxKind, U256},
-        };
-
-        let transaction = TxEip1559 {
-            chain_id: tx.chain_id,
-            nonce: tx.nonce,
-            gas_limit: tx.gas_limit,
-            max_fee_per_gas: tx.max_fee_per_gas,
-            max_priority_fee_per_gas: tx.max_priority_fee_per_gas,
-            to: TxKind::Call(Address::from(tx.to)),
-            value: U256::from_be_bytes(tx.value),
-            input: Bytes::from(tx.input),
-            access_list: AccessList::default(),
-        };
-
-        let sig = self
-            .evm
-            .ask(ClientSignTransaction {
-                client_id: tx.client_id,
-                wallet_address: Address::from(tx.wallet_address),
-                transaction,
-            })
-            .await
-            .map_err(|e| Error::ExecutionFailed(format!("sign one-off tx: {e}")))?;
-
-        let mut conn = self.db.get().await.map_err(Error::DatabaseConnection)?;
-        one_off_transaction::store_signature(proposal_id, &sig, &mut conn)
-            .await
-            .map_err(|e| Error::ExecutionFailed(format!("store proposal result: {e}")))?;
-
-        Ok(())
-    }
-
-    async fn execute_approve_persistent_grant(
-        &self,
-        grant: persistent_grant::Settings,
-    ) -> Result<(), Error> {
-        use crate::{
-            actors::evm::OperatorCreateGrant,
-            evm::policies::{
-                SharedGrantSettings, SpecificGrant, TransactionRateLimit, VolumeRateLimit,
-                ether_transfer, token_transfers,
-            },
-        };
-        use alloy::primitives::{Address, U256};
-        use chrono::Duration;
-
-        let volume = |limit: persistent_grant::VolumeLimit| VolumeRateLimit {
-            max_volume: U256::from_be_bytes(limit.max_volume),
-            window: Duration::seconds(limit.window_secs),
-        };
-
-        let basic = SharedGrantSettings {
-            wallet_access_id: grant.wallet_access_id,
-            chain: grant.chain_id,
-            valid_from: grant
-                .valid_from_secs
-                .and_then(|s| chrono::DateTime::from_timestamp(s, 0)),
-            valid_until: grant
-                .valid_until_secs
-                .and_then(|s| chrono::DateTime::from_timestamp(s, 0)),
-            max_gas_fee_per_gas: grant.max_gas_fee_per_gas.map(U256::from_be_bytes),
-            max_priority_fee_per_gas: grant.max_priority_fee_per_gas.map(U256::from_be_bytes),
-            rate_limit: grant.rate_limit.map(|r| TransactionRateLimit {
-                count: r.count,
-                window: Duration::seconds(r.window_secs),
-            }),
-        };
-
-        let grant = match grant.specific {
-            persistent_grant::Specific::EtherTransfer { targets, limit } => {
-                SpecificGrant::EtherTransfer(ether_transfer::Settings {
-                    target: targets.into_iter().map(Address::from).collect(),
-                    limit: volume(limit),
-                })
             }
-            persistent_grant::Specific::TokenTransfer {
+            VoteOutcome::Pending => {}
-                token_contract,
+        }
-                receiver,
-                volume_limits,
-            } => SpecificGrant::TokenTransfer(token_transfers::Settings {
-                token_contract: Address::from(token_contract),
-                target: receiver.map(Address::from),
-                volume_limits: volume_limits.into_iter().map(volume).collect(),
-            }),
-        };
 
-        self.evm
+        Ok(outcome)
-            .ask(OperatorCreateGrant { basic, grant })
-            .await
-            .map_err(|e| Error::ExecutionFailed(format!("create grant: {e}")))?;
-
-        Ok(())
     }
 
-    async fn execute_approve_sdk_client(&self, client_id: i32) -> Result<(), Error> {
+    /// Marks the proposal approved and hands the outcome to whoever owns that kind.
-        use crate::{crypto::integrity, peers::client::ClientCredentials};
+    ///
-        use arbiter_crypto::authn;
+    /// The outcome is published, not executed: this actor coordinates voting and nothing
+    /// else. Executors subscribe on the bus, so a vote is answered once the quorum is
+    /// recorded rather than once the effect has landed.
+    async fn announce_approval(&self, proposal: &Proposal) -> Result<(), Error> {
+        self.store
+            .set_status(proposal.id, ProposalStatus::Approved)
+            .await?;
 
-        let mut conn = self.db.get().await.map_err(Error::DatabaseConnection)?;
+        let kind = self.store.load_kind(proposal.id, proposal.kind).await?;
+        let _ = self
+            .events
+            .tell(Publish(ProposalApproved {
+                id: proposal.id,
+                kind,
+            }))
+            .await;
 
-        let pubkey_bytes: Vec<u8> = schema::program_client::table
+        Ok(())
-            .find(client_id)
-            .select(schema::program_client::public_key)
-            .first(&mut conn)
-            .await
-            .map_err(|e| Error::ExecutionFailed(format!("client not found: {e}")))?;
-
-        let pubkey = authn::PublicKey::try_from(pubkey_bytes.as_slice())
-            .map_err(|()| Error::ExecutionFailed("invalid client public key".to_owned()))?;
-
-        let creds = ClientCredentials { pubkey };
-
-        integrity::sign_entity(&mut conn, &self.vault, &creds, client_id)
-            .await
-            .map_err(|e| {
-                error!(?e, "Failed to sign integrity envelope for client");
-                Error::ExecutionFailed(e.to_string())
-            })
     }
 }
+
+#[cfg(test)]
+mod tests;

server/crates/arbiter-server/src/actors/proposal_manager/events.rs

@@ -0,0 +1,11 @@
+use crate::db::{models::ProposalId, proposal::ProposalKind};
+
+/// Published once a proposal reaches its approval threshold.
+///
+/// Executors subscribe on the global `MessageBus` and act on the kinds they own;
+/// `ProposalManager` does not know who acts on an outcome, or whether anyone does.
+#[derive(Debug, Clone)]
+pub struct ProposalApproved {
+    pub id: ProposalId,
+    pub kind: ProposalKind,
+}

server/crates/arbiter-server/src/actors/proposal_manager/store.rs

@@ -0,0 +1,409 @@
+//! Database access for [`super::ProposalManager`], behind a trait.
+//!
+//! The actor holds a `dyn ProposalStore` rather than a connection pool, so the quorum
+//! rules can be exercised against a mock instead of a live SQLite file.
+
+use super::{Error, ProposalSummary, WAKEUP_DELAY_SECS};
+use crate::db::{
+    self,
+    functions::unixepoch,
+    models::{
+        NewProposal, NewProposalVote, NewRecoveryProposalVote, NewRecoveryWakeupRequest,
+        OperatorIdentityId, Proposal, ProposalId, ProposalStatus, RecoveryOperatorIdentityId,
+        SqliteTimestamp,
+    },
+    proposal::{ProposalKind, ProposalKindTag},
+    schema,
+};
+
+use async_trait::async_trait;
+use chrono::Utc;
+use diesel::{
+    ExpressionMethods as _, QueryDsl,
+    dsl::{exists, select},
+};
+use diesel_async::{AsyncConnection as _, RunQueryDsl};
+use std::collections::HashMap;
+use strum::IntoDiscriminant as _;
+
+/// Everything the quorum rules need to know about one proposal's votes.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Tally {
+    pub approve: i64,
+    pub reject: i64,
+    pub total_ordinary: i64,
+    pub total_recovery: i64,
+}
+
+#[cfg_attr(test, mockall::automock)]
+#[async_trait]
+pub trait ProposalStore: Send + Sync + 'static {
+    /// Writes the proposal and its kind-specific rows in one transaction.
+    async fn create(
+        &self,
+        kind: ProposalKind,
+        initiator_id: OperatorIdentityId,
+        expires_at: SqliteTimestamp,
+    ) -> Result<ProposalId, Error>;
+
+    async fn load(&self, id: ProposalId) -> Result<Proposal, Error>;
+
+    async fn load_kind(&self, id: ProposalId, tag: ProposalKindTag) -> Result<ProposalKind, Error>;
+
+    async fn has_voted(
+        &self,
+        id: ProposalId,
+        operator_id: OperatorIdentityId,
+    ) -> Result<bool, Error>;
+
+    async fn has_recovery_voted(
+        &self,
+        id: ProposalId,
+        recovery_operator_id: RecoveryOperatorIdentityId,
+    ) -> Result<bool, Error>;
+
+    async fn operator_public_key(&self, id: OperatorIdentityId) -> Result<Vec<u8>, Error>;
+
+    async fn recovery_operator_public_key(
+        &self,
+        id: RecoveryOperatorIdentityId,
+    ) -> Result<Vec<u8>, Error>;
+
+    async fn record_vote(&self, vote: NewProposalVote) -> Result<(), Error>;
+
+    async fn record_recovery_vote(&self, vote: NewRecoveryProposalVote) -> Result<(), Error>;
+
+    /// Vote counts for one proposal, alongside the size of each electorate.
+    async fn tally(&self, id: ProposalId) -> Result<Tally, Error>;
+
+    async fn set_status(&self, id: ProposalId, status: ProposalStatus) -> Result<(), Error>;
+
+    /// Pending, unexpired proposals this operator has not voted on yet.
+    async fn pending_for(
+        &self,
+        operator_id: OperatorIdentityId,
+    ) -> Result<Vec<ProposalSummary>, Error>;
+
+    /// True once an uncancelled wake-up request has outlived the dispute window.
+    async fn is_recovery_active(&self) -> Result<bool, Error>;
+
+    /// True while any wake-up request stands, whether or not the window has elapsed.
+    async fn has_uncancelled_wakeup(&self) -> Result<bool, Error>;
+
+    async fn request_wakeup(&self, operator_id: OperatorIdentityId) -> Result<(), Error>;
+
+    /// Returns false when there was no uncancelled request to cancel.
+    async fn cancel_wakeup(&self, operator_id: OperatorIdentityId) -> Result<bool, Error>;
+}
+
+pub struct DieselProposalStore {
+    db: db::DatabasePool,
+}
+
+impl DieselProposalStore {
+    pub const fn new(db: db::DatabasePool) -> Self {
+        Self { db }
+    }
+}
+
+/// `NotFound` means the row is absent, which every caller reports as its own error.
+fn missing(absent: Error) -> impl FnOnce(diesel::result::Error) -> Error {
+    move |e| match e {
+        diesel::result::Error::NotFound => absent,
+        other => Error::DatabaseQuery(other),
+    }
+}
+
+#[async_trait]
+impl ProposalStore for DieselProposalStore {
+    async fn create(
+        &self,
+        kind: ProposalKind,
+        initiator_id: OperatorIdentityId,
+        expires_at: SqliteTimestamp,
+    ) -> Result<ProposalId, Error> {
+        let id = self
+            .db
+            .get()
+            .await?
+            .transaction(async |conn| {
+                let id: ProposalId = diesel::insert_into(schema::proposal::table)
+                    .values(&NewProposal {
+                        kind: kind.discriminant(),
+                        initiator_id,
+                        expires_at,
+                    })
+                    .returning(schema::proposal::id)
+                    .get_result(conn)
+                    .await?;
+                db::proposal::insert_kind(conn, id, &kind).await?;
+                Ok::<_, diesel::result::Error>(id)
+            })
+            .await?;
+
+        Ok(id)
+    }
+
+    async fn load(&self, id: ProposalId) -> Result<Proposal, Error> {
+        let mut conn = self.db.get().await?;
+        schema::proposal::table
+            .find(id)
+            .first(&mut conn)
+            .await
+            .map_err(missing(Error::ProposalNotFound))
+    }
+
+    async fn load_kind(&self, id: ProposalId, tag: ProposalKindTag) -> Result<ProposalKind, Error> {
+        let mut conn = self.db.get().await?;
+        db::proposal::load_kind(&mut conn, id, tag)
+            .await
+            .map_err(Error::from)
+    }
+
+    async fn has_voted(
+        &self,
+        id: ProposalId,
+        operator_id: OperatorIdentityId,
+    ) -> Result<bool, Error> {
+        let mut conn = self.db.get().await?;
+        select(exists(
+            schema::proposal_vote::table
+                .filter(schema::proposal_vote::proposal_id.eq(id))
+                .filter(schema::proposal_vote::operator_id.eq(operator_id)),
+        ))
+        .get_result(&mut conn)
+        .await
+        .map_err(Error::from)
+    }
+
+    async fn has_recovery_voted(
+        &self,
+        id: ProposalId,
+        recovery_operator_id: RecoveryOperatorIdentityId,
+    ) -> Result<bool, Error> {
+        let mut conn = self.db.get().await?;
+        select(exists(
+            schema::recovery_proposal_vote::table
+                .filter(schema::recovery_proposal_vote::proposal_id.eq(id))
+                .filter(
+                    schema::recovery_proposal_vote::recovery_operator_id.eq(recovery_operator_id),
+                ),
+        ))
+        .get_result(&mut conn)
+        .await
+        .map_err(Error::from)
+    }
+
+    async fn operator_public_key(&self, id: OperatorIdentityId) -> Result<Vec<u8>, Error> {
+        let mut conn = self.db.get().await?;
+        schema::operator_identity::table
+            .find(id)
+            .select(schema::operator_identity::public_key)
+            .first(&mut conn)
+            .await
+            .map_err(missing(Error::OperatorNotFound))
+    }
+
+    async fn recovery_operator_public_key(
+        &self,
+        id: RecoveryOperatorIdentityId,
+    ) -> Result<Vec<u8>, Error> {
+        let mut conn = self.db.get().await?;
+        schema::recovery_operator_identity::table
+            .find(id)
+            .select(schema::recovery_operator_identity::public_key)
+            .first(&mut conn)
+            .await
+            .map_err(missing(Error::OperatorNotFound))
+    }
+
+    async fn record_vote(&self, vote: NewProposalVote) -> Result<(), Error> {
+        let mut conn = self.db.get().await?;
+        diesel::insert_into(schema::proposal_vote::table)
+            .values(&vote)
+            .execute(&mut conn)
+            .await?;
+        Ok(())
+    }
+
+    async fn record_recovery_vote(&self, vote: NewRecoveryProposalVote) -> Result<(), Error> {
+        let mut conn = self.db.get().await?;
+        diesel::insert_into(schema::recovery_proposal_vote::table)
+            .values(&vote)
+            .execute(&mut conn)
+            .await?;
+        Ok(())
+    }
+
+    async fn tally(&self, id: ProposalId) -> Result<Tally, Error> {
+        let mut conn = self.db.get().await?;
+
+        let ordinary_approve: i64 = schema::proposal_vote::table
+            .filter(schema::proposal_vote::proposal_id.eq(id))
+            .filter(schema::proposal_vote::approve.eq(true))
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        let recovery_approve: i64 = schema::recovery_proposal_vote::table
+            .filter(schema::recovery_proposal_vote::proposal_id.eq(id))
+            .filter(schema::recovery_proposal_vote::approve.eq(true))
+            .count()
+            .get_result(&mut conn)
+            .await?;
+
+        let ordinary_reject: i64 = schema::proposal_vote::table
+            .filter(schema::proposal_vote::proposal_id.eq(id))
+            .filter(schema::proposal_vote::approve.eq(false))
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        let recovery_reject: i64 = schema::recovery_proposal_vote::table
+            .filter(schema::recovery_proposal_vote::proposal_id.eq(id))
+            .filter(schema::recovery_proposal_vote::approve.eq(false))
+            .count()
+            .get_result(&mut conn)
+            .await?;
+
+        let total_ordinary: i64 = schema::operator_identity::table
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        let total_recovery: i64 = schema::recovery_operator_identity::table
+            .count()
+            .get_result(&mut conn)
+            .await?;
+
+        Ok(Tally {
+            approve: ordinary_approve + recovery_approve,
+            reject: ordinary_reject + recovery_reject,
+            total_ordinary,
+            total_recovery,
+        })
+    }
+
+    async fn set_status(&self, id: ProposalId, status: ProposalStatus) -> Result<(), Error> {
+        let mut conn = self.db.get().await?;
+        diesel::update(schema::proposal::table.find(id))
+            .set(schema::proposal::status.eq(status))
+            .execute(&mut conn)
+            .await?;
+        Ok(())
+    }
+
+    async fn pending_for(
+        &self,
+        operator_id: OperatorIdentityId,
+    ) -> Result<Vec<ProposalSummary>, Error> {
+        #[expect(
+            clippy::cast_possible_truncation,
+            clippy::as_conversions,
+            reason = "fixme! #84; this will break in 2038"
+        )]
+        let now_ts = Utc::now().timestamp() as i32;
+
+        let mut conn = self.db.get().await?;
+
+        let voted_ids: Vec<ProposalId> = schema::proposal_vote::table
+            .filter(schema::proposal_vote::operator_id.eq(operator_id))
+            .select(schema::proposal_vote::proposal_id)
+            .load(&mut conn)
+            .await?;
+
+        let proposals: Vec<Proposal> = schema::proposal::table
+            .filter(schema::proposal::status.eq(ProposalStatus::Pending))
+            .filter(schema::proposal::expires_at.gt(now_ts))
+            .filter(diesel::dsl::not(schema::proposal::id.eq_any(&voted_ids)))
+            .load(&mut conn)
+            .await?;
+
+        let ids: Vec<ProposalId> = proposals.iter().map(|p| p.id).collect();
+        let tallies: Vec<(ProposalId, bool, i64)> = schema::proposal_vote::table
+            .filter(schema::proposal_vote::proposal_id.eq_any(&ids))
+            .group_by((
+                schema::proposal_vote::proposal_id,
+                schema::proposal_vote::approve,
+            ))
+            .select((
+                schema::proposal_vote::proposal_id,
+                schema::proposal_vote::approve,
+                diesel::dsl::count_star(),
+            ))
+            .load(&mut conn)
+            .await?;
+
+        let mut by_proposal: HashMap<ProposalId, (i64, i64)> = HashMap::new();
+        for (proposal_id, approve, count) in tallies {
+            let entry = by_proposal.entry(proposal_id).or_insert((0, 0));
+            if approve {
+                entry.0 += count;
+            } else {
+                entry.1 += count;
+            }
+        }
+
+        Ok(proposals
+            .into_iter()
+            .map(|p| {
+                let (approve_count, reject_count) =
+                    by_proposal.get(&p.id).copied().unwrap_or((0, 0));
+                ProposalSummary {
+                    id: p.id,
+                    kind: p.kind,
+                    initiator_id: p.initiator_id,
+                    expires_at: p.expires_at,
+                    approve_count,
+                    reject_count,
+                }
+            })
+            .collect())
+    }
+
+    async fn is_recovery_active(&self) -> Result<bool, Error> {
+        let mut conn = self.db.get().await?;
+        select(exists(
+            schema::recovery_wakeup_request::table
+                .filter(schema::recovery_wakeup_request::cancelled_at.is_null())
+                .filter(
+                    schema::recovery_wakeup_request::requested_at
+                        .le(unixepoch("now") - WAKEUP_DELAY_SECS),
+                ),
+        ))
+        .get_result(&mut conn)
+        .await
+        .map_err(Error::from)
+    }
+
+    async fn has_uncancelled_wakeup(&self) -> Result<bool, Error> {
+        let mut conn = self.db.get().await?;
+        select(exists(schema::recovery_wakeup_request::table.filter(
+            schema::recovery_wakeup_request::cancelled_at.is_null(),
+        )))
+        .get_result(&mut conn)
+        .await
+        .map_err(Error::from)
+    }
+
+    async fn request_wakeup(&self, operator_id: OperatorIdentityId) -> Result<(), Error> {
+        let mut conn = self.db.get().await?;
+        diesel::insert_into(schema::recovery_wakeup_request::table)
+            .values(&NewRecoveryWakeupRequest {
+                requested_by: operator_id,
+            })
+            .execute(&mut conn)
+            .await?;
+        Ok(())
+    }
+
+    async fn cancel_wakeup(&self, operator_id: OperatorIdentityId) -> Result<bool, Error> {
+        let mut conn = self.db.get().await?;
+        let rows = diesel::update(schema::recovery_wakeup_request::table)
+            .filter(schema::recovery_wakeup_request::cancelled_at.is_null())
+            .set((
+                schema::recovery_wakeup_request::cancelled_by.eq(Some(operator_id)),
+                schema::recovery_wakeup_request::cancelled_at.eq(Some(SqliteTimestamp::now())),
+            ))
+            .execute(&mut conn)
+            .await?;
+        Ok(rows > 0)
+    }
+}

server/crates/arbiter-server/src/actors/proposal_manager/tests.rs

@@ -0,0 +1,222 @@
+//! The quorum rules, exercised without a database.
+//!
+//! These assertions are the point of [`super::store::ProposalStore`]: until the actor took
+//! its data through a trait, checking that two of three operators carry an ordinary
+//! proposal meant opening SQLite and registering operators first.
+
+use super::{
+    ProposalManager, VoteOutcome,
+    store::{MockProposalStore, Tally},
+};
+use crate::{
+    actors::GlobalActors,
+    crypto::governance::vote_message,
+    db::{
+        models::{OperatorIdentityId, Proposal, ProposalId, ProposalStatus, SqliteTimestamp},
+        proposal::ProposalKindTag,
+    },
+};
+use arbiter_crypto::authn::{SigningContext, SigningKey};
+use chrono::{Duration, Utc};
+use std::sync::Arc;
+
+const fn tally(approve: i64, reject: i64, ordinary: i64, recovery: i64) -> Tally {
+    Tally {
+        approve,
+        reject,
+        total_ordinary: ordinary,
+        total_recovery: recovery,
+    }
+}
+
+#[test]
+fn simple_majority_approves_at_two_of_three() {
+    assert_eq!(
+        ProposalManager::evaluate_quorum(&tally(2, 0, 3, 0), false),
+        VoteOutcome::Approved
+    );
+}
+
+#[test]
+fn one_of_three_is_not_yet_a_majority() {
+    assert_eq!(
+        ProposalManager::evaluate_quorum(&tally(1, 0, 3, 0), false),
+        VoteOutcome::Pending
+    );
+}
+
+#[test]
+fn full_quorum_kind_needs_every_voter() {
+    assert_eq!(
+        ProposalManager::evaluate_quorum(&tally(2, 0, 3, 0), true),
+        VoteOutcome::Pending,
+        "two of three must not carry a key-rotation proposal"
+    );
+}
+
+#[test]
+fn recovery_voters_count_towards_full_quorum() {
+    assert_eq!(
+        ProposalManager::evaluate_quorum(&tally(3, 0, 2, 1), true),
+        VoteOutcome::Approved
+    );
+    assert_eq!(
+        ProposalManager::evaluate_quorum(&tally(2, 0, 2, 1), true),
+        VoteOutcome::Pending,
+        "the sleeping recovery operator still owes a vote"
+    );
+}
+
+#[test]
+fn rejection_is_decided_once_approval_is_unreachable() {
+    // Threshold is 2 of 3, so two rejections leave at most one approval available.
+    assert_eq!(
+        ProposalManager::evaluate_quorum(&tally(0, 2, 3, 0), false),
+        VoteOutcome::Rejected
+    );
+    assert_eq!(
+        ProposalManager::evaluate_quorum(&tally(0, 1, 3, 0), false),
+        VoteOutcome::Pending,
+        "one rejection still leaves two approvals reachable"
+    );
+}
+
+#[test]
+fn a_single_rejection_sinks_a_full_quorum_proposal() {
+    assert_eq!(
+        ProposalManager::evaluate_quorum(&tally(2, 1, 3, 0), true),
+        VoteOutcome::Rejected
+    );
+}
+
+fn pending_proposal(id: ProposalId, kind: ProposalKindTag) -> Proposal {
+    let now = Utc::now();
+    Proposal {
+        id,
+        kind,
+        initiator_id: OperatorIdentityId::from_raw(1),
+        created_at: SqliteTimestamp::from(now),
+        expires_at: SqliteTimestamp::from(now + Duration::days(1)),
+        status: ProposalStatus::Pending,
+    }
+}
+
+/// The mock earns its keep here: reaching quorum must flip the stored status to
+/// `Approved` exactly once. Signature verification stays real -- only the database is
+/// stubbed out.
+#[tokio::test]
+async fn reaching_quorum_marks_the_proposal_approved() {
+    let id = ProposalId::from_raw(1);
+    let voter = OperatorIdentityId::from_raw(1);
+    let key = SigningKey::generate();
+    let signature = key
+        .sign_message(&vote_message(id, true), SigningContext::GovernanceVote)
+        .expect("signing a vote must succeed");
+    let public_key = key.public_key().to_bytes();
+
+    let mut store = MockProposalStore::new();
+    store
+        .expect_load()
+        .returning(move |id| Ok(pending_proposal(id, ProposalKindTag::TriggerRekey)));
+    store.expect_has_voted().returning(|_, _| Ok(false));
+    store
+        .expect_operator_public_key()
+        .returning(move |_| Ok(public_key.clone()));
+    store.expect_record_vote().returning(|_| Ok(()));
+    store.expect_is_recovery_active().returning(|| Ok(false));
+    store.expect_tally().returning(|_| Ok(tally(1, 0, 1, 0)));
+    store
+        .expect_set_status()
+        .withf(move |got, status| *got == id && *status == ProposalStatus::Approved)
+        .times(1)
+        .returning(|_, _| Ok(()));
+    store
+        .expect_load_kind()
+        .returning(|_, _| Ok(crate::db::proposal::ProposalKind::TriggerRekey));
+
+    let mut manager =
+        ProposalManager::with_store(Arc::new(store), GlobalActors::spawn_message_bus());
+
+    let outcome = manager
+        .cast_vote(id, voter, true, signature.to_bytes())
+        .await
+        .expect("a valid vote must be accepted");
+
+    assert_eq!(outcome, VoteOutcome::Approved);
+}
+
+/// A vote that does not reach the threshold must leave the stored status alone.
+#[tokio::test]
+async fn a_vote_short_of_quorum_does_not_touch_the_status() {
+    let id = ProposalId::from_raw(7);
+    let voter = OperatorIdentityId::from_raw(2);
+    let key = SigningKey::generate();
+    let signature = key
+        .sign_message(&vote_message(id, true), SigningContext::GovernanceVote)
+        .expect("signing a vote must succeed");
+    let public_key = key.public_key().to_bytes();
+
+    let mut store = MockProposalStore::new();
+    store
+        .expect_load()
+        .returning(move |id| Ok(pending_proposal(id, ProposalKindTag::ApproveSdkClient)));
+    store.expect_has_voted().returning(|_, _| Ok(false));
+    store
+        .expect_operator_public_key()
+        .returning(move |_| Ok(public_key.clone()));
+    store.expect_record_vote().returning(|_| Ok(()));
+    store.expect_is_recovery_active().returning(|| Ok(false));
+    store.expect_tally().returning(|_| Ok(tally(1, 0, 3, 0)));
+    store.expect_set_status().never();
+
+    let mut manager =
+        ProposalManager::with_store(Arc::new(store), GlobalActors::spawn_message_bus());
+
+    let outcome = manager
+        .cast_vote(id, voter, true, signature.to_bytes())
+        .await
+        .expect("a valid vote must be accepted");
+
+    assert_eq!(outcome, VoteOutcome::Pending);
+}
+
+/// A sleeping recovery electorate must not raise the bar for an ordinary proposal.
+#[tokio::test]
+async fn sleeping_recovery_operators_do_not_count_towards_quorum() {
+    let id = ProposalId::from_raw(9);
+    let voter = OperatorIdentityId::from_raw(3);
+    let key = SigningKey::generate();
+    let signature = key
+        .sign_message(&vote_message(id, true), SigningContext::GovernanceVote)
+        .expect("signing a vote must succeed");
+    let public_key = key.public_key().to_bytes();
+
+    let mut store = MockProposalStore::new();
+    store
+        .expect_load()
+        .returning(move |id| Ok(pending_proposal(id, ProposalKindTag::ApproveSdkClient)));
+    store.expect_has_voted().returning(|_, _| Ok(false));
+    store
+        .expect_operator_public_key()
+        .returning(move |_| Ok(public_key.clone()));
+    store.expect_record_vote().returning(|_| Ok(()));
+    store.expect_is_recovery_active().returning(|| Ok(false));
+    // Two recovery operators exist but are asleep, so the threshold stays at 1 of 1.
+    store.expect_tally().returning(|_| Ok(tally(1, 0, 1, 2)));
+    store.expect_set_status().times(1).returning(|_, _| Ok(()));
+    store.expect_load_kind().returning(|_, _| {
+        Ok(crate::db::proposal::ProposalKind::ApproveSdkClient(
+            crate::db::proposal::approve_sdk_client::Settings { client_id: 1 },
+        ))
+    });
+
+    let mut manager =
+        ProposalManager::with_store(Arc::new(store), GlobalActors::spawn_message_bus());
+
+    let outcome = manager
+        .cast_vote(id, voter, true, signature.to_bytes())
+        .await
+        .expect("a valid vote must be accepted");
+
+    assert_eq!(outcome, VoteOutcome::Approved);
+}

server/crates/arbiter-server/src/actors/vault/mod.rs

@@ -1,12 +1,14 @@
 use crate::{
+    actors::proposal_manager::events::ProposalApproved,
     crypto::{
         KeyCell,
         encryption::v1::{self, Nonce},
-        integrity::v1::HmacSha256,
+        integrity::{self, v1::HmacSha256},
     },
     db::{
         self,
         models::{self, RootKeyHistory, RootKeyHistoryId},
+        proposal::ProposalKind,
         schema,
     },
 };
@@ -19,7 +21,7 @@ use diesel::{
 };
 use diesel_async::{AsyncConnection, RunQueryDsl};
 use hmac::{KeyInit as _, Mac as _};
-use kameo::{Actor, Reply, actor::ActorRef, messages};
+use kameo::{Actor, Reply, actor::ActorRef, messages, prelude::Message};
 use kameo_actors::message_bus::{MessageBus, Publish};
 use strum::{EnumDiscriminants, IntoDiscriminant};
 use tracing::{error, info};
@@ -461,6 +463,62 @@ impl Vault {
     }
 }
 
+impl Message<ProposalApproved> for Vault {
+    type Reply = ();
+
+    /// Every subscriber sees every approval and acts only on the kinds it owns.
+    async fn handle(
+        &mut self,
+        msg: ProposalApproved,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        let ProposalKind::ApproveSdkClient(settings) = msg.kind else {
+            return;
+        };
+
+        if let Err(error) = self.approve_sdk_client(settings.client_id).await {
+            error!(
+                ?error,
+                proposal_id = msg.id.to_raw(),
+                "Failed to execute an approved proposal"
+            );
+        }
+    }
+}
+
+impl Vault {
+    /// Attests an approved SDK client with the root key.
+    ///
+    /// Builds the envelope from its parts rather than calling `integrity::sign_entity`,
+    /// which would have this actor ask itself for a signature and deadlock.
+    async fn approve_sdk_client(&mut self, client_id: i32) -> Result<(), Error> {
+        use crate::peers::client::ClientCredentials;
+        use arbiter_crypto::authn;
+
+        // Cloned so the connection does not hold a borrow of `self` across `sign_integrity`.
+        let db = self.db.clone();
+        let mut conn = db.get().await?;
+
+        let pubkey_bytes: Vec<u8> = schema::program_client::table
+            .find(client_id)
+            .select(schema::program_client::public_key)
+            .first(&mut conn)
+            .await?;
+
+        let pubkey =
+            authn::PublicKey::try_from(pubkey_bytes.as_slice()).map_err(|()| Error::InvalidKey)?;
+        let credentials = ClientCredentials { pubkey };
+
+        let (entity_id, mac_input) = integrity::envelope_input(&credentials, client_id);
+        let (key_version, mac) = self.sign_integrity(mac_input)?;
+
+        integrity::store_envelope::<ClientCredentials>(&mut conn, entity_id, key_version, mac)
+            .await?;
+
+        Ok(())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::actors::GlobalActors;

server/crates/arbiter-server/src/actors/vault_coordinator/mod.rs

@@ -3,14 +3,21 @@ use std::collections::HashMap;
 use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
 use diesel::{ExpressionMethods as _, QueryDsl};
 use diesel_async::RunQueryDsl;
-use kameo::{Actor, actor::ActorRef, messages};
+use kameo::{Actor, actor::ActorRef, messages, prelude::Message};
 use rand_core::{OsRng, RngCore as _};
 use tracing::error;
 
 use crate::{
-    actors::vault::{Bootstrap, RekeyRootKey, TryUnseal, Vault},
+    actors::{
+        proposal_manager::events::ProposalApproved,
+        vault::{Bootstrap, RekeyRootKey, TryUnseal, Vault},
+    },
     crypto::{KeyCell, derive_key, encryption::v1::Nonce, shamir, shamir::shamir_threshold},
-    db::{self, models, schema},
+    db::{
+        self, models,
+        proposal::{ProposalKind, replace_operator},
+        schema,
+    },
 };
 
 #[derive(Debug, thiserror::Error)]
@@ -719,3 +726,55 @@ impl VaultCoordinator {
         self.do_finalize_rekey().await
     }
 }
+
+impl Message<ProposalApproved> for VaultCoordinator {
+    type Reply = ();
+
+    /// Every subscriber sees every approval and acts only on the kinds it owns.
+    async fn handle(
+        &mut self,
+        msg: ProposalApproved,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        let result = match msg.kind {
+            ProposalKind::ReplaceOperator(settings) => self.replace_operator(&settings).await,
+            ProposalKind::TriggerRekey => self.start_rekey().await,
+            _ => return,
+        };
+
+        if let Err(error) = result {
+            error!(
+                ?error,
+                proposal_id = msg.id.to_raw(),
+                "Failed to execute an approved proposal"
+            );
+        }
+    }
+}
+
+impl VaultCoordinator {
+    /// Replaces the operator's public key in place, keeping their id and history, drops the
+    /// share that key no longer matches, then begins a coordinated re-key (§3.3).
+    async fn replace_operator(
+        &mut self,
+        settings: &replace_operator::Settings,
+    ) -> Result<(), Error> {
+        let mut conn = self.db.get().await?;
+
+        diesel::update(schema::operator_identity::table)
+            .filter(schema::operator_identity::id.eq(settings.old_operator_id))
+            .set(schema::operator_identity::public_key.eq(&settings.new_pubkey))
+            .execute(&mut conn)
+            .await?;
+
+        // Drop the stale Shamir share; finalize_rekey stores a fresh one.
+        diesel::delete(schema::operator::table)
+            .filter(schema::operator::id.eq(Some(settings.old_operator_id)))
+            .execute(&mut conn)
+            .await?;
+
+        drop(conn);
+
+        self.start_rekey().await
+    }
+}

server/crates/arbiter-server/src/crypto/integrity/v1.rs

@@ -2,7 +2,7 @@ use crate::{
     actors::vault::{self, GetState, SignIntegrity, Vault, VerifyIntegrity},
     db::{
         self,
-        models::{IntegrityEnvelope, NewIntegrityEnvelope},
+        models::{IntegrityEnvelope, NewIntegrityEnvelope, RootKeyHistoryId},
         schema::integrity_envelope,
     },
 };
@@ -109,11 +109,7 @@ pub async fn sign_entity<E: Integrable>(
     entity: &E,
     entity_id: impl IntoId,
 ) -> Result<(), Error> {
-    let payload_hash = payload_hash(&entity);
+    let (entity_id, mac_input) = envelope_input::<E>(entity, entity_id);
-
-    let entity_id = entity_id.into_id();
-
-    let mac_input = build_mac_input(E::KIND, &entity_id, E::VERSION, &payload_hash);
 
     let (key_version, mac) =
         vault
@@ -124,6 +120,31 @@ pub async fn sign_entity<E: Integrable>(
                 _ => Error::VaultSend,
             })?;
 
+    store_envelope::<E>(conn, entity_id, key_version, mac)
+        .await
+        .map_err(db::DatabaseError::from)?;
+
+    Ok(())
+}
+
+/// The entity id and the bytes the root key covers, as a pair.
+///
+/// Split out of [`sign_entity`] so the `Vault` actor can build an envelope from inside a
+/// message handler, where asking itself for a signature would deadlock.
+pub fn envelope_input<E: Integrable>(entity: &E, entity_id: impl IntoId) -> (Vec<u8>, Vec<u8>) {
+    let payload_hash = payload_hash(entity);
+    let entity_id = entity_id.into_id();
+    let mac_input = build_mac_input(E::KIND, &entity_id, E::VERSION, &payload_hash);
+    (entity_id, mac_input)
+}
+
+/// Stores the integrity envelope for one entity, replacing any envelope it already has.
+pub async fn store_envelope<E: Integrable>(
+    conn: &mut impl AsyncConnection<Backend = Sqlite>,
+    entity_id: Vec<u8>,
+    key_version: RootKeyHistoryId,
+    mac: Vec<u8>,
+) -> Result<(), diesel::result::Error> {
     insert_into(integrity_envelope::table)
         .values(NewIntegrityEnvelope {
             entity_kind: E::KIND.to_owned(),
@@ -143,8 +164,7 @@ pub async fn sign_entity<E: Integrable>(
             integrity_envelope::mac.eq(mac),
         ))
         .execute(conn)
-        .await
+        .await?;
-        .map_err(db::DatabaseError::from)?;
 
     Ok(())
 }

server/crates/arbiter-server/src/grpc/operator/governance.rs

@@ -217,6 +217,9 @@ async fn handle_vote(
         Err(kameo::error::SendError::HandlerError(ProposalError::ProposalNotFound)) => {
             return Err(Status::not_found("Proposal not found"));
         }
+        Err(kameo::error::SendError::HandlerError(ProposalError::Unavailable)) => {
+            return Err(Status::unavailable("Proposal manager is unavailable"));
+        }
         Err(e) => {
             warn!(?e, "cast_vote failed");
             return Err(Status::internal("Failed to cast vote"));

server/crates/arbiter-server/src/peers/operator/session/handlers.rs

@@ -318,7 +318,7 @@ impl OperatorSession {
             .await
             .map_err(|err| match err {
                 SendError::HandlerError(e) => e,
-                _ => crate::actors::proposal_manager::Error::ExecutionFailed("actor unavailable".to_owned()),
+                _ => crate::actors::proposal_manager::Error::Unavailable,
             })
     }
 

server/crates/arbiter-server/tests/governance.rs

@@ -27,6 +27,25 @@ use arbiter_server::db::schema::{
 use diesel::{ExpressionMethods, QueryDsl, insert_into};
 use diesel_async::RunQueryDsl;
 
+/// Retries `probe` until it yields a value, then returns it.
+///
+/// Outcome execution is asynchronous: `CastVote` answers as soon as the quorum is
+/// recorded, and the actor that owns the kind runs afterwards off the message bus. Tests
+/// therefore wait for the effect instead of reading the database straight after the vote.
+async fn eventually<T, F, Fut>(what: &str, mut probe: F) -> T
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Option<T>>,
+{
+    for _ in 0..100 {
+        if let Some(value) = probe().await {
+            return value;
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+    }
+    panic!("{what} did not happen within 2s");
+}
+
 async fn register_operator(db: &db::DatabasePool, pubkey: &authn::PublicKey) -> OperatorIdentityId {
     let mut conn = db.get().await.unwrap();
     insert_into(operator_identity::table)
@@ -628,14 +647,20 @@ async fn approve_sdk_client_writes_integrity_envelope() {
 
     assert_eq!(outcome, VoteOutcome::Approved);
 
-    let mut conn = db.get().await.unwrap();
+    eventually("the client's integrity envelope", || {
-    let count: i64 = integrity_envelope::table
+        let db = db.clone();
-        .filter(integrity_envelope::entity_kind.eq("client_credentials"))
+        async move {
-        .count()
+            let mut conn = db.get().await.unwrap();
-        .get_result(&mut conn)
+            let count: i64 = integrity_envelope::table
-        .await
+                .filter(integrity_envelope::entity_kind.eq("client_credentials"))
-        .unwrap();
+                .count()
-    assert_eq!(count, 1);
+                .get_result(&mut conn)
+                .await
+                .unwrap();
+            (count == 1).then_some(())
+        }
+    })
+    .await;
 }
 
 #[tokio::test]
@@ -682,15 +707,21 @@ async fn grant_wallet_access_on_quorum_approval() {
 
     assert_eq!(outcome, VoteOutcome::Approved);
 
-    let mut conn = db.get().await.unwrap();
+    eventually("the wallet access row", || {
-    let count: i64 = evm_wallet_access::table
+        let db = db.clone();
-        .filter(evm_wallet_access::wallet_id.eq(wallet_id))
+        async move {
-        .filter(evm_wallet_access::client_id.eq(client_id))
+            let mut conn = db.get().await.unwrap();
-        .count()
+            let count: i64 = evm_wallet_access::table
-        .get_result(&mut conn)
+                .filter(evm_wallet_access::wallet_id.eq(wallet_id))
-        .await
+                .filter(evm_wallet_access::client_id.eq(client_id))
-        .unwrap();
+                .count()
-    assert_eq!(count, 1);
+                .get_result(&mut conn)
+                .await
+                .unwrap();
+            (count == 1).then_some(())
+        }
+    })
+    .await;
 }
 
 #[tokio::test]
@@ -767,14 +798,20 @@ async fn approve_persistent_grant_creates_basic_grant_row() {
 
     assert_eq!(outcome, VoteOutcome::Approved);
 
-    let mut conn = db.get().await.unwrap();
+    eventually("the basic grant row", || {
-    let count: i64 = evm_basic_grant::table
+        let db = db.clone();
-        .filter(evm_basic_grant::wallet_access_id.eq(wallet_access_id))
+        async move {
-        .count()
+            let mut conn = db.get().await.unwrap();
-        .get_result(&mut conn)
+            let count: i64 = evm_basic_grant::table
-        .await
+                .filter(evm_basic_grant::wallet_access_id.eq(wallet_access_id))
-        .unwrap();
+                .count()
-    assert_eq!(count, 1);
+                .get_result(&mut conn)
+                .await
+                .unwrap();
+            (count == 1).then_some(())
+        }
+    })
+    .await;
 }
 
 #[tokio::test]
@@ -881,17 +918,23 @@ async fn approve_one_off_transaction_stores_result() {
 
     assert_eq!(outcome, VoteOutcome::Approved);
 
-    let mut conn = db.get().await.unwrap();
+    let (r, s, y_parity): (Vec<u8>, Vec<u8>, i32) = eventually("the transaction signature", || {
-    let (r, s, y_parity): (Vec<u8>, Vec<u8>, i32) = proposal_one_off_transaction_result::table
+        let db = db.clone();
-        .find(proposal_id)
+        async move {
-        .select((
+            let mut conn = db.get().await.unwrap();
-            proposal_one_off_transaction_result::r,
+            proposal_one_off_transaction_result::table
-            proposal_one_off_transaction_result::s,
+                .find(proposal_id)
-            proposal_one_off_transaction_result::y_parity,
+                .select((
-        ))
+                    proposal_one_off_transaction_result::r,
-        .first(&mut conn)
+                    proposal_one_off_transaction_result::s,
-        .await
+                    proposal_one_off_transaction_result::y_parity,
-        .expect("an approved transaction must leave its signature");
+                ))
+                .first(&mut conn)
+                .await
+                .ok()
+        }
+    })
+    .await;
 
     assert_eq!(r.len(), 32, "r must be a 32-byte scalar");
     assert_eq!(s.len(), 32, "s must be a 32-byte scalar");
@@ -944,23 +987,30 @@ async fn replace_operator_updates_pubkey_and_starts_rekey() {
 
     assert_eq!(outcome, VoteOutcome::Approved);
 
+    eventually("the operator's public key to be replaced", || {
+        let db = db.clone();
+        let new_pubkey = new_pubkey.clone();
+        async move {
+            let mut conn = db.get().await.unwrap();
+            let stored: Vec<u8> = operator_identity::table
+                .filter(operator_identity::id.eq(op_id))
+                .select(operator_identity::public_key)
+                .first(&mut conn)
+                .await
+                .unwrap();
+            (stored == new_pubkey).then_some(())
+        }
+    })
+    .await;
+
+    // The old identity row is updated in place, so no second operator appears.
     let mut conn = db.get().await.unwrap();
-    // The old identity row is updated in-place; count stays the same.
     let count: i64 = operator_identity::table
         .count()
         .get_result(&mut conn)
         .await
         .unwrap();
     assert_eq!(count, 1);
-
-    // Verify the public key was updated to the new one.
-    let stored_pubkey: Vec<u8> = operator_identity::table
-        .filter(operator_identity::id.eq(op_id))
-        .select(operator_identity::public_key)
-        .first(&mut conn)
-        .await
-        .unwrap();
-    assert_eq!(stored_pubkey, new_pubkey.clone());
 }
 
 #[tokio::test]