refactor(proposal): publish approved proposals on the bus instead of executing them

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,9 +1,5 @@
 use crate::{
-    actors::{
+    actors::proposal_manager::events::ProposalApproved,
-        evm::EvmActor,
-        vault::Vault,
-        vault_coordinator::{StartRekey, VaultCoordinator},
-    },
     crypto::governance,
     db::{
         self,
@@ -13,7 +9,7 @@ use crate::{
             OperatorIdentityId, Proposal, ProposalId, ProposalStatus, RecoveryOperatorIdentityId,
             SqliteTimestamp,
         },
-        proposal::{ProposalKind, ProposalKindTag, one_off_transaction, persistent_grant},
+        proposal::{ProposalKind, ProposalKindTag},
         schema,
     },
 };
@@ -24,9 +20,12 @@ use diesel::{
 };
 use diesel_async::{AsyncConnection as _, RunQueryDsl};
 use kameo::{Actor, actor::ActorRef, messages};
+use kameo_actors::message_bus::{MessageBus, Publish};
 use std::collections::HashMap;
 use strum::IntoDiscriminant as _;
-use tracing::{error, warn};
+use tracing::warn;
+
+pub mod events;
 
 pub const DEFAULT_TTL_SECS: u32 = 7 * 24 * 60 * 60; // 7 days
 pub const MAX_TTL_SECS: u32 = DEFAULT_TTL_SECS;
@@ -58,8 +57,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")]
@@ -83,24 +82,12 @@ pub struct ProposalSummary {
 #[derive(Actor)]
 pub struct ProposalManager {
     pub(crate) db: db::DatabasePool,
-    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 const fn new(db: db::DatabasePool, events: ActorRef<MessageBus>) -> Self {
-        db: db::DatabasePool,
+        Self { db, events }
-        vault: ActorRef<Vault>,
-        evm: ActorRef<EvmActor>,
-        vault_coordinator: ActorRef<VaultCoordinator>,
-    ) -> Self {
-        Self {
-            db,
-            vault,
-            evm,
-            vault_coordinator,
-        }
     }
 }
 
@@ -343,12 +330,7 @@ impl ProposalManager {
         let threshold_i64 = threshold as i64;
 
         if approve_count >= threshold_i64 {
-            diesel::update(schema::proposal::table.find(proposal_id))
+            self.announce_approval(&mut conn, &proposal).await?;
-                .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);
         }
 
@@ -505,12 +487,7 @@ impl ProposalManager {
         let approve_count = ordinary_approve + recovery_approve;
 
         if approve_count >= threshold_i64 {
-            diesel::update(schema::proposal::table.find(proposal_id))
+            self.announce_approval(&mut conn, &proposal).await?;
-                .set(schema::proposal::status.eq(ProposalStatus::Approved))
-                .execute(&mut conn)
-                .await?;
-            drop(conn);
-            self.execute_proposal(&proposal).await?;
             return Ok(VoteOutcome::Approved);
         }
 
@@ -568,222 +545,30 @@ impl ProposalManager {
         .map_err(Error::from)
     }
 
-    async fn execute_proposal(&self, proposal: &Proposal) -> Result<(), Error> {
+    /// Marks the proposal approved and hands the outcome to whoever owns that kind.
-        let mut conn = self.db.get().await?;
+    ///
-        let kind = db::proposal::load_kind(&mut conn, proposal.id, proposal.kind).await?;
+    /// The outcome is published, not executed: this actor coordinates voting and nothing
-        drop(conn);
+    /// else. Executors subscribe on the bus, so a vote is answered once the quorum is
-
+    /// recorded rather than once the effect has landed.
-        match kind {
+    async fn announce_approval(
-            ProposalKind::ApproveSdkClient(s) => self.execute_approve_sdk_client(s.client_id).await,
-            ProposalKind::GrantWalletAccess(s) => {
-                self.execute_grant_wallet_access(s.wallet_id, s.client_id)
-                    .await
-            }
-            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(
         &self,
-        wallet_id: i32,
+        conn: &mut db::DatabaseConnection,
-        client_id: i32,
+        proposal: &Proposal,
     ) -> Result<(), Error> {
-        use crate::db::models::EvmWalletId;
+        diesel::update(schema::proposal::table.find(proposal.id))
+            .set(schema::proposal::status.eq(ProposalStatus::Approved))
+            .execute(conn)
+            .await?;
 
-        let mut conn = self.db.get().await.map_err(Error::DatabaseConnection)?;
+        let kind = db::proposal::load_kind(conn, proposal.id, proposal.kind).await?;
-
+        let _ = self
-        diesel::insert_into(schema::evm_wallet_access::table)
+            .events
-            .values((
+            .tell(Publish(ProposalApproved {
-                schema::evm_wallet_access::wallet_id.eq(EvmWalletId::from_raw(wallet_id)),
+                id: proposal.id,
-                schema::evm_wallet_access::client_id.eq(client_id),
+                kind,
-            ))
+            }))
-            .execute(&mut conn)
+            .await;
-            .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 {
-                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
-            .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> {
-        use crate::{crypto::integrity, peers::client::ClientCredentials};
-        use arbiter_crypto::authn;
-
-        let mut conn = self.db.get().await.map_err(Error::DatabaseConnection)?;
-
-        let pubkey_bytes: Vec<u8> = schema::program_client::table
-            .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())
-            })
-    }
 }

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/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

@@ -26,6 +26,26 @@ use arbiter_server::db::schema::{
 };
 use diesel::{ExpressionMethods, QueryDsl, insert_into};
 use diesel_async::RunQueryDsl;
+use std::future::Future;
+
+/// 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();
@@ -628,14 +648,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 +708,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 +799,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 +919,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 +988,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]