feat(vault)!: implement full Shamir re-key flow and governance execution (§3.3–§3.5)

- Add `rekey.proto` with `ContributePassphrase` / `ContributeRecoveryPassphrase` / `RekeyResult`
- Wire `rekey` as a 4th vault stream payload in `vault.proto` and gRPC dispatch
- Add `RekeyRootKey` message to `Vault` actor: generates new random seal key, re-encrypts root key, writes new `root_key_history` row
- Add `StartRekey`, `ContributeRekey`, `ContributeRecoveryRekey` messages to `VaultCoordinator`; `finalize_rekey` uses threshold-1 fast path identical to bootstrap
- `execute_replace_operator` now UPDATEs `operator_identity.public_key` in-place (avoids FK constraint violation), deletes stale `operator` share row, then triggers `StartRekey`
- `execute_update_shamir_parameters` triggers `StartRekey` instead of warning stub
- `ProposalKind::ReplaceOperator` carries `old_operator_id`; encode/decode updated accordingly
- `GlobalActors::spawn` extracts `vault_coordinator` before `Ok(Self { … })` so it can be cloned into `ProposalManager::new`
- Add `handle_rekey` in session handlers forwarding passphrase contributions to `VaultCoordinator`
- Fix test: rename `replace_operator_inserts_identity_row` → `replace_operator_updates_pubkey_and_starts_rekey`, assert count stays 1 and pubkey is updated

protobufs/operator/governance.proto

@@ -12,19 +12,20 @@ message Request {
 
 message CreateProposalRequest {
   oneof kind {
-    ApproveSdkClientPayload    approve_sdk_client    = 1;
-    GrantWalletAccessPayload   grant_wallet_access   = 3;
-    ApproveServerUpdatePayload approve_server_update = 4;
-    ReplaceOperatorPayload          replace_operator          = 5;
-    UpdateShamirParametersPayload   update_shamir_parameters  = 6;
-    ApprovePersistentGrantPayload   approve_persistent_grant  = 7;
+    ApproveSdkClientPayload         approve_sdk_client          = 1;
+    GrantWalletAccessPayload        grant_wallet_access         = 3;
+    ApproveServerUpdatePayload      approve_server_update       = 4;
+    ReplaceOperatorPayload          replace_operator            = 5;
+    UpdateShamirParametersPayload   update_shamir_parameters    = 6;
+    ApprovePersistentGrantPayload   approve_persistent_grant    = 7;
     ApproveOneOffTransactionPayload approve_one_off_transaction = 8;
   }
   optional uint32 ttl_secs = 2;
 }
 
 message ReplaceOperatorPayload {
-  bytes new_pubkey = 1;
+  int32 old_operator_id = 1;
+  bytes new_pubkey      = 2;
 }
 
 message UpdateShamirParametersPayload {
@@ -52,9 +53,9 @@ message QueryPendingRequest {}
 
 message Response {
   oneof payload {
-    CreateProposalResponse created  = 1;
-    VoteResponse           voted    = 2;
-    QueryPendingResponse   pending  = 3;
+    CreateProposalResponse created = 1;
+    VoteResponse           voted   = 2;
+    QueryPendingResponse   pending = 3;
   }
 }
 
@@ -67,10 +68,10 @@ message VoteResponse {
 }
 
 enum VoteOutcome {
-  VOTE_OUTCOME_UNSPECIFIED  = 0;
-  VOTE_OUTCOME_PENDING      = 1;
-  VOTE_OUTCOME_APPROVED     = 2;
-  VOTE_OUTCOME_REJECTED     = 3;
+  VOTE_OUTCOME_UNSPECIFIED = 0;
+  VOTE_OUTCOME_PENDING     = 1;
+  VOTE_OUTCOME_APPROVED    = 2;
+  VOTE_OUTCOME_REJECTED    = 3;
 }
 
 message ProposalSummary {
@@ -97,39 +98,39 @@ message VolumeLimitProto {
 }
 
 message EtherTransferSpecProto {
-  repeated bytes  targets = 1;
-  VolumeLimitProto limit  = 2;
+  repeated bytes   targets = 1;
+  VolumeLimitProto limit   = 2;
 }
 
 message TokenTransferSpecProto {
-  bytes            token_contract = 1;
-  optional bytes   target         = 2;
-  repeated VolumeLimitProto volume_limits = 3;
+  bytes                     token_contract = 1;
+  optional bytes            target         = 2;
+  repeated VolumeLimitProto volume_limits  = 3;
 }
 
 message ApproveOneOffTransactionPayload {
-  int32  client_id              = 1;
-  bytes  wallet_address         = 2;
-  uint64 chain_id               = 3;
-  uint64 nonce                  = 4;
-  uint64 gas_limit              = 5;
-  bytes  max_fee_per_gas        = 6;
+  int32  client_id                = 1;
+  bytes  wallet_address           = 2;
+  uint64 chain_id                 = 3;
+  uint64 nonce                    = 4;
+  uint64 gas_limit                = 5;
+  bytes  max_fee_per_gas          = 6;
   bytes  max_priority_fee_per_gas = 7;
-  bytes  to                     = 8;
-  bytes  value                  = 9;
-  bytes  input                  = 10;
+  bytes  to                       = 8;
+  bytes  value                    = 9;
+  bytes  input                    = 10;
 }
 
 message ApprovePersistentGrantPayload {
-  int32  wallet_access_id           = 1;
-  uint64 chain_id                   = 2;
-  optional int64 valid_from_secs    = 3;
-  optional int64 valid_until_secs   = 4;
-  optional bytes max_gas_fee_per_gas         = 5;
-  optional bytes max_priority_fee_per_gas    = 6;
-  optional TransactionRateLimitProto rate_limit = 7;
+  int32                              wallet_access_id         = 1;
+  uint64                             chain_id                 = 2;
+  optional int64                     valid_from_secs          = 3;
+  optional int64                     valid_until_secs         = 4;
+  optional bytes                     max_gas_fee_per_gas      = 5;
+  optional bytes                     max_priority_fee_per_gas = 6;
+  optional TransactionRateLimitProto rate_limit               = 7;
   oneof specific {
-    EtherTransferSpecProto  ether_transfer = 8;
-    TokenTransferSpecProto  token_transfer = 9;
+    EtherTransferSpecProto ether_transfer = 8;
+    TokenTransferSpecProto token_transfer = 9;
   }
 }

protobufs/operator/vault/rekey.proto

@@ -0,0 +1,30 @@
+syntax = "proto3";
+
+package arbiter.operator.vault.rekey;
+
+message ContributePassphrase {
+  bytes passphrase = 1;
+}
+
+message ContributeRecoveryPassphrase {
+  int32 recovery_operator_id = 1;
+  bytes passphrase           = 2;
+}
+
+enum RekeyResult {
+  REKEY_RESULT_UNSPECIFIED            = 0;
+  REKEY_RESULT_SUCCESS                = 1;
+  REKEY_RESULT_AWAITING_CONTRIBUTIONS = 2;
+  REKEY_RESULT_NOT_IN_PROGRESS        = 3;
+}
+
+message Request {
+  oneof payload {
+    ContributePassphrase         contribute_passphrase          = 1;
+    ContributeRecoveryPassphrase contribute_recovery_passphrase = 2;
+  }
+}
+
+message Response {
+  RekeyResult result = 1;
+}

protobufs/operator/vault/vault.proto

@@ -5,20 +5,23 @@ package arbiter.operator.vault;
 import "google/protobuf/empty.proto";
 import "shared/vault.proto";
 import "operator/vault/bootstrap.proto";
+import "operator/vault/rekey.proto";
 import "operator/vault/unseal.proto";
 
 message Request {
   oneof payload {
     google.protobuf.Empty query_state = 1;
-    unseal.Request unseal = 2;
-    bootstrap.Request bootstrap = 3;
+    unseal.Request        unseal      = 2;
+    bootstrap.Request     bootstrap   = 3;
+    rekey.Request         rekey       = 4;
   }
 }
 
 message Response {
   oneof payload {
-    arbiter.shared.VaultState state = 1;
-    unseal.Response unseal = 2;
-    bootstrap.Response bootstrap = 3;
+    arbiter.shared.VaultState state     = 1;
+    unseal.Response           unseal    = 2;
+    bootstrap.Response        bootstrap = 3;
+    rekey.Response            rekey     = 4;
   }
 }

server/clippy.toml

@@ -26,3 +26,5 @@ trait-assoc-item-kinds-order = [
     "type",
     "fn",
 ] # community tested standard
+
+too-many-lines-threshold = 150

server/crates/arbiter-proto/src/lib.rs

@@ -38,6 +38,10 @@ pub mod proto {
                 tonic::include_proto!("arbiter.operator.vault.bootstrap");
             }
 
+            pub mod rekey {
+                tonic::include_proto!("arbiter.operator.vault.rekey");
+            }
+
             pub mod unseal {
                 tonic::include_proto!("arbiter.operator.vault.unseal");
             }

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

@@ -51,18 +51,20 @@ impl GlobalActors {
         let key_holder = Vault::spawn(Vault::new(db.clone(), message_bus.clone()).await?);
         let operator_registry = OperatorRegistry::spawn(OperatorRegistry::default());
         let evm = EvmActor::spawn(EvmActor::new(key_holder.clone(), db.clone()));
+        let vault_coordinator = VaultCoordinator::spawn(VaultCoordinator::new(
+            db.clone(),
+            key_holder.clone(),
+        ));
         Ok(Self {
             bootstrapper: Bootstrapper::spawn(Bootstrapper::new(&db).await?),
-            vault_coordinator: VaultCoordinator::spawn(VaultCoordinator::new(
-                db.clone(),
-                key_holder.clone(),
-            )),
             proposal_manager: ProposalManager::spawn(ProposalManager::new(
                 db,
                 key_holder.clone(),
                 evm.clone(),
+                vault_coordinator.clone(),
             )),
             vault: key_holder,
+            vault_coordinator,
             flow_coordinator: FlowCoordinator::spawn(FlowCoordinator::new(
                 operator_registry.clone(),
             )),

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

@@ -1,10 +1,14 @@
 use crate::{
-    actors::{evm::EvmActor, vault::Vault},
+    actors::{
+        evm::EvmActor,
+        vault::Vault,
+        vault_coordinator::{StartRekey, VaultCoordinator},
+    },
     db::{
         self,
         models::{
-            NewProposal, NewProposalVote, NewRecoveryProposalVote,
-            NewRecoveryWakeupRequest, Proposal, ProposalStatus, SqliteTimestamp,
+            NewProposal, NewProposalVote, NewRecoveryProposalVote, NewRecoveryWakeupRequest,
+            Proposal, ProposalStatus, SqliteTimestamp,
         },
         schema,
     },
@@ -13,34 +17,65 @@ use chrono::Utc;
 use diesel::{ExpressionMethods as _, QueryDsl};
 use diesel_async::RunQueryDsl;
 use kameo::{actor::ActorRef, messages};
+use strum::{Display, EnumString, IntoStaticStr};
 use tracing::{error, warn};
 
 pub const DEFAULT_TTL_SECS: i64 = 7 * 24 * 60 * 60; // 7 days
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Display, EnumString, IntoStaticStr)]
+#[strum(serialize_all = "snake_case")]
+pub enum ProposalKindTag {
+    ApproveSdkClient,
+    GrantWalletAccess,
+    ApproveServerUpdate,
+    ReplaceOperator,
+    UpdateShamirParameters,
+    ApprovePersistentGrant,
+    ApproveOneOffTransaction,
+}
+
 #[derive(Debug, Clone)]
 pub enum ProposalKind {
-    ApproveSdkClient { client_id: i32 },
-    GrantWalletAccess { wallet_id: i32, client_id: i32 },
+    ApproveSdkClient {
+        client_id: i32,
+    },
+    GrantWalletAccess {
+        wallet_id: i32,
+        client_id: i32,
+    },
     ApproveServerUpdate,
-    ReplaceOperator { new_pubkey: Vec<u8> },
-    UpdateShamirParameters { new_n: u8 },
-    ApprovePersistentGrant { payload_bytes: Vec<u8> },
-    ApproveOneOffTransaction { payload_bytes: Vec<u8> },
+    ReplaceOperator {
+        old_operator_id: i32,
+        new_pubkey: Vec<u8>,
+    },
+    UpdateShamirParameters {
+        new_n: u8,
+    },
+    ApprovePersistentGrant {
+        payload_bytes: Vec<u8>,
+    },
+    ApproveOneOffTransaction {
+        payload_bytes: Vec<u8>,
+    },
 }
 
 impl ProposalKind {
-    pub const fn kind_str(&self) -> &'static str {
+    pub const fn tag(&self) -> ProposalKindTag {
         match self {
-            Self::ApproveSdkClient { .. } => "approve_sdk_client",
-            Self::GrantWalletAccess { .. } => "grant_wallet_access",
-            Self::ApproveServerUpdate => "approve_server_update",
-            Self::ReplaceOperator { .. } => "replace_operator",
-            Self::UpdateShamirParameters { .. } => "update_shamir_parameters",
-            Self::ApprovePersistentGrant { .. } => "approve_persistent_grant",
-            Self::ApproveOneOffTransaction { .. } => "approve_one_off_transaction",
+            Self::ApproveSdkClient { .. } => ProposalKindTag::ApproveSdkClient,
+            Self::GrantWalletAccess { .. } => ProposalKindTag::GrantWalletAccess,
+            Self::ApproveServerUpdate => ProposalKindTag::ApproveServerUpdate,
+            Self::ReplaceOperator { .. } => ProposalKindTag::ReplaceOperator,
+            Self::UpdateShamirParameters { .. } => ProposalKindTag::UpdateShamirParameters,
+            Self::ApprovePersistentGrant { .. } => ProposalKindTag::ApprovePersistentGrant,
+            Self::ApproveOneOffTransaction { .. } => ProposalKindTag::ApproveOneOffTransaction,
         }
     }
 
+    pub fn kind_str(&self) -> &'static str {
+        self.tag().into()
+    }
+
     pub fn encode_payload(&self) -> Vec<u8> {
         match self {
             Self::ApproveSdkClient { client_id } => client_id.to_be_bytes().to_vec(),
@@ -54,35 +89,45 @@ impl ProposalKind {
                 buf
             }
             Self::ApproveServerUpdate => vec![],
-            Self::ReplaceOperator { new_pubkey } => {
+            Self::ReplaceOperator {
+                old_operator_id,
+                new_pubkey,
+            } => {
                 let len = u32::try_from(new_pubkey.len()).expect("pubkey len fits in u32");
-                let mut buf = Vec::with_capacity(4 + new_pubkey.len());
+                let mut buf = Vec::with_capacity(4 + 4 + new_pubkey.len());
+                buf.extend_from_slice(&old_operator_id.to_be_bytes());
                 buf.extend_from_slice(&len.to_be_bytes());
                 buf.extend_from_slice(new_pubkey);
                 buf
             }
             Self::UpdateShamirParameters { new_n } => vec![*new_n],
-            Self::ApprovePersistentGrant { payload_bytes } => payload_bytes.clone(),
-            Self::ApproveOneOffTransaction { payload_bytes } => payload_bytes.clone(),
+            Self::ApprovePersistentGrant { payload_bytes }
+            | Self::ApproveOneOffTransaction { payload_bytes } => payload_bytes.clone(),
         }
     }
 
     /// Key-rotation proposals require every operator to approve (§3.3).
     #[must_use]
     pub fn requires_full_quorum(kind: &str) -> bool {
-        matches!(kind, "replace_operator" | "update_shamir_parameters")
+        matches!(
+            kind.parse::<ProposalKindTag>(),
+            Ok(ProposalKindTag::ReplaceOperator | ProposalKindTag::UpdateShamirParameters)
+        )
     }
 
     pub fn decode(kind: &str, payload: &[u8]) -> Result<Self, String> {
-        match kind {
-            "approve_sdk_client" => {
+        let tag = kind
+            .parse::<ProposalKindTag>()
+            .map_err(|_| format!("unknown proposal kind: {kind}"))?;
+        match tag {
+            ProposalKindTag::ApproveSdkClient => {
                 let bytes = <[u8; 4]>::try_from(payload)
                     .map_err(|_| "invalid payload for approve_sdk_client".to_owned())?;
                 Ok(Self::ApproveSdkClient {
                     client_id: i32::from_be_bytes(bytes),
                 })
             }
-            "grant_wallet_access" => {
+            ProposalKindTag::GrantWalletAccess => {
                 let bytes = <[u8; 8]>::try_from(payload)
                     .map_err(|_| "invalid payload for grant_wallet_access".to_owned())?;
                 Ok(Self::GrantWalletAccess {
@@ -90,9 +135,13 @@ impl ProposalKind {
                     client_id: i32::from_be_bytes(bytes[4..].try_into().unwrap()),
                 })
             }
-            "approve_server_update" => Ok(Self::ApproveServerUpdate),
-            "replace_operator" => {
-                let (len_bytes, rest) = payload
+            ProposalKindTag::ApproveServerUpdate => Ok(Self::ApproveServerUpdate),
+            ProposalKindTag::ReplaceOperator => {
+                let (id_bytes, rest) = payload
+                    .split_first_chunk::<4>()
+                    .ok_or_else(|| "replace_operator payload too short".to_owned())?;
+                let old_operator_id = i32::from_be_bytes(*id_bytes);
+                let (len_bytes, rest) = rest
                     .split_first_chunk::<4>()
                     .ok_or_else(|| "replace_operator payload too short".to_owned())?;
                 let len = u32::from_be_bytes(*len_bytes);
@@ -101,21 +150,23 @@ impl ProposalKind {
                     .get(..len)
                     .ok_or_else(|| "replace_operator payload truncated".to_owned())?
                     .to_vec();
-                Ok(Self::ReplaceOperator { new_pubkey })
+                Ok(Self::ReplaceOperator {
+                    old_operator_id,
+                    new_pubkey,
+                })
             }
-            "update_shamir_parameters" => {
+            ProposalKindTag::UpdateShamirParameters => {
                 let &[new_n] = payload else {
                     return Err("invalid payload for update_shamir_parameters".to_owned());
                 };
                 Ok(Self::UpdateShamirParameters { new_n })
             }
-            "approve_persistent_grant" => Ok(Self::ApprovePersistentGrant {
+            ProposalKindTag::ApprovePersistentGrant => Ok(Self::ApprovePersistentGrant {
                 payload_bytes: payload.to_vec(),
             }),
-            "approve_one_off_transaction" => Ok(Self::ApproveOneOffTransaction {
+            ProposalKindTag::ApproveOneOffTransaction => Ok(Self::ApproveOneOffTransaction {
                 payload_bytes: payload.to_vec(),
             }),
-            other => Err(format!("unknown proposal kind: {other}")),
         }
     }
 }
@@ -169,6 +220,7 @@ pub struct ProposalManager {
     pub(crate) db: db::DatabasePool,
     pub(crate) vault: ActorRef<Vault>,
     pub(crate) evm: ActorRef<EvmActor>,
+    pub(crate) vault_coordinator: ActorRef<VaultCoordinator>,
 }
 
 impl ProposalManager {
@@ -176,8 +228,14 @@ impl ProposalManager {
         db: db::DatabasePool,
         vault: ActorRef<Vault>,
         evm: ActorRef<EvmActor>,
+        vault_coordinator: ActorRef<VaultCoordinator>,
     ) -> Self {
-        Self { db, vault, evm }
+        Self {
+            db,
+            vault,
+            evm,
+            vault_coordinator,
+        }
     }
 }
 
@@ -496,8 +554,7 @@ impl ProposalManager {
             .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())),
+                schema::recovery_wakeup_request::cancelled_at.eq(Some(SqliteTimestamp::now())),
             ))
             .execute(&mut conn)
             .await?;
@@ -530,7 +587,7 @@ impl ProposalManager {
                 other => Error::DatabaseQuery(other),
             })?;
 
-        if proposal.kind != "replace_operator" {
+        if proposal.kind.parse::<ProposalKindTag>() != Ok(ProposalKindTag::ReplaceOperator) {
             return Err(Error::NotAllowedForRecoveryOperator);
         }
 
@@ -651,9 +708,7 @@ impl ProposalManager {
     const WAKEUP_DELAY_SECS: i32 = 14 * 24 * 60 * 60;
 
     /// Returns true when an uncancelled wakeup request has passed the 14-day dispute window.
-    async fn is_recovery_active_conn(
-        conn: &mut db::DatabaseConnection,
-    ) -> Result<bool, Error> {
+    async fn is_recovery_active_conn(conn: &mut db::DatabaseConnection) -> Result<bool, Error> {
         let count: i64 = schema::recovery_wakeup_request::table
             .filter(schema::recovery_wakeup_request::cancelled_at.is_null())
             .filter(
@@ -671,9 +726,7 @@ impl ProposalManager {
     }
 
     /// Returns true when there is any uncancelled wakeup request (pending or active).
-    async fn has_uncancelled_wakeup(
-        conn: &mut db::DatabaseConnection,
-    ) -> Result<bool, Error> {
+    async fn has_uncancelled_wakeup(conn: &mut db::DatabaseConnection) -> Result<bool, Error> {
         let count: i64 = schema::recovery_wakeup_request::table
             .filter(schema::recovery_wakeup_request::cancelled_at.is_null())
             .count()
@@ -694,11 +747,15 @@ impl ProposalManager {
                 client_id,
             } => self.execute_grant_wallet_access(wallet_id, client_id).await,
             ProposalKind::ApproveServerUpdate => Ok(()),
-            ProposalKind::ReplaceOperator { new_pubkey } => {
-                self.execute_replace_operator(new_pubkey).await
+            ProposalKind::ReplaceOperator {
+                old_operator_id,
+                new_pubkey,
+            } => {
+                self.execute_replace_operator(old_operator_id, new_pubkey)
+                    .await
             }
             ProposalKind::UpdateShamirParameters { new_n } => {
-                self.execute_update_shamir_parameters(new_n)
+                self.execute_update_shamir_parameters(new_n).await
             }
             ProposalKind::ApprovePersistentGrant { payload_bytes } => {
                 self.execute_approve_persistent_grant(payload_bytes).await
@@ -731,26 +788,45 @@ impl ProposalManager {
         Ok(())
     }
 
-    async fn execute_replace_operator(&self, new_pubkey: Vec<u8>) -> Result<(), Error> {
+    /// 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: i32,
+        new_pubkey: Vec<u8>,
+    ) -> Result<(), Error> {
         let mut conn = self.db.get().await.map_err(Error::DatabaseConnection)?;
-        diesel::insert_into(schema::operator_identity::table)
-            .values(schema::operator_identity::public_key.eq(&new_pubkey))
+
+        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!("replace operator: {e}")))?;
+            .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(())
     }
 
-    #[expect(
-        clippy::unused_self,
-        clippy::unnecessary_wraps,
-        reason = "signature must match other execute_* methods"
-    )]
-    fn execute_update_shamir_parameters(&self, new_n: u8) -> Result<(), Error> {
-        warn!(
-            new_n,
-            "UpdateShamirParameters approved; Shamir re-keying must be performed out-of-band"
-        );
+    /// Triggers a Shamir re-key with the current operator set (§3.3).
+    async fn execute_update_shamir_parameters(&self, _new_n: u8) -> Result<(), Error> {
+        self.vault_coordinator
+            .ask(StartRekey {})
+            .await
+            .map_err(|e| Error::ExecutionFailed(format!("start rekey: {e}")))?;
         Ok(())
     }
 

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

@@ -275,6 +275,59 @@ impl Vault {
         Ok(())
     }
 
+    /// Re-encrypts the root key with `new_seal_key` and records a new root_key_history row.
+    /// Called after a Shamir re-key so the old seal key is no longer sufficient to unseal.
+    #[message]
+    pub async fn rekey_root_key(&mut self, mut new_seal_key: KeyCell) -> Result<(), Error> {
+        let Unsealed {
+            root_key,
+            root_key_history_id,
+        } = Self::expect_unsealed(&mut self.state)?;
+
+        let new_nonce = Nonce::default();
+        let new_salt = v1::generate_salt();
+
+        let new_ciphertext: Vec<u8> = root_key.0.read_inline(|rk| {
+            new_seal_key
+                .encrypt(&new_nonce, v1::ROOT_KEY_TAG, rk.as_slice())
+                .map_err(|err| {
+                    error!(?err, "Fatal rekey error");
+                    Error::Encryption(err)
+                })
+        })?;
+
+        let data_encryption_nonce = Nonce::default();
+
+        let mut conn = self.db.get().await?;
+        let new_root_key_history_id: i32 = conn
+            .transaction(async |conn| {
+                let new_id = insert_into(schema::root_key_history::table)
+                    .values(&models::NewRootKeyHistory {
+                        ciphertext: new_ciphertext,
+                        tag: v1::ROOT_KEY_TAG.to_vec(),
+                        root_key_encryption_nonce: new_nonce.to_vec(),
+                        data_encryption_nonce: data_encryption_nonce.to_vec(),
+                        schema_version: 1,
+                        salt: new_salt.to_vec(),
+                    })
+                    .returning(schema::root_key_history::id)
+                    .get_result::<i32>(&mut *conn)
+                    .await?;
+
+                update(schema::arbiter_settings::table)
+                    .set(schema::arbiter_settings::root_key_id.eq(new_id))
+                    .execute(&mut *conn)
+                    .await?;
+
+                Result::<_, diesel::result::Error>::Ok(new_id)
+            })
+            .await?;
+
+        *root_key_history_id = RootKeyHistoryId::from_raw(new_root_key_history_id);
+        info!("Vault root key rekeyed successfully");
+        Ok(())
+    }
+
     #[message]
     pub async fn seal(&mut self) -> Result<(), Error> {
         let Unsealed {

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

@@ -8,7 +8,7 @@ use rand_core::{OsRng, RngCore as _};
 use tracing::error;
 
 use crate::{
-    actors::vault::{Bootstrap, TryUnseal, Vault},
+    actors::vault::{Bootstrap, RekeyRootKey, TryUnseal, Vault},
     crypto::{KeyCell, derive_key, encryption::v1::Nonce, shamir, shamir::shamir_threshold},
     db::{self, models, schema},
 };
@@ -19,6 +19,8 @@ pub enum Error {
     AlreadyBootstrapping,
     #[error("Already coordinating an unseal")]
     AlreadyUnsealing,
+    #[error("Rekey not in progress")]
+    NotRekeying,
     #[error("Bootstrap not in progress")]
     NotBootstrapping,
     #[error("Unseal not in progress")]
@@ -60,6 +62,15 @@ enum CoordinatorState {
         ordinary_passphrases: HashMap<i32, Vec<u8>>,
         recovery_passphrases: HashMap<i32, Vec<u8>>,
     },
+    /// Shamir re-key after `replace_operator` or `update_shamir_parameters` is approved (§3.3).
+    /// Collects new passphrases from all current operators, then generates a fresh seal key,
+    /// re-splits it, and re-encrypts the vault root key.
+    Rekeying {
+        ordinary_count: usize,
+        recovery_count: usize,
+        passphrases: HashMap<i32, Vec<u8>>,
+        recovery_passphrases: HashMap<i32, Vec<u8>>,
+    },
 }
 
 #[derive(Actor)]
@@ -102,17 +113,17 @@ fn encrypt_share(
 fn decrypt_share(
     passphrase_bytes: Vec<u8>,
     encrypted_share: Vec<u8>,
-    share_nonce_bytes: Vec<u8>,
-    share_salt: Vec<u8>,
+    share_nonce_bytes: &[u8],
+    share_salt: &[u8],
     operator_id: i32,
 ) -> Result<Vec<u8>, Error> {
-    let nonce = Nonce::try_from(share_nonce_bytes.as_slice()).map_err(|()| {
+    let nonce = Nonce::try_from(share_nonce_bytes).map_err(|()| {
         error!(operator_id, "Invalid nonce in DB");
         Error::BrokenDatabase
     })?;
 
     let mut passphrase_cell = SafeCell::new(passphrase_bytes);
-    let mut share_seal_key = derive_key(&mut passphrase_cell, &share_salt);
+    let mut share_seal_key = derive_key(&mut passphrase_cell, share_salt);
 
     let mut share_buffer = SafeCell::new(encrypted_share);
     share_seal_key
@@ -123,8 +134,8 @@ fn decrypt_share(
 }
 
 /// §3.4: Split the seal key across ordinary + recovery operators.
-/// Threshold = shamir_threshold(ordinary_count); total shares = ordinary + recovery.
-/// When ordinary_count == 1 (threshold = 1), vsss-rs does not support a proper split,
+/// Threshold = `shamir_threshold(ordinary_count)`; total shares = ordinary + recovery.
+/// When `ordinary_count` == 1 (threshold = 1), vsss-rs does not support a proper split,
 /// so each share is the seal key itself — any single participant can reconstruct.
 async fn finalize_bootstrap(
     db: db::DatabasePool,
@@ -146,7 +157,7 @@ async fn finalize_bootstrap(
         shamir::split_key(threshold, total, &seal_key_bytes, OsRng)
             .map_err(|e| Error::Shamir(e.to_string()))?
     } else {
-        (0..total).map(|_| seal_key_bytes.to_vec()).collect()
+        std::iter::repeat_with(|| seal_key_bytes.to_vec()).take(total).collect()
     };
 
     let seal_key = KeyCell::from(seal_key_bytes);
@@ -155,9 +166,10 @@ async fn finalize_bootstrap(
     let mut shares_iter = shares.into_iter();
 
     for (operator_id_raw, passphrase_bytes) in ordinary_passphrases {
-        let share = shares_iter.next().expect("split_key returned enough shares");
-        let (encrypted_share, nonce_bytes, share_salt) =
-            encrypt_share(passphrase_bytes, &share)?;
+        let share = shares_iter
+            .next()
+            .expect("split_key returned enough shares");
+        let (encrypted_share, nonce_bytes, share_salt) = encrypt_share(passphrase_bytes, &share)?;
 
         diesel::replace_into(schema::operator::table)
             .values((
@@ -173,9 +185,10 @@ async fn finalize_bootstrap(
     }
 
     for (recovery_id_raw, passphrase_bytes) in recovery_passphrases {
-        let share = shares_iter.next().expect("split_key returned enough shares");
-        let (encrypted_share, nonce_bytes, share_salt) =
-            encrypt_share(passphrase_bytes, &share)?;
+        let share = shares_iter
+            .next()
+            .expect("split_key returned enough shares");
+        let (encrypted_share, nonce_bytes, share_salt) = encrypt_share(passphrase_bytes, &share)?;
 
         diesel::replace_into(schema::recovery_operator::table)
             .values((
@@ -190,13 +203,10 @@ async fn finalize_bootstrap(
             .await?;
     }
 
-    vault
-        .ask(Bootstrap { seal_key })
-        .await
-        .map_err(|err| {
-            error!(?err, "Vault bootstrap failed");
-            Error::VaultError
-        })?;
+    vault.ask(Bootstrap { seal_key }).await.map_err(|err| {
+        error!(?err, "Vault bootstrap failed");
+        Error::VaultError
+    })?;
 
     Ok(())
 }
@@ -235,8 +245,8 @@ async fn finalize_unseal(
         shares.push(decrypt_share(
             passphrase_bytes,
             encrypted_share,
-            share_nonce_bytes,
-            share_salt,
+            &share_nonce_bytes,
+            &share_salt,
             operator_id_raw,
         )?);
     }
@@ -257,8 +267,8 @@ async fn finalize_unseal(
         shares.push(decrypt_share(
             passphrase_bytes,
             encrypted_share,
-            share_nonce_bytes,
-            share_salt,
+            &share_nonce_bytes,
+            &share_salt,
             recovery_id_raw,
         )?);
     }
@@ -266,19 +276,100 @@ async fn finalize_unseal(
     // When threshold==1, shares are raw 32-byte seal keys (vsss-rs cannot split 1-of-N).
     // Any single decrypted share is the key itself.
     let seal_key_bytes: [u8; 32] = if threshold <= 1 {
-        let raw = shares.into_iter().next().ok_or_else(|| Error::Shamir("No shares available".into()))?;
-        raw.try_into().map_err(|_| Error::Shamir("Invalid share length".into()))?
+        let raw = shares
+            .into_iter()
+            .next()
+            .ok_or_else(|| Error::Shamir("No shares available".into()))?;
+        raw.try_into()
+            .map_err(|_| Error::Shamir("Invalid share length".into()))?
     } else {
         shamir::combine_shares(&shares).map_err(|e| Error::Shamir(e.to_string()))?
     };
 
     let seal_key = KeyCell::from(seal_key_bytes);
 
+    vault.ask(TryUnseal { seal_key }).await.map_err(|err| {
+        error!(?err, "Vault unseal failed");
+        Error::VaultError
+    })?;
+
+    Ok(())
+}
+
+/// §3.3: Generate a fresh seal key, split across current operators, re-encrypt the vault root key.
+/// Called after `replace_operator` or `update_shamir_parameters` is approved and all contributors submit.
+async fn finalize_rekey(
+    db: db::DatabasePool,
+    vault: ActorRef<Vault>,
+    ordinary_passphrases: HashMap<i32, Vec<u8>>,
+    recovery_passphrases: HashMap<i32, Vec<u8>>,
+) -> Result<(), Error> {
+    let ordinary_count = ordinary_passphrases.len();
+    let recovery_count = recovery_passphrases.len();
+    let total = ordinary_count + recovery_count;
+    let threshold = shamir_threshold(ordinary_count);
+
+    let mut new_seal_key_bytes = [0u8; 32];
+    OsRng.fill_bytes(&mut new_seal_key_bytes);
+
+    let shares: Vec<Vec<u8>> = if threshold >= 2 {
+        shamir::split_key(threshold, total, &new_seal_key_bytes, OsRng)
+            .map_err(|e| Error::Shamir(e.to_string()))?
+    } else {
+        std::iter::repeat_with(|| new_seal_key_bytes.to_vec())
+            .take(total)
+            .collect()
+    };
+
+    let mut conn = db.get().await?;
+    let mut shares_iter = shares.into_iter();
+
+    for (operator_id_raw, passphrase_bytes) in ordinary_passphrases {
+        let share = shares_iter
+            .next()
+            .expect("split_key returned enough shares");
+        let (encrypted_share, nonce_bytes, share_salt) = encrypt_share(passphrase_bytes, &share)?;
+
+        diesel::replace_into(schema::operator::table)
+            .values((
+                schema::operator::id.eq(Some(operator_id_raw)),
+                schema::operator::share.eq(&encrypted_share),
+                schema::operator::share_nonce.eq(&nonce_bytes),
+                schema::operator::share_salt.eq(&share_salt),
+                schema::operator::created_at.eq(models::SqliteTimestamp::now()),
+                schema::operator::updated_at.eq(models::SqliteTimestamp::now()),
+            ))
+            .execute(&mut conn)
+            .await?;
+    }
+
+    for (recovery_id_raw, passphrase_bytes) in recovery_passphrases {
+        let share = shares_iter
+            .next()
+            .expect("split_key returned enough shares");
+        let (encrypted_share, nonce_bytes, share_salt) = encrypt_share(passphrase_bytes, &share)?;
+
+        diesel::replace_into(schema::recovery_operator::table)
+            .values((
+                schema::recovery_operator::id.eq(recovery_id_raw),
+                schema::recovery_operator::share.eq(&encrypted_share),
+                schema::recovery_operator::share_nonce.eq(&nonce_bytes),
+                schema::recovery_operator::share_salt.eq(&share_salt),
+                schema::recovery_operator::created_at.eq(models::SqliteTimestamp::now()),
+                schema::recovery_operator::updated_at.eq(models::SqliteTimestamp::now()),
+            ))
+            .execute(&mut conn)
+            .await?;
+    }
+
+    drop(conn);
+
+    let new_seal_key = KeyCell::from(new_seal_key_bytes);
     vault
-        .ask(TryUnseal { seal_key })
+        .ask(RekeyRootKey { new_seal_key })
         .await
         .map_err(|err| {
-            error!(?err, "Vault unseal failed");
+            error!(?err, "Vault rekey failed");
             Error::VaultError
         })?;
 
@@ -486,8 +577,7 @@ impl VaultCoordinator {
                 .count()
                 .get_result(&mut conn)
                 .await?;
-            let threshold =
-                shamir_threshold(usize::try_from(ordinary_count).unwrap_or_default());
+            let threshold = shamir_threshold(usize::try_from(ordinary_count).unwrap_or_default());
             self.state = CoordinatorState::Unsealing {
                 threshold,
                 ordinary_passphrases: HashMap::new(),
@@ -517,4 +607,115 @@ impl VaultCoordinator {
         .await?;
         Ok(true)
     }
+
+    async fn do_finalize_rekey(&mut self) -> Result<bool, Error> {
+        let CoordinatorState::Rekeying {
+            passphrases,
+            recovery_passphrases,
+            ..
+        } = std::mem::replace(&mut self.state, CoordinatorState::Idle)
+        else {
+            unreachable!()
+        };
+
+        finalize_rekey(
+            self.db.clone(),
+            self.vault.clone(),
+            passphrases,
+            recovery_passphrases,
+        )
+        .await?;
+        Ok(true)
+    }
+}
+
+#[messages]
+impl VaultCoordinator {
+    /// Begin Shamir re-key after a key-rotation proposal is approved (§3.3).
+    /// Queries the current operator and recovery operator counts from the DB,
+    /// then transitions to Rekeying state awaiting contributions from all of them.
+    #[message]
+    pub async fn start_rekey(&mut self) -> Result<(), Error> {
+        if !matches!(self.state, CoordinatorState::Idle) {
+            return Err(Error::AlreadyBootstrapping);
+        }
+        let mut conn = self.db.get().await?;
+        let ordinary_count: i64 = schema::operator_identity::table
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        let recovery_count: i64 = schema::recovery_operator_identity::table
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        self.state = CoordinatorState::Rekeying {
+            ordinary_count: ordinary_count as usize,
+            recovery_count: recovery_count as usize,
+            passphrases: HashMap::new(),
+            recovery_passphrases: HashMap::new(),
+        };
+        Ok(())
+    }
+
+    /// Contribute an ordinary operator passphrase for the re-key.
+    /// Returns Ok(true) when all contributors have submitted and the re-key is complete.
+    #[message]
+    pub async fn contribute_rekey(
+        &mut self,
+        operator_id: i32,
+        mut passphrase: SafeCell<Vec<u8>>,
+    ) -> Result<bool, Error> {
+        let CoordinatorState::Rekeying {
+            ordinary_count,
+            recovery_count,
+            passphrases,
+            recovery_passphrases,
+        } = &mut self.state
+        else {
+            return Err(Error::NotRekeying);
+        };
+
+        if passphrases.contains_key(&operator_id) {
+            return Err(Error::DuplicateContribution);
+        }
+
+        passphrases.insert(operator_id, passphrase.read().to_vec());
+
+        if passphrases.len() < *ordinary_count || recovery_passphrases.len() < *recovery_count {
+            return Ok(false);
+        }
+
+        self.do_finalize_rekey().await
+    }
+
+    /// Contribute a recovery operator passphrase for the re-key.
+    /// Returns Ok(true) when all contributors have submitted and the re-key is complete.
+    #[message]
+    pub async fn contribute_recovery_rekey(
+        &mut self,
+        recovery_operator_id: i32,
+        mut passphrase: SafeCell<Vec<u8>>,
+    ) -> Result<bool, Error> {
+        let CoordinatorState::Rekeying {
+            ordinary_count,
+            recovery_count,
+            passphrases,
+            recovery_passphrases,
+        } = &mut self.state
+        else {
+            return Err(Error::NotRekeying);
+        };
+
+        if recovery_passphrases.contains_key(&recovery_operator_id) {
+            return Err(Error::DuplicateContribution);
+        }
+
+        recovery_passphrases.insert(recovery_operator_id, passphrase.read().to_vec());
+
+        if passphrases.len() < *ordinary_count || recovery_passphrases.len() < *recovery_count {
+            return Ok(false);
+        }
+
+        self.do_finalize_rekey().await
+    }
 }

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

@@ -54,20 +54,28 @@ async fn handle_create(
         },
         Some(ProtoKind::ApproveServerUpdate(_)) => ProposalKind::ApproveServerUpdate,
         Some(ProtoKind::ReplaceOperator(p)) => ProposalKind::ReplaceOperator {
-            new_pubkey: p.new_pubkey.try_into()
-                .map_err(|_| Status::invalid_argument("replace_operator: pubkey must be 32 bytes"))?,
+            old_operator_id: p.old_operator_id,
+            new_pubkey: p.new_pubkey,
         },
         Some(ProtoKind::UpdateShamirParameters(p)) => ProposalKind::UpdateShamirParameters {
-            #[expect(clippy::cast_possible_truncation, clippy::as_conversions, reason = "new_n is always a small operator count")]
+            #[expect(
+                clippy::cast_possible_truncation,
+                clippy::as_conversions,
+                reason = "new_n is always a small operator count"
+            )]
             new_n: p.new_n as u8,
         },
         Some(ProtoKind::ApprovePersistentGrant(p)) => {
             use prost::Message as _;
-            ProposalKind::ApprovePersistentGrant { payload_bytes: p.encode_to_vec() }
+            ProposalKind::ApprovePersistentGrant {
+                payload_bytes: p.encode_to_vec(),
+            }
         }
         Some(ProtoKind::ApproveOneOffTransaction(p)) => {
             use prost::Message as _;
-            ProposalKind::ApproveOneOffTransaction { payload_bytes: p.encode_to_vec() }
+            ProposalKind::ApproveOneOffTransaction {
+                payload_bytes: p.encode_to_vec(),
+            }
         }
         None => return Err(Status::invalid_argument("Missing proposal kind")),
     };

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

@@ -1,12 +1,20 @@
 use crate::{
     actors::vault::VaultState,
-    peers::operator::{OperatorSession, session::handlers::HandleQueryVaultState},
+    peers::operator::{
+        OperatorSession,
+        session::handlers::{
+            HandleContributeRecoveryRekeyPassphrase, HandleContributeRekeyPassphrase,
+            HandleQueryVaultState,
+        },
+    },
 };
 use arbiter_proto::{
     proto::operator::{
         operator_response::Payload as OperatorResponsePayload,
         vault::{
-            self as proto_vault, request::Payload as VaultRequestPayload,
+            self as proto_vault,
+            rekey::{self as proto_rekey, RekeyResult as ProtoRekeyResult},
+            request::Payload as VaultRequestPayload,
             response::Payload as VaultResponsePayload,
         },
     },
@@ -33,6 +41,7 @@ pub(super) async fn dispatch(
 
     match payload {
         VaultRequestPayload::QueryState(()) => handle_query_vault_state(actor).await,
+        VaultRequestPayload::Rekey(req) => handle_rekey(actor, req).await,
         VaultRequestPayload::Unseal(_) | VaultRequestPayload::Bootstrap(_) => {
             Err(Status::permission_denied(
                 "Vault is already unsealed; unseal/bootstrap not permitted in session",
@@ -41,6 +50,51 @@ pub(super) async fn dispatch(
     }
 }
 
+async fn handle_rekey(
+    actor: &ActorRef<OperatorSession>,
+    req: proto_rekey::Request,
+) -> Result<Option<OperatorResponsePayload>, Status> {
+    use arbiter_proto::proto::operator::vault::rekey::request::Payload as RekeyPayload;
+
+    let payload = req
+        .payload
+        .ok_or_else(|| Status::invalid_argument("Missing rekey payload"))?;
+
+    let done: bool = match payload {
+        RekeyPayload::ContributePassphrase(cp) => actor
+            .ask(HandleContributeRekeyPassphrase {
+                passphrase: cp.passphrase,
+            })
+            .await
+            .map_err(|e| {
+                warn!(?e, "rekey passphrase contribution failed");
+                Status::internal("Rekey contribution failed")
+            })?,
+        RekeyPayload::ContributeRecoveryPassphrase(crp) => actor
+            .ask(HandleContributeRecoveryRekeyPassphrase {
+                recovery_operator_id: crp.recovery_operator_id,
+                passphrase: crp.passphrase,
+            })
+            .await
+            .map_err(|e| {
+                warn!(?e, "rekey recovery passphrase contribution failed");
+                Status::internal("Rekey recovery contribution failed")
+            })?,
+    };
+
+    let proto_result = if done {
+        ProtoRekeyResult::Success
+    } else {
+        ProtoRekeyResult::AwaitingContributions
+    };
+
+    Ok(Some(wrap_vault_response(VaultResponsePayload::Rekey(
+        proto_rekey::Response {
+            result: proto_result.into(),
+        },
+    ))))
+}
+
 async fn handle_query_vault_state(
     actor: &ActorRef<OperatorSession>,
 ) -> Result<Option<OperatorResponsePayload>, Status> {

server/crates/arbiter-server/src/grpc/operator/vault_gate/inbound.rs

@@ -53,6 +53,9 @@ impl TryConvert for VaultRequestPayload {
             Self::QueryState(()) => Ok(vault_gate::Inbound::HandleVaultState),
             Self::Unseal(req) => req.try_convert(),
             Self::Bootstrap(req) => req.try_convert(),
+            Self::Rekey(_) => Err(Status::permission_denied(
+                "Rekey requires an authenticated session",
+            )),
         }
     }
 }

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

@@ -335,3 +335,46 @@ impl OperatorSession {
             .unwrap_or_default()
     }
 }
+
+#[messages]
+impl OperatorSession {
+    #[message]
+    pub(crate) async fn handle_contribute_rekey_passphrase(
+        &mut self,
+        passphrase: Vec<u8>,
+    ) -> Result<bool, Error> {
+        use crate::actors::vault_coordinator::ContributeRekey;
+        use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
+
+        let operator_id = self.credentials.id;
+        self.props
+            .actors
+            .vault_coordinator
+            .ask(ContributeRekey {
+                operator_id,
+                passphrase: SafeCell::new(passphrase),
+            })
+            .await
+            .map_err(|_| Error::internal("VaultCoordinator unavailable"))
+    }
+
+    #[message]
+    pub(crate) async fn handle_contribute_recovery_rekey_passphrase(
+        &mut self,
+        recovery_operator_id: i32,
+        passphrase: Vec<u8>,
+    ) -> Result<bool, Error> {
+        use crate::actors::vault_coordinator::ContributeRecoveryRekey;
+        use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
+
+        self.props
+            .actors
+            .vault_coordinator
+            .ask(ContributeRecoveryRekey {
+                recovery_operator_id,
+                passphrase: SafeCell::new(passphrase),
+            })
+            .await
+            .map_err(|_| Error::internal("VaultCoordinator unavailable"))
+    }
+}

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

@@ -733,7 +733,7 @@ async fn approve_one_off_transaction_stores_result() {
 }
 
 #[tokio::test]
-async fn replace_operator_inserts_identity_row() {
+async fn replace_operator_updates_pubkey_and_starts_rekey() {
     let db = db::create_test_pool().await;
     let actors = GlobalActors::spawn(db.clone()).await.unwrap();
     actors
@@ -751,7 +751,7 @@ async fn replace_operator_inserts_identity_row() {
     let proposal_id = actors
         .proposal_manager
         .ask(CreateProposal {
-            kind: ProposalKind::ReplaceOperator { new_pubkey },
+            kind: ProposalKind::ReplaceOperator { old_operator_id: op_id, new_pubkey: new_pubkey.clone() },
             initiator_id: op_id,
             ttl_secs: None,
         })
@@ -774,12 +774,22 @@ async fn replace_operator_inserts_identity_row() {
     assert_eq!(outcome, VoteOutcome::QuorumApproved);
 
     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, 2); // original + new
+    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]
@@ -843,7 +853,7 @@ async fn key_rotation_requires_full_quorum() {
     let proposal_id = actors
         .proposal_manager
         .ask(CreateProposal {
-            kind: ProposalKind::ReplaceOperator { new_pubkey },
+            kind: ProposalKind::ReplaceOperator { old_operator_id: 1, new_pubkey },
             initiator_id: op1,
             ttl_secs: None,
         })
@@ -925,7 +935,7 @@ async fn recovery_vote_rejected_when_sleeping() {
     let proposal_id = actors
         .proposal_manager
         .ask(CreateProposal {
-            kind: ProposalKind::ReplaceOperator { new_pubkey },
+            kind: ProposalKind::ReplaceOperator { old_operator_id: 1, new_pubkey },
             initiator_id: op_id,
             ttl_secs: None,
         })
@@ -1072,7 +1082,7 @@ async fn recovery_operator_vote_contributes_to_replace_quorum() {
     let proposal_id = actors
         .proposal_manager
         .ask(CreateProposal {
-            kind: ProposalKind::ReplaceOperator { new_pubkey },
+            kind: ProposalKind::ReplaceOperator { old_operator_id: 1, new_pubkey },
             initiator_id: op_id,
             ttl_secs: None,
         })