WIP: kameo::messages wiring for transport generalization

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

@@ -58,7 +58,6 @@
 use std::marker::PhantomData;
 
 use async_trait::async_trait;
-use kameo::{error::Infallible, prelude::*};
 
 /// Errors returned by transport adapters implementing [`Bi`].
 #[derive(thiserror::Error, Debug)]
@@ -192,29 +191,3 @@ where
 }
 
 pub mod grpc;
-
-#[derive(thiserror::Error, Debug)]
-pub enum ForwardError<I> {
-    #[error("Transport error: {0}")]
-    Transport(#[from] Error),
-    #[error("Actor delivery error: {0}")]
-    Actor(SendError<I>),
-}
-
-pub async fn forward_to_actor<Transport, Inbound, Outbound, Handler>(
-    transport: &mut Transport,
-    actor: &ActorRef<Handler>,
-) -> Result<(), ForwardError<Inbound>>
-where
-    Transport: Bi<Inbound, <Outbound as Reply>::Ok>,
-    Handler: Actor + Message<Inbound, Reply = Outbound>,
-    Inbound: Send + 'static,
-    Outbound: Send + 'static + Reply<Error = Infallible>, // `Infallible` to enforce contract that `Outbound` carries handler-level error
-{
-    while let Some(request) = transport.recv().await {
-        let resp = actor.ask(request).await.map_err(ForwardError::Actor)?;
-        transport.send(resp).await?
-    }
-
-    Err(Error::ChannelClosed.into())
-}

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

@@ -1,9 +1,9 @@
-use crate::{actors::vault::{self, GetState}, crypto::integrity::hashing::Hashable};
+use crate::{
+    actors::vault::{self, GetState},
+    crypto::integrity::hashing::Hashable,
+};
 use hmac::Hmac;
 use sha2::Sha256;
-use std::future::Future;
-use std::ops::Deref;
-use std::pin::Pin;
 
 use diesel::{ExpressionMethods as _, QueryDsl, dsl::insert_into, sqlite::Sqlite};
 use diesel_async::{AsyncConnection, RunQueryDsl};
@@ -11,23 +11,16 @@ use kameo::{actor::ActorRef, error::SendError};
 use sha2::Digest as _;
 
 pub mod hashing;
-pub mod verified;
 
 use crate::{
     actors::vault::{SignIntegrity, Vault, VerifyIntegrity},
     db::{
         self,
-        models::{IntegrityEnvelope as IntegrityEnvelopeRow, NewIntegrityEnvelope},
+        models::{IntegrityEnvelope, NewIntegrityEnvelope},
         schema::integrity_envelope,
     },
 };
 
-pub const CURRENT_PAYLOAD_VERSION: i32 = 1;
-pub const INTEGRITY_SUBKEY_TAG: &[u8] = b"arbiter/db-integrity-key/v1";
-
-pub type HmacSha256 = Hmac<Sha256>;
-pub use self::verified::{Nested, VerificationOrigin, Verified};
-
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
     #[error("Database error: {0}")]
@@ -56,90 +49,71 @@ pub enum Error {
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
-#[must_use]
 pub enum AttestationStatus {
     Attested,
     Unavailable,
 }
 
+pub const CURRENT_PAYLOAD_VERSION: i32 = 1;
+pub const INTEGRITY_SUBKEY_TAG: &[u8] = b"arbiter/db-integrity-key/v1";
+
+pub type HmacSha256 = Hmac<Sha256>;
+
 pub trait Integrable: Hashable {
     const KIND: &'static str;
     const VERSION: i32 = 1;
 }
 
-impl<T: Integrable> Integrable for &T {
+fn payload_hash(payload: &impl Hashable) -> [u8; 32] {
-    const KIND: &'static str = T::KIND;
+    let mut hasher = Sha256::new();
-    const VERSION: i32 = T::VERSION;
+    payload.hash(&mut hasher);
+    hasher.finalize().into()
 }
 
-#[derive(Debug, Clone)]
+fn push_len_prefixed(out: &mut Vec<u8>, bytes: &[u8]) {
-pub struct EntityId(Vec<u8>);
+    out.extend_from_slice(&(bytes.len() as u32).to_be_bytes());
+    out.extend_from_slice(bytes);
+}
 
-impl Deref for EntityId {
+fn build_mac_input(
-    type Target = [u8];
+    entity_kind: &str,
+    entity_id: &[u8],
+    payload_version: i32,
+    payload_hash: &[u8; 32],
+) -> Vec<u8> {
+    let mut out = Vec::with_capacity(8 + entity_kind.len() + entity_id.len() + 32);
+    push_len_prefixed(&mut out, entity_kind.as_bytes());
+    push_len_prefixed(&mut out, entity_id);
+    out.extend_from_slice(&payload_version.to_be_bytes());
+    out.extend_from_slice(payload_hash);
+    out
+}
 
-    fn deref(&self) -> &Self::Target {
+pub trait IntoId {
-        &self.0
+    fn into_id(self) -> Vec<u8>;
+}
+
+impl IntoId for i32 {
+    fn into_id(self) -> Vec<u8> {
+        self.to_be_bytes().to_vec()
     }
 }
 
-impl From<i32> for EntityId {
+impl IntoId for &'_ [u8] {
-    fn from(value: i32) -> Self {
+    fn into_id(self) -> Vec<u8> {
-        Self(value.to_be_bytes().to_vec())
+        self.to_vec()
     }
 }
 
-impl From<&'_ [u8]> for EntityId {
+pub async fn sign_entity<E: Integrable>(
-    fn from(bytes: &'_ [u8]) -> Self {
-        Self(bytes.to_vec())
-    }
-}
-
-pub async fn lookup_verified<E, Id, C, F, Fut>(
-    conn: &mut C,
-    vault: &ActorRef<Vault>,
-    entity_id: Id,
-    load: F,
-) -> Result<VerifiedEntity<E, Id>, Error>
-where
-    C: AsyncConnection<Backend = Sqlite>,
-    E: Integrable,
-    Id: Into<EntityId> + Clone,
-    F: FnOnce(&mut C) -> Fut,
-    Fut: Future<Output = Result<E, db::DatabaseError>>,
-{
-    let entity = load(conn).await?;
-    verify_entity(conn, vault, entity, entity_id).await
-}
-
-pub async fn lookup_verified_from_query<E, Id, C, F>(
-    conn: &mut C,
-    vault: &ActorRef<Vault>,
-    load: F,
-) -> Result<VerifiedEntity<E, Id>, Error>
-where
-    C: AsyncConnection<Backend = Sqlite> + Send,
-    E: Integrable,
-    Id: Into<EntityId> + Clone,
-    F: for<'a> FnOnce(
-        &'a mut C,
-    ) -> Pin<
-        Box<dyn Future<Output = Result<(Id, E), db::DatabaseError>> + Send + 'a>,
-    >,
-{
-    let (entity_id, entity) = load(conn).await?;
-    verify_entity(conn, vault, entity, entity_id).await
-}
-
-pub async fn sign_entity<E: Integrable, Id: Into<EntityId> + Clone>(
     conn: &mut impl AsyncConnection<Backend = Sqlite>,
     vault: &ActorRef<Vault>,
     entity: &E,
-    as_entity_id: Id,
+    entity_id: impl IntoId,
-) -> Result<Verified<Id, Nested<E>>, Error> {
+) -> Result<(), Error> {
-    let payload_hash = payload_hash(entity);
+    let payload_hash = payload_hash(&entity);
 
-    let entity_id = as_entity_id.clone().into();
+    let entity_id = entity_id.into_id();
 
     let mac_input = build_mac_input(E::KIND, &entity_id, E::VERSION, &payload_hash);
 
@@ -155,7 +129,7 @@ pub async fn sign_entity<E: Integrable, Id: Into<EntityId> + Clone>(
     insert_into(integrity_envelope::table)
         .values(NewIntegrityEnvelope {
             entity_kind: E::KIND.to_owned(),
-            entity_id: entity_id.to_vec(),
+            entity_id,
             payload_version: E::VERSION,
             key_version,
             mac: mac.to_vec(),
@@ -174,19 +148,19 @@ pub async fn sign_entity<E: Integrable, Id: Into<EntityId> + Clone>(
         .await
         .map_err(db::DatabaseError::from)?;
 
-    Ok(Verified::<Id, Nested<E>>::new(as_entity_id))
+    Ok(())
 }
 
-pub async fn check_entity_attestation<E: Integrable>(
+pub async fn verify_entity<E: Integrable>(
     conn: &mut impl AsyncConnection<Backend = Sqlite>,
     vault: &ActorRef<Vault>,
     entity: &E,
-    entity_id: impl Into<EntityId>,
+    entity_id: impl IntoId,
 ) -> Result<AttestationStatus, Error> {
-    let entity_id = entity_id.into();
+    let entity_id = entity_id.into_id();
-    let envelope: IntegrityEnvelopeRow = integrity_envelope::table
+    let envelope: IntegrityEnvelope = integrity_envelope::table
         .filter(integrity_envelope::entity_kind.eq(E::KIND))
-        .filter(integrity_envelope::entity_id.eq(&*entity_id))
+        .filter(integrity_envelope::entity_id.eq(&entity_id))
         .first(conn)
         .await
         .map_err(|err| match err {
@@ -204,7 +178,7 @@ pub async fn check_entity_attestation<E: Integrable>(
         });
     }
 
-    let payload_hash = payload_hash(entity);
+    let payload_hash = payload_hash(&entity);
     let mac_input = build_mac_input(E::KIND, &entity_id, envelope.payload_version, &payload_hash);
 
     let result = vault
@@ -220,111 +194,24 @@ pub async fn check_entity_attestation<E: Integrable>(
         Ok(false) => Err(Error::MacMismatch {
             entity_kind: E::KIND,
         }),
-        Err(SendError::HandlerError(vault::Error::Sealed)) => Ok(AttestationStatus::Unavailable),
+        Err(SendError::HandlerError(vault::Error::Sealed)) => {
+            Ok(AttestationStatus::Unavailable)
+        }
         Err(_) => Err(Error::VaultSend),
     }
 }
 
-#[derive(Debug, Clone)]
-#[repr(C)]
-pub struct VerifiedEntity<E, Id> {
-    pub entity: Verified<E>,
-    pub entity_id: Verified<Id, Nested<E>>,
-}
-
-impl<E, Id> Deref for VerifiedEntity<E, Id> {
-    type Target = Verified<E>;
-
-    fn deref(&self) -> &Self::Target {
-        &self.entity
-    }
-}
-
-pub async fn verify_entity<E: Integrable, Id: Into<EntityId> + Clone>(
-    conn: &mut impl AsyncConnection<Backend = Sqlite>,
-    vault: &ActorRef<Vault>,
-    entity: E,
-    entity_id: Id,
-) -> Result<VerifiedEntity<E, Id>, Error> {
-    match check_entity_attestation(conn, vault, &entity, entity_id.clone()).await? {
-        AttestationStatus::Attested => Ok(VerifiedEntity {
-            entity: Verified::new(entity),
-            entity_id: Verified::new(entity_id),
-        }),
-        AttestationStatus::Unavailable => Err(Error::Vault(vault::Error::Sealed)),
-    }
-}
-
-pub async fn verify_entity_ref<'e, E: Integrable, Id: Into<EntityId> + Clone>(
-    conn: &mut impl AsyncConnection<Backend = Sqlite>,
-    vault: &ActorRef<Vault>,
-    entity: &'e E,
-    entity_id: Id,
-) -> Result<Verified<VerifiedEntity<&'e E, Id>, Nested<E>>, Error> {
-    match check_entity_attestation(conn, vault, entity, entity_id.clone()).await? {
-        AttestationStatus::Attested => Ok(Verified::<VerifiedEntity<&'e E, Id>, Nested<E>>::new(
-            VerifiedEntity {
-                entity: Verified::new(entity),
-                entity_id: Verified::new(entity_id),
-            },
-        )),
-        AttestationStatus::Unavailable => Err(Error::Vault(vault::Error::Sealed)),
-    }
-}
-
-pub async fn delete_envelope<E: Integrable>(
-    conn: &mut impl AsyncConnection<Backend = Sqlite>,
-    entity_id: impl Into<EntityId>,
-) -> Result<usize, Error> {
-    let entity_id = entity_id.into();
-
-    let affected = diesel::delete(
-        integrity_envelope::table
-            .filter(integrity_envelope::entity_kind.eq(E::KIND))
-            .filter(integrity_envelope::entity_id.eq(&*entity_id)),
-    )
-    .execute(conn)
-    .await
-    .map_err(db::DatabaseError::from)?;
-
-    Ok(affected)
-}
-
 pub async fn is_signing_available(vault: &ActorRef<Vault>) -> Result<bool, Error> {
     let state = vault.ask(GetState).await.map_err(|_| Error::VaultSend)?;
     Ok(matches!(state, vault::VaultState::Unsealed))
 }
 
-fn payload_hash(payload: &impl Hashable) -> [u8; 32] {
-    let mut hasher = Sha256::new();
-    payload.hash(&mut hasher);
-    hasher.finalize().into()
-}
-
-fn build_mac_input(
-    entity_kind: &str,
-    entity_id: &[u8],
-    payload_version: i32,
-    payload_hash: &[u8; 32],
-) -> Vec<u8> {
-    let mut out = Vec::with_capacity(8 + entity_kind.len() + entity_id.len() + 32);
-    push_len_prefixed(&mut out, entity_kind.as_bytes());
-    push_len_prefixed(&mut out, entity_id);
-    out.extend_from_slice(&payload_version.to_be_bytes());
-    out.extend_from_slice(payload_hash);
-    out
-}
-
-fn push_len_prefixed(out: &mut Vec<u8>, bytes: &[u8]) {
-    out.extend_from_slice(&(bytes.len() as u32).to_be_bytes());
-    out.extend_from_slice(bytes);
-}
-
 #[cfg(test)]
 mod tests {
     use diesel::{ExpressionMethods as _, QueryDsl};
     use diesel_async::RunQueryDsl;
     use kameo::{actor::ActorRef, prelude::Spawn};
+
     use sha2::Digest;
 
     use crate::{
@@ -337,7 +224,7 @@ mod tests {
     use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
 
     use super::hashing::Hashable;
-    use super::{Error, Integrable, check_entity_attestation, sign_entity};
+    use super::{Error, Integrable, sign_entity, verify_entity};
 
     #[derive(Clone)]
     struct DummyEntity {
@@ -385,8 +272,7 @@ mod tests {
 
         sign_entity(&mut conn, &vault, &entity, ENTITY_ID)
             .await
-            .unwrap()
+            .unwrap();
-            .drop_verification_provenance();
 
         let count: i64 = schema::integrity_envelope::table
             .filter(schema::integrity_envelope::entity_kind.eq("dummy_entity"))
@@ -397,11 +283,9 @@ mod tests {
             .unwrap();
 
         assert_eq!(count, 1, "envelope row must be created exactly once");
-
+        verify_entity(&mut conn, &vault, &entity, ENTITY_ID)
-        let status = check_entity_attestation(&mut conn, &vault, &entity, ENTITY_ID)
             .await
             .unwrap();
-        assert!(matches!(status, super::AttestationStatus::Attested));
     }
 
     #[tokio::test]
@@ -419,8 +303,7 @@ mod tests {
 
         sign_entity(&mut conn, &vault, &entity, ENTITY_ID)
             .await
-            .unwrap()
+            .unwrap();
-            .drop_verification_provenance();
 
         diesel::update(schema::integrity_envelope::table)
             .filter(schema::integrity_envelope::entity_kind.eq("dummy_entity"))
@@ -430,7 +313,35 @@ mod tests {
             .await
             .unwrap();
 
-        let err = check_entity_attestation(&mut conn, &vault, &entity, ENTITY_ID)
+        let err = verify_entity(&mut conn, &vault, &entity, ENTITY_ID)
+            .await
+            .unwrap_err();
+        assert!(matches!(err, Error::MacMismatch { .. }));
+    }
+
+    #[tokio::test]
+    async fn changed_payload_fails_verification() {
+        let db = db::create_test_pool().await;
+        let vault = bootstrapped_vault(&db).await;
+        let mut conn = db.get().await.unwrap();
+
+        const ENTITY_ID: &[u8] = b"entity-id-21";
+
+        let entity = DummyEntity {
+            payload_version: 1,
+            payload: b"payload-v1".to_vec(),
+        };
+
+        sign_entity(&mut conn, &vault, &entity, ENTITY_ID)
+            .await
+            .unwrap();
+
+        let tampered = DummyEntity {
+            payload: b"payload-v1-but-tampered".to_vec(),
+            ..entity
+        };
+
+        let err = verify_entity(&mut conn, &vault, &tampered, ENTITY_ID)
             .await
             .unwrap_err();
         assert!(matches!(err, Error::MacMismatch { .. }));

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

@@ -1,151 +0,0 @@
-use std::ops::Deref;
-
-use super::Integrable;
-
-mod private {
-    pub trait Sealed {}
-}
-
-/// Marker trait for type-level verification provenance.
-///
-/// This trait is intentionally sealed so external code cannot invent arbitrary
-/// provenance tags and bypass the intended type-level guarantees.
-pub trait VerificationOrigin: private::Sealed {
-    type Origin: VerificationOrigin;
-}
-
-/// Root provenance marker for values directly produced by integrity APIs.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
-pub struct Root;
-
-impl private::Sealed for Root {}
-impl VerificationOrigin for Root {
-    type Origin = Self;
-}
-
-/// Nested provenance marker carrying the source integrable type and previous
-/// provenance marker in the chain.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
-pub struct Nested<From, P: VerificationOrigin = Root>(core::marker::PhantomData<(From, P)>);
-
-impl<T, P: VerificationOrigin> private::Sealed for Nested<T, P> {}
-impl<T, P: VerificationOrigin> VerificationOrigin for Nested<T, P> {
-    type Origin = P::Origin;
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)] 
-// #[derive(Copy)] // fixme!: soundness: Unimplemented Copy helps to avoid accidentally origin-unqualifying due to Deref impl.
-#[repr(transparent)]
-#[must_use = "Verified<T> is a proof-bearing wrapper; use self.drop_verification_provenance() to explicitly discard integrity provenance when needed"]
-pub struct Verified<T, O: VerificationOrigin = Root> {
-    inner: T,
-    origin: core::marker::PhantomData<O>,
-}
-
-impl<T, O: VerificationOrigin> AsRef<T> for Verified<T, O> {
-    fn as_ref(&self) -> &T {
-        &self.inner
-    }
-}
-
-impl<T, N: Integrable, O: VerificationOrigin> Deref for Verified<T, Nested<N, O>> {
-    type Target = Verified<T, O::Origin>;
-
-    fn deref(&self) -> &Self::Target {
-        // SAFETY: `Verified<T, _>` is `#[repr(transparent)]` over `T`, so
-        // `&Verified<T, Nested<U, O>>` and `&Verified<T, O::Origin>` have identical layout.
-        unsafe { &*(self as *const Self as *const Verified<T, O::Origin>) }
-    }
-}
-impl<T> Deref for Verified<T, Root> {
-    type Target = T;
-
-    fn deref(&self) -> &Self::Target {
-        AsRef::as_ref(self)
-    }
-}
-
-impl<T, O: VerificationOrigin> Verified<T, O> {
-    /// Unwraps the verified value, discarding the integrity provenance.
-    pub fn drop_verification_provenance(self) -> T {
-        self.inner
-    }
-
-    /// Downgrades the origin provenance by recursively resolving the terminal
-    /// origin of the verification chain.
-    pub fn unqualify_origin(self) -> Verified<T, O::Origin> {
-        Verified {
-            inner: self.inner,
-            origin: core::marker::PhantomData,
-        }
-    }
-
-    /// Constructs a `Verified<T>` by wrapping a `T`.
-    #[cfg(not(test))]
-    pub(super) const fn new(value: T) -> Self {
-        Self {
-            inner: value,
-            origin: core::marker::PhantomData,
-        }
-    }
-    #[cfg(test)]
-    pub(crate) const fn new(value: T) -> Self {
-        Self {
-            inner: value,
-            origin: core::marker::PhantomData,
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    use crate::crypto::integrity::v1::hashing::Hashable;
-    use hmac::digest::Digest;
-    use std::mem::{align_of, size_of};
-
-    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
-    struct Marker;
-
-    impl Hashable for Marker {
-        fn hash<H: Digest>(&self, hasher: &mut H) {
-            hasher.update(b"marker");
-        }
-    }
-
-    impl Integrable for Marker {
-        const KIND: &'static str = "marker";
-    }
-
-    #[test]
-    fn verified_root_exposes_inner_value() {
-        let verified = Verified::<_, Root>::new("root-value");
-
-        assert_eq!(verified.as_ref(), &"root-value");
-        assert_eq!(*verified, "root-value");
-        assert_eq!(verified.drop_verification_provenance(), "root-value");
-        assert_eq!(size_of::<Verified<&str>>(), size_of::<&str>());
-        assert_eq!(align_of::<Verified<&str>>(), align_of::<&str>());
-    }
-
-    #[test]
-    fn nested_verified_derefs_back_to_root() {
-        let verified: Verified<_, Nested<Marker, Nested<Marker>>> = Verified::new("nested-value");
-
-        let _: &Verified<&str, Root> = &verified;
-        let root_view: Verified<&str, Root> = verified.unqualify_origin();
-
-        assert_eq!(root_view.as_ref(), &"nested-value");
-    }
-
-    #[test]
-    fn nested_verified_can_be_unqualified_to_root() {
-        let verified: Verified<_, Nested<Marker>> = Verified::new("nested-value");
-
-        let downgraded = verified.unqualify_origin();
-
-        assert_eq!(downgraded.as_ref(), &"nested-value");
-        assert_eq!(downgraded.drop_verification_provenance(), "nested-value");
-    }
-}

server/crates/arbiter-server/src/grpc/client/auth.rs

@@ -22,9 +22,8 @@ use tonic::Status;
 use tracing::warn;
 
 use crate::{
-    crypto::integrity::{Nested, Verified},
     grpc::request_tracker::RequestTracker,
-    peers::client::{self, ClientConnection, ClientCredentials, auth},
+    peers::client::{self, ClientConnection, auth},
 };
 
 pub struct AuthTransportAdapter<'a> {
@@ -198,7 +197,7 @@ pub async fn start(
     conn: &mut ClientConnection,
     bi: &mut GrpcBi<ClientRequest, ClientResponse>,
     request_tracker: &mut RequestTracker,
-) -> Result<Verified<i32, Nested<ClientCredentials>>, auth::Error> {
+) -> Result<i32, auth::Error> {
     let mut transport = AuthTransportAdapter::new(bi, request_tracker);
     client::auth::authenticate(conn, &mut transport).await
 }

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

@@ -1,4 +1,4 @@
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, oneshot};
 
 use arbiter_proto::{
     proto::user_agent::{
@@ -9,13 +9,17 @@ use arbiter_proto::{
     transport::{Error as TransportError, Receiver, Sender, grpc::GrpcBi},
 };
 use async_trait::async_trait;
-use kameo::actor::ActorRef;
+use kameo::actor::{ActorRef, Spawn as _};
 use tonic::Status;
 use tracing::{error, info, warn};
 
 use crate::{
+    crypto::integrity,
     grpc::request_tracker::RequestTracker,
-    peers::user_agent::{OutOfBand, UserAgentConnection, UserAgentSession},
+    peers::user_agent::{
+        Credentials, OutOfBand, UserAgentConnection, UserAgentSession,
+        vault_gate::VaultGate,
+    },
 };
 
 mod auth;
@@ -125,23 +129,115 @@ pub async fn start(
 ) {
     let mut request_tracker = RequestTracker::default();
 
-    let (oob_sender, oob_receiver) = mpsc::channel(16);
+    let auth_creds = match auth::start(&mut conn, &mut bi, &mut request_tracker).await {
-    let oob_adapter = OutOfBandAdapter(oob_sender);
+        Ok(creds) => creds,
-
-    let actor = {
-        let transport = auth::AuthTransportAdapter::new(&mut bi, &mut request_tracker);
-        match crate::peers::user_agent::start(&mut conn, transport, Box::new(oob_adapter)).await {
-            Ok(actor) => actor,
         Err(e) => {
-                warn!(error = ?e, "User agent connection failed");
+            warn!(error = ?e, "Authentication failed");
             return;
         }
-        }
     };
 
-    info!("User agent session established");
+    info!(pubkey = ?auth_creds.creds.pubkey, "User authenticated successfully");
 
-    dispatch_loop(bi, actor.clone(), oob_receiver, request_tracker).await;
+    let creds = if integrity::is_signing_available(&conn.actors.vault)
-    actor.kill();
+        .await
+        .unwrap_or(false)
+    {
+        // Vault is unsealed; integrity was verified during auth — promote directly.
+        auth_creds.creds
+    } else {
+        // Vault is sealed/unbootstrapped; run the VaultGate phase.
+        let (promotion_tx, promotion_rx) = oneshot::channel();
+        let gate = VaultGate::spawn(VaultGate::new(
+            auth_creds,
+            conn.actors.clone(),
+            conn.db.clone(),
+            promotion_tx,
+        ));
+
+        let result = vault_gate_loop(&mut bi, &gate, &mut request_tracker, promotion_rx).await;
+        gate.kill();
+
+        match result {
+            Some(creds) => creds,
+            None => return,
+        }
+    };
+
+    let (oob_sender, oob_receiver) = mpsc::channel(16);
+    let oob_adapter = OutOfBandAdapter(oob_sender);
+
+    let actor = UserAgentSession::spawn(UserAgentSession::new(conn, creds, Box::new(oob_adapter)));
+    let actor_for_cleanup = actor.clone();
+
+    dispatch_loop(bi, actor, oob_receiver, request_tracker).await;
+    actor_for_cleanup.kill();
 }
 
+async fn vault_gate_loop(
+    bi: &mut GrpcBi<UserAgentRequest, UserAgentResponse>,
+    gate: &ActorRef<VaultGate>,
+    request_tracker: &mut RequestTracker,
+    mut promotion_rx: oneshot::Receiver<Result<Credentials, crate::peers::user_agent::vault_gate::Error>>,
+) -> Option<Credentials> {
+    loop {
+        tokio::select! {
+            result = &mut promotion_rx => {
+                return match result {
+                    Ok(Ok(creds)) => Some(creds),
+                    Ok(Err(e)) => {
+                        warn!(error = ?e, "VaultGate promotion failed");
+                        None
+                    }
+                    Err(_) => {
+                        warn!("VaultGate promotion channel closed unexpectedly");
+                        None
+                    }
+                };
+            }
+
+            message = bi.recv() => {
+                let Some(message) = message else { return None; };
+
+                let conn = match message {
+                    Ok(conn) => conn,
+                    Err(err) => {
+                        warn!(error = ?err, "Failed to receive request during vault gate phase");
+                        return None;
+                    }
+                };
+
+                let request_id = match request_tracker.request(conn.id) {
+                    Ok(id) => id,
+                    Err(err) => {
+                        let _ = bi.send(Err(err)).await;
+                        return None;
+                    }
+                };
+
+                let Some(payload) = conn.payload else {
+                    let _ = bi.send(Err(Status::invalid_argument("Missing request payload"))).await;
+                    return None;
+                };
+
+                let response = match payload {
+                    UserAgentRequestPayload::Vault(req) => vault_gate::dispatch(gate, req).await,
+                    _ => Err(Status::permission_denied("Only vault operations are permitted before unsealing")),
+                };
+
+                match response {
+                    Ok(Some(payload)) => {
+                        if bi.send(Ok(UserAgentResponse { id: Some(request_id), payload: Some(payload) })).await.is_err() {
+                            return None;
+                        }
+                    }
+                    Ok(None) => {}
+                    Err(status) => {
+                        let _ = bi.send(Err(status)).await;
+                        return None;
+                    }
+                }
+            }
+        }
+    }
+}

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

@@ -1,3 +1,4 @@
+#![forbid(unsafe_code)]
 use crate::context::ServerContext;
 
 pub mod actors;

server/crates/arbiter-server/src/peers/client/auth.rs

@@ -18,7 +18,7 @@ use crate::{
         flow_coordinator::{self, RequestClientApproval},
         vault::Vault,
     },
-    crypto::integrity::{self, Nested, Verified},
+    crypto::integrity::{self, AttestationStatus},
     db::{
         self,
         models::{ProgramClientMetadata, SqliteTimestamp},
@@ -104,6 +104,44 @@ async fn get_current_nonce_and_id(
         })
 }
 
+async fn verify_integrity(
+    db: &db::DatabasePool,
+    vault: &ActorRef<Vault>,
+    pubkey: &authn::PublicKey,
+) -> Result<(), Error> {
+    let mut db_conn = db.get().await.map_err(|e| {
+        error!(error = ?e, "Database pool error");
+        Error::DatabasePoolUnavailable
+    })?;
+
+    let (id, nonce) = get_current_nonce_and_id(db, pubkey).await?.ok_or_else(|| {
+        error!("Client not found during integrity verification");
+        Error::DatabaseOperationFailed
+    })?;
+
+    let attestation = integrity::verify_entity(
+        &mut db_conn,
+        vault,
+        &ClientCredentials {
+            pubkey: pubkey.clone(),
+            nonce,
+        },
+        id,
+    )
+    .await
+    .map_err(|e| {
+        error!(?e, "Integrity verification failed");
+        Error::IntegrityCheckFailed
+    })?;
+
+    if attestation != AttestationStatus::Attested {
+        error!("Integrity attestation unavailable for client {id}");
+        return Err(Error::IntegrityCheckFailed);
+    }
+
+    Ok(())
+}
+
 /// Atomically increments the nonce and re-signs the integrity envelope.
 /// Returns the new nonce, which is used as the challenge nonce.
 async fn create_nonce(
@@ -176,7 +214,7 @@ async fn insert_client(
     vault: &ActorRef<Vault>,
     pubkey: &authn::PublicKey,
     metadata: &ClientMetadata,
-) -> Result<Verified<i32, Nested<ClientCredentials>>, Error> {
+) -> Result<i32, Error> {
     use crate::db::schema::{client_metadata, program_client};
     let pubkey = pubkey.clone();
     let metadata = metadata.clone();
@@ -213,7 +251,7 @@ async fn insert_client(
                 .get_result::<i32>(conn)
                 .await?;
 
-            let verified_id = integrity::sign_entity(
+            integrity::sign_entity(
                 conn,
                 &vault,
                 &ClientCredentials {
@@ -228,7 +266,7 @@ async fn insert_client(
                 Error::DatabaseOperationFailed
             })?;
 
-            Ok(verified_id)
+            Ok(client_id)
         })
     })
     .await
@@ -236,7 +274,7 @@ async fn insert_client(
 
 async fn sync_client_metadata(
     db: &db::DatabasePool,
-    client_id: &Verified<i32, Nested<ClientCredentials>>,
+    client_id: i32,
     metadata: &ClientMetadata,
 ) -> Result<(), Error> {
     use crate::db::schema::{client_metadata, client_metadata_history};
@@ -253,7 +291,7 @@ async fn sync_client_metadata(
         Box::pin(async move {
             let (current_metadata_id, current): (i32, ProgramClientMetadata) =
                 program_client::table
-                    .find(client_id.as_ref())
+                    .find(client_id)
                     .inner_join(client_metadata::table)
                     .select((
                         program_client::metadata_id,
@@ -272,7 +310,7 @@ async fn sync_client_metadata(
             insert_into(client_metadata_history::table)
                 .values((
                     client_metadata_history::metadata_id.eq(current_metadata_id),
-                    client_metadata_history::client_id.eq(client_id.as_ref()),
+                    client_metadata_history::client_id.eq(client_id),
                 ))
                 .execute(conn)
                 .await?;
@@ -287,7 +325,7 @@ async fn sync_client_metadata(
                 .get_result::<i32>(conn)
                 .await?;
 
-            update(program_client::table.find(client_id.as_ref()))
+            update(program_client::table.find(client_id))
                 .set((
                     program_client::metadata_id.eq(metadata_id),
                     program_client::updated_at.eq(now),
@@ -342,10 +380,7 @@ where
     Ok(())
 }
 
-pub async fn authenticate<T>(
+pub async fn authenticate<T>(props: &mut ClientConnection, transport: &mut T) -> Result<i32, Error>
-    props: &mut ClientConnection,
-    transport: &mut T,
-) -> Result<Verified<i32, Nested<ClientCredentials>>, Error>
 where
     T: Bi<Inbound, Result<Outbound, Error>> + Send + ?Sized,
 {
@@ -354,27 +389,9 @@ where
     };
 
     let client_id = match get_current_nonce_and_id(&props.db, &pubkey).await? {
-        Some((id, nonce)) => {
+        Some((id, _)) => {
-            let mut db_conn = props.db.get().await.map_err(|e| {
+            verify_integrity(&props.db, &props.actors.vault, &pubkey).await?;
-                error!(error = ?e, "Database pool error");
+            id
-                Error::DatabasePoolUnavailable
-            })?;
-
-            integrity::verify_entity(
-                &mut db_conn,
-                &props.actors.vault,
-                ClientCredentials {
-                    pubkey: pubkey.clone(),
-                    nonce,
-                },
-                id,
-            )
-            .await
-            .map_err(|e| {
-                error!(?e, "Integrity verification failed");
-                Error::IntegrityCheckFailed
-            })?
-            .entity_id
         }
         None => {
             approve_new_client(
@@ -389,7 +406,7 @@ where
         }
     };
 
-    sync_client_metadata(&props.db, &client_id, &metadata).await?;
+    sync_client_metadata(&props.db, client_id, &metadata).await?;
     let challenge_nonce = create_nonce(&props.db, &props.actors.vault, &pubkey).await?;
     challenge_client(transport, pubkey, challenge_nonce).await?;
 

server/crates/arbiter-server/src/peers/client/session.rs

@@ -10,24 +10,19 @@ use crate::{
         flow_coordinator::RegisterClient,
         vault::VaultState,
     },
-    crypto::integrity::{Nested, Verified},
     db,
     evm::VetError,
 };
 
 use super::ClientConnection;
-use super::ClientCredentials;
 
 pub struct ClientSession {
     props: ClientConnection,
-    client_id: Verified<i32, Nested<ClientCredentials>>,
+    client_id: i32,
 }
 
 impl ClientSession {
-    pub(crate) fn new(
+    pub(crate) fn new(props: ClientConnection, client_id: i32) -> Self {
-        props: ClientConnection,
-        client_id: Verified<i32, Nested<ClientCredentials>>,
-    ) -> Self {
         Self { props, client_id }
     }
 }
@@ -60,7 +55,7 @@ impl ClientSession {
             .actors
             .evm
             .ask(ClientSignTransaction {
-                client_id: *self.client_id.as_ref(),
+                client_id: self.client_id,
                 wallet_address,
                 transaction,
             })
@@ -98,12 +93,11 @@ impl Actor for ClientSession {
 }
 
 impl ClientSession {
-    #[cfg(test)]
     pub fn new_test(db: db::DatabasePool, actors: GlobalActors) -> Self {
         let props = ClientConnection::new(db, actors);
         Self {
             props,
-            client_id: Verified::new(0),
+            client_id: 0,
         }
     }
 }

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

@@ -7,7 +7,7 @@ use kameo::actor::ActorRef;
 use tracing::error;
 
 use super::Error;
-use crate::{crypto::integrity::{Nested, Verified}, peers::user_agent::auth::Outbound};
+use crate::peers::user_agent::auth::Outbound;
 use crate::{
     actors::{bootstrap::ConsumeToken, vault::Vault},
     crypto::integrity,
@@ -131,7 +131,7 @@ async fn resign_credentials(
     id: i32,
     pubkey: &authn::PublicKey,
     new_nonce: i32,
-) -> Result<Verified<i32, Nested<AuthCredentials>>, Error> {
+) -> Result<(), Error> {
     integrity::sign_entity(
         conn,
         vault,

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

@@ -1,7 +1,7 @@
 use crate::{
     actors::GlobalActors,
     crypto::integrity::{self, Integrable},
-    db::{self, DatabaseError},
+    db,
     peers::client::ClientProfile,
 };
 use arbiter_crypto::authn;
@@ -83,8 +83,6 @@ pub enum Error {
     VaultGate(#[from] vault_gate::Error),
     #[error("transport closed unexpectedly")]
     Transport,
-    #[error("database error: {0}")]
-    Database(DatabaseError),
     #[error("internal: {0}")]
     Internal(String),
 }
@@ -106,13 +104,13 @@ where
 {
     let auth_creds = authenticate(props, &mut transport).await?;
 
-    let creds = match integrity::is_signing_available(&props.actors.vault)
+    let creds = if integrity::is_signing_available(&props.actors.vault)
         .await
-        .map_err(|_| Error::Internal("Integrity verification failed".into()))?
+        .unwrap_or(false)
     {
-        // credentials were checked by `auth` stage
+        auth_creds.creds
-        true => auth_creds.creds,
+    } else {
-        false => run_vault_gate(props, &mut transport, auth_creds).await?,
+        run_vault_gate(props, &mut transport, auth_creds).await?
     };
 
     Ok(UserAgentSession::spawn(UserAgentSession::new(

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

@@ -267,7 +267,7 @@ impl Message<events::Unsealed> for VaultGate {
     ) -> Self::Reply {
         let result = async {
             let mut conn = self.db.get().await.map_err(|_| Error::internal("DB unavailable"))?;
-            match integrity::check_entity_attestation(
+            match integrity::verify_entity(
                 &mut conn,
                 &self.actors.vault,
                 &self.auth_creds,