feat(server): ProposalKind::ReplaceOperator

protobufs/operator/governance.proto

@@ -15,10 +15,15 @@ message CreateProposalRequest {
     ApproveSdkClientPayload    approve_sdk_client    = 1;
     GrantWalletAccessPayload   grant_wallet_access   = 3;
     ApproveServerUpdatePayload approve_server_update = 4;
+    ReplaceOperatorPayload     replace_operator      = 5;
   }
   optional uint32 ttl_secs = 2;
 }
 
+message ReplaceOperatorPayload {
+  bytes new_pubkey = 1;
+}
+
 message ApproveServerUpdatePayload {}
 
 message ApproveSdkClientPayload {

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

@@ -19,6 +19,7 @@ pub enum ProposalKind {
     ApproveSdkClient { client_id: i32 },
     GrantWalletAccess { wallet_id: i32, client_id: i32 },
     ApproveServerUpdate,
+    ReplaceOperator { new_pubkey: Vec<u8> },
 }
 
 impl ProposalKind {
@@ -27,6 +28,7 @@ impl ProposalKind {
             Self::ApproveSdkClient { .. } => "approve_sdk_client",
             Self::GrantWalletAccess { .. } => "grant_wallet_access",
             Self::ApproveServerUpdate => "approve_server_update",
+            Self::ReplaceOperator { .. } => "replace_operator",
         }
     }
 
@@ -40,6 +42,14 @@ impl ProposalKind {
                 buf
             }
             Self::ApproveServerUpdate => vec![],
+            Self::ReplaceOperator { new_pubkey } => {
+                #[expect(clippy::cast_possible_truncation, reason = "pubkey is always 32 bytes")]
+                let len = new_pubkey.len() as u32;
+                let mut buf = Vec::with_capacity(4 + new_pubkey.len());
+                buf.extend_from_slice(&len.to_be_bytes());
+                buf.extend_from_slice(new_pubkey);
+                buf
+            }
         }
     }
 
@@ -61,6 +71,19 @@ impl ProposalKind {
                 })
             }
             "approve_server_update" => Ok(Self::ApproveServerUpdate),
+            "replace_operator" => {
+                let (len_bytes, rest) = payload
+                    .split_first_chunk::<4>()
+                    .ok_or_else(|| "replace_operator payload too short".to_owned())?;
+                let len = u32::from_be_bytes(*len_bytes);
+                let len = usize::try_from(len).unwrap_or(usize::MAX);
+                if rest.len() < len {
+                    return Err("replace_operator payload truncated".to_owned());
+                }
+                Ok(Self::ReplaceOperator {
+                    new_pubkey: rest[..len].to_vec(),
+                })
+            }
             other => Err(format!("unknown proposal kind: {other}")),
         }
     }
@@ -389,6 +412,9 @@ impl ProposalManager {
                 self.execute_grant_wallet_access(wallet_id, client_id).await
             }
             ProposalKind::ApproveServerUpdate => Ok(()),
+            ProposalKind::ReplaceOperator { new_pubkey } => {
+                self.execute_replace_operator(new_pubkey).await
+            }
         }
     }
 
@@ -409,6 +435,16 @@ impl ProposalManager {
         Ok(())
     }
 
+    async fn execute_replace_operator(&self, 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))
+            .execute(&mut conn)
+            .await
+            .map_err(|e| Error::ExecutionFailed(format!("replace operator: {e}")))?;
+        Ok(())
+    }
+
     async fn execute_approve_sdk_client(&self, client_id: i32) -> Result<(), Error> {
         use arbiter_crypto::authn;
         use crate::{

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

@@ -53,6 +53,9 @@ async fn handle_create(
             client_id: p.client_id,
         },
         Some(ProtoKind::ApproveServerUpdate(_)) => ProposalKind::ApproveServerUpdate,
+        Some(ProtoKind::ReplaceOperator(p)) => ProposalKind::ReplaceOperator {
+            new_pubkey: p.new_pubkey,
+        },
         None => return Err(Status::invalid_argument("Missing proposal kind")),
     };
     let ttl_secs = req.ttl_secs.map(i64::from);

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

@@ -500,6 +500,56 @@ async fn grant_wallet_access_on_quorum_approval() {
     assert_eq!(count, 1);
 }
 
+#[tokio::test]
+async fn replace_operator_inserts_identity_row() {
+    let db = db::create_test_pool().await;
+    let actors = GlobalActors::spawn(db.clone()).await.unwrap();
+    actors
+        .vault
+        .ask(Bootstrap { seal_key: KeyCell::from([0u8; 32]) })
+        .await
+        .unwrap();
+
+    let signing_key = authn::SigningKey::generate();
+    let op_id = register_operator(&db, &signing_key.public_key()).await;
+
+    let new_op_key = authn::SigningKey::generate();
+    let new_pubkey = new_op_key.public_key().to_bytes().to_vec();
+
+    let proposal_id = actors
+        .proposal_manager
+        .ask(CreateProposal {
+            kind: ProposalKind::ReplaceOperator { new_pubkey },
+            initiator_id: op_id,
+            ttl_secs: None,
+        })
+        .await
+        .unwrap();
+
+    let msg = make_vote_message(proposal_id, true);
+    let sig = signing_key.sign_message(&msg, GOVERNANCE_CONTEXT).unwrap();
+    let outcome = actors
+        .proposal_manager
+        .ask(CastVote {
+            proposal_id,
+            operator_id: op_id,
+            approve: true,
+            signature: sig.to_bytes(),
+        })
+        .await
+        .unwrap();
+
+    assert_eq!(outcome, VoteOutcome::QuorumApproved);
+
+    let mut conn = db.get().await.unwrap();
+    let count: i64 = operator_identity::table
+        .count()
+        .get_result(&mut conn)
+        .await
+        .unwrap();
+    assert_eq!(count, 2); // original + new
+}
+
 #[tokio::test]
 async fn approve_server_update_reaches_quorum() {
     let db = db::create_test_pool().await;