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,13 +12,26 @@ message Request {
 
 message CreateProposalRequest {
   oneof kind {
-    ApproveSdkClientPayload    approve_sdk_client    = 1;
-    GrantWalletAccessPayload   grant_wallet_access   = 3;
-    ApproveServerUpdatePayload approve_server_update = 4;
+    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 {
+  int32 old_operator_id = 1;
+  bytes new_pubkey      = 2;
+}
+
+message UpdateShamirParametersPayload {
+  uint32 new_n = 1;
+}
+
 message ApproveServerUpdatePayload {}
 
 message ApproveSdkClientPayload {
@@ -40,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;
   }
 }
 
@@ -55,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 {
@@ -73,3 +86,51 @@ message ProposalSummary {
 message QueryPendingResponse {
   repeated ProposalSummary proposals = 1;
 }
+
+message TransactionRateLimitProto {
+  uint32 count       = 1;
+  int64  window_secs = 2;
+}
+
+message VolumeLimitProto {
+  bytes max_volume  = 1;
+  int64 window_secs = 2;
+}
+
+message EtherTransferSpecProto {
+  repeated bytes   targets = 1;
+  VolumeLimitProto limit   = 2;
+}
+
+message TokenTransferSpecProto {
+  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;
+  bytes  max_priority_fee_per_gas = 7;
+  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;
+  oneof specific {
+    EtherTransferSpecProto ether_transfer = 8;
+    TokenTransferSpecProto token_transfer = 9;
+  }
+}

protobufs/operator/vault/bootstrap.proto

@@ -9,13 +9,19 @@ message BootstrapEncryptedKey {
 }
 
 message DeclareCommittee {
-  uint32 count = 1;
+  uint32 count          = 1;
+  uint32 recovery_count = 2;
 }
 
 message ContributePassphrase {
   bytes passphrase = 1;
 }
 
+message ContributeRecoveryPassphrase {
+  int32 recovery_operator_id = 1;
+  bytes passphrase            = 2;
+}
+
 enum BootstrapResult {
   BOOTSTRAP_RESULT_UNSPECIFIED = 0;
   BOOTSTRAP_RESULT_SUCCESS = 1;
@@ -26,9 +32,10 @@ enum BootstrapResult {
 
 message Request {
   oneof payload {
-    BootstrapEncryptedKey encrypted_key = 2;
-    DeclareCommittee declare_committee = 3;
-    ContributePassphrase contribute_passphrase = 4;
+    BootstrapEncryptedKey        encrypted_key                    = 2;
+    DeclareCommittee             declare_committee                = 3;
+    ContributePassphrase         contribute_passphrase            = 4;
+    ContributeRecoveryPassphrase contribute_recovery_passphrase   = 5;
   }
 }
 

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/unseal.proto

@@ -19,6 +19,11 @@ message ContributePassphrase {
   bytes passphrase = 1;
 }
 
+message ContributeRecoveryPassphrase {
+  int32 recovery_operator_id = 1;
+  bytes passphrase            = 2;
+}
+
 enum UnsealResult {
   UNSEAL_RESULT_UNSPECIFIED = 0;
   UNSEAL_RESULT_SUCCESS = 1;
@@ -29,9 +34,10 @@ enum UnsealResult {
 
 message Request {
   oneof payload {
-    UnsealStart start = 1;
-    UnsealEncryptedKey encrypted_key = 2;
-    ContributePassphrase contribute_passphrase = 3;
+    UnsealStart                  start                          = 1;
+    UnsealEncryptedKey           encrypted_key                  = 2;
+    ContributePassphrase         contribute_passphrase          = 3;
+    ContributeRecoveryPassphrase contribute_recovery_passphrase = 4;
   }
 }
 

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/Cargo.lock

@@ -769,6 +769,7 @@ dependencies = [
  "mutants",
  "pem",
  "proptest",
+ "prost",
  "prost-types",
  "rand 0.10.1",
  "rand_core 0.6.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/Cargo.toml

@@ -42,6 +42,7 @@ pem = "3.0.6"
 sha2.workspace = true
 hmac.workspace = true
 alloy.workspace = true
+prost.workspace = true
 prost-types.workspace = true
 arbiter-tokens-registry.path = "../arbiter-tokens-registry"
 anyhow = "1.0.102"

server/crates/arbiter-server/migrations/2026-02-14-171124-0000_init/up.sql

@@ -237,3 +237,52 @@ create table if not exists proposal_vote (
     voted_at    integer not null default(unixepoch('now')),
     unique (proposal_id, operator_id)
 ) STRICT;
+
+
+create table if not exists proposal_result (
+    proposal_id integer not null primary key references proposal(id) on delete cascade,
+    data        blob    not null,
+    created_at  integer not null default(unixepoch('now'))
+) STRICT;
+
+-- ===============================
+-- Recovery Operators (§3.4/§3.5/§3.6)
+-- ===============================
+
+-- Encrypted Shamir shares for recovery operators (mirrors the `operator` table).
+create table if not exists recovery_operator (
+    id          integer not null primary key references recovery_operator_identity(id) on delete restrict,
+    share       blob    not null,
+    share_nonce blob    not null,
+    share_salt  blob    not null,
+    created_at  integer not null default(unixepoch('now')),
+    updated_at  integer not null default(unixepoch('now'))
+) STRICT;
+
+create table if not exists recovery_operator_identity (
+    id         integer not null primary key,
+    public_key blob    not null unique,
+    created_at integer not null default(unixepoch('now')),
+    updated_at integer not null default(unixepoch('now'))
+) STRICT;
+
+-- One active wakeup request at a time. A request is pending when cancelled_at IS NULL
+-- and requested_at + 14 days > now. It becomes active (recovery live) after 14 days.
+create table if not exists recovery_wakeup_request (
+    id           integer not null primary key,
+    requested_by integer not null references operator_identity(id) on delete restrict,
+    requested_at integer not null default(unixepoch('now')),
+    cancelled_by integer references operator_identity(id) on delete restrict,
+    cancelled_at integer
+) STRICT;
+
+-- Votes cast by recovery operators; only allowed on replace_operator proposals.
+create table if not exists recovery_proposal_vote (
+    id                   integer not null primary key,
+    proposal_id          integer not null references proposal(id) on delete cascade,
+    recovery_operator_id integer not null references recovery_operator_identity(id) on delete restrict,
+    approve              integer not null check (approve in (0, 1)),
+    signature            blob    not null,
+    voted_at             integer not null default(unixepoch('now')),
+    unique (proposal_id, recovery_operator_id)
+) STRICT;

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

@@ -50,23 +50,27 @@ impl GlobalActors {
         let message_bus = Self::spawn_message_bus();
         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?),
-            evm: EvmActor::spawn(EvmActor::new(key_holder.clone(), db.clone())),
-            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(),
             )),
             operator_registry,
             events: message_bus,
+            evm,
         })
     }
 }

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

@@ -1,8 +1,15 @@
 use crate::{
-    actors::vault::Vault,
+    actors::{
+        evm::EvmActor,
+        vault::Vault,
+        vault_coordinator::{StartRekey, VaultCoordinator},
+    },
     db::{
         self,
-        models::{NewProposal, NewProposalVote, Proposal, ProposalStatus, SqliteTimestamp},
+        models::{
+            NewProposal, NewProposalVote, NewRecoveryProposalVote, NewRecoveryWakeupRequest,
+            Proposal, ProposalStatus, SqliteTimestamp,
+        },
         schema,
     },
 };
@@ -10,49 +17,117 @@ 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 {
+        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::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(),
-            Self::GrantWalletAccess { wallet_id, client_id } => {
+            Self::GrantWalletAccess {
+                wallet_id,
+                client_id,
+            } => {
                 let mut buf = Vec::with_capacity(8);
                 buf.extend_from_slice(&wallet_id.to_be_bytes());
                 buf.extend_from_slice(&client_id.to_be_bytes());
                 buf
             }
             Self::ApproveServerUpdate => vec![],
+            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 + 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 }
+            | 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.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 {
@@ -60,8 +135,38 @@ impl ProposalKind {
                     client_id: i32::from_be_bytes(bytes[4..].try_into().unwrap()),
                 })
             }
-            "approve_server_update" => Ok(Self::ApproveServerUpdate),
-            other => Err(format!("unknown proposal kind: {other}")),
+            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);
+                let len = usize::try_from(len).unwrap_or(usize::MAX);
+                let new_pubkey = rest
+                    .get(..len)
+                    .ok_or_else(|| "replace_operator payload truncated".to_owned())?
+                    .to_vec();
+                Ok(Self::ReplaceOperator {
+                    old_operator_id,
+                    new_pubkey,
+                })
+            }
+            ProposalKindTag::UpdateShamirParameters => {
+                let &[new_n] = payload else {
+                    return Err("invalid payload for update_shamir_parameters".to_owned());
+                };
+                Ok(Self::UpdateShamirParameters { new_n })
+            }
+            ProposalKindTag::ApprovePersistentGrant => Ok(Self::ApprovePersistentGrant {
+                payload_bytes: payload.to_vec(),
+            }),
+            ProposalKindTag::ApproveOneOffTransaction => Ok(Self::ApproveOneOffTransaction {
+                payload_bytes: payload.to_vec(),
+            }),
         }
     }
 }
@@ -91,6 +196,14 @@ pub enum Error {
     DatabaseQuery(#[from] diesel::result::Error),
     #[error("Execution failed: {0}")]
     ExecutionFailed(String),
+    #[error("Recovery operators are sleeping")]
+    RecoveryNotActive,
+    #[error("Recovery operators may only vote on operator replacement")]
+    NotAllowedForRecoveryOperator,
+    #[error("A recovery wake-up is already pending or active")]
+    WakeupAlreadyPending,
+    #[error("No active recovery wake-up to cancel")]
+    NoActiveWakeup,
 }
 
 #[derive(Debug)]
@@ -106,11 +219,23 @@ pub struct ProposalSummary {
 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 {
-    pub const fn new(db: db::DatabasePool, vault: ActorRef<Vault>) -> Self {
-        Self { db, vault }
+    pub const fn new(
+        db: db::DatabasePool,
+        vault: ActorRef<Vault>,
+        evm: ActorRef<EvmActor>,
+        vault_coordinator: ActorRef<VaultCoordinator>,
+    ) -> Self {
+        Self {
+            db,
+            vault,
+            evm,
+            vault_coordinator,
+        }
     }
 }
 
@@ -118,10 +243,7 @@ impl kameo::Actor for ProposalManager {
     type Args = Self;
     type Error = ();
 
-    async fn on_start(
-        args: Self::Args,
-        actor_ref: ActorRef<Self>,
-    ) -> Result<Self, Self::Error> {
+    async fn on_start(args: Self::Args, actor_ref: ActorRef<Self>) -> Result<Self, Self::Error> {
         let weak = actor_ref.downgrade();
         tokio::spawn(async move {
             loop {
@@ -326,27 +448,56 @@ impl ProposalManager {
             .count()
             .get_result(&mut conn)
             .await?;
+        let recovery_active = Self::is_recovery_active_conn(&mut conn).await?;
+        let total_recovery: i64 = if recovery_active {
+            schema::recovery_operator_identity::table
+                .count()
+                .get_result(&mut conn)
+                .await?
+        } else {
+            0
+        };
         #[expect(
             clippy::cast_possible_truncation,
             clippy::cast_sign_loss,
             clippy::as_conversions,
             reason = "operator count is always a small positive integer"
         )]
-        let threshold = crate::crypto::shamir::shamir_threshold(total_operators as usize);
+        let threshold = if ProposalKind::requires_full_quorum(&proposal.kind) {
+            // §3.3: key-rotation proposals require every eligible voter to approve
+            // §3.5: when recovery is active, recovery operators also vote on replace_operator
+            (total_operators + total_recovery) as usize
+        } else {
+            crate::crypto::shamir::shamir_threshold(total_operators as usize)
+        };
 
-        let approve_count: i64 = schema::proposal_vote::table
+        let ordinary_approve: i64 = schema::proposal_vote::table
             .filter(schema::proposal_vote::proposal_id.eq(proposal_id))
             .filter(schema::proposal_vote::approve.eq(true))
             .count()
             .get_result(&mut conn)
             .await?;
+        let recovery_approve: i64 = schema::recovery_proposal_vote::table
+            .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id))
+            .filter(schema::recovery_proposal_vote::approve.eq(true))
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        let approve_count = ordinary_approve + recovery_approve;
 
-        let reject_count: i64 = schema::proposal_vote::table
+        let ordinary_reject: i64 = schema::proposal_vote::table
             .filter(schema::proposal_vote::proposal_id.eq(proposal_id))
             .filter(schema::proposal_vote::approve.eq(false))
             .count()
             .get_result(&mut conn)
             .await?;
+        let recovery_reject: i64 = schema::recovery_proposal_vote::table
+            .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id))
+            .filter(schema::recovery_proposal_vote::approve.eq(false))
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        let reject_count = ordinary_reject + recovery_reject;
 
         #[expect(
             clippy::cast_possible_wrap,
@@ -365,7 +516,183 @@ impl ProposalManager {
             return Ok(VoteOutcome::QuorumApproved);
         }
 
-        if reject_count > total_operators - threshold_i64 {
+        let total_eligible = total_operators + total_recovery;
+        if reject_count > total_eligible - threshold_i64 {
+            diesel::update(schema::proposal::table.find(proposal_id))
+                .set(schema::proposal::status.eq(ProposalStatus::Rejected))
+                .execute(&mut conn)
+                .await?;
+            return Ok(VoteOutcome::QuorumRejected);
+        }
+
+        Ok(VoteOutcome::Pending)
+    }
+
+    /// §3.6: Any ordinary operator may request recovery wake-up.
+    /// Fails if a wake-up is already pending or active.
+    #[message]
+    pub async fn request_recovery_wakeup(&mut self, operator_id: i32) -> Result<(), Error> {
+        let mut conn = self.db.get().await?;
+        if Self::has_uncancelled_wakeup(&mut conn).await? {
+            return Err(Error::WakeupAlreadyPending);
+        }
+        diesel::insert_into(schema::recovery_wakeup_request::table)
+            .values(&NewRecoveryWakeupRequest {
+                requested_by: operator_id,
+            })
+            .execute(&mut conn)
+            .await?;
+        Ok(())
+    }
+
+    /// §3.6: Any ordinary operator may cancel a pending wake-up request.
+    /// Fails if there is no uncancelled request.
+    #[message]
+    pub async fn cancel_recovery_wakeup(&mut self, operator_id: i32) -> Result<(), Error> {
+        let mut conn = self.db.get().await?;
+        let rows_updated = diesel::update(schema::recovery_wakeup_request::table)
+            .filter(schema::recovery_wakeup_request::cancelled_at.is_null())
+            .set((
+                schema::recovery_wakeup_request::cancelled_by.eq(Some(operator_id)),
+                schema::recovery_wakeup_request::cancelled_at.eq(Some(SqliteTimestamp::now())),
+            ))
+            .execute(&mut conn)
+            .await?;
+        if rows_updated == 0 {
+            return Err(Error::NoActiveWakeup);
+        }
+        Ok(())
+    }
+
+    /// §3.5: Recovery operators may only vote on operator replacement proposals.
+    /// §3.6: Voting is gated behind recovery being active (14-day window elapsed).
+    #[message]
+    pub async fn cast_recovery_vote(
+        &mut self,
+        proposal_id: i32,
+        recovery_operator_id: i32,
+        approve: bool,
+        signature: Vec<u8>,
+    ) -> Result<VoteOutcome, Error> {
+        use arbiter_crypto::authn::{self, GOVERNANCE_CONTEXT};
+
+        let mut conn = self.db.get().await?;
+
+        let proposal: Proposal = schema::proposal::table
+            .find(proposal_id)
+            .first(&mut conn)
+            .await
+            .map_err(|e| match e {
+                diesel::result::Error::NotFound => Error::ProposalNotFound,
+                other => Error::DatabaseQuery(other),
+            })?;
+
+        if proposal.kind.parse::<ProposalKindTag>() != Ok(ProposalKindTag::ReplaceOperator) {
+            return Err(Error::NotAllowedForRecoveryOperator);
+        }
+
+        if !Self::is_recovery_active_conn(&mut conn).await? {
+            return Err(Error::RecoveryNotActive);
+        }
+
+        let existing: i64 = schema::recovery_proposal_vote::table
+            .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id))
+            .filter(schema::recovery_proposal_vote::recovery_operator_id.eq(recovery_operator_id))
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        if existing > 0 {
+            return Err(Error::AlreadyVoted);
+        }
+
+        if proposal.status != ProposalStatus::Pending {
+            return Err(Error::ProposalNotPending);
+        }
+
+        let pubkey_bytes: Vec<u8> = schema::recovery_operator_identity::table
+            .find(recovery_operator_id)
+            .select(schema::recovery_operator_identity::public_key)
+            .first(&mut conn)
+            .await
+            .map_err(|e| match e {
+                diesel::result::Error::NotFound => Error::OperatorNotFound,
+                other => Error::DatabaseQuery(other),
+            })?;
+
+        let pubkey = authn::PublicKey::try_from(pubkey_bytes.as_slice())
+            .map_err(|()| Error::InvalidSignature)?;
+
+        let mut vote_msg = Vec::with_capacity(9);
+        vote_msg.extend_from_slice(&i64::from(proposal_id).to_be_bytes());
+        vote_msg.push(u8::from(approve));
+
+        let auth_sig = authn::Signature::try_from(signature.as_slice())
+            .map_err(|()| Error::InvalidSignature)?;
+
+        if !pubkey.verify_message(&vote_msg, GOVERNANCE_CONTEXT, &auth_sig) {
+            return Err(Error::InvalidSignature);
+        }
+
+        diesel::insert_into(schema::recovery_proposal_vote::table)
+            .values(&NewRecoveryProposalVote {
+                proposal_id,
+                recovery_operator_id,
+                approve,
+                signature,
+            })
+            .execute(&mut conn)
+            .await?;
+
+        // Quorum: all ordinary + all recovery operators must approve (§3.3 + §3.5)
+        let total_ordinary: i64 = schema::operator_identity::table
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        let total_recovery: i64 = schema::recovery_operator_identity::table
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        let threshold_i64 = total_ordinary + total_recovery;
+
+        let ordinary_approve: i64 = schema::proposal_vote::table
+            .filter(schema::proposal_vote::proposal_id.eq(proposal_id))
+            .filter(schema::proposal_vote::approve.eq(true))
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        let recovery_approve: i64 = schema::recovery_proposal_vote::table
+            .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id))
+            .filter(schema::recovery_proposal_vote::approve.eq(true))
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        let approve_count = ordinary_approve + recovery_approve;
+
+        if approve_count >= threshold_i64 {
+            diesel::update(schema::proposal::table.find(proposal_id))
+                .set(schema::proposal::status.eq(ProposalStatus::Approved))
+                .execute(&mut conn)
+                .await?;
+            drop(conn);
+            self.execute_proposal(&proposal).await?;
+            return Ok(VoteOutcome::QuorumApproved);
+        }
+
+        let recovery_reject: i64 = schema::recovery_proposal_vote::table
+            .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id))
+            .filter(schema::recovery_proposal_vote::approve.eq(false))
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        let ordinary_reject: i64 = schema::proposal_vote::table
+            .filter(schema::proposal_vote::proposal_id.eq(proposal_id))
+            .filter(schema::proposal_vote::approve.eq(false))
+            .count()
+            .get_result(&mut conn)
+            .await?;
+        let reject_count = ordinary_reject + recovery_reject;
+
+        if reject_count > threshold_i64 - approve_count - reject_count {
             diesel::update(schema::proposal::table.find(proposal_id))
                 .set(schema::proposal::status.eq(ProposalStatus::Rejected))
                 .execute(&mut conn)
@@ -378,6 +705,36 @@ impl ProposalManager {
 }
 
 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> {
+        let count: i64 = schema::recovery_wakeup_request::table
+            .filter(schema::recovery_wakeup_request::cancelled_at.is_null())
+            .filter(
+                schema::recovery_wakeup_request::requested_at.le(diesel::dsl::sql::<
+                    diesel::sql_types::Integer,
+                >(&format!(
+                    "unixepoch('now') - {}",
+                    Self::WAKEUP_DELAY_SECS
+                ))),
+            )
+            .count()
+            .get_result(conn)
+            .await?;
+        Ok(count > 0)
+    }
+
+    /// Returns true when there is any uncancelled wakeup request (pending or active).
+    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()
+            .get_result(conn)
+            .await?;
+        Ok(count > 0)
+    }
+
     async fn execute_proposal(&self, proposal: &Proposal) -> Result<(), Error> {
         let kind = ProposalKind::decode(&proposal.kind, &proposal.payload)
             .map_err(Error::ExecutionFailed)?;
@@ -385,14 +742,36 @@ impl ProposalManager {
             ProposalKind::ApproveSdkClient { client_id } => {
                 self.execute_approve_sdk_client(client_id).await
             }
-            ProposalKind::GrantWalletAccess { wallet_id, client_id } => {
-                self.execute_grant_wallet_access(wallet_id, client_id).await
-            }
+            ProposalKind::GrantWalletAccess {
+                wallet_id,
+                client_id,
+            } => self.execute_grant_wallet_access(wallet_id, client_id).await,
             ProposalKind::ApproveServerUpdate => Ok(()),
+            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).await
+            }
+            ProposalKind::ApprovePersistentGrant { payload_bytes } => {
+                self.execute_approve_persistent_grant(payload_bytes).await
+            }
+            ProposalKind::ApproveOneOffTransaction { payload_bytes } => {
+                self.execute_approve_one_off_transaction(proposal.id, payload_bytes)
+                    .await
+            }
         }
     }
 
-    async fn execute_grant_wallet_access(&self, wallet_id: i32, client_id: i32) -> Result<(), Error> {
+    async fn execute_grant_wallet_access(
+        &self,
+        wallet_id: i32,
+        client_id: i32,
+    ) -> Result<(), Error> {
         use crate::db::models::EvmWalletId;
 
         let mut conn = self.db.get().await.map_err(Error::DatabaseConnection)?;
@@ -409,12 +788,204 @@ impl ProposalManager {
         Ok(())
     }
 
-    async fn execute_approve_sdk_client(&self, client_id: i32) -> Result<(), Error> {
-        use arbiter_crypto::authn;
-        use crate::{
-            crypto::integrity,
-            peers::client::ClientCredentials,
+    /// 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::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_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(())
+    }
+
+    async fn execute_approve_one_off_transaction(
+        &self,
+        proposal_id: i32,
+        payload_bytes: Vec<u8>,
+    ) -> Result<(), Error> {
+        use crate::actors::evm::ClientSignTransaction;
+        use crate::db::models::NewProposalResult;
+        use alloy::{
+            consensus::TxEip1559,
+            eips::eip2930::AccessList,
+            primitives::{Address, Bytes, TxKind, U256},
         };
+        use arbiter_proto::proto::operator::governance::ApproveOneOffTransactionPayload;
+        use prost::Message as _;
+
+        let p = ApproveOneOffTransactionPayload::decode(payload_bytes.as_slice())
+            .map_err(|e| Error::ExecutionFailed(format!("decode one-off tx payload: {e}")))?;
+
+        let wallet_address = Address::from_slice(p.wallet_address.as_slice());
+        let to = Address::from_slice(p.to.as_slice());
+
+        let transaction = TxEip1559 {
+            chain_id: p.chain_id,
+            nonce: p.nonce,
+            gas_limit: p.gas_limit,
+            max_fee_per_gas: u128::from_be_bytes(
+                p.max_fee_per_gas
+                    .as_slice()
+                    .try_into()
+                    .map_err(|_| Error::ExecutionFailed("invalid max_fee_per_gas".to_owned()))?,
+            ),
+            max_priority_fee_per_gas: u128::from_be_bytes(
+                p.max_priority_fee_per_gas
+                    .as_slice()
+                    .try_into()
+                    .map_err(|_| {
+                        Error::ExecutionFailed("invalid max_priority_fee_per_gas".to_owned())
+                    })?,
+            ),
+            to: TxKind::Call(to),
+            value: U256::from_be_slice(p.value.as_slice()),
+            input: Bytes::from(p.input),
+            access_list: AccessList::default(),
+        };
+
+        let sig = self
+            .evm
+            .ask(ClientSignTransaction {
+                client_id: p.client_id,
+                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)?;
+        diesel::insert_into(schema::proposal_result::table)
+            .values(NewProposalResult {
+                proposal_id,
+                data: sig.as_bytes().to_vec(),
+            })
+            .execute(&mut conn)
+            .await
+            .map_err(|e| Error::ExecutionFailed(format!("store proposal result: {e}")))?;
+
+        Ok(())
+    }
+
+    async fn execute_approve_persistent_grant(&self, payload_bytes: Vec<u8>) -> Result<(), Error> {
+        use crate::{
+            actors::evm::OperatorCreateGrant,
+            evm::policies::{
+                SharedGrantSettings, SpecificGrant, TransactionRateLimit, VolumeRateLimit,
+                ether_transfer, token_transfers,
+            },
+        };
+        use alloy::primitives::{Address, U256};
+        use arbiter_proto::proto::operator::governance::{
+            ApprovePersistentGrantPayload, approve_persistent_grant_payload::Specific,
+        };
+        use chrono::Duration;
+        use prost::Message as _;
+
+        let payload = ApprovePersistentGrantPayload::decode(payload_bytes.as_slice())
+            .map_err(|e| Error::ExecutionFailed(format!("decode grant payload: {e}")))?;
+
+        let basic = SharedGrantSettings {
+            wallet_access_id: payload.wallet_access_id,
+            chain: payload.chain_id,
+            valid_from: payload
+                .valid_from_secs
+                .and_then(|s| chrono::DateTime::from_timestamp(s, 0)),
+            valid_until: payload
+                .valid_until_secs
+                .and_then(|s| chrono::DateTime::from_timestamp(s, 0)),
+            max_gas_fee_per_gas: payload
+                .max_gas_fee_per_gas
+                .map(|b| U256::from_be_slice(b.as_slice())),
+            max_priority_fee_per_gas: payload
+                .max_priority_fee_per_gas
+                .map(|b| U256::from_be_slice(b.as_slice())),
+            rate_limit: payload.rate_limit.map(|r| TransactionRateLimit {
+                count: r.count,
+                window: Duration::seconds(r.window_secs),
+            }),
+        };
+
+        let grant = match payload.specific {
+            Some(Specific::EtherTransfer(spec)) => {
+                let target: Vec<Address> = spec
+                    .targets
+                    .iter()
+                    .map(|b| Address::from_slice(b.as_slice()))
+                    .collect();
+                let limit = spec
+                    .limit
+                    .map(|l| VolumeRateLimit {
+                        max_volume: U256::from_be_slice(l.max_volume.as_slice()),
+                        window: Duration::seconds(l.window_secs),
+                    })
+                    .ok_or_else(|| {
+                        Error::ExecutionFailed("missing ether transfer limit".to_owned())
+                    })?;
+                SpecificGrant::EtherTransfer(ether_transfer::Settings { target, limit })
+            }
+            Some(Specific::TokenTransfer(spec)) => {
+                let token_contract = Address::from_slice(spec.token_contract.as_slice());
+                let target = spec.target.map(|b| Address::from_slice(b.as_slice()));
+                let volume_limits: Vec<VolumeRateLimit> = spec
+                    .volume_limits
+                    .iter()
+                    .map(|l| VolumeRateLimit {
+                        max_volume: U256::from_be_slice(l.max_volume.as_slice()),
+                        window: Duration::seconds(l.window_secs),
+                    })
+                    .collect();
+                SpecificGrant::TokenTransfer(token_transfers::Settings {
+                    token_contract,
+                    target,
+                    volume_limits,
+                })
+            }
+            None => return Err(Error::ExecutionFailed("missing grant specific".to_owned())),
+        };
+
+        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)?;
 

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")]
@@ -39,6 +41,8 @@ pub enum Error {
     Encryption,
     #[error("Vault error")]
     VaultError,
+    #[error("Two-operator vaults require at least one recovery share")]
+    TwoOperatorsRequireRecovery,
     #[error("Broken database")]
     BrokenDatabase,
 }
@@ -49,11 +53,23 @@ enum CoordinatorState {
     Idle,
     Bootstrapping {
         declared_count: usize,
+        recovery_count: usize,
         passphrases: HashMap<i32, Vec<u8>>,
+        recovery_passphrases: HashMap<i32, Vec<u8>>,
     },
     Unsealing {
         threshold: usize,
+        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>>,
     },
 }
 
@@ -76,42 +92,84 @@ impl VaultCoordinator {
 
 const SHARE_AAD: &[u8] = b"arbiter/shamir-share/v1";
 
+fn encrypt_share(
+    passphrase_bytes: Vec<u8>,
+    share: &[u8],
+) -> Result<(Vec<u8>, Vec<u8>, Vec<u8>), Error> {
+    let mut share_salt = vec![0u8; 32];
+    OsRng.fill_bytes(&mut share_salt);
+
+    let mut passphrase_cell = SafeCell::new(passphrase_bytes);
+    let mut share_seal_key = derive_key(&mut passphrase_cell, &share_salt);
+
+    let nonce = Nonce::default();
+    let encrypted_share = share_seal_key
+        .encrypt(&nonce, SHARE_AAD, share)
+        .map_err(|_| Error::Encryption)?;
+
+    Ok((encrypted_share, nonce.to_vec(), share_salt))
+}
+
+fn decrypt_share(
+    passphrase_bytes: Vec<u8>,
+    encrypted_share: Vec<u8>,
+    share_nonce_bytes: &[u8],
+    share_salt: &[u8],
+    operator_id: i32,
+) -> Result<Vec<u8>, Error> {
+    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_buffer = SafeCell::new(encrypted_share);
+    share_seal_key
+        .decrypt_in_place(&nonce, SHARE_AAD, &mut share_buffer)
+        .map_err(|_| Error::InvalidPassphrase)?;
+
+    Ok(share_buffer.read().clone())
+}
+
+/// §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,
+/// so each share is the seal key itself — any single participant can reconstruct.
 async fn finalize_bootstrap(
     db: db::DatabasePool,
     vault: ActorRef<Vault>,
-    passphrases: HashMap<i32, Vec<u8>>,
+    ordinary_passphrases: HashMap<i32, Vec<u8>>,
+    recovery_passphrases: HashMap<i32, Vec<u8>>,
 ) -> Result<(), Error> {
-    let total = passphrases.len();
-    let threshold = shamir_threshold(total);
+    let ordinary_count = ordinary_passphrases.len();
+    let recovery_count = recovery_passphrases.len();
+    let total = ordinary_count + recovery_count;
+    let threshold = shamir_threshold(ordinary_count);
 
-    // Generate random 32-byte seal key
     let mut seal_key_bytes = [0u8; 32];
     OsRng.fill_bytes(&mut seal_key_bytes);
 
-    // Split seal key into shares using Shamir (OsRng from rand_core 0.6, compatible with vsss-rs)
-    let shares = shamir::split_key(threshold, total, &seal_key_bytes, OsRng)
-        .map_err(|e| Error::Shamir(e.to_string()))?;
+    // threshold == 1 means any single share reconstructs the key (degenerate split).
+    // vsss-rs requires threshold >= 2, so we store the key directly in this case.
+    let shares: Vec<Vec<u8>> = if threshold >= 2 {
+        shamir::split_key(threshold, total, &seal_key_bytes, OsRng)
+            .map_err(|e| Error::Shamir(e.to_string()))?
+    } else {
+        std::iter::repeat_with(|| seal_key_bytes.to_vec()).take(total).collect()
+    };
 
     let seal_key = KeyCell::from(seal_key_bytes);
 
     let mut conn = db.get().await?;
+    let mut shares_iter = shares.into_iter();
 
-    for ((operator_id_raw, passphrase_bytes), share) in passphrases.into_iter().zip(shares) {
-        // Generate a fresh share_salt for this operator
-        let mut share_salt = vec![0u8; 32];
-        OsRng.fill_bytes(&mut share_salt);
-
-        // Derive share encryption key from passphrase + salt
-        let mut passphrase_cell = SafeCell::new(passphrase_bytes);
-        let mut share_seal_key = derive_key(&mut passphrase_cell, &share_salt);
-
-        // Encrypt this operator's share
-        let nonce = Nonce::default();
-        let encrypted_share = share_seal_key
-            .encrypt(&nonce, SHARE_AAD, &share)
-            .map_err(|_| Error::Encryption)?;
-
-        let nonce_bytes = nonce.to_vec();
+    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((
@@ -126,26 +184,52 @@ async fn finalize_bootstrap(
             .await?;
     }
 
-    vault
-        .ask(Bootstrap { seal_key })
-        .await
-        .map_err(|err| {
-            error!(?err, "Vault bootstrap failed");
-            Error::VaultError
-        })?;
+    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?;
+    }
+
+    vault.ask(Bootstrap { seal_key }).await.map_err(|err| {
+        error!(?err, "Vault bootstrap failed");
+        Error::VaultError
+    })?;
 
     Ok(())
 }
 
+/// §3.5: Unseal using any threshold-sized mix of ordinary + recovery shares.
 async fn finalize_unseal(
     db: db::DatabasePool,
     vault: ActorRef<Vault>,
-    passphrases: HashMap<i32, Vec<u8>>,
+    ordinary_passphrases: HashMap<i32, Vec<u8>>,
+    recovery_passphrases: HashMap<i32, Vec<u8>>,
 ) -> Result<(), Error> {
     let mut conn = db.get().await?;
+
+    // Determine whether shares were stored as raw keys (threshold=1) or vsss-rs splits (threshold>=2).
+    let ordinary_operator_count: i64 = schema::operator::table
+        .count()
+        .get_result(&mut conn)
+        .await?;
+    let threshold = shamir_threshold(ordinary_operator_count as usize);
+
     let mut shares: Vec<Vec<u8>> = Vec::new();
 
-    for (operator_id_raw, passphrase_bytes) in passphrases {
+    for (operator_id_raw, passphrase_bytes) in ordinary_passphrases {
         let (encrypted_share, share_nonce_bytes, share_salt): (Vec<u8>, Vec<u8>, Vec<u8>) =
             schema::operator::table
                 .filter(schema::operator::id.eq(Some(operator_id_raw)))
@@ -158,33 +242,134 @@ async fn finalize_unseal(
                 .await
                 .map_err(|_| Error::OperatorNotFound)?;
 
-        let nonce = Nonce::try_from(share_nonce_bytes.as_slice()).map_err(|()| {
-            error!(operator_id = operator_id_raw, "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_buffer = SafeCell::new(encrypted_share);
-        share_seal_key
-            .decrypt_in_place(&nonce, SHARE_AAD, &mut share_buffer)
-            .map_err(|_| Error::InvalidPassphrase)?;
-
-        let decrypted_share = share_buffer.read().clone();
-        shares.push(decrypted_share);
+        shares.push(decrypt_share(
+            passphrase_bytes,
+            encrypted_share,
+            &share_nonce_bytes,
+            &share_salt,
+            operator_id_raw,
+        )?);
     }
 
-    let seal_key_bytes =
-        shamir::combine_shares(&shares).map_err(|e| Error::Shamir(e.to_string()))?;
+    for (recovery_id_raw, passphrase_bytes) in recovery_passphrases {
+        let (encrypted_share, share_nonce_bytes, share_salt): (Vec<u8>, Vec<u8>, Vec<u8>) =
+            schema::recovery_operator::table
+                .find(recovery_id_raw)
+                .select((
+                    schema::recovery_operator::share,
+                    schema::recovery_operator::share_nonce,
+                    schema::recovery_operator::share_salt,
+                ))
+                .first(&mut conn)
+                .await
+                .map_err(|_| Error::OperatorNotFound)?;
+
+        shares.push(decrypt_share(
+            passphrase_bytes,
+            encrypted_share,
+            &share_nonce_bytes,
+            &share_salt,
+            recovery_id_raw,
+        )?);
+    }
+
+    // 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()))?
+    } 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
         })?;
 
@@ -200,20 +385,26 @@ impl VaultCoordinator {
         &mut self,
         operator_id: i32,
         declared_count: usize,
+        recovery_count: usize,
     ) -> Result<(), Error> {
         let _ = operator_id; // fixme!: any authenticated operator may announce the committee size. the first call wins
         if !matches!(self.state, CoordinatorState::Idle) {
             return Err(Error::AlreadyBootstrapping);
         }
+        if declared_count == 2 && recovery_count == 0 {
+            return Err(Error::TwoOperatorsRequireRecovery);
+        }
         self.state = CoordinatorState::Bootstrapping {
             declared_count,
+            recovery_count,
             passphrases: HashMap::new(),
+            recovery_passphrases: HashMap::new(),
         };
         Ok(())
     }
 
-    /// Phase 2 of multi-operator bootstrap: contribute a passphrase.
-    /// Returns Ok(true) when all operators contributed and bootstrap finalized.
+    /// Phase 2 of multi-operator bootstrap: ordinary operator contributes a passphrase.
+    /// Returns Ok(true) when all ordinary + recovery operators contributed and bootstrap finalized.
     #[message]
     pub async fn contribute_bootstrap(
         &mut self,
@@ -222,7 +413,9 @@ impl VaultCoordinator {
     ) -> Result<bool, Error> {
         let CoordinatorState::Bootstrapping {
             declared_count,
+            recovery_count,
             passphrases,
+            recovery_passphrases,
         } = &mut self.state
         else {
             return Err(Error::NotBootstrapping);
@@ -232,25 +425,81 @@ impl VaultCoordinator {
             return Err(Error::DuplicateContribution);
         }
 
-        // Extract bytes immediately so state stays Sync
         let passphrase_bytes = passphrase.read().to_vec();
         passphrases.insert(operator_id, passphrase_bytes);
 
-        if passphrases.len() < *declared_count {
+        if passphrases.len() < *declared_count || recovery_passphrases.len() < *recovery_count {
             return Ok(false);
         }
 
-        let CoordinatorState::Bootstrapping { passphrases, .. } =
-            std::mem::replace(&mut self.state, CoordinatorState::Idle)
+        let CoordinatorState::Bootstrapping {
+            passphrases,
+            recovery_passphrases,
+            ..
+        } = std::mem::replace(&mut self.state, CoordinatorState::Idle)
         else {
             unreachable!()
         };
 
-        finalize_bootstrap(self.db.clone(), self.vault.clone(), passphrases).await?;
+        finalize_bootstrap(
+            self.db.clone(),
+            self.vault.clone(),
+            passphrases,
+            recovery_passphrases,
+        )
+        .await?;
         Ok(true)
     }
 
-    /// Contribute a passphrase for vault unseal.
+    /// Phase 2 of multi-operator bootstrap: recovery operator contributes a passphrase.
+    /// Returns Ok(true) when all contributors are in and bootstrap finalized.
+    #[message]
+    pub async fn contribute_recovery_bootstrap(
+        &mut self,
+        recovery_operator_id: i32,
+        mut passphrase: SafeCell<Vec<u8>>,
+    ) -> Result<bool, Error> {
+        let CoordinatorState::Bootstrapping {
+            declared_count,
+            recovery_count,
+            passphrases,
+            recovery_passphrases,
+        } = &mut self.state
+        else {
+            return Err(Error::NotBootstrapping);
+        };
+
+        if recovery_passphrases.contains_key(&recovery_operator_id) {
+            return Err(Error::DuplicateContribution);
+        }
+
+        let passphrase_bytes = passphrase.read().to_vec();
+        recovery_passphrases.insert(recovery_operator_id, passphrase_bytes);
+
+        if passphrases.len() < *declared_count || recovery_passphrases.len() < *recovery_count {
+            return Ok(false);
+        }
+
+        let CoordinatorState::Bootstrapping {
+            passphrases,
+            recovery_passphrases,
+            ..
+        } = std::mem::replace(&mut self.state, CoordinatorState::Idle)
+        else {
+            unreachable!()
+        };
+
+        finalize_bootstrap(
+            self.db.clone(),
+            self.vault.clone(),
+            passphrases,
+            recovery_passphrases,
+        )
+        .await?;
+        Ok(true)
+    }
+
+    /// Contribute a passphrase for vault unseal (ordinary operator).
     /// Returns Ok(true) when threshold reached and vault is unsealed.
     #[message]
     pub async fn contribute_unseal(
@@ -258,46 +507,215 @@ impl VaultCoordinator {
         operator_id: i32,
         mut passphrase: SafeCell<Vec<u8>>,
     ) -> Result<bool, Error> {
-        if matches!(self.state, CoordinatorState::Idle) {
-            let mut conn = self.db.get().await?;
-            let count: i64 = schema::operator::table
-                .count()
-                .get_result(&mut conn)
-                .await?;
-            let threshold = shamir_threshold(usize::try_from(count).unwrap_or_default());
-
-            self.state = CoordinatorState::Unsealing {
-                threshold,
-                passphrases: HashMap::new(),
-            };
-        }
+        self.ensure_unsealing_state().await?;
 
         let CoordinatorState::Unsealing {
             threshold,
-            passphrases,
+            ordinary_passphrases,
+            recovery_passphrases,
         } = &mut self.state
         else {
             return Err(Error::NotUnsealing);
         };
 
+        if ordinary_passphrases.contains_key(&operator_id) {
+            return Err(Error::DuplicateContribution);
+        }
+
+        let passphrase_bytes = passphrase.read().to_vec();
+        ordinary_passphrases.insert(operator_id, passphrase_bytes);
+
+        if ordinary_passphrases.len() + recovery_passphrases.len() < *threshold {
+            return Ok(false);
+        }
+
+        self.do_finalize_unseal().await
+    }
+
+    /// Contribute a passphrase for vault unseal (recovery operator, §3.5).
+    /// Recovery operators may contribute during unseal when recovery is active.
+    /// Returns Ok(true) when threshold reached and vault is unsealed.
+    #[message]
+    pub async fn contribute_recovery_unseal(
+        &mut self,
+        recovery_operator_id: i32,
+        mut passphrase: SafeCell<Vec<u8>>,
+    ) -> Result<bool, Error> {
+        self.ensure_unsealing_state().await?;
+
+        let CoordinatorState::Unsealing {
+            threshold,
+            ordinary_passphrases,
+            recovery_passphrases,
+        } = &mut self.state
+        else {
+            return Err(Error::NotUnsealing);
+        };
+
+        if recovery_passphrases.contains_key(&recovery_operator_id) {
+            return Err(Error::DuplicateContribution);
+        }
+
+        let passphrase_bytes = passphrase.read().to_vec();
+        recovery_passphrases.insert(recovery_operator_id, passphrase_bytes);
+
+        if ordinary_passphrases.len() + recovery_passphrases.len() < *threshold {
+            return Ok(false);
+        }
+
+        self.do_finalize_unseal().await
+    }
+}
+
+impl VaultCoordinator {
+    /// Initializes `CoordinatorState::Unsealing` on first call if still `Idle`.
+    /// Threshold is based on ordinary operator count only (§3.4).
+    async fn ensure_unsealing_state(&mut self) -> Result<(), Error> {
+        if matches!(self.state, CoordinatorState::Idle) {
+            let mut conn = self.db.get().await?;
+            let ordinary_count: i64 = schema::operator::table
+                .count()
+                .get_result(&mut conn)
+                .await?;
+            let threshold = shamir_threshold(usize::try_from(ordinary_count).unwrap_or_default());
+            self.state = CoordinatorState::Unsealing {
+                threshold,
+                ordinary_passphrases: HashMap::new(),
+                recovery_passphrases: HashMap::new(),
+            };
+        }
+        Ok(())
+    }
+
+    /// Moves state back to Idle and calls finalize_unseal.
+    async fn do_finalize_unseal(&mut self) -> Result<bool, Error> {
+        let CoordinatorState::Unsealing {
+            ordinary_passphrases,
+            recovery_passphrases,
+            ..
+        } = std::mem::replace(&mut self.state, CoordinatorState::Idle)
+        else {
+            unreachable!()
+        };
+
+        finalize_unseal(
+            self.db.clone(),
+            self.vault.clone(),
+            ordinary_passphrases,
+            recovery_passphrases,
+        )
+        .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);
         }
 
-        let passphrase_bytes = passphrase.read().to_vec();
-        passphrases.insert(operator_id, passphrase_bytes);
+        passphrases.insert(operator_id, passphrase.read().to_vec());
 
-        if passphrases.len() < *threshold {
+        if passphrases.len() < *ordinary_count || recovery_passphrases.len() < *recovery_count {
             return Ok(false);
         }
 
-        let CoordinatorState::Unsealing { passphrases, .. } =
-            std::mem::replace(&mut self.state, CoordinatorState::Idle)
+        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 {
-            unreachable!()
+            return Err(Error::NotRekeying);
         };
 
-        finalize_unseal(self.db.clone(), self.vault.clone(), passphrases).await?;
-        Ok(true)
+        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/db/models.rs

@@ -520,3 +520,26 @@ pub struct NewProposalVote {
     pub approve: bool,
     pub signature: Vec<u8>,
 }
+
+
+#[derive(Debug, Insertable)]
+#[diesel(table_name = schema::proposal_result, check_for_backend(Sqlite))]
+pub struct NewProposalResult {
+    pub proposal_id: i32,
+    pub data: Vec<u8>,
+}
+
+#[derive(Debug, Insertable)]
+#[diesel(table_name = schema::recovery_proposal_vote, check_for_backend(Sqlite))]
+pub struct NewRecoveryProposalVote {
+    pub proposal_id: i32,
+    pub recovery_operator_id: i32,
+    pub approve: bool,
+    pub signature: Vec<u8>,
+}
+
+#[derive(Debug, Insertable)]
+#[diesel(table_name = schema::recovery_wakeup_request, check_for_backend(Sqlite))]
+pub struct NewRecoveryWakeupRequest {
+    pub requested_by: i32,
+}

server/crates/arbiter-server/src/db/schema.rs

@@ -184,6 +184,55 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    proposal_result (proposal_id) {
+        proposal_id -> Integer,
+        data -> Binary,
+        created_at -> Integer,
+    }
+}
+
+diesel::table! {
+    recovery_operator (id) {
+        id -> Integer,
+        share -> Binary,
+        share_nonce -> Binary,
+        share_salt -> Binary,
+        created_at -> Integer,
+        updated_at -> Integer,
+    }
+}
+
+diesel::table! {
+    recovery_operator_identity (id) {
+        id -> Integer,
+        public_key -> Binary,
+        created_at -> Integer,
+        updated_at -> Integer,
+    }
+}
+
+diesel::table! {
+    recovery_wakeup_request (id) {
+        id -> Integer,
+        requested_by -> Integer,
+        requested_at -> Integer,
+        cancelled_by -> Nullable<Integer>,
+        cancelled_at -> Nullable<Integer>,
+    }
+}
+
+diesel::table! {
+    recovery_proposal_vote (id) {
+        id -> Integer,
+        proposal_id -> Integer,
+        recovery_operator_id -> Integer,
+        approve -> Bool,
+        signature -> Binary,
+        voted_at -> Integer,
+    }
+}
+
 diesel::table! {
     proposal_vote (id) {
         id -> Integer,
@@ -249,11 +298,21 @@ diesel::joinable!(evm_wallet_access -> program_client (client_id));
 diesel::joinable!(operator -> operator_identity (id));
 diesel::joinable!(program_client -> client_metadata (metadata_id));
 diesel::joinable!(proposal -> operator_identity (initiator_id));
+diesel::joinable!(proposal_result -> proposal (proposal_id));
 diesel::joinable!(proposal_vote -> proposal (proposal_id));
 diesel::joinable!(proposal_vote -> operator_identity (operator_id));
+diesel::joinable!(recovery_operator -> recovery_operator_identity (id));
+diesel::joinable!(recovery_proposal_vote -> proposal (proposal_id));
+diesel::joinable!(recovery_proposal_vote -> recovery_operator_identity (recovery_operator_id));
+diesel::joinable!(recovery_wakeup_request -> operator_identity (requested_by));
 
 diesel::allow_tables_to_appear_in_same_query!(
     aead_encrypted,
+    proposal_result,
+    recovery_operator,
+    recovery_operator_identity,
+    recovery_wakeup_request,
+    recovery_proposal_vote,
     arbiter_settings,
     client_metadata,
     client_metadata_history,

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

@@ -53,6 +53,30 @@ async fn handle_create(
             client_id: p.client_id,
         },
         Some(ProtoKind::ApproveServerUpdate(_)) => ProposalKind::ApproveServerUpdate,
+        Some(ProtoKind::ReplaceOperator(p)) => ProposalKind::ReplaceOperator {
+            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"
+            )]
+            new_n: p.new_n as u8,
+        },
+        Some(ProtoKind::ApprovePersistentGrant(p)) => {
+            use prost::Message as _;
+            ProposalKind::ApprovePersistentGrant {
+                payload_bytes: p.encode_to_vec(),
+            }
+        }
+        Some(ProtoKind::ApproveOneOffTransaction(p)) => {
+            use prost::Message as _;
+            ProposalKind::ApproveOneOffTransaction {
+                payload_bytes: p.encode_to_vec(),
+            }
+        }
         None => return Err(Status::invalid_argument("Missing proposal kind")),
     };
     let ttl_secs = req.ttl_secs.map(i64::from);

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

@@ -2,6 +2,7 @@ use crate::{
     grpc::{Convert, TryConvert},
     peers::operator::vault_gate::{
         self as vault_gate, HandleBootstrapEncryptedKey, HandleContributeBootstrapPassphrase,
+        HandleContributeRecoveryBootstrapPassphrase, HandleContributeRecoveryUnsealPassphrase,
         HandleContributeUnsealPassphrase, HandleDeclareCommittee, HandleHandshake,
         HandleUnsealEncryptedKey,
     },
@@ -52,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",
+            )),
         }
     }
 }
@@ -82,6 +86,14 @@ impl TryConvert for UnsealRequestPayload {
                     },
                 ),
             ),
+            Self::ContributeRecoveryPassphrase(crp) => Ok(
+                vault_gate::Inbound::HandleContributeRecoveryUnsealPassphrase(
+                    HandleContributeRecoveryUnsealPassphrase {
+                        recovery_operator_id: crp.recovery_operator_id,
+                        passphrase: crp.passphrase,
+                    },
+                ),
+            ),
         }
     }
 }
@@ -132,6 +144,7 @@ impl TryConvert for BootstrapRequestPayload {
             Self::DeclareCommittee(dc) => Ok(
                 vault_gate::Inbound::HandleDeclareCommittee(HandleDeclareCommittee {
                     count: dc.count as usize,
+                    recovery_count: dc.recovery_count as usize,
                 }),
             ),
             Self::ContributePassphrase(cp) => Ok(
@@ -141,6 +154,14 @@ impl TryConvert for BootstrapRequestPayload {
                     },
                 ),
             ),
+            Self::ContributeRecoveryPassphrase(crp) => Ok(
+                vault_gate::Inbound::HandleContributeRecoveryBootstrapPassphrase(
+                    HandleContributeRecoveryBootstrapPassphrase {
+                        recovery_operator_id: crp.recovery_operator_id,
+                        passphrase: crp.passphrase,
+                    },
+                ),
+            ),
         }
     }
 }

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

@@ -131,6 +131,19 @@ impl TryConvert for vault_gate::Outbound {
                 };
                 Ok(wrap_bootstrap_response(proto_result))
             }
+            Self::HandleContributeRecoveryBootstrapPassphrase(result) => {
+                let proto_result = match result {
+                    Ok(true) => ProtoBootstrapResult::Success,
+                    Ok(false) => ProtoBootstrapResult::AwaitingContributions,
+                    Err(err) => {
+                        warn!(?err, "contribute recovery bootstrap passphrase failed");
+                        return Err(Status::internal(
+                            "Failed to contribute recovery bootstrap passphrase",
+                        ));
+                    }
+                };
+                Ok(wrap_bootstrap_response(proto_result))
+            }
             Self::HandleContributeUnsealPassphrase(result) => {
                 let proto_result = match result {
                     Ok(true) => ProtoUnsealResult::Success,
@@ -144,6 +157,21 @@ impl TryConvert for vault_gate::Outbound {
                     proto_result.into(),
                 )))
             }
+            Self::HandleContributeRecoveryUnsealPassphrase(result) => {
+                let proto_result = match result {
+                    Ok(true) => ProtoUnsealResult::Success,
+                    Ok(false) => ProtoUnsealResult::AwaitingContributions,
+                    Err(err) => {
+                        warn!(?err, "contribute recovery unseal passphrase failed");
+                        return Err(Status::internal(
+                            "Failed to contribute recovery unseal passphrase",
+                        ));
+                    }
+                };
+                Ok(wrap_unseal_response(UnsealResponsePayload::Result(
+                    proto_result.into(),
+                )))
+            }
         }
     }
 }

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/src/peers/operator/vault_gate/mod.rs

@@ -3,7 +3,10 @@ use crate::{
     actors::{
         GlobalActors,
         vault::{self, Bootstrap, GetState, TryUnseal, VaultState, events},
-        vault_coordinator::{ContributeBootstrap, ContributeUnseal, StartBootstrap},
+        vault_coordinator::{
+            ContributeBootstrap, ContributeRecoveryBootstrap, ContributeRecoveryUnseal,
+            ContributeUnseal, StartBootstrap,
+        },
     },
     crypto::{KeyCell, integrity::{self}},
     db::DatabasePool,
@@ -234,12 +237,17 @@ impl VaultGate {
     }
 
     #[message]
-    pub async fn handle_declare_committee(&mut self, count: usize) -> Result<(), Error> {
+    pub async fn handle_declare_committee(
+        &mut self,
+        count: usize,
+        recovery_count: usize,
+    ) -> Result<(), Error> {
         self.actors
             .vault_coordinator
             .ask(StartBootstrap {
                 operator_id: self.auth_creds.id,
                 declared_count: count,
+                recovery_count,
             })
             .await
             .map_err(|_| Error::internal("VaultCoordinator unavailable"))
@@ -261,6 +269,23 @@ impl VaultGate {
             .map_err(|_| Error::internal("VaultCoordinator unavailable"))
     }
 
+    #[message]
+    pub async fn handle_contribute_recovery_bootstrap_passphrase(
+        &mut self,
+        recovery_operator_id: i32,
+        passphrase: Vec<u8>,
+    ) -> Result<bool, Error> {
+        let passphrase_cell = SafeCell::new(passphrase);
+        self.actors
+            .vault_coordinator
+            .ask(ContributeRecoveryBootstrap {
+                recovery_operator_id,
+                passphrase: passphrase_cell,
+            })
+            .await
+            .map_err(|_| Error::internal("VaultCoordinator unavailable"))
+    }
+
     #[message]
     pub async fn handle_contribute_unseal_passphrase(
         &mut self,
@@ -276,6 +301,23 @@ impl VaultGate {
             .await
             .map_err(|_| Error::internal("VaultCoordinator unavailable"))
     }
+
+    #[message]
+    pub async fn handle_contribute_recovery_unseal_passphrase(
+        &mut self,
+        recovery_operator_id: i32,
+        passphrase: Vec<u8>,
+    ) -> Result<bool, Error> {
+        let passphrase_cell = SafeCell::new(passphrase);
+        self.actors
+            .vault_coordinator
+            .ask(ContributeRecoveryUnseal {
+                recovery_operator_id,
+                passphrase: passphrase_cell,
+            })
+            .await
+            .map_err(|_| Error::internal("VaultCoordinator unavailable"))
+    }
 }
 
 impl Message<events::Bootstrapped> for VaultGate {

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

@@ -2,13 +2,20 @@ use arbiter_crypto::authn::{self, GOVERNANCE_CONTEXT};
 use arbiter_server::{
     actors::{
         GlobalActors,
-        proposal_manager::{CastVote, CreateProposal, Error as ProposalError, ExpireStale, ProposalKind, QueryPending, VoteOutcome},
+        proposal_manager::{
+            CancelRecoveryWakeup, CastRecoveryVote, CastVote, CreateProposal,
+            Error as ProposalError, ExpireStale, ProposalKind, QueryPending,
+            RequestRecoveryWakeup, VoteOutcome,
+        },
     },
     crypto::KeyCell,
     db,
 };
 use arbiter_server::actors::vault::Bootstrap;
-use arbiter_server::db::schema::{aead_encrypted, evm_wallet, evm_wallet_access, operator_identity};
+use arbiter_server::db::schema::{
+    aead_encrypted, evm_basic_grant, evm_wallet, evm_wallet_access, operator_identity,
+    proposal_result, recovery_operator_identity,
+};
 use diesel::{ExpressionMethods, QueryDsl, insert_into};
 use diesel_async::RunQueryDsl;
 
@@ -22,6 +29,28 @@ async fn register_operator(db: &db::DatabasePool, pubkey: &authn::PublicKey) ->
         .unwrap()
 }
 
+async fn register_recovery_operator(db: &db::DatabasePool, pubkey: &authn::PublicKey) -> i32 {
+    let mut conn = db.get().await.unwrap();
+    insert_into(recovery_operator_identity::table)
+        .values(recovery_operator_identity::public_key.eq(pubkey.to_bytes()))
+        .returning(recovery_operator_identity::id)
+        .get_result::<i32>(&mut conn)
+        .await
+        .unwrap()
+}
+
+/// Backdates a wakeup request so it appears to have passed the 14-day window.
+async fn insert_active_wakeup(db: &db::DatabasePool, operator_id: i32) {
+    let mut conn = db.get().await.unwrap();
+    diesel::sql_query(format!(
+        "INSERT INTO recovery_wakeup_request (requested_by, requested_at) \
+         VALUES ({operator_id}, unixepoch('now') - 14*24*3600 - 1)"
+    ))
+    .execute(&mut conn)
+    .await
+    .unwrap();
+}
+
 fn make_vote_message(proposal_id: i32, approve: bool) -> Vec<u8> {
     let mut msg = Vec::with_capacity(9);
     msg.extend_from_slice(&(proposal_id as i64).to_be_bytes());
@@ -500,6 +529,356 @@ async fn grant_wallet_access_on_quorum_approval() {
     assert_eq!(count, 1);
 }
 
+#[tokio::test]
+async fn approve_persistent_grant_creates_basic_grant_row() {
+    use arbiter_proto::proto::operator::governance::{
+        ApprovePersistentGrantPayload, EtherTransferSpecProto, VolumeLimitProto,
+        approve_persistent_grant_payload::Specific,
+    };
+    use prost::Message as _;
+
+    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;
+
+    // Insert a dummy wallet and client, then a wallet_access row
+    let wallet_id = insert_evm_wallet(&db).await;
+    let client_key = authn::SigningKey::generate();
+    let client_id = insert_unapproved_client(&db, &client_key.public_key()).await;
+
+    let mut conn = db.get().await.unwrap();
+    let wallet_access_id: i32 = insert_into(evm_wallet_access::table)
+        .values((
+            evm_wallet_access::wallet_id.eq(wallet_id),
+            evm_wallet_access::client_id.eq(client_id),
+        ))
+        .returning(evm_wallet_access::id)
+        .get_result(&mut conn)
+        .await
+        .unwrap();
+    drop(conn);
+
+    let payload = ApprovePersistentGrantPayload {
+        wallet_access_id,
+        chain_id: 1,
+        valid_from_secs: None,
+        valid_until_secs: None,
+        max_gas_fee_per_gas: None,
+        max_priority_fee_per_gas: None,
+        rate_limit: None,
+        specific: Some(Specific::EtherTransfer(EtherTransferSpecProto {
+            targets: vec![vec![0u8; 20]],
+            limit: Some(VolumeLimitProto {
+                max_volume: alloy::primitives::U256::from(1_000_000u64).to_be_bytes_vec(),
+                window_secs: 86400,
+            }),
+        })),
+    };
+
+    let proposal_id = actors
+        .proposal_manager
+        .ask(CreateProposal {
+            kind: ProposalKind::ApprovePersistentGrant { payload_bytes: payload.encode_to_vec() },
+            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 = evm_basic_grant::table
+        .filter(evm_basic_grant::wallet_access_id.eq(wallet_access_id))
+        .count()
+        .get_result(&mut conn)
+        .await
+        .unwrap();
+    assert_eq!(count, 1);
+}
+
+#[tokio::test]
+async fn approve_one_off_transaction_stores_result() {
+    use arbiter_proto::proto::operator::governance::ApproveOneOffTransactionPayload;
+    use arbiter_server::actors::evm::{Generate, OperatorCreateGrant};
+    use arbiter_server::evm::policies::{
+        SharedGrantSettings, SpecificGrant, VolumeRateLimit, ether_transfer,
+    };
+    use alloy::primitives::{Address, U256};
+    use chrono::Duration;
+    use prost::Message as _;
+
+    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;
+
+    // Create a real encrypted wallet
+    let (wallet_id, wallet_address) = actors.evm.ask(Generate {}).await.unwrap();
+
+    // Create a client and wallet_access
+    let client_key = authn::SigningKey::generate();
+    let client_id = insert_unapproved_client(&db, &client_key.public_key()).await;
+
+    let mut conn = db.get().await.unwrap();
+    let wallet_access_id: i32 = insert_into(evm_wallet_access::table)
+        .values((
+            evm_wallet_access::wallet_id.eq(wallet_id),
+            evm_wallet_access::client_id.eq(client_id),
+        ))
+        .returning(evm_wallet_access::id)
+        .get_result(&mut conn)
+        .await
+        .unwrap();
+    drop(conn);
+
+    // Create a grant that permits ether transfer to address zero
+    let to_address = Address::ZERO;
+    actors
+        .evm
+        .ask(OperatorCreateGrant {
+            basic: SharedGrantSettings {
+                wallet_access_id,
+                chain: 1,
+                valid_from: None,
+                valid_until: None,
+                max_gas_fee_per_gas: None,
+                max_priority_fee_per_gas: None,
+                rate_limit: None,
+            },
+            grant: SpecificGrant::EtherTransfer(ether_transfer::Settings {
+                target: vec![to_address],
+                limit: VolumeRateLimit {
+                    max_volume: U256::from(1_000_000_000_000_000_000u128),
+                    window: Duration::hours(24),
+                },
+            }),
+        })
+        .await
+        .unwrap();
+
+    // Encode the one-off transaction payload
+    let payload = ApproveOneOffTransactionPayload {
+        client_id,
+        wallet_address: wallet_address.as_slice().to_vec(),
+        chain_id: 1,
+        nonce: 0,
+        gas_limit: 21000,
+        max_fee_per_gas: 1u128.to_be_bytes().to_vec(),
+        max_priority_fee_per_gas: 1u128.to_be_bytes().to_vec(),
+        to: to_address.as_slice().to_vec(),
+        value: U256::from(1u64).to_be_bytes_vec(),
+        input: vec![],
+    };
+
+    let proposal_id = actors
+        .proposal_manager
+        .ask(CreateProposal {
+            kind: ProposalKind::ApproveOneOffTransaction { payload_bytes: payload.encode_to_vec() },
+            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 = proposal_result::table
+        .filter(proposal_result::proposal_id.eq(proposal_id))
+        .count()
+        .get_result(&mut conn)
+        .await
+        .unwrap();
+    assert_eq!(count, 1);
+}
+
+#[tokio::test]
+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
+        .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();
+
+    let proposal_id = actors
+        .proposal_manager
+        .ask(CreateProposal {
+            kind: ProposalKind::ReplaceOperator { old_operator_id: op_id, new_pubkey: new_pubkey.clone() },
+            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();
+    // 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]
+async fn update_shamir_parameters_reaches_quorum() {
+    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 proposal_id = actors
+        .proposal_manager
+        .ask(CreateProposal {
+            kind: ProposalKind::UpdateShamirParameters { new_n: 5 },
+            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);
+}
+
+#[tokio::test]
+async fn key_rotation_requires_full_quorum() {
+    // §3.3: ReplaceOperator needs all 3 operators to approve, not just shamir_threshold(3)=2
+    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 key1 = authn::SigningKey::generate();
+    let key2 = authn::SigningKey::generate();
+    let key3 = authn::SigningKey::generate();
+    let op1 = register_operator(&db, &key1.public_key()).await;
+    let op2 = register_operator(&db, &key2.public_key()).await;
+    let op3 = register_operator(&db, &key3.public_key()).await;
+
+    let new_pubkey = authn::SigningKey::generate().public_key().to_bytes();
+    let proposal_id = actors
+        .proposal_manager
+        .ask(CreateProposal {
+            kind: ProposalKind::ReplaceOperator { old_operator_id: 1, new_pubkey },
+            initiator_id: op1,
+            ttl_secs: None,
+        })
+        .await
+        .unwrap();
+
+    let cast = |op_id, key: &authn::SigningKey| {
+        let actors = actors.clone();
+        let sig = key.sign_message(&make_vote_message(proposal_id, true), GOVERNANCE_CONTEXT).unwrap();
+        async move {
+            actors
+                .proposal_manager
+                .ask(CastVote { proposal_id, operator_id: op_id, approve: true, signature: sig.to_bytes() })
+                .await
+                .unwrap()
+        }
+    };
+
+    // With shamir_threshold(3)=2, two approvals would suffice for a normal proposal.
+    // For key rotation, they must not.
+    assert_eq!(cast(op1, &key1).await, VoteOutcome::Pending);
+    assert_eq!(cast(op2, &key2).await, VoteOutcome::Pending);
+    assert_eq!(cast(op3, &key3).await, VoteOutcome::QuorumApproved);
+}
+
 #[tokio::test]
 async fn approve_server_update_reaches_quorum() {
     let db = db::create_test_pool().await;
@@ -538,3 +917,204 @@ async fn approve_server_update_reaches_quorum() {
 
     assert_eq!(outcome, VoteOutcome::QuorumApproved);
 }
+
+// ─── §3.5 / §3.6 Recovery Operator tests ──────────────────────────────────
+
+#[tokio::test]
+async fn recovery_vote_rejected_when_sleeping() {
+    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 op_key = authn::SigningKey::generate();
+    let op_id = register_operator(&db, &op_key.public_key()).await;
+    let rec_key = authn::SigningKey::generate();
+    let rec_id = register_recovery_operator(&db, &rec_key.public_key()).await;
+
+    let new_pubkey = authn::SigningKey::generate().public_key().to_bytes();
+    let proposal_id = actors
+        .proposal_manager
+        .ask(CreateProposal {
+            kind: ProposalKind::ReplaceOperator { old_operator_id: 1, new_pubkey },
+            initiator_id: op_id,
+            ttl_secs: None,
+        })
+        .await
+        .unwrap();
+
+    let msg = make_vote_message(proposal_id, true);
+    let sig = rec_key.sign_message(&msg, GOVERNANCE_CONTEXT).unwrap();
+    let err = actors
+        .proposal_manager
+        .ask(CastRecoveryVote {
+            proposal_id,
+            recovery_operator_id: rec_id,
+            approve: true,
+            signature: sig.to_bytes(),
+        })
+        .await
+        .unwrap_err();
+
+    assert!(
+        matches!(err, kameo::error::SendError::HandlerError(ProposalError::RecoveryNotActive)),
+        "expected RecoveryNotActive, got {err:?}"
+    );
+}
+
+#[tokio::test]
+async fn recovery_vote_blocked_on_non_replace_proposal() {
+    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 op_key = authn::SigningKey::generate();
+    let op_id = register_operator(&db, &op_key.public_key()).await;
+    let rec_key = authn::SigningKey::generate();
+    let rec_id = register_recovery_operator(&db, &rec_key.public_key()).await;
+
+    insert_active_wakeup(&db, op_id).await;
+
+    let client_key = authn::SigningKey::generate();
+    let client_id = insert_unapproved_client(&db, &client_key.public_key()).await;
+    let proposal_id = actors
+        .proposal_manager
+        .ask(CreateProposal {
+            kind: ProposalKind::ApproveSdkClient { client_id },
+            initiator_id: op_id,
+            ttl_secs: None,
+        })
+        .await
+        .unwrap();
+
+    let msg = make_vote_message(proposal_id, true);
+    let sig = rec_key.sign_message(&msg, GOVERNANCE_CONTEXT).unwrap();
+    let err = actors
+        .proposal_manager
+        .ask(CastRecoveryVote {
+            proposal_id,
+            recovery_operator_id: rec_id,
+            approve: true,
+            signature: sig.to_bytes(),
+        })
+        .await
+        .unwrap_err();
+
+    assert!(
+        matches!(
+            err,
+            kameo::error::SendError::HandlerError(ProposalError::NotAllowedForRecoveryOperator)
+        ),
+        "expected NotAllowedForRecoveryOperator, got {err:?}"
+    );
+}
+
+#[tokio::test]
+async fn recovery_wakeup_can_be_cancelled() {
+    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 key = authn::SigningKey::generate();
+    let op_id = register_operator(&db, &key.public_key()).await;
+
+    actors
+        .proposal_manager
+        .ask(RequestRecoveryWakeup { operator_id: op_id })
+        .await
+        .unwrap();
+
+    actors
+        .proposal_manager
+        .ask(CancelRecoveryWakeup { operator_id: op_id })
+        .await
+        .unwrap();
+
+    // Second request must succeed (previous one was cancelled)
+    actors
+        .proposal_manager
+        .ask(RequestRecoveryWakeup { operator_id: op_id })
+        .await
+        .unwrap();
+}
+
+#[tokio::test]
+async fn recovery_wakeup_prevents_duplicate_request() {
+    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 key = authn::SigningKey::generate();
+    let op_id = register_operator(&db, &key.public_key()).await;
+
+    actors
+        .proposal_manager
+        .ask(RequestRecoveryWakeup { operator_id: op_id })
+        .await
+        .unwrap();
+
+    let err = actors
+        .proposal_manager
+        .ask(RequestRecoveryWakeup { operator_id: op_id })
+        .await
+        .unwrap_err();
+
+    assert!(
+        matches!(err, kameo::error::SendError::HandlerError(ProposalError::WakeupAlreadyPending)),
+        "expected WakeupAlreadyPending, got {err:?}"
+    );
+}
+
+#[tokio::test]
+async fn recovery_operator_vote_contributes_to_replace_quorum() {
+    // 1 ordinary operator + 1 recovery operator; replace_operator needs both.
+    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 op_key = authn::SigningKey::generate();
+    let op_id = register_operator(&db, &op_key.public_key()).await;
+    let rec_key = authn::SigningKey::generate();
+    let rec_id = register_recovery_operator(&db, &rec_key.public_key()).await;
+
+    insert_active_wakeup(&db, op_id).await;
+
+    let new_pubkey = authn::SigningKey::generate().public_key().to_bytes();
+    let proposal_id = actors
+        .proposal_manager
+        .ask(CreateProposal {
+            kind: ProposalKind::ReplaceOperator { old_operator_id: 1, new_pubkey },
+            initiator_id: op_id,
+            ttl_secs: None,
+        })
+        .await
+        .unwrap();
+
+    // Ordinary operator approves — still pending (needs recovery too)
+    let msg = make_vote_message(proposal_id, true);
+    let sig = op_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::Pending);
+
+    // Recovery operator approves — now quorum is reached
+    let sig = rec_key.sign_message(&msg, GOVERNANCE_CONTEXT).unwrap();
+    let outcome = actors
+        .proposal_manager
+        .ask(CastRecoveryVote {
+            proposal_id,
+            recovery_operator_id: rec_id,
+            approve: true,
+            signature: sig.to_bytes(),
+        })
+        .await
+        .unwrap();
+    assert_eq!(outcome, VoteOutcome::QuorumApproved);
+}

server/crates/arbiter-server/tests/vault/lifecycle.rs

@@ -3,14 +3,19 @@ use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
 use arbiter_server::{
     actors::{
         GlobalActors,
-        vault::{Error, Vault},
+        vault::{Error, GetState, Vault, VaultState},
+        vault_coordinator::{
+            ContributeBootstrap, ContributeRecoveryBootstrap, ContributeRecoveryUnseal,
+            Error as CoordinatorError, StartBootstrap, VaultCoordinator,
+        },
     },
     crypto::{KeyCell, encryption::v1::{Nonce, ROOT_KEY_TAG}},
     db::{self, models, schema},
 };
 
-use diesel::{QueryDsl, SelectableHelper};
+use diesel::{ExpressionMethods, QueryDsl, SelectableHelper, insert_into};
 use diesel_async::RunQueryDsl;
+use kameo::actor::Spawn as _;
 
 #[tokio::test]
 #[test_log::test]
@@ -139,3 +144,127 @@ async fn test_unseal_wrong_then_correct_password() {
     let mut decrypted = actor.decrypt(aead_id).await.unwrap();
     assert_eq!(*decrypted.read(), plaintext);
 }
+
+#[tokio::test]
+#[test_log::test]
+async fn two_operator_vault_requires_recovery_share() {
+    let db = db::create_test_pool().await;
+    let bus = GlobalActors::spawn_message_bus();
+    let vault_ref = Vault::spawn(Vault::new(db.clone(), bus).await.unwrap());
+    let coordinator = VaultCoordinator::spawn(VaultCoordinator::new(db, vault_ref));
+
+    let err = coordinator
+        .ask(StartBootstrap {
+            operator_id: 1,
+            declared_count: 2,
+            recovery_count: 0,
+        })
+        .await
+        .unwrap_err();
+
+    assert!(
+        matches!(
+            err,
+            kameo::error::SendError::HandlerError(CoordinatorError::TwoOperatorsRequireRecovery)
+        ),
+        "expected TwoOperatorsRequireRecovery, got {err:?}"
+    );
+}
+
+/// §3.4: Bootstrap with 1 ordinary + 1 recovery operator produces a valid 1-of-2 Shamir split.
+/// Both ordinary and recovery shares are stored; the vault can be unsealed with either one.
+#[tokio::test]
+#[test_log::test]
+async fn recovery_share_stored_and_used_for_unseal() {
+    let db = db::create_test_pool().await;
+    let bus = GlobalActors::spawn_message_bus();
+    let vault_ref = Vault::spawn(Vault::new(db.clone(), bus).await.unwrap());
+    let coordinator = VaultCoordinator::spawn(VaultCoordinator::new(db.clone(), vault_ref.clone()));
+
+    // Register one ordinary operator and one recovery operator in the DB
+    let ordinary_id: i32 = {
+        let mut conn = db.get().await.unwrap();
+        insert_into(schema::operator_identity::table)
+            .values(schema::operator_identity::public_key.eq(vec![1u8; 32]))
+            .returning(schema::operator_identity::id)
+            .get_result(&mut conn)
+            .await
+            .unwrap()
+    };
+    let recovery_id: i32 = {
+        let mut conn = db.get().await.unwrap();
+        insert_into(schema::recovery_operator_identity::table)
+            .values(schema::recovery_operator_identity::public_key.eq(vec![2u8; 32]))
+            .returning(schema::recovery_operator_identity::id)
+            .get_result(&mut conn)
+            .await
+            .unwrap()
+    };
+
+    // Declare committee: 1 ordinary + 1 recovery
+    coordinator
+        .ask(StartBootstrap {
+            operator_id: ordinary_id,
+            declared_count: 1,
+            recovery_count: 1,
+        })
+        .await
+        .unwrap();
+
+    // Recovery operator contributes first — bootstrap should not finalize yet
+    let done = coordinator
+        .ask(ContributeRecoveryBootstrap {
+            recovery_operator_id: recovery_id,
+            passphrase: SafeCell::new(b"recovery-pass".to_vec()),
+        })
+        .await
+        .unwrap();
+    assert!(!done, "should not finalize with only recovery passphrase");
+
+    // Ordinary operator contributes — now bootstrap finalizes
+    let done = coordinator
+        .ask(ContributeBootstrap {
+            operator_id: ordinary_id,
+            passphrase: SafeCell::new(b"ordinary-pass".to_vec()),
+        })
+        .await
+        .unwrap();
+    assert!(done, "should finalize once all contributors are in");
+
+    // After bootstrap, vault is Unsealed (seal key still in memory).
+    let state = vault_ref.ask(GetState {}).await.unwrap();
+    assert_eq!(state, VaultState::Unsealed);
+
+    // Verify recovery_operator row was created
+    let recovery_share_count: i64 = {
+        let mut conn = db.get().await.unwrap();
+        schema::recovery_operator::table
+            .count()
+            .get_result(&mut conn)
+            .await
+            .unwrap()
+    };
+    assert_eq!(recovery_share_count, 1);
+
+    // Simulate restart: drop vault and coordinator, create fresh vault (comes up Sealed).
+    drop(coordinator);
+    drop(vault_ref);
+    let bus2 = GlobalActors::spawn_message_bus();
+    let vault_ref2 = Vault::spawn(Vault::new(db.clone(), bus2).await.unwrap());
+    let state = vault_ref2.ask(GetState {}).await.unwrap();
+    assert_eq!(state, VaultState::Sealed);
+
+    // §3.5: Unseal using ONLY the recovery operator share (threshold = shamir_threshold(1) = 1).
+    let coordinator2 = VaultCoordinator::spawn(VaultCoordinator::new(db.clone(), vault_ref2.clone()));
+    let done = coordinator2
+        .ask(ContributeRecoveryUnseal {
+            recovery_operator_id: recovery_id,
+            passphrase: SafeCell::new(b"recovery-pass".to_vec()),
+        })
+        .await
+        .unwrap();
+    assert!(done, "recovery share alone should satisfy threshold");
+
+    let state = vault_ref2.ask(GetState {}).await.unwrap();
+    assert_eq!(state, VaultState::Unsealed);
+}