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

@@ -4,6 +4,7 @@ package arbiter.operator;
 
 import "operator/auth.proto";
 import "operator/evm.proto";
+import "operator/governance.proto";
 import "operator/sdk_client.proto";
 import "operator/vault/vault.proto";
 
@@ -14,6 +15,7 @@ message OperatorRequest {
     vault.Request           vault       = 2;
     evm.Request             evm         = 3;
     sdk_client.Request      sdk_client  = 4;
+    governance.Request      governance  = 5;
   }
 }
 
@@ -24,5 +26,6 @@ message OperatorResponse {
     vault.Response          vault       = 2;
     evm.Response            evm         = 3;
     sdk_client.Response     sdk_client  = 4;
+    governance.Response     governance  = 5;
   }
 }

protobufs/operator/governance.proto

@@ -0,0 +1,136 @@
+syntax = "proto3";
+
+package arbiter.operator.governance;
+
+message Request {
+  oneof payload {
+    CreateProposalRequest create = 1;
+    CastVoteRequest       vote   = 2;
+    QueryPendingRequest   query  = 3;
+  }
+}
+
+message CreateProposalRequest {
+  oneof kind {
+    ApproveSdkClientPayload         approve_sdk_client          = 1;
+    GrantWalletAccessPayload        grant_wallet_access         = 3;
+    ApproveServerUpdatePayload      approve_server_update       = 4;
+    ReplaceOperatorPayload          replace_operator            = 5;
+    UpdateShamirParametersPayload   update_shamir_parameters    = 6;
+    ApprovePersistentGrantPayload   approve_persistent_grant    = 7;
+    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 {
+  int32 client_id = 1;
+}
+
+message GrantWalletAccessPayload {
+  int32 wallet_id = 1;
+  int32 client_id = 2;
+}
+
+message CastVoteRequest {
+  int32 proposal_id = 1;
+  bool  approve     = 2;
+  bytes signature   = 3;
+}
+
+message QueryPendingRequest {}
+
+message Response {
+  oneof payload {
+    CreateProposalResponse created = 1;
+    VoteResponse           voted   = 2;
+    QueryPendingResponse   pending = 3;
+  }
+}
+
+message CreateProposalResponse {
+  int32 proposal_id = 1;
+}
+
+message VoteResponse {
+  VoteOutcome outcome = 1;
+}
+
+enum VoteOutcome {
+  VOTE_OUTCOME_UNSPECIFIED = 0;
+  VOTE_OUTCOME_PENDING     = 1;
+  VOTE_OUTCOME_APPROVED    = 2;
+  VOTE_OUTCOME_REJECTED    = 3;
+}
+
+message ProposalSummary {
+  int32  id            = 1;
+  string kind          = 2;
+  int32  initiator_id  = 3;
+  int64  expires_at    = 4;
+  int64  approve_count = 5;
+  int64  reject_count  = 6;
+}
+
+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

@@ -8,15 +8,35 @@ message BootstrapEncryptedKey {
   bytes associated_data = 3;
 }
 
+message DeclareCommittee {
+  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;
   BOOTSTRAP_RESULT_ALREADY_BOOTSTRAPPED = 2;
   BOOTSTRAP_RESULT_INVALID_KEY = 3;
+  BOOTSTRAP_RESULT_AWAITING_CONTRIBUTIONS = 4;
 }
 
 message Request {
+  oneof payload {
     BootstrapEncryptedKey        encrypted_key                    = 2;
+    DeclareCommittee             declare_committee                = 3;
+    ContributePassphrase         contribute_passphrase            = 4;
+    ContributeRecoveryPassphrase contribute_recovery_passphrase   = 5;
+  }
 }
 
 message Response {

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

@@ -15,17 +15,29 @@ message UnsealEncryptedKey {
   bytes associated_data = 3;
 }
 
+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;
   UNSEAL_RESULT_INVALID_KEY = 2;
   UNSEAL_RESULT_UNBOOTSTRAPPED = 3;
+  UNSEAL_RESULT_AWAITING_CONTRIBUTIONS = 4;
 }
 
 message Request {
   oneof payload {
     UnsealStart                  start                          = 1;
     UnsealEncryptedKey           encrypted_key                  = 2;
+    ContributePassphrase         contribute_passphrase          = 3;
+    ContributeRecoveryPassphrase contribute_recovery_passphrase = 4;
   }
 }
 

protobufs/operator/vault/vault.proto

@@ -5,6 +5,7 @@ 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 {
@@ -12,6 +13,7 @@ message Request {
     google.protobuf.Empty query_state = 1;
     unseal.Request        unseal      = 2;
     bootstrap.Request     bootstrap   = 3;
+    rekey.Request         rekey       = 4;
   }
 }
 
@@ -20,5 +22,6 @@ message Response {
     arbiter.shared.VaultState state     = 1;
     unseal.Response           unseal    = 2;
     bootstrap.Response        bootstrap = 3;
+    rekey.Response            rekey     = 4;
   }
 }

protobufs/shared/vault.proto

@@ -5,7 +5,8 @@ package arbiter.shared;
 enum VaultState {
   VAULT_STATE_UNSPECIFIED = 0;
   VAULT_STATE_UNBOOTSTRAPPED = 1;
-  VAULT_STATE_SEALED = 2;
-  VAULT_STATE_UNSEALED = 3;
-  VAULT_STATE_ERROR = 4;
+  VAULT_STATE_BOOSTRAPPING = 2;
+  VAULT_STATE_SEALED = 3;
+  VAULT_STATE_UNSEALED = 4;
+  VAULT_STATE_ERROR = 5;
 }

server/Cargo.toml

@@ -6,25 +6,25 @@ resolver = "3"
 
 
 [workspace.dependencies]
-alloy = "2.0.0"
+alloy = "2.0.4"
 async-trait = "0.1.89"
 base64 = "0.22.1"
 chrono = { version = "0.4.44", features = ["serde"] }
 futures = "0.3.32"
 k256 = { version = "0.13.4", features = ["ecdsa", "pkcs8"] }
-kameo = {git = "https://github.com/hdbg/kameo.git", rev = "805b417"}
-kameo_actors = {git = "https://github.com/hdbg/kameo.git", rev = "805b417"}
+kameo = {git = "https://github.com/hdbg/kameo.git", rev = "3e18ba2"}
+kameo_actors = {git = "https://github.com/hdbg/kameo.git", rev = "3e18ba2"}
 hmac = "0.13.0"
 miette = { version = "7.6.0", features = ["fancy", "serde"] }
-ml-dsa = { version = "0.1.0-rc.8", features = ["zeroize"] }
+ml-dsa = { version = "0.1.0-rc.9", features = ["zeroize"] }
 mutants = "0.0.4"
 prost = "0.14.3"
 prost-types = { version = "0.14.3", features = ["chrono"] }
 rand = "0.10.1"
 rcgen = { version = "0.14.7", features = [ "aws_lc_rs",  "pem", "x509-parser", "zeroize" ], default-features = false }
 rstest = "0.26.1"
-rustls = { version = "0.23.38", features = ["aws-lc-rs", "logging", "prefer-post-quantum", "std"], default-features = false }
-rustls-pki-types = "1.14.0"
+rustls = { version = "0.23.40", features = ["aws-lc-rs", "logging", "prefer-post-quantum", "std"], default-features = false }
+rustls-pki-types = "1.14.1"
 sha2 = "0.11"
 smlang = "0.8.0"
 thiserror = "2.0.18"
@@ -106,7 +106,6 @@ indexing_slicing = "warn"
 infinite_loop = "warn"
 inline_asm_x86_att_syntax = "warn"
 inline_asm_x86_intel_syntax = "warn"
-integer_division = "warn"
 large_include_file = "warn"
 lossy_float_literal = "warn"
 map_with_unused_argument_over_ranges = "warn"

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-client/Cargo.toml

@@ -21,7 +21,7 @@ tokio.workspace = true
 tokio-stream.workspace = true
 thiserror.workspace = true
 http = "1.4.0"
-rustls-webpki = { version = "0.103.12", features = ["aws-lc-rs"] }
+rustls-webpki = { version = "0.103.13", features = ["aws-lc-rs"] }
 async-trait.workspace = true
 chrono.workspace = true
 

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

@@ -8,6 +8,7 @@ use rand::RngExt;
 
 pub static CLIENT_CONTEXT: &[u8] = b"arbiter_client";
 pub static OPERATOR_CONTEXT: &[u8] = b"arbiter_operator";
+pub static GOVERNANCE_CONTEXT: &[u8] = b"arbiter_governance_vote";
 
 const NONCE_SIZE: usize = 32;
 
@@ -90,6 +91,11 @@ impl PublicKey {
         self.0
             .verify_with_context(&challenge, context, &signature.0)
     }
+
+    #[must_use]
+    pub fn verify_message(&self, message: &[u8], context: &[u8], signature: &Signature) -> bool {
+        self.0.verify_with_context(message, context, &signature.0)
+    }
 }
 
 impl Signature {

server/crates/arbiter-crypto/src/safecell.rs

@@ -22,7 +22,7 @@ pub trait SafeCellHandle<T> {
     fn read(&mut self) -> Self::CellRead<'_>;
     fn write(&mut self) -> Self::CellWrite<'_>;
 
-    fn new_inline<F>(f: F) -> Self
+    fn new_inline_default<F>(f: F) -> Self
     where
         Self: Sized,
         T: Default,
@@ -36,6 +36,14 @@ pub trait SafeCellHandle<T> {
         cell
     }
 
+    fn new_inline<F>(f: Box<F>) -> Self
+    where
+        Self: Sized,
+        F: for<'a> FnOnce() -> T,
+    {
+        Self::new(f())
+    }
+
     #[inline(always)]
     fn read_inline<F, R>(&mut self, f: F) -> R
     where

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

@@ -23,6 +23,10 @@ pub mod proto {
             tonic::include_proto!("arbiter.operator.evm");
         }
 
+        pub mod governance {
+            tonic::include_proto!("arbiter.operator.governance");
+        }
+
         pub mod sdk_client {
             tonic::include_proto!("arbiter.operator.sdk_client");
         }
@@ -34,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

@@ -9,8 +9,8 @@ license = "Apache-2.0"
 workspace = true
 
 [dependencies]
-diesel = { version = "2.3.7", features = ["chrono", "returning_clauses_for_sqlite_3_35", "serde_json", "time", "uuid"] }
-diesel-async = { version = "0.8.0", features = [
+diesel = { version = "2.3.9", features = ["chrono", "returning_clauses_for_sqlite_3_35", "serde_json", "time", "uuid"] }
+diesel-async = { version = "0.9.0", features = [
     "bb8",
     "migrations",
     "sqlite",
@@ -27,7 +27,7 @@ tokio.workspace = true
 rustls.workspace = true
 smlang.workspace = true
 thiserror.workspace = true
-diesel_migrations = { version = "2.3.1", features = ["sqlite"] }
+diesel_migrations = { version = "2.3.2", features = ["sqlite"] }
 async-trait.workspace = true
 tokio-stream.workspace = true
 rand.workspace = true
@@ -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"
@@ -50,7 +51,8 @@ subtle = "2.6.1"
 x25519-dalek.workspace = true
 k256.workspace = true
 kameo_actors.workspace = true
-blahaj = "0.6.0"
+vsss-rs = "5.4.0"
+rand_core = "0.6"
 
 [dev-dependencies]
 proptest = "1.11.0"

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

@@ -49,13 +49,14 @@ create table if not exists operator_identity (
     created_at integer not null default(unixepoch ('now')),
     updated_at integer not null default(unixepoch ('now'))
 ) STRICT;
-create unique index if not exists uniq_operator_client_public_key on operator_identity (public_key);
+create unique index if not exists uniq_operator_identity_public_key on operator_identity (public_key);
 
 create table if not exists operator (
     id integer primary key references operator_identity(id) on delete restrict, -- same id as operator_identity
 
     share blob not null,
     share_nonce blob not null,
+    share_salt blob not null default (randomblob(32)),
 
     created_at integer not null default(unixepoch ('now')),
     updated_at integer not null default(unixepoch ('now'))
@@ -215,3 +216,73 @@ create table if not exists integrity_envelope (
 ) STRICT;
 
 create unique index if not exists uniq_integrity_envelope_entity on integrity_envelope (entity_kind, entity_id);
+
+create table if not exists proposal (
+    id           integer not null primary key,
+    kind         text    not null,
+    payload      blob    not null,
+    initiator_id integer not null references operator_identity(id) on delete restrict,
+    created_at   integer not null default(unixepoch('now')),
+    expires_at   integer not null,
+    status       text    not null default 'pending'
+        check (status in ('pending', 'approved', 'rejected', 'expired'))
+) STRICT;
+
+create table if not exists proposal_vote (
+    id          integer not null primary key,
+    proposal_id integer not null references proposal(id) on delete cascade,
+    operator_id integer not null references 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, 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/bootstrap.rs

@@ -48,7 +48,7 @@ impl Bootstrapper {
         let row_count: i64 = {
             let mut conn = db.get().await?;
 
-            schema::operator_identity::table
+            schema::operator::table
                 .count()
                 .get_result(&mut conn)
                 .await?

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

@@ -160,29 +160,23 @@ impl EvmActor {
     }
 
     #[message]
-    #[expect(clippy::unused_async, reason = "reserved for impl")]
-    pub async fn operator_delete_grant(&mut self, _grant_id: i32) -> Result<(), Error> {
-        // let mut conn = self.db.get().await.map_err(DatabaseError::from)?;
-        // let vault = self.vault.clone();
+    pub async fn operator_delete_grant(&mut self, grant_id: i32) -> Result<(), Error> {
+        let mut conn = self.db.get().await.map_err(DatabaseError::from)?;
 
-        // diesel_async::AsyncConnection::transaction(&mut conn, |conn| {
-        //     Box::pin(async move {
-        //         diesel::update(schema::evm_basic_grant::table)
-        //             .filter(schema::evm_basic_grant::id.eq(grant_id))
-        //             .set(schema::evm_basic_grant::revoked_at.eq(SqliteTimestamp::now()))
-        //             .execute(conn)
-        //             .await?;
+        let affected = diesel::update(schema::evm_basic_grant::table)
+            .filter(schema::evm_basic_grant::id.eq(grant_id))
+            .set(schema::evm_basic_grant::revoked_at.eq(models::SqliteTimestamp::now()))
+            .execute(&mut conn)
+            .await
+            .map_err(DatabaseError::from)?;
 
-        //         let signed = integrity::evm::load_signed_grant_by_basic_id(conn, grant_id).await?;
+        if affected == 0 {
+            return Err(Error::Database(DatabaseError::from(
+                diesel::result::Error::NotFound,
+            )));
+        }
 
-        //         diesel::result::QueryResult::Ok(())
-        //     })
-        // })
-        // .await
-        // .map_err(DatabaseError::from)?;
-
-        // Ok(())
-        todo!()
+        Ok(())
     }
 
     #[message]

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

@@ -1,7 +1,8 @@
 use crate::{
     actors::{
         bootstrap::Bootstrapper, evm::EvmActor, flow_coordinator::FlowCoordinator,
-        operator_registry::OperatorRegistry, vault::Vault,
+        operator_registry::OperatorRegistry, proposal_manager::ProposalManager, vault::Vault,
+        vault_coordinator::VaultCoordinator,
     },
     db,
 };
@@ -14,7 +15,9 @@ pub mod bootstrap;
 pub mod evm;
 pub mod flow_coordinator;
 pub mod operator_registry;
+pub mod proposal_manager;
 pub mod vault;
+pub mod vault_coordinator;
 
 #[derive(Error, Debug)]
 pub enum SpawnError {
@@ -30,9 +33,11 @@ pub enum SpawnError {
 pub struct GlobalActors {
     pub vault: ActorRef<Vault>,
     pub bootstrapper: ActorRef<Bootstrapper>,
+    pub vault_coordinator: ActorRef<VaultCoordinator>,
     pub flow_coordinator: ActorRef<FlowCoordinator>,
     pub operator_registry: ActorRef<OperatorRegistry>,
     pub evm: ActorRef<EvmActor>,
+    pub proposal_manager: ActorRef<ProposalManager>,
     pub events: ActorRef<MessageBus>,
 }
 
@@ -45,15 +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)),
+            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/vault/mod.rs

@@ -1,13 +1,13 @@
 use crate::{
     crypto::{
-        KeyCell, derive_key,
+        KeyCell,
         encryption::v1::{self, Nonce},
         integrity::v1::HmacSha256,
     },
     db::{
         self,
         models::{self, RootKeyHistory, RootKeyHistoryId},
-        schema::{self},
+        schema,
     },
 };
 use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
@@ -90,7 +90,6 @@ pub struct Vault {
     events: ActorRef<MessageBus>,
 }
 
-#[messages]
 impl Vault {
     pub async fn new(db: db::DatabasePool, events: ActorRef<MessageBus>) -> Result<Self, Error> {
         let state = {
@@ -113,7 +112,7 @@ impl Vault {
         Ok(Self { db, state, events })
     }
 
-    // Exclusive transaction to avoid race condtions if multiple vaults write
+    // Exclusive transaction to avoid race conditions if multiple vaults write
     // additional layer of protection against nonce-reuse
     async fn get_new_nonce(
         pool: &db::DatabasePool,
@@ -122,12 +121,11 @@ impl Vault {
         let mut conn = pool.get().await?;
 
         let nonce = conn
-            .exclusive_transaction(|conn| {
-                Box::pin(async move {
+            .exclusive_transaction(async |conn| {
                 let current_nonce: Vec<u8> = schema::root_key_history::table
                     .filter(schema::root_key_history::id.eq(root_key_id))
                     .select(schema::root_key_history::data_encryption_nonce)
-                        .first(conn)
+                    .first(&mut *conn)
                     .await?;
 
                 let mut nonce = Nonce::try_from(current_nonce.as_slice()).map_err(|()| {
@@ -142,12 +140,11 @@ impl Vault {
                 update(schema::root_key_history::table)
                     .filter(schema::root_key_history::id.eq(root_key_id))
                     .set(schema::root_key_history::data_encryption_nonce.eq(nonce.to_vec()))
-                        .execute(conn)
+                    .execute(&mut *conn)
                     .await?;
 
                 Result::<_, Error>::Ok(nonce)
             })
-            })
             .await?;
 
         Ok(nonce)
@@ -160,57 +157,60 @@ impl Vault {
             State::Sealed { .. } => Err(Error::Sealed),
         }
     }
+}
 
+#[messages]
+impl Vault {
     #[message]
-    pub async fn bootstrap(&mut self, seal_key_raw: SafeCell<Vec<u8>>) -> Result<(), Error> {
-        if !matches!(self.state, State::Unbootstrapped) {
+    pub async fn bootstrap(&mut self, mut seal_key: KeyCell) -> Result<(), Error> {
+        if !matches!(&self.state, State::Unbootstrapped) {
             return Err(Error::AlreadyBootstrapped);
         }
-        let salt = v1::generate_salt();
-        let mut seal_key = derive_key(seal_key_raw, &salt);
+
         let mut root_key = KeyCell::new_secure_random();
 
         // Zero nonces are fine because they are one-time
         let root_key_nonce = Nonce::default();
         let data_encryption_nonce = Nonce::default();
 
-        let root_key_ciphertext: Vec<u8> = root_key.0.read_inline(|reader| {
-            let root_key_reader = reader.as_slice();
+        // Generate salt (kept for schema compat)
+        let root_key_salt = v1::generate_salt();
+
+        let root_key_ciphertext: Vec<u8> = root_key.0.read_inline(|rk| {
             seal_key
-                .encrypt(&root_key_nonce, v1::ROOT_KEY_TAG, root_key_reader)
+                .encrypt(&root_key_nonce, v1::ROOT_KEY_TAG, rk.as_slice())
                 .map_err(|err| {
                     error!(?err, "Fatal bootstrap error");
                     Error::Encryption(err)
                 })
         })?;
 
+        let data_encryption_nonce_bytes = data_encryption_nonce.to_vec();
         let mut conn = self.db.get().await?;
 
-        let data_encryption_nonce_bytes = data_encryption_nonce.to_vec();
         let root_key_history_id = conn
-            .transaction(|conn| {
-                Box::pin(async move {
-                    let root_key_history_id: RootKeyHistoryId =
-                        insert_into(schema::root_key_history::table)
+            .transaction(async |conn| {
+                let root_key_history_id = insert_into(schema::root_key_history::table)
                     .values(&models::NewRootKeyHistory {
-                                ciphertext: root_key_ciphertext,
+                        ciphertext: root_key_ciphertext.clone(),
                         tag: v1::ROOT_KEY_TAG.to_vec(),
                         root_key_encryption_nonce: root_key_nonce.to_vec(),
-                                data_encryption_nonce: data_encryption_nonce_bytes,
+                        data_encryption_nonce: data_encryption_nonce_bytes.clone(),
                         schema_version: 1,
-                                salt: salt.to_vec(),
+                        salt: root_key_salt.to_vec(),
                     })
                     .returning(schema::root_key_history::id)
-                            .get_result(conn)
+                    .get_result(&mut *conn)
                     .await?;
 
                 update(schema::arbiter_settings::table)
                     .set(schema::arbiter_settings::root_key_id.eq(root_key_history_id))
-                        .execute(conn)
+                    .execute(&mut *conn)
                     .await?;
 
-                    Result::<_, diesel::result::Error>::Ok(root_key_history_id)
-                })
+                Result::<_, diesel::result::Error>::Ok(RootKeyHistoryId::from_raw(
+                    root_key_history_id,
+                ))
             })
             .await?;
 
@@ -226,52 +226,47 @@ impl Vault {
     }
 
     #[message]
-    pub async fn try_unseal(&mut self, seal_key_raw: SafeCell<Vec<u8>>) -> Result<(), Error> {
+    pub async fn try_unseal(&mut self, mut seal_key: KeyCell) -> Result<(), Error> {
         let State::Sealed {
             root_key_history_id,
         } = &self.state
         else {
             return Err(Error::NotBootstrapped);
         };
+        let root_key_history_id = *root_key_history_id;
 
-        // We don't want to hold connection while doing expensive KDF work
+        // We don't want to hold connection while doing expensive work
         let current_key = {
             let mut conn = self.db.get().await?;
             schema::root_key_history::table
-                .filter(schema::root_key_history::id.eq(*root_key_history_id))
+                .filter(schema::root_key_history::id.eq(root_key_history_id))
                 .select(RootKeyHistory::as_select())
                 .first(&mut conn)
                 .await?
         };
 
-        let salt = &current_key.salt;
-        let salt = v1::Salt::try_from(salt.as_slice()).map_err(|_| {
-            error!("Broken database: invalid salt for root key");
-            Error::BrokenDatabase
-        })?;
-        let mut seal_key = derive_key(seal_key_raw, &salt);
-
-        let mut root_key = SafeCell::new(current_key.ciphertext.clone());
-
         let nonce =
             Nonce::try_from(current_key.root_key_encryption_nonce.as_slice()).map_err(|()| {
                 error!("Broken database: invalid nonce for root key");
                 Error::BrokenDatabase
             })?;
 
+        let mut root_key_bytes = SafeCell::new(current_key.ciphertext.clone());
         seal_key
-            .decrypt_in_place(&nonce, v1::ROOT_KEY_TAG, &mut root_key)
+            .decrypt_in_place(&nonce, v1::ROOT_KEY_TAG, &mut root_key_bytes)
             .map_err(|err| {
                 error!(?err, "Failed to unseal root key: invalid seal key");
                 Error::InvalidKey
             })?;
 
+        let root_key = KeyCell::try_from(root_key_bytes).map_err(|()| {
+            error!("Broken database: invalid encryption key size");
+            Error::BrokenDatabase
+        })?;
+
         self.state = State::Unsealed(Unsealed {
             root_key_history_id: current_key.id,
-            root_key: KeyCell::try_from(root_key).map_err(|err| {
-                error!(?err, "Broken database: invalid encryption key size");
-                Error::BrokenDatabase
-            })?,
+            root_key,
         });
 
         info!("Vault unsealed successfully");
@@ -280,6 +275,77 @@ 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 {
+            root_key_history_id,
+            ..
+        } = Self::expect_unsealed(&mut self.state)?;
+
+        self.state = State::Sealed {
+            root_key_history_id: *root_key_history_id,
+        };
+        let _ = self.events.tell(Publish(events::VaultResealed)).await;
+        Ok(())
+    }
+}
+
+// Server-side cryptographic operations
+#[messages]
+impl Vault {
     #[message]
     pub async fn decrypt(&mut self, aead_id: i32) -> Result<SafeCell<Vec<u8>>, Error> {
         let Unsealed { root_key, .. } = Self::expect_unsealed(&mut self.state)?;
@@ -357,11 +423,9 @@ impl Vault {
             root_key_history_id,
         } = Self::expect_unsealed(&mut self.state)?;
 
-        let mut hmac = root_key
-            .0
-            .read_inline(|k| match HmacSha256::new_from_slice(k) {
-                Ok(v) => v,
-                Err(_) => unreachable!("HMAC accepts keys of any size"),
+        let mut hmac = root_key.0.read_inline(|k| {
+            HmacSha256::new_from_slice(k)
+                .unwrap_or_else(|_| unreachable!("HMAC accepts keys of any size"))
         });
         hmac.update(&root_key_history_id.to_raw().to_be_bytes());
         hmac.update(&mac_input);
@@ -386,36 +450,20 @@ impl Vault {
             return Ok(false);
         }
 
-        let mut hmac = root_key
-            .0
-            .read_inline(|k| match HmacSha256::new_from_slice(k) {
-                Ok(v) => v,
-                Err(_) => unreachable!("HMAC accepts keys of any size"),
+        let mut hmac = root_key.0.read_inline(|k| {
+            HmacSha256::new_from_slice(k)
+                .unwrap_or_else(|_| unreachable!("HMAC accepts keys of any size"))
         });
         hmac.update(&key_version.to_raw().to_be_bytes());
         hmac.update(&mac_input);
 
         Ok(hmac.verify_slice(&expected_mac).is_ok())
     }
-
-    #[message]
-    pub async fn seal(&mut self) -> Result<(), Error> {
-        let Unsealed {
-            root_key_history_id,
-            ..
-        } = Self::expect_unsealed(&mut self.state)?;
-
-        self.state = State::Sealed {
-            root_key_history_id: *root_key_history_id,
-        };
-        let _ = self.events.tell(Publish(events::VaultResealed)).await;
-        Ok(())
-    }
 }
 
 #[cfg(test)]
 mod tests {
-    use crate::{actors::GlobalActors, db::models::RootKeyHistory};
+    use crate::actors::GlobalActors;
     use arbiter_crypto::safecell::SafeCellHandle as _;
 
     use super::*;
@@ -424,8 +472,7 @@ mod tests {
         let mut actor = Vault::new(db.clone(), GlobalActors::spawn_message_bus())
             .await
             .unwrap();
-        let seal_key = SafeCell::new(b"test-seal-key".to_vec());
-        actor.bootstrap(seal_key).await.unwrap();
+        actor.bootstrap(KeyCell::from([0u8; 32])).await.unwrap();
         actor
     }
 
@@ -434,12 +481,12 @@ mod tests {
     async fn nonce_monotonic_even_when_nonce_allocation_interleaves() {
         let db = db::create_test_pool().await;
         let mut actor = bootstrapped_actor(&db).await;
-        let root_key_history_id = match actor.state {
-            State::Unsealed(Unsealed {
+        let State::Unsealed(Unsealed {
             root_key_history_id,
             ..
-            }) => root_key_history_id,
-            _ => panic!("expected unsealed state"),
+        }) = actor.state
+        else {
+            panic!("expected unsealed state");
         };
 
         let n1 = Vault::get_new_nonce(&db, root_key_history_id)

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

@@ -0,0 +1,721 @@
+use std::collections::HashMap;
+
+use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
+use diesel::{ExpressionMethods as _, QueryDsl};
+use diesel_async::RunQueryDsl;
+use kameo::{Actor, actor::ActorRef, messages};
+use rand_core::{OsRng, RngCore as _};
+use tracing::error;
+
+use crate::{
+    actors::vault::{Bootstrap, RekeyRootKey, TryUnseal, Vault},
+    crypto::{KeyCell, derive_key, encryption::v1::Nonce, shamir, shamir::shamir_threshold},
+    db::{self, models, schema},
+};
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("Already coordinating a bootstrap")]
+    AlreadyBootstrapping,
+    #[error("Already coordinating an unseal")]
+    AlreadyUnsealing,
+    #[error("Rekey not in progress")]
+    NotRekeying,
+    #[error("Bootstrap not in progress")]
+    NotBootstrapping,
+    #[error("Unseal not in progress")]
+    NotUnsealing,
+    #[error("Operator already contributed")]
+    DuplicateContribution,
+    #[error("Operator not found in database")]
+    OperatorNotFound,
+    #[error("Invalid passphrase (decryption failed)")]
+    InvalidPassphrase,
+    #[error("Shamir error: {0}")]
+    Shamir(String),
+    #[error("Database connection error: {0}")]
+    DatabaseConnection(#[from] db::PoolError),
+    #[error("Database query error: {0}")]
+    DatabaseQuery(#[from] diesel::result::Error),
+    #[error("Encryption error")]
+    Encryption,
+    #[error("Vault error")]
+    VaultError,
+    #[error("Two-operator vaults require at least one recovery share")]
+    TwoOperatorsRequireRecovery,
+    #[error("Broken database")]
+    BrokenDatabase,
+}
+
+// Passphrases stored as plain Vec<u8> (not SafeCell) so CoordinatorState is Sync.
+// They are ephemeral and dropped immediately after use.
+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>>,
+    },
+}
+
+#[derive(Actor)]
+pub struct VaultCoordinator {
+    db: db::DatabasePool,
+    vault: ActorRef<Vault>,
+    state: CoordinatorState,
+}
+
+impl VaultCoordinator {
+    pub const fn new(db: db::DatabasePool, vault: ActorRef<Vault>) -> Self {
+        Self {
+            db,
+            vault,
+            state: CoordinatorState::Idle,
+        }
+    }
+}
+
+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>,
+    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 seal_key_bytes = [0u8; 32];
+    OsRng.fill_bytes(&mut seal_key_bytes);
+
+    // 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) 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?;
+    }
+
+    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>,
+    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 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)))
+                .select((
+                    schema::operator::share,
+                    schema::operator::share_nonce,
+                    schema::operator::share_salt,
+                ))
+                .first(&mut conn)
+                .await
+                .map_err(|_| Error::OperatorNotFound)?;
+
+        shares.push(decrypt_share(
+            passphrase_bytes,
+            encrypted_share,
+            &share_nonce_bytes,
+            &share_salt,
+            operator_id_raw,
+        )?);
+    }
+
+    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(RekeyRootKey { new_seal_key })
+        .await
+        .map_err(|err| {
+            error!(?err, "Vault rekey failed");
+            Error::VaultError
+        })?;
+
+    Ok(())
+}
+
+#[messages]
+impl VaultCoordinator {
+    /// Phase 1 of multi-operator bootstrap: declare the committee size.
+    #[message]
+    #[expect(clippy::unused_async, reason = "kameo requires messages to be async")]
+    pub async fn start_bootstrap(
+        &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: 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,
+        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 passphrases.contains_key(&operator_id) {
+            return Err(Error::DuplicateContribution);
+        }
+
+        let passphrase_bytes = passphrase.read().to_vec();
+        passphrases.insert(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)
+    }
+
+    /// 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(
+        &mut self,
+        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 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);
+        }
+
+        passphrases.insert(operator_id, passphrase.read().to_vec());
+
+        if passphrases.len() < *ordinary_count || recovery_passphrases.len() < *recovery_count {
+            return Ok(false);
+        }
+
+        self.do_finalize_rekey().await
+    }
+
+    /// Contribute a recovery operator passphrase for the re-key.
+    /// Returns Ok(true) when all contributors have submitted and the re-key is complete.
+    #[message]
+    pub async fn contribute_recovery_rekey(
+        &mut self,
+        recovery_operator_id: i32,
+        mut passphrase: SafeCell<Vec<u8>>,
+    ) -> Result<bool, Error> {
+        let CoordinatorState::Rekeying {
+            ordinary_count,
+            recovery_count,
+            passphrases,
+            recovery_passphrases,
+        } = &mut self.state
+        else {
+            return Err(Error::NotRekeying);
+        };
+
+        if recovery_passphrases.contains_key(&recovery_operator_id) {
+            return Err(Error::DuplicateContribution);
+        }
+
+        recovery_passphrases.insert(recovery_operator_id, passphrase.read().to_vec());
+
+        if passphrases.len() < *ordinary_count || recovery_passphrases.len() < *recovery_count {
+            return Ok(false);
+        }
+
+        self.do_finalize_rekey().await
+    }
+}

server/crates/arbiter-server/src/context/tls.rs

@@ -174,8 +174,7 @@ impl TlsManager {
 
         {
             let mut conn = db.get().await?;
-            conn.transaction(|conn| {
-                Box::pin(async {
+            conn.transaction(async |conn| {
                 let new_tls_history = NewTlsHistory {
                     cert: new_cert.cert.pem(),
                     cert_key: new_cert.cert_key.serialize_pem(),
@@ -186,17 +185,16 @@ impl TlsManager {
                 let inserted_tls_history: i32 = diesel::insert_into(tls_history::table)
                     .values(&new_tls_history)
                     .returning(tls_history::id)
-                        .get_result(conn)
+                    .get_result(&mut *conn)
                     .await?;
 
                 diesel::update(arbiter_settings::table)
                     .set(arbiter_settings::tls_id.eq(inserted_tls_history))
-                        .execute(conn)
+                    .execute(&mut *conn)
                     .await?;
 
                 Result::<_, diesel::result::Error>::Ok(())
             })
-            })
             .await?;
         }
 

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

@@ -61,12 +61,12 @@ mod tests {
     #[test]
     fn derive_seal_key_deterministic() {
         static PASSWORD: &[u8] = b"password";
-        let password = SafeCell::new(PASSWORD.to_vec());
-        let password2 = SafeCell::new(PASSWORD.to_vec());
+        let mut password = SafeCell::new(PASSWORD.to_vec());
+        let mut password2 = SafeCell::new(PASSWORD.to_vec());
         let salt = generate_salt();
 
-        let mut key1 = derive_key(password, &salt);
-        let mut key2 = derive_key(password2, &salt);
+        let mut key1 = derive_key(&mut password, &salt);
+        let mut key2 = derive_key(&mut password2, &salt);
 
         let key1_reader = key1.0.read();
         let key2_reader = key2.0.read();
@@ -77,10 +77,10 @@ mod tests {
     #[test]
     fn successful_derive() {
         static PASSWORD: &[u8] = b"password";
-        let password = SafeCell::new(PASSWORD.to_vec());
+        let mut password = SafeCell::new(PASSWORD.to_vec());
         let salt = generate_salt();
 
-        let mut key = derive_key(password, &salt);
+        let mut key = derive_key(&mut password, &salt);
         let key_reader = key.0.read();
 
         assert_ne!(key_reader.as_slice(), &[0u8; 32][..]);

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

@@ -215,8 +215,6 @@ mod tests {
         },
         db::{self, schema},
     };
-    use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
-
     use super::{Error, Integrable, sign_entity, verify_entity};
     #[derive(Clone, arbiter_macros::Hashable)]
     struct DummyEntity {
@@ -235,7 +233,7 @@ mod tests {
         );
         actor
             .ask(Bootstrap {
-                seal_key_raw: SafeCell::new(b"integrity-test-seal-key".to_vec()),
+                seal_key: crate::crypto::KeyCell::from([0u8; 32]),
             })
             .await
             .unwrap();

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

@@ -1,5 +1,5 @@
 use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
-use encryption::v1::{Nonce, Salt};
+use encryption::v1::Nonce;
 
 use argon2::{Algorithm, Argon2};
 use chacha20poly1305::{
@@ -13,6 +13,7 @@ use rand::{
 
 pub mod encryption;
 pub mod integrity;
+pub mod shamir;
 
 pub struct KeyCell(pub SafeCell<Key>);
 impl From<SafeCell<Key>> for KeyCell {
@@ -20,6 +21,15 @@ impl From<SafeCell<Key>> for KeyCell {
         Self(value)
     }
 }
+impl From<[u8; 32]> for KeyCell {
+    fn from(bytes: [u8; 32]) -> Self {
+        let cell = SafeCell::new_inline_default(|key: &mut Key| {
+            key.copy_from_slice(&bytes);
+        });
+        Self(cell)
+    }
+}
+
 impl TryFrom<SafeCell<Vec<u8>>> for KeyCell {
     type Error = ();
 
@@ -28,7 +38,7 @@ impl TryFrom<SafeCell<Vec<u8>>> for KeyCell {
         if value.len() != size_of::<Key>() {
             return Err(());
         }
-        let cell = SafeCell::new_inline(|cell_write: &mut Key| {
+        let cell = SafeCell::new_inline_default(|cell_write: &mut Key| {
             cell_write.copy_from_slice(&value);
         });
         Ok(Self(cell))
@@ -37,7 +47,7 @@ impl TryFrom<SafeCell<Vec<u8>>> for KeyCell {
 
 impl KeyCell {
     pub fn new_secure_random() -> Self {
-        let key = SafeCell::new_inline(|key_buffer: &mut Key| {
+        let key = SafeCell::new_inline_default(|key_buffer: &mut Key| {
             let mut rng = StdRng::try_from_rng(&mut SysRng)
                 .expect("Rng failure is unrecoverable and should panic");
             rng.fill_bytes(key_buffer);
@@ -94,7 +104,7 @@ impl KeyCell {
 }
 
 /// Derive a fixed-length key from the password using Argon2id, which is designed for password hashing and key derivation.
-pub fn derive_key(mut password: SafeCell<Vec<u8>>, salt: &Salt) -> KeyCell {
+pub fn derive_key(password: &mut SafeCell<Vec<u8>>, salt: &[u8]) -> KeyCell {
     let params = {
         #[cfg(debug_assertions)]
         {
@@ -132,10 +142,10 @@ mod tests {
     #[test]
     fn encrypt_decrypt() {
         static PASSWORD: &[u8] = b"password";
-        let password = SafeCell::new(PASSWORD.to_vec());
+        let mut password = SafeCell::new(PASSWORD.to_vec());
         let salt = generate_salt();
 
-        let mut key = derive_key(password, &salt);
+        let mut key = derive_key(&mut password, &salt);
         let nonce = Nonce(*b"unique nonce 123 1231233"); // 24 bytes for XChaCha20Poly1305
         let associated_data = b"associated data";
         let mut buffer = b"secret data".to_vec();

server/crates/arbiter-server/src/crypto/shamir.rs

@@ -0,0 +1,41 @@
+use vsss_rs::Gf256;
+
+#[derive(Debug, thiserror::Error)]
+pub enum ShamirError {
+    #[error("Failed to split key: {0}")]
+    Split(String),
+    #[error("Failed to combine shares: {0}")]
+    Combine(String),
+}
+
+/// Split `key` into `total` shares where any `threshold` shares can reconstruct it.
+/// Each returned Vec<u8> is a share with format [`identifier_byte`, `value_bytes`...].
+pub fn split_key(
+    threshold: usize,
+    total: usize,
+    key: &[u8; 32],
+    rng: impl rand_core::RngCore + rand_core::CryptoRng,
+) -> Result<Vec<Vec<u8>>, ShamirError> {
+    Gf256::split_array(threshold, total, key.as_slice(), rng)
+        .map_err(|e| ShamirError::Split(format!("{e:?}")))
+}
+
+/// Returns the minimum number of shares required to reconstruct the secret
+/// for a committee of `n` operators.
+#[must_use]
+pub const fn shamir_threshold(n: usize) -> usize {
+    match n {
+        0 => panic!("No operators"),
+        1 => 1,
+        2 => 2,
+        n => n / 2 + 1,
+    }
+}
+
+/// Reconstruct the secret from `threshold` or more shares.
+pub fn combine_shares(shares: &[Vec<u8>]) -> Result<[u8; 32], ShamirError> {
+    let bytes = Gf256::combine_array(shares)
+        .map_err(|e| ShamirError::Combine(format!("{e:?}")))?;
+    <[u8; 32]>::try_from(bytes.as_slice())
+        .map_err(|_| ShamirError::Combine("unexpected reconstructed key length".to_owned()))
+}

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

@@ -15,10 +15,11 @@ use restructed::Models;
 pub mod types {
     use chrono::{DateTime, Utc};
     use diesel::{
+        backend::Backend,
         deserialize::{FromSql, FromSqlRow},
         expression::AsExpression,
         serialize::{IsNull, ToSql},
-        sql_types::Integer,
+        sql_types::{Integer, Text},
         sqlite::{Sqlite, SqliteType},
     };
 
@@ -61,7 +62,7 @@ pub mod types {
 
     impl FromSql<Integer, Sqlite> for SqliteTimestamp {
         fn from_sql(
-            mut bytes: <Sqlite as diesel::backend::Backend>::RawValue<'_>,
+            mut bytes: <Sqlite as Backend>::RawValue<'_>,
         ) -> diesel::deserialize::Result<Self> {
             let Some(SqliteType::Long) = bytes.value_type() else {
                 return Err(format!(
@@ -141,6 +142,45 @@ pub mod types {
     declare_id!(TlsHistoryId);
     declare_id!(EvmWalletId);
     declare_id!(ClientId);
+
+    #[derive(Debug, Clone, PartialEq, Eq, AsExpression, FromSqlRow)]
+    #[diesel(sql_type = Text)]
+    pub enum ProposalStatus {
+        Pending,
+        Approved,
+        Rejected,
+        Expired,
+    }
+
+    impl ToSql<Text, Sqlite> for ProposalStatus {
+        fn to_sql<'b>(
+            &'b self,
+            out: &mut diesel::serialize::Output<'b, '_, Sqlite>,
+        ) -> diesel::serialize::Result {
+            let s: &str = match self {
+                Self::Pending => "pending",
+                Self::Approved => "approved",
+                Self::Rejected => "rejected",
+                Self::Expired => "expired",
+            };
+            <str as ToSql<Text, Sqlite>>::to_sql(s, out)
+        }
+    }
+
+    impl FromSql<Text, Sqlite> for ProposalStatus {
+        fn from_sql(
+            bytes: <Sqlite as Backend>::RawValue<'_>,
+        ) -> diesel::deserialize::Result<Self> {
+            let s = <String as FromSql<Text, Sqlite>>::from_sql(bytes)?;
+            match s.as_str() {
+                "pending" => Ok(Self::Pending),
+                "approved" => Ok(Self::Approved),
+                "rejected" => Ok(Self::Rejected),
+                "expired" => Ok(Self::Expired),
+                other => Err(format!("Unknown proposal status: {other}").into()),
+            }
+        }
+    }
 }
 pub use types::*;
 
@@ -271,8 +311,8 @@ pub struct ProgramClient {
 }
 
 #[derive(Queryable, Debug)]
-#[diesel(table_name = schema::operator_identity, check_for_backend(Sqlite))]
-pub struct OperatorIdentity {
+#[diesel(table_name = schema::operator_client, check_for_backend(Sqlite))]
+pub struct OperatorClient {
     pub id: OperatorIdentityId,
     pub public_key: Vec<u8>,
     pub created_at: SqliteTimestamp,
@@ -285,6 +325,7 @@ pub struct Operator {
     pub id: OperatorId,
     pub share: Vec<u8>,
     pub share_nonce: Vec<u8>,
+    pub share_salt: Vec<u8>,
     pub created_at: SqliteTimestamp,
     pub updated_at: SqliteTimestamp,
 }
@@ -437,3 +478,68 @@ pub struct IntegrityEnvelope {
     pub signed_at: SqliteTimestamp,
     pub created_at: SqliteTimestamp,
 }
+
+#[derive(Debug, Queryable, Selectable, Identifiable)]
+#[diesel(table_name = schema::proposal, check_for_backend(Sqlite))]
+pub struct Proposal {
+    pub id: i32,
+    pub kind: String,
+    pub payload: Vec<u8>,
+    pub initiator_id: i32,
+    pub created_at: SqliteTimestamp,
+    pub expires_at: SqliteTimestamp,
+    pub status: ProposalStatus,
+}
+
+#[derive(Debug, Insertable)]
+#[diesel(table_name = schema::proposal, check_for_backend(Sqlite))]
+pub struct NewProposal {
+    pub kind: String,
+    pub payload: Vec<u8>,
+    pub initiator_id: i32,
+    // status defaults to 'pending' at the DB layer
+    pub expires_at: SqliteTimestamp,
+}
+
+#[derive(Debug, Queryable, Selectable, Identifiable)]
+#[diesel(table_name = schema::proposal_vote, check_for_backend(Sqlite))]
+pub struct ProposalVote {
+    pub id: i32,
+    pub proposal_id: i32,
+    pub operator_id: i32,
+    pub approve: bool,
+    pub signature: Vec<u8>,
+    pub voted_at: SqliteTimestamp,
+}
+
+#[derive(Debug, Insertable)]
+#[diesel(table_name = schema::proposal_vote, check_for_backend(Sqlite))]
+pub struct NewProposalVote {
+    pub proposal_id: i32,
+    pub operator_id: i32,
+    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

@@ -157,6 +157,7 @@ diesel::table! {
         id -> Nullable<Integer>,
         share -> Binary,
         share_nonce -> Binary,
+        share_salt -> Binary,
         created_at -> Integer,
         updated_at -> Integer,
     }
@@ -171,6 +172,78 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    proposal (id) {
+        id -> Integer,
+        kind -> Text,
+        payload -> Binary,
+        initiator_id -> Integer,
+        created_at -> Integer,
+        expires_at -> Integer,
+        status -> Text,
+    }
+}
+
+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,
+        proposal_id -> Integer,
+        operator_id -> Integer,
+        approve -> Bool,
+        signature -> Binary,
+        voted_at -> Integer,
+    }
+}
+
 diesel::table! {
     program_client (id) {
         id -> Integer,
@@ -224,9 +297,22 @@ diesel::joinable!(evm_wallet_access -> evm_wallet (wallet_id));
 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,
@@ -244,6 +330,8 @@ diesel::allow_tables_to_appear_in_same_query!(
     operator,
     operator_identity,
     program_client,
+    proposal,
+    proposal_vote,
     root_key_history,
     tls_history,
 );

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

@@ -179,8 +179,7 @@ impl Engine {
         }
 
         if run_kind == RunKind::Execution {
-            conn.transaction(|conn| {
-                Box::pin(async move {
+            conn.transaction(async |conn| {
                 let log_id: i32 = insert_into(evm_transaction_log::table)
                     .values(&NewEvmTransactionLog {
                         grant_id: grant.common_settings_id,
@@ -190,14 +189,13 @@ impl Engine {
                         signed_at: Utc::now().into(),
                     })
                     .returning(evm_transaction_log::id)
-                        .get_result(conn)
+                    .get_result(&mut *conn)
                     .await?;
 
-                    P::record_transaction(&context, meaning, log_id, &grant, conn).await?;
+                P::record_transaction(&context, meaning, log_id, &grant, &mut *conn).await?;
 
                 QueryResult::Ok(())
             })
-            })
             .await
             .map_err(DatabaseError::from)?;
         }
@@ -222,8 +220,7 @@ impl Engine {
         let vault = self.vault.clone();
 
         let id = conn
-            .transaction(|conn| {
-                Box::pin(async move {
+            .transaction(async |conn| {
                 use schema::evm_basic_grant;
 
                 #[expect(
@@ -259,18 +256,17 @@ impl Engine {
                         revoked_at: None,
                     })
                     .returning(evm_basic_grant::all_columns)
-                        .get_result(conn)
+                    .get_result(&mut *conn)
                     .await?;
 
-                    P::create_grant(&basic_grant, &full_grant.specific, conn).await?;
+                P::create_grant(&basic_grant, &full_grant.specific, &mut *conn).await?;
 
-                    integrity::sign_entity(conn, &vault, &full_grant, basic_grant.id)
+                integrity::sign_entity(&mut *conn, &vault, &full_grant, basic_grant.id)
                     .await
                     .map_err(|_| diesel::result::Error::RollbackTransaction)?;
 
                 QueryResult::Ok(basic_grant.id)
             })
-            })
             .await?;
 
         Ok(id)

server/crates/arbiter-server/src/evm/safe_signer.rs

@@ -44,7 +44,7 @@ impl std::fmt::Debug for SafeSigner {
 /// Returns the protected key bytes and the derived Ethereum address.
 pub fn generate(rng: &mut impl rand::Rng) -> (SafeCell<[u8; 32]>, Address) {
     loop {
-        let mut cell = SafeCell::new_inline(|w: &mut [u8; 32]| {
+        let mut cell = SafeCell::new_inline_default(|w: &mut [u8; 32]| {
             rng.fill_bytes(w);
         });
 

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

@@ -19,6 +19,7 @@ use tracing::{error, info, warn};
 
 mod auth;
 mod evm;
+mod governance;
 mod inbound;
 mod outbound;
 mod sdk_client;
@@ -115,6 +116,7 @@ async fn dispatch_inner(
             warn!("Unsupported post-auth operator auth request");
             Err(Status::invalid_argument("Unsupported operator request"))
         }
+        OperatorRequestPayload::Governance(req) => governance::dispatch(actor, req).await,
     }
 }
 

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

@@ -0,0 +1,155 @@
+use crate::{
+    actors::proposal_manager::{Error as ProposalError, ProposalKind, VoteOutcome},
+    peers::operator::{
+        OperatorSession,
+        session::handlers::{HandleCastVote, HandleCreateProposal, HandleQueryPending},
+    },
+};
+use arbiter_proto::proto::operator::{
+    governance::{
+        self as proto_gov, CreateProposalRequest, QueryPendingRequest, QueryPendingResponse,
+        VoteOutcome as ProtoVoteOutcome, create_proposal_request::Kind as ProtoKind,
+        request::Payload as GovRequestPayload, response::Payload as GovResponsePayload,
+    },
+    operator_response::Payload as OperatorResponsePayload,
+};
+use kameo::actor::ActorRef;
+use tonic::Status;
+use tracing::warn;
+
+const fn wrap(payload: GovResponsePayload) -> OperatorResponsePayload {
+    OperatorResponsePayload::Governance(proto_gov::Response {
+        payload: Some(payload),
+    })
+}
+
+pub(super) async fn dispatch(
+    actor: &ActorRef<OperatorSession>,
+    req: proto_gov::Request,
+) -> Result<Option<OperatorResponsePayload>, Status> {
+    let Some(payload) = req.payload else {
+        return Err(Status::invalid_argument(
+            "Missing governance request payload",
+        ));
+    };
+
+    match payload {
+        GovRequestPayload::Create(req) => handle_create(actor, req).await,
+        GovRequestPayload::Vote(req) => handle_vote(actor, req).await,
+        GovRequestPayload::Query(QueryPendingRequest {}) => handle_query(actor).await,
+    }
+}
+
+async fn handle_create(
+    actor: &ActorRef<OperatorSession>,
+    req: CreateProposalRequest,
+) -> Result<Option<OperatorResponsePayload>, Status> {
+    let kind = match req.kind {
+        Some(ProtoKind::ApproveSdkClient(p)) => ProposalKind::ApproveSdkClient {
+            client_id: p.client_id,
+        },
+        Some(ProtoKind::GrantWalletAccess(p)) => ProposalKind::GrantWalletAccess {
+            wallet_id: p.wallet_id,
+            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);
+
+    let proposal_id = actor
+        .ask(HandleCreateProposal { kind, ttl_secs })
+        .await
+        .map_err(|e| {
+            warn!(?e, "create_proposal failed");
+            Status::internal("Failed to create proposal")
+        })?;
+
+    Ok(Some(wrap(GovResponsePayload::Created(
+        proto_gov::CreateProposalResponse { proposal_id },
+    ))))
+}
+
+async fn handle_vote(
+    actor: &ActorRef<OperatorSession>,
+    req: proto_gov::CastVoteRequest,
+) -> Result<Option<OperatorResponsePayload>, Status> {
+    let result = actor
+        .ask(HandleCastVote {
+            proposal_id: req.proposal_id,
+            approve: req.approve,
+            signature: req.signature,
+        })
+        .await;
+
+    let outcome = match result {
+        Ok(VoteOutcome::Pending) => ProtoVoteOutcome::Pending,
+        Ok(VoteOutcome::QuorumApproved) => ProtoVoteOutcome::Approved,
+        Ok(VoteOutcome::QuorumRejected) => ProtoVoteOutcome::Rejected,
+        Err(kameo::error::SendError::HandlerError(ProposalError::AlreadyVoted)) => {
+            return Err(Status::invalid_argument("Already voted on this proposal"));
+        }
+        Err(kameo::error::SendError::HandlerError(ProposalError::InvalidSignature)) => {
+            return Err(Status::invalid_argument("Invalid vote signature"));
+        }
+        Err(kameo::error::SendError::HandlerError(ProposalError::ProposalNotFound)) => {
+            return Err(Status::not_found("Proposal not found"));
+        }
+        Err(e) => {
+            warn!(?e, "cast_vote failed");
+            return Err(Status::internal("Failed to cast vote"));
+        }
+    };
+
+    Ok(Some(wrap(GovResponsePayload::Voted(
+        proto_gov::VoteResponse {
+            outcome: outcome.into(),
+        },
+    ))))
+}
+
+async fn handle_query(
+    actor: &ActorRef<OperatorSession>,
+) -> Result<Option<OperatorResponsePayload>, Status> {
+    let summaries = actor.ask(HandleQueryPending {}).await.unwrap_or_default();
+
+    let proposals = summaries
+        .into_iter()
+        .map(|s| proto_gov::ProposalSummary {
+            id: s.id,
+            kind: s.kind,
+            initiator_id: s.initiator_id,
+            expires_at: s.expires_at.0.timestamp(),
+            approve_count: s.approve_count,
+            reject_count: s.reject_count,
+        })
+        .collect();
+
+    Ok(Some(wrap(GovResponsePayload::Pending(
+        QueryPendingResponse { proposals },
+    ))))
+}

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

@@ -1,5 +1,5 @@
 use crate::{
-    db::models::{EvmWalletAccess, EvmWalletId},
+    db::models::EvmWalletAccess,
     evm::policies::{SharedGrantSettings, SpecificGrant, TransactionRateLimit, VolumeRateLimit},
     grpc::Convert,
 };

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

@@ -1,5 +1,5 @@
 use crate::{
-    db::models::{ClientId, NewEvmWalletAccess},
+    db::models::NewEvmWalletAccess,
     grpc::Convert,
     peers::operator::{
         OperatorSession, OutOfBand,

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

@@ -1,16 +1,24 @@
 use crate::{
     actors::vault::VaultState,
-    peers::operator::{OperatorSession, session::handlers::HandleQueryVaultState},
+    peers::operator::{
+        OperatorSession,
+        session::handlers::{
+            HandleContributeRecoveryRekeyPassphrase, HandleContributeRekeyPassphrase,
+            HandleQueryVaultState,
+        },
+    },
 };
 use arbiter_proto::{
-    proto::shared::VaultState as ProtoVaultState,
     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,
         },
     },
+    proto::shared::VaultState as ProtoVaultState,
 };
 
 use kameo::actor::ActorRef;
@@ -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

@@ -1,14 +1,17 @@
 use crate::{
     grpc::{Convert, TryConvert},
     peers::operator::vault_gate::{
-        self as vault_gate, HandleBootstrapEncryptedKey, HandleHandshake, HandleUnsealEncryptedKey,
+        self as vault_gate, HandleBootstrapEncryptedKey, HandleContributeBootstrapPassphrase,
+        HandleContributeRecoveryBootstrapPassphrase, HandleContributeRecoveryUnsealPassphrase,
+        HandleContributeUnsealPassphrase, HandleDeclareCommittee, HandleHandshake,
+        HandleUnsealEncryptedKey,
     },
 };
 use arbiter_proto::proto::operator::{
     operator_request::Payload as OperatorRequestPayload,
     vault::{
         self as proto_vault,
-        bootstrap::{self as proto_bootstrap},
+        bootstrap::{self as proto_bootstrap, request::Payload as BootstrapRequestPayload},
         request::Payload as VaultRequestPayload,
         unseal::{self as proto_unseal, request::Payload as UnsealRequestPayload},
     },
@@ -50,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",
+            )),
         }
     }
 }
@@ -73,6 +79,21 @@ impl TryConvert for UnsealRequestPayload {
         match self {
             Self::Start(start) => start.try_convert(),
             Self::EncryptedKey(key) => Ok(key.convert()),
+            Self::ContributePassphrase(cp) => Ok(
+                vault_gate::Inbound::HandleContributeUnsealPassphrase(
+                    HandleContributeUnsealPassphrase {
+                        passphrase: cp.passphrase,
+                    },
+                ),
+            ),
+            Self::ContributeRecoveryPassphrase(crp) => Ok(
+                vault_gate::Inbound::HandleContributeRecoveryUnsealPassphrase(
+                    HandleContributeRecoveryUnsealPassphrase {
+                        recovery_operator_id: crp.recovery_operator_id,
+                        passphrase: crp.passphrase,
+                    },
+                ),
+            ),
         }
     }
 }
@@ -107,12 +128,44 @@ impl TryConvert for proto_bootstrap::Request {
     type Error = Status;
 
     fn try_convert(self) -> Result<vault_gate::Inbound, Status> {
-        self.encrypted_key
-            .ok_or_else(|| Status::invalid_argument("Missing bootstrap encrypted key"))?
+        self.payload
+            .ok_or_else(|| Status::invalid_argument("Missing bootstrap payload"))?
             .try_convert()
     }
 }
 
+impl TryConvert for BootstrapRequestPayload {
+    type Output = vault_gate::Inbound;
+    type Error = Status;
+
+    fn try_convert(self) -> Result<vault_gate::Inbound, Status> {
+        match self {
+            Self::EncryptedKey(key) => key.try_convert(),
+            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(
+                vault_gate::Inbound::HandleContributeBootstrapPassphrase(
+                    HandleContributeBootstrapPassphrase {
+                        passphrase: cp.passphrase,
+                    },
+                ),
+            ),
+            Self::ContributeRecoveryPassphrase(crp) => Ok(
+                vault_gate::Inbound::HandleContributeRecoveryBootstrapPassphrase(
+                    HandleContributeRecoveryBootstrapPassphrase {
+                        recovery_operator_id: crp.recovery_operator_id,
+                        passphrase: crp.passphrase,
+                    },
+                ),
+            ),
+        }
+    }
+}
+
 impl TryConvert for proto_bootstrap::BootstrapEncryptedKey {
     type Output = vault_gate::Inbound;
     type Error = Status;

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

@@ -4,7 +4,6 @@ use crate::{
     peers::operator::vault_gate::{self as vault_gate},
 };
 use arbiter_proto::proto::{
-    shared::VaultState as ProtoVaultState,
     operator::{
         operator_response::Payload as OperatorResponsePayload,
         vault::{
@@ -17,6 +16,7 @@ use arbiter_proto::proto::{
             },
         },
     },
+    shared::VaultState as ProtoVaultState,
 };
 
 use tonic::Status;
@@ -110,6 +110,68 @@ impl TryConvert for vault_gate::Outbound {
                 };
                 Ok(wrap_bootstrap_response(proto_result))
             }
+            Self::HandleDeclareCommittee(result) => {
+                let proto_result = match result {
+                    Ok(()) => ProtoBootstrapResult::Success,
+                    Err(err) => {
+                        warn!(?err, "declare committee failed");
+                        return Err(Status::internal("Failed to declare committee"));
+                    }
+                };
+                Ok(wrap_bootstrap_response(proto_result))
+            }
+            Self::HandleContributeBootstrapPassphrase(result) => {
+                let proto_result = match result {
+                    Ok(true) => ProtoBootstrapResult::Success,
+                    Ok(false) => ProtoBootstrapResult::AwaitingContributions,
+                    Err(err) => {
+                        warn!(?err, "contribute bootstrap passphrase failed");
+                        return Err(Status::internal("Failed to contribute bootstrap passphrase"));
+                    }
+                };
+                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,
+                    Ok(false) => ProtoUnsealResult::AwaitingContributions,
+                    Err(err) => {
+                        warn!(?err, "contribute unseal passphrase failed");
+                        return Err(Status::internal("Failed to contribute unseal passphrase"));
+                    }
+                };
+                Ok(wrap_unseal_response(UnsealResponsePayload::Result(
+                    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/client/auth.rs

@@ -171,10 +171,7 @@ async fn insert_client(
         Error::DatabasePoolUnavailable
     })?;
 
-    conn.exclusive_transaction(|conn| {
-        let vault = vault.clone();
-        let pubkey = pubkey.clone();
-        Box::pin(async move {
+    conn.exclusive_transaction(async |conn| {
         let metadata_id = insert_into(client_metadata::table)
             .values((
                 client_metadata::name.eq(&metadata.name),
@@ -182,7 +179,7 @@ async fn insert_client(
                 client_metadata::version.eq(&metadata.version),
             ))
             .returning(client_metadata::id)
-                .get_result::<i32>(conn)
+            .get_result::<i32>(&mut *conn)
             .await?;
 
         let client_id = insert_into(program_client::table)
@@ -192,12 +189,12 @@ async fn insert_client(
             ))
             .on_conflict_do_nothing()
             .returning(program_client::id)
-                .get_result::<i32>(conn)
+            .get_result::<i32>(&mut *conn)
             .await?;
 
         integrity::sign_entity(
-                conn,
-                &vault,
+            &mut *conn,
+            vault,
             &ClientCredentials {
                 pubkey: pubkey.clone(),
             },
@@ -211,7 +208,6 @@ async fn insert_client(
 
         Ok(client_id)
     })
-    })
     .await
 }
 
@@ -229,18 +225,15 @@ async fn sync_client_metadata(
         Error::DatabasePoolUnavailable
     })?;
 
-    conn.exclusive_transaction(|conn| {
-        let metadata = metadata.clone();
-        Box::pin(async move {
-            let (current_metadata_id, current): (i32, ProgramClientMetadata) =
-                program_client::table
+    conn.exclusive_transaction(async |conn| {
+        let (current_metadata_id, current): (i32, ProgramClientMetadata) = program_client::table
             .find(client_id)
             .inner_join(client_metadata::table)
             .select((
                 program_client::metadata_id,
                 ProgramClientMetadata::as_select(),
             ))
-                    .first(conn)
+            .first(&mut *conn)
             .await?;
 
         let unchanged = current.name == metadata.name
@@ -255,7 +248,7 @@ async fn sync_client_metadata(
                 client_metadata_history::metadata_id.eq(current_metadata_id),
                 client_metadata_history::client_id.eq(client_id),
             ))
-                .execute(conn)
+            .execute(&mut *conn)
             .await?;
 
         let metadata_id = insert_into(client_metadata::table)
@@ -265,7 +258,7 @@ async fn sync_client_metadata(
                 client_metadata::version.eq(&metadata.version),
             ))
             .returning(client_metadata::id)
-                .get_result::<i32>(conn)
+            .get_result::<i32>(&mut *conn)
             .await?;
 
         update(program_client::table.find(client_id))
@@ -273,12 +266,11 @@ async fn sync_client_metadata(
                 program_client::metadata_id.eq(metadata_id),
                 program_client::updated_at.eq(now),
             ))
-                .execute(conn)
+            .execute(&mut *conn)
             .await?;
 
         Ok::<(), diesel::result::Error>(())
     })
-    })
     .await
     .map_err(|e| {
         error!(error = ?e, "Database error");
@@ -306,7 +298,7 @@ where
 
     let signature = expect_message(transport, |req: Inbound| match req {
         Inbound::AuthChallengeSolution { signature } => Some(signature),
-        _ => None,
+        Inbound::AuthChallengeRequest { .. } => None,
     })
     .await
     .map_err(|e| {

server/crates/arbiter-server/src/peers/operator/auth/state.rs

@@ -14,19 +14,19 @@ use diesel::{ExpressionMethods as _, OptionalExtension as _, QueryDsl};
 use diesel_async::RunQueryDsl;
 use tracing::error;
 
-pub(super) struct ChallengeRequest {
-    pub(super) pubkey: authn::PublicKey,
-    pub(super) bootstrap_token: Option<String>,
+pub(crate) struct ChallengeRequest {
+    pub(crate) pubkey: authn::PublicKey,
+    pub(crate) bootstrap_token: Option<String>,
 }
 
-pub(super) struct ChallengeContext {
-    pub(super) challenge: AuthChallenge,
-    pub(super) pubkey: authn::PublicKey,
-    pub(super) bootstrap_token: Option<String>,
+pub struct ChallengeContext {
+    pub challenge: AuthChallenge,
+    pub pubkey: authn::PublicKey,
+    pub bootstrap_token: Option<String>,
 }
 
-pub(super) struct ChallengeSolution {
-    pub(super) solution: Vec<u8>,
+pub(crate) struct ChallengeSolution {
+    pub(crate) solution: Vec<u8>,
 }
 
 smlang::statemachine!(
@@ -127,8 +127,6 @@ where
         })
     }
 
-    #[allow(missing_docs)]
-    #[allow(clippy::unused_unit)]
     async fn verify_solution(
         &mut self,
         ChallengeContext {

server/crates/arbiter-server/src/peers/operator/mod.rs

@@ -180,6 +180,7 @@ where
 
     Ok(OperatorSession::spawn(OperatorSession::new(
         props.clone(),
+        creds.clone(),
         oob_sender,
     )))
 }

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

@@ -2,8 +2,8 @@ use super::{Error, OperatorSession};
 use crate::{
     actors::{
         evm::{
-            ClientSignTransaction, Generate, ListWallets, OperatorCreateGrant, OperatorListGrants,
-            SignTransactionError as EvmSignError,
+            ClientSignTransaction, Generate, ListWallets, OperatorCreateGrant, OperatorDeleteGrant,
+            OperatorListGrants, SignTransactionError as EvmSignError,
         },
         flow_coordinator::client_connect_approval::ClientApprovalAnswer,
         vault::VaultState,
@@ -122,22 +122,23 @@ impl OperatorSession {
     }
 
     #[message]
-    pub(crate) fn handle_grant_delete(&mut self, grant_id: i32) -> Result<(), GrantMutationError> {
-        // match self
-        //     .props
-        //     .actors
-        //     .evm
-        //     .ask(OperatorDeleteGrant { grant_id })
-        //     .await
-        // {
-        //     Ok(()) => Ok(()),
-        //     Err(err) => {
-        //         error!(?err, "EVM grant delete failed");
-        //         Err(GrantMutationError::Internal)
-        //     }
-        // }
-        let _ = grant_id;
-        todo!()
+    pub(crate) async fn handle_grant_delete(
+        &mut self,
+        grant_id: i32,
+    ) -> Result<(), GrantMutationError> {
+        match self
+            .props
+            .actors
+            .evm
+            .ask(OperatorDeleteGrant { grant_id })
+            .await
+        {
+            Ok(()) => Ok(()),
+            Err(err) => {
+                error!(?err, "EVM grant delete failed");
+                Err(GrantMutationError::Internal)
+            }
+        }
     }
 
     #[message]
@@ -175,21 +176,19 @@ impl OperatorSession {
         entries: Vec<NewEvmWalletAccess>,
     ) -> Result<(), Error> {
         let mut conn = self.props.db.get().await?;
-        conn.transaction(|conn| {
-            Box::pin(async move {
+        conn.transaction(async |conn| {
             use crate::db::schema::evm_wallet_access;
 
             for entry in entries {
                 diesel::insert_into(evm_wallet_access::table)
                     .values(&entry)
                     .on_conflict_do_nothing()
-                        .execute(conn)
+                    .execute(&mut *conn)
                     .await?;
             }
 
             Result::<_, Error>::Ok(())
         })
-        })
         .await?;
         Ok(())
     }
@@ -200,19 +199,17 @@ impl OperatorSession {
         entries: Vec<i32>,
     ) -> Result<(), Error> {
         let mut conn = self.props.db.get().await?;
-        conn.transaction(|conn| {
-            Box::pin(async move {
+        conn.transaction(async |conn| {
             use crate::db::schema::evm_wallet_access;
             for entry in entries {
                 diesel::delete(evm_wallet_access::table)
                     .filter(evm_wallet_access::wallet_id.eq(entry))
-                        .execute(conn)
+                    .execute(&mut *conn)
                     .await?;
             }
 
             Result::<_, Error>::Ok(())
         })
-        })
         .await?;
         Ok(())
     }
@@ -221,8 +218,8 @@ impl OperatorSession {
     pub(crate) async fn handle_list_wallet_access(
         &mut self,
     ) -> Result<Vec<EvmWalletAccess>, Error> {
-        let mut conn = self.props.db.get().await?;
         use crate::db::schema::evm_wallet_access;
+        let mut conn = self.props.db.get().await?;
         let access_entries = evm_wallet_access::table
             .select(EvmWalletAccess::as_select())
             .load::<_>(&mut conn)
@@ -282,3 +279,102 @@ impl OperatorSession {
         Ok(clients)
     }
 }
+
+#[messages]
+impl OperatorSession {
+    #[message]
+    pub(crate) async fn handle_create_proposal(
+        &mut self,
+        kind: crate::actors::proposal_manager::ProposalKind,
+        ttl_secs: Option<i64>,
+    ) -> Result<i32, Error> {
+        use crate::actors::proposal_manager::CreateProposal;
+        let initiator_id = self.credentials.id;
+        self.props
+            .actors
+            .proposal_manager
+            .ask(CreateProposal { kind, initiator_id, ttl_secs })
+            .await
+            .map_err(|e| {
+                error!(?e, "create_proposal failed");
+                Error::internal("Failed to create proposal")
+            })
+    }
+
+    #[message]
+    pub(crate) async fn handle_cast_vote(
+        &mut self,
+        proposal_id: i32,
+        approve: bool,
+        signature: Vec<u8>,
+    ) -> Result<crate::actors::proposal_manager::VoteOutcome, crate::actors::proposal_manager::Error> {
+        use crate::actors::proposal_manager::CastVote;
+        let operator_id = self.credentials.id;
+        self.props
+            .actors
+            .proposal_manager
+            .ask(CastVote { proposal_id, operator_id, approve, signature })
+            .await
+            .map_err(|err| match err {
+                SendError::HandlerError(e) => e,
+                _ => crate::actors::proposal_manager::Error::ExecutionFailed("actor unavailable".to_owned()),
+            })
+    }
+
+    #[message]
+    pub(crate) async fn handle_query_pending(
+        &mut self,
+    ) -> Vec<crate::actors::proposal_manager::ProposalSummary> {
+        use crate::actors::proposal_manager::QueryPending;
+        let operator_id = self.credentials.id;
+        self.props
+            .actors
+            .proposal_manager
+            .ask(QueryPending { operator_id })
+            .await
+            .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/session/mod.rs

@@ -1,4 +1,4 @@
-use super::{OutOfBand, OperatorConnection};
+use super::{Credentials, OutOfBand, OperatorConnection};
 use crate::{
     actors::{
         flow_coordinator::client_connect_approval::ClientApprovalController,
@@ -51,6 +51,7 @@ pub struct PendingClientApproval {
 
 pub struct OperatorSession {
     props: OperatorConnection,
+    credentials: Credentials,
     sender: Box<dyn Sender<OutOfBand>>,
 
     pending_client_approvals: HashMap<Vec<u8>, PendingClientApproval>,
@@ -59,11 +60,12 @@ pub struct OperatorSession {
 pub mod handlers;
 
 impl OperatorSession {
-    pub(crate) fn new(props: OperatorConnection, sender: Box<dyn Sender<OutOfBand>>) -> Self {
+    pub(crate) fn new(props: OperatorConnection, credentials: Credentials, sender: Box<dyn Sender<OutOfBand>>) -> Self {
         Self {
             props,
+            credentials,
             sender,
-            pending_client_approvals: Default::default(),
+            pending_client_approvals: HashMap::default(),
         }
     }
 }

server/crates/arbiter-server/src/peers/operator/vault_gate/mod.rs

@@ -3,8 +3,12 @@ use crate::{
     actors::{
         GlobalActors,
         vault::{self, Bootstrap, GetState, TryUnseal, VaultState, events},
+        vault_coordinator::{
+            ContributeBootstrap, ContributeRecoveryBootstrap, ContributeRecoveryUnseal,
+            ContributeUnseal, StartBootstrap,
         },
-    crypto::integrity::{self},
+    },
+    crypto::{KeyCell, integrity::{self}},
     db::DatabasePool,
 };
 use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
@@ -17,6 +21,9 @@ use tokio::sync::oneshot;
 use tracing::{error, info};
 use x25519_dalek::{EphemeralSecret, PublicKey, SharedSecret};
 
+pub use VaultGateMessage as Inbound;
+pub use VaultGateMessageReply as Outbound;
+
 pub mod state;
 
 #[derive(Debug, thiserror::Error)]
@@ -98,11 +105,9 @@ impl VaultGate {
         nonce: &[u8],
         ciphertext: &[u8],
         associated_data: &[u8],
-    ) -> Result<SafeCell<Vec<u8>>, ()> {
+    ) -> Result<KeyCell, ()> {
         let nonce = XNonce::from_slice(nonce);
-
         let cipher = XChaCha20Poly1305::new(secret.as_bytes().into());
-
         let mut key_buffer = SafeCell::new(ciphertext.to_vec());
 
         let decryption_result = key_buffer.write_inline(|write_handle| {
@@ -110,7 +115,9 @@ impl VaultGate {
         });
 
         match decryption_result {
-            Ok(()) => Ok(key_buffer),
+            Ok(()) => KeyCell::try_from(key_buffer).map_err(|()| {
+                error!("Decrypted key material has unexpected length");
+            }),
             Err(err) => {
                 error!(?err, "Failed to decrypt encrypted key material");
                 Err(())
@@ -119,7 +126,7 @@ impl VaultGate {
     }
 }
 
-#[messages(messages = Inbound, replies = Outbound)]
+#[messages(enum)]
 impl VaultGate {
     #[message]
     pub fn handle_handshake(
@@ -152,17 +159,14 @@ impl VaultGate {
             return Err(Error::State);
         };
 
-        let Ok(seal_key_buffer) = Self::decrypt_key(secret, &nonce, &ciphertext, &associated_data)
-        else {
+        let Ok(seal_key) = Self::decrypt_key(secret, &nonce, &ciphertext, &associated_data) else {
             return Err(Error::InvalidKey);
         };
 
         match self
             .actors
             .vault
-            .ask(TryUnseal {
-                seal_key_raw: seal_key_buffer,
-            })
+            .ask(TryUnseal { seal_key })
             .await
         {
             Ok(()) => {
@@ -192,17 +196,14 @@ impl VaultGate {
             return Err(Error::State);
         };
 
-        let Ok(seal_key_buffer) = Self::decrypt_key(secret, &nonce, &ciphertext, &associated_data)
-        else {
+        let Ok(seal_key) = Self::decrypt_key(secret, &nonce, &ciphertext, &associated_data) else {
             return Err(Error::InvalidKey);
         };
 
         match self
             .actors
             .vault
-            .ask(Bootstrap {
-                seal_key_raw: seal_key_buffer,
-            })
+            .ask(Bootstrap { seal_key })
             .await
         {
             Ok(()) => {
@@ -234,6 +235,89 @@ impl VaultGate {
 
         Ok(answer)
     }
+
+    #[message]
+    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"))
+    }
+
+    #[message]
+    pub async fn handle_contribute_bootstrap_passphrase(
+        &mut self,
+        passphrase: Vec<u8>,
+    ) -> Result<bool, Error> {
+        let passphrase_cell = SafeCell::new(passphrase);
+        self.actors
+            .vault_coordinator
+            .ask(ContributeBootstrap {
+                operator_id: self.auth_creds.id,
+                passphrase: passphrase_cell,
+            })
+            .await
+            .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,
+        passphrase: Vec<u8>,
+    ) -> Result<bool, Error> {
+        let passphrase_cell = SafeCell::new(passphrase);
+        self.actors
+            .vault_coordinator
+            .ask(ContributeUnseal {
+                operator_id: self.auth_creds.id,
+                passphrase: passphrase_cell,
+            })
+            .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/client/auth.rs

@@ -1,8 +1,5 @@
 use super::common::ChannelTransport;
-use arbiter_crypto::{
-    authn::{self, AuthChallenge, CLIENT_CONTEXT},
-    safecell::{SafeCell, SafeCellHandle as _},
-};
+use arbiter_crypto::authn::{self, AuthChallenge, CLIENT_CONTEXT};
 use arbiter_proto::{
     ClientMetadata,
     transport::{Receiver, Sender},
@@ -100,7 +97,7 @@ async fn spawn_test_actors(db: &db::DatabasePool) -> GlobalActors {
     actors
         .vault
         .ask(Bootstrap {
-            seal_key_raw: SafeCell::new(b"test-seal-key".to_vec()),
+            seal_key: arbiter_server::crypto::KeyCell::from([0u8; 32]),
         })
         .await
         .unwrap();

server/crates/arbiter-server/tests/common/mod.rs

@@ -2,7 +2,6 @@
     dead_code,
     reason = "Common test utilities that may not be used in every test"
 )]
-use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
 use arbiter_proto::transport::{Bi, Error, Receiver, Sender};
 use arbiter_server::{
     actors::{GlobalActors, vault::Vault},
@@ -19,7 +18,7 @@ pub(crate) async fn bootstrapped_vault(db: &db::DatabasePool) -> Vault {
         .await
         .unwrap();
     actor
-        .bootstrap(SafeCell::new(b"test-seal-key".to_vec()))
+        .bootstrap(arbiter_server::crypto::KeyCell::from([0u8; 32]))
         .await
         .unwrap();
     actor

server/crates/arbiter-server/tests/operator/auth.rs

@@ -1,8 +1,5 @@
 use super::common::ChannelTransport;
-use arbiter_crypto::{
-    authn::{self, AuthChallenge, OPERATOR_CONTEXT},
-    safecell::{SafeCell, SafeCellHandle as _},
-};
+use arbiter_crypto::authn::{self, AuthChallenge, OPERATOR_CONTEXT};
 use arbiter_proto::transport::{Error as TransportError, Receiver, Sender};
 use arbiter_server::{
     actors::{GlobalActors, bootstrap::GetToken, vault::Bootstrap},
@@ -157,7 +154,7 @@ pub async fn bootstrap_token_auth() {
     actors
         .vault
         .ask(Bootstrap {
-            seal_key_raw: SafeCell::new(b"test-seal-key".to_vec()),
+            seal_key: arbiter_server::crypto::KeyCell::from([0u8; 32]),
         })
         .await
         .unwrap();
@@ -275,7 +272,7 @@ pub async fn challenge_auth() {
     actors
         .vault
         .ask(Bootstrap {
-            seal_key_raw: SafeCell::new(b"test-seal-key".to_vec()),
+            seal_key: arbiter_server::crypto::KeyCell::from([0u8; 32]),
         })
         .await
         .unwrap();
@@ -361,7 +358,7 @@ pub async fn challenge_auth_rejects_integrity_tag_mismatch_when_unsealed() {
     actors
         .vault
         .ask(Bootstrap {
-            seal_key_raw: SafeCell::new(b"test-seal-key".to_vec()),
+            seal_key: arbiter_server::crypto::KeyCell::from([0u8; 32]),
         })
         .await
         .unwrap();
@@ -434,7 +431,7 @@ pub async fn challenge_auth_rejects_invalid_signature() {
     actors
         .vault
         .ask(Bootstrap {
-            seal_key_raw: SafeCell::new(b"test-seal-key".to_vec()),
+            seal_key: arbiter_server::crypto::KeyCell::from([0u8; 32]),
         })
         .await
         .unwrap();

server/crates/arbiter-server/tests/operator/unseal.rs

@@ -1,7 +1,4 @@
-use arbiter_crypto::{
-    authn,
-    safecell::{SafeCell, SafeCellHandle as _},
-};
+use arbiter_crypto::authn;
 use arbiter_server::{
     actors::{
         GlobalActors,
@@ -22,7 +19,7 @@ use tokio::sync::oneshot;
 use x25519_dalek::{EphemeralSecret, PublicKey};
 
 async fn setup_sealed_gate(
-    seal_key: &[u8],
+    seal_key: &[u8; 32],
 ) -> (
     db::DatabasePool,
     kameo::actor::ActorRef<VaultGate>,
@@ -34,7 +31,7 @@ async fn setup_sealed_gate(
     actors
         .vault
         .ask(Bootstrap {
-            seal_key_raw: SafeCell::new(seal_key.to_vec()),
+            seal_key: arbiter_server::crypto::KeyCell::from(*seal_key),
         })
         .await
         .unwrap();
@@ -50,7 +47,7 @@ async fn setup_sealed_gate(
 
 async fn client_dh_encrypt(
     gate: &kameo::actor::ActorRef<VaultGate>,
-    key_to_send: &[u8],
+    key_to_send: &[u8; 32],
 ) -> HandleUnsealEncryptedKey {
     let client_secret = EphemeralSecret::random();
     let client_public = PublicKey::from(&client_secret);
@@ -83,7 +80,7 @@ async fn client_dh_encrypt(
 #[tokio::test]
 #[test_log::test]
 pub async fn unseal_success() {
-    let seal_key = b"test-seal-key";
+    let seal_key = b"test-seal-key-padded-to-32bytes!";
     let (_db, gate, _promotion_rx) = setup_sealed_gate(seal_key).await;
 
     let encrypted_key = client_dh_encrypt(&gate, seal_key).await;
@@ -95,10 +92,10 @@ pub async fn unseal_success() {
 #[tokio::test]
 #[test_log::test]
 pub async fn unseal_wrong_seal_key() {
-    let seal_key = b"test-seal-key";
+    let seal_key = b"test-seal-key-padded-to-32bytes!";
     let (_db, gate, _promotion_rx) = setup_sealed_gate(seal_key).await;
 
-    let encrypted_key = client_dh_encrypt(&gate, b"wrong-key").await;
+    let encrypted_key = client_dh_encrypt(&gate, b"wrong-key-padded-to-32-bytes!!!!").await;
 
     let response = gate.ask(encrypted_key).await;
     assert!(matches!(
@@ -112,7 +109,7 @@ pub async fn unseal_wrong_seal_key() {
 #[tokio::test]
 #[test_log::test]
 pub async fn unseal_corrupted_ciphertext() {
-    let seal_key = b"test-seal-key";
+    let seal_key = b"test-seal-key-padded-to-32bytes!";
     let (_db, gate, _promotion_rx) = setup_sealed_gate(seal_key).await;
 
     let client_secret = EphemeralSecret::random();
@@ -143,11 +140,11 @@ pub async fn unseal_corrupted_ciphertext() {
 #[tokio::test]
 #[test_log::test]
 pub async fn unseal_retry_after_invalid_key() {
-    let seal_key = b"real-seal-key";
+    let seal_key = b"real-seal-key-padded-to-32bytes!";
     let (_db, gate, _promotion_rx) = setup_sealed_gate(seal_key).await;
 
     {
-        let encrypted_key = client_dh_encrypt(&gate, b"wrong-key").await;
+        let encrypted_key = client_dh_encrypt(&gate, b"wrong-key-padded-to-32-bytes!!!!").await;
 
         let response = gate.ask(encrypted_key).await;
         assert!(matches!(

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

@@ -166,7 +166,7 @@ async fn decrypt_roundtrip_after_high_concurrency() {
         .await
         .unwrap();
     decryptor
-        .try_unseal(SafeCell::new(b"test-seal-key".to_vec()))
+        .try_unseal(arbiter_server::crypto::KeyCell::from([0u8; 32]))
         .await
         .unwrap();
 

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::encryption::v1::{Nonce, ROOT_KEY_TAG},
+    },
+    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]
@@ -20,7 +25,7 @@ async fn test_bootstrap() {
         .await
         .unwrap();
 
-    let seal_key = SafeCell::new(b"test-seal-key".to_vec());
+    let seal_key = KeyCell::from([0u8; 32]);
     actor.bootstrap(seal_key).await.unwrap();
 
     let mut conn = db.get().await.unwrap();
@@ -43,7 +48,7 @@ async fn test_bootstrap_rejects_double() {
     let db = db::create_test_pool().await;
     let mut actor = common::bootstrapped_vault(&db).await;
 
-    let seal_key2 = SafeCell::new(b"test-seal-key".to_vec());
+    let seal_key2 = KeyCell::from([0u8; 32]);
     let err = actor.bootstrap(seal_key2).await.unwrap_err();
     assert!(matches!(err, Error::AlreadyBootstrapped));
 }
@@ -105,7 +110,7 @@ async fn test_unseal_correct_password() {
     let mut actor = Vault::new(db.clone(), GlobalActors::spawn_message_bus())
         .await
         .unwrap();
-    let seal_key = SafeCell::new(b"test-seal-key".to_vec());
+    let seal_key = KeyCell::from([0u8; 32]);
     actor.try_unseal(seal_key).await.unwrap();
 
     let mut decrypted = actor.decrypt(aead_id).await.unwrap();
@@ -129,13 +134,137 @@ async fn test_unseal_wrong_then_correct_password() {
         .await
         .unwrap();
 
-    let bad_key = SafeCell::new(b"wrong-password".to_vec());
+    let bad_key = KeyCell::from([1u8; 32]);
     let err = actor.try_unseal(bad_key).await.unwrap_err();
     assert!(matches!(err, Error::InvalidKey));
 
-    let good_key = SafeCell::new(b"test-seal-key".to_vec());
+    let good_key = KeyCell::from([0u8; 32]);
     actor.try_unseal(good_key).await.unwrap();
 
     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);
+}