feat(integrtity): introduce zero cost wrapper Verified

server/Cargo.toml

@@ -26,7 +26,6 @@ thiserror = "2.0.18"
 async-trait = "0.1.89"
 futures = "0.3.32"
 tokio-stream = { version = "0.1.18", features = ["full"] }
-kameo = "0.20"
 prost-types = { version = "0.14.3", features = ["chrono"] }
 x25519-dalek = { version = "2.0.1", features = ["getrandom"] }
 rstest = "0.26.1"
@@ -47,3 +46,6 @@ miette = { version = "7.6.0", features = ["fancy", "serde"] }
 mutants = "0.0.4"
 ml-dsa = { version = "0.1.0-rc.8", features = ["zeroize"] }
 base64 = "0.22.1"
+kameo = {git = "https://github.com/hdbg/kameo.git", rev = "805b417"}
+kameo_actors = {git = "https://github.com/hdbg/kameo.git", rev = "805b417"}
+

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

@@ -58,6 +58,7 @@
 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)]
@@ -106,6 +107,36 @@ pub trait Receiver<Inbound>: Send + Sync {
 /// any built-in correlation mechanism between inbound and outbound items.
 pub trait Bi<Inbound, Outbound>: Sender<Outbound> + Receiver<Inbound> + Send + Sync {}
 
+#[async_trait]
+impl<T, Outbound> Sender<Outbound> for &mut T
+where
+    T: Sender<Outbound> + ?Sized,
+    Outbound: Send + 'static,
+{
+    async fn send(&mut self, item: Outbound) -> Result<(), Error> {
+        (**self).send(item).await
+    }
+}
+
+#[async_trait]
+impl<T, Inbound> Receiver<Inbound> for &mut T
+where
+    T: Receiver<Inbound> + ?Sized,
+    Inbound: Send + 'static,
+{
+    async fn recv(&mut self) -> Option<Inbound> {
+        (**self).recv().await
+    }
+}
+
+impl<T, Inbound, Outbound> Bi<Inbound, Outbound> for &mut T
+where
+    T: Bi<Inbound, Outbound> + ?Sized,
+    Inbound: Send + 'static,
+    Outbound: Send + 'static,
+{
+}
+
 pub trait SplittableBi<Inbound, Outbound>: Bi<Inbound, Outbound> {
     type Sender: Sender<Outbound>;
     type Receiver: Receiver<Inbound>;
@@ -161,3 +192,29 @@ 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/Cargo.toml

@@ -58,7 +58,7 @@ ml-dsa.workspace = true
 ed25519-dalek.workspace = true
 x25519-dalek.workspace = true
 k256.workspace = true
-kameo_actors = "0.5.0"
+kameo_actors.workspace = true
 
 [dev-dependencies]
 insta = "1.46.3"

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,16 +11,23 @@ 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, NewIntegrityEnvelope},
+        models::{IntegrityEnvelope as IntegrityEnvelopeRow, 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}")]
@@ -49,71 +56,90 @@ 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;
 }
 
-fn payload_hash(payload: &impl Hashable) -> [u8; 32] {
-    let mut hasher = Sha256::new();
-    payload.hash(&mut hasher);
-    hasher.finalize().into()
+impl<T: Integrable> Integrable for &T {
+    const KIND: &'static str = T::KIND;
+    const VERSION: i32 = T::VERSION;
 }
 
-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);
-}
+#[derive(Debug, Clone)]
+pub struct EntityId(Vec<u8>);
 
-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
-}
+impl Deref for EntityId {
+    type Target = [u8];
 
-pub trait IntoId {
-    fn into_id(self) -> Vec<u8>;
-}
-
-impl IntoId for i32 {
-    fn into_id(self) -> Vec<u8> {
-        self.to_be_bytes().to_vec()
+    fn deref(&self) -> &Self::Target {
+        &self.0
     }
 }
 
-impl IntoId for &'_ [u8] {
-    fn into_id(self) -> Vec<u8> {
-        self.to_vec()
+impl From<i32> for EntityId {
+    fn from(value: i32) -> Self {
+        Self(value.to_be_bytes().to_vec())
     }
 }
 
-pub async fn sign_entity<E: Integrable>(
+impl From<&'_ [u8]> for EntityId {
+    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,
-    entity_id: impl IntoId,
-) -> Result<(), Error> {
-    let payload_hash = payload_hash(&entity);
+    as_entity_id: Id,
+) -> Result<Verified<Id, Nested<E>>, Error> {
+    let payload_hash = payload_hash(entity);
 
-    let entity_id = entity_id.into_id();
+    let entity_id = as_entity_id.clone().into();
 
     let mac_input = build_mac_input(E::KIND, &entity_id, E::VERSION, &payload_hash);
 
@@ -129,7 +155,7 @@ pub async fn sign_entity<E: Integrable>(
     insert_into(integrity_envelope::table)
         .values(NewIntegrityEnvelope {
             entity_kind: E::KIND.to_owned(),
-            entity_id,
+            entity_id: entity_id.to_vec(),
             payload_version: E::VERSION,
             key_version,
             mac: mac.to_vec(),
@@ -148,19 +174,19 @@ pub async fn sign_entity<E: Integrable>(
         .await
         .map_err(db::DatabaseError::from)?;
 
-    Ok(())
+    Ok(Verified::<Id, Nested<E>>::new(as_entity_id))
 }
 
-pub async fn verify_entity<E: Integrable>(
+pub async fn check_entity_attestation<E: Integrable>(
     conn: &mut impl AsyncConnection<Backend = Sqlite>,
     vault: &ActorRef<Vault>,
     entity: &E,
-    entity_id: impl IntoId,
+    entity_id: impl Into<EntityId>,
 ) -> Result<AttestationStatus, Error> {
-    let entity_id = entity_id.into_id();
-    let envelope: IntegrityEnvelope = integrity_envelope::table
+    let entity_id = entity_id.into();
+    let envelope: IntegrityEnvelopeRow = 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 {
@@ -178,7 +204,7 @@ pub async fn verify_entity<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
@@ -194,24 +220,111 @@ pub async fn verify_entity<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::{
@@ -224,7 +337,7 @@ mod tests {
     use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
 
     use super::hashing::Hashable;
-    use super::{Error, Integrable, sign_entity, verify_entity};
+    use super::{Error, Integrable, check_entity_attestation, sign_entity};
 
     #[derive(Clone)]
     struct DummyEntity {
@@ -272,7 +385,8 @@ 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"))
@@ -283,9 +397,11 @@ 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]
@@ -303,7 +419,8 @@ 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"))
@@ -313,35 +430,7 @@ mod tests {
             .await
             .unwrap();
 
-        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)
+        let err = check_entity_attestation(&mut conn, &vault, &entity, ENTITY_ID)
             .await
             .unwrap_err();
         assert!(matches!(err, Error::MacMismatch { .. }));

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

@@ -0,0 +1,151 @@
+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,8 +22,9 @@ use tonic::Status;
 use tracing::warn;
 
 use crate::{
+    crypto::integrity::{Nested, Verified},
     grpc::request_tracker::RequestTracker,
-    peers::client::{self, ClientConnection, auth},
+    peers::client::{self, ClientConnection, ClientCredentials, auth},
 };
 
 pub struct AuthTransportAdapter<'a> {
@@ -197,7 +198,7 @@ pub async fn start(
     conn: &mut ClientConnection,
     bi: &mut GrpcBi<ClientRequest, ClientResponse>,
     request_tracker: &mut RequestTracker,
-) -> Result<i32, auth::Error> {
+) -> Result<Verified<i32, Nested<ClientCredentials>>, 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, oneshot};
+use tokio::sync::mpsc;
 
 use arbiter_proto::{
     proto::user_agent::{
@@ -9,17 +9,13 @@ use arbiter_proto::{
     transport::{Error as TransportError, Receiver, Sender, grpc::GrpcBi},
 };
 use async_trait::async_trait;
-use kameo::actor::{ActorRef, Spawn as _};
+use kameo::actor::ActorRef;
 use tonic::Status;
 use tracing::{error, info, warn};
 
 use crate::{
-    crypto::integrity,
     grpc::request_tracker::RequestTracker,
-    peers::user_agent::{
-        Credentials, OutOfBand, UserAgentConnection, UserAgentSession,
-        vault_gate::VaultGate,
-    },
+    peers::user_agent::{OutOfBand, UserAgentConnection, UserAgentSession},
 };
 
 mod auth;
@@ -129,115 +125,23 @@ pub async fn start(
 ) {
     let mut request_tracker = RequestTracker::default();
 
-    let auth_creds = match auth::start(&mut conn, &mut bi, &mut request_tracker).await {
-        Ok(creds) => creds,
-        Err(e) => {
-            warn!(error = ?e, "Authentication failed");
-            return;
-        }
-    };
-
-    info!(pubkey = ?auth_creds.creds.pubkey, "User authenticated successfully");
-
-    let creds = if integrity::is_signing_available(&conn.actors.vault)
-        .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;
-                    }
-                }
+    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");
+                return;
             }
         }
-    }
+    };
+
+    info!("User agent session established");
+
+    dispatch_loop(bi, actor.clone(), oob_receiver, request_tracker).await;
+    actor.kill();
 }
+

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

@@ -23,8 +23,8 @@ use crate::{
 };
 
 pub struct AuthTransportAdapter<'a> {
-    bi: &'a mut GrpcBi<UserAgentRequest, UserAgentResponse>,
-    request_tracker: &'a mut RequestTracker,
+    pub(super) bi: &'a mut GrpcBi<UserAgentRequest, UserAgentResponse>,
+    pub(super) request_tracker: &'a mut RequestTracker,
 }
 
 impl<'a> AuthTransportAdapter<'a> {
@@ -38,19 +38,35 @@ impl<'a> AuthTransportAdapter<'a> {
         }
     }
 
-    async fn send_user_agent_response(
+    pub(super) fn bi_mut(&mut self) -> &mut GrpcBi<UserAgentRequest, UserAgentResponse> {
+        self.bi
+    }
+
+    pub(super) fn tracker_mut(&mut self) -> &mut RequestTracker {
+        self.request_tracker
+    }
+
+    pub(super) async fn send_response_payload(
         &mut self,
-        payload: AuthResponsePayload,
+        payload: UserAgentResponsePayload,
     ) -> Result<(), TransportError> {
         self.bi
             .send(Ok(UserAgentResponse {
                 id: Some(self.request_tracker.current_request_id()),
-                payload: Some(UserAgentResponsePayload::Auth(proto_auth::Response {
-                    payload: Some(payload),
-                })),
+                payload: Some(payload),
             }))
             .await
     }
+
+    async fn send_user_agent_response(
+        &mut self,
+        payload: AuthResponsePayload,
+    ) -> Result<(), TransportError> {
+        self.send_response_payload(UserAgentResponsePayload::Auth(proto_auth::Response {
+            payload: Some(payload),
+        }))
+        .await
+    }
 }
 
 #[async_trait]
@@ -168,6 +184,6 @@ pub async fn start(
     bi: &mut GrpcBi<UserAgentRequest, UserAgentResponse>,
     request_tracker: &mut RequestTracker,
 ) -> Result<AuthCredentials, auth::Error> {
-    let transport = AuthTransportAdapter::new(bi, request_tracker);
-    auth::authenticate(conn, transport).await
+    let mut transport = AuthTransportAdapter::new(bi, request_tracker);
+    auth::authenticate(conn, &mut transport).await
 }

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

@@ -1,25 +1,28 @@
-use arbiter_proto::proto::user_agent::{
-    user_agent_response::Payload as UserAgentResponsePayload,
-    vault::{
-        self as proto_vault,
-        bootstrap::{
-            self as proto_bootstrap, BootstrapResult as ProtoBootstrapResult,
-        },
-        request::Payload as VaultRequestPayload,
-        response::Payload as VaultResponsePayload,
-        unseal::{
-            self as proto_unseal, UnsealResult as ProtoUnsealResult, UnsealStart,
-            request::Payload as UnsealRequestPayload, response::Payload as UnsealResponsePayload,
+use arbiter_proto::{
+    proto::user_agent::{
+        user_agent_request::Payload as UserAgentRequestPayload,
+        user_agent_response::Payload as UserAgentResponsePayload,
+        vault::{
+            self as proto_vault,
+            bootstrap::{self as proto_bootstrap, BootstrapResult as ProtoBootstrapResult},
+            request::Payload as VaultRequestPayload,
+            response::Payload as VaultResponsePayload,
+            unseal::{
+                self as proto_unseal, UnsealResult as ProtoUnsealResult,
+                request::Payload as UnsealRequestPayload,
+                response::Payload as UnsealResponsePayload,
+            },
         },
     },
+    transport::{Bi, Error as TransportError, Receiver, Sender},
 };
-use kameo::{actor::ActorRef, error::SendError};
+use async_trait::async_trait;
 use tonic::Status;
 use tracing::warn;
 
+use super::auth::AuthTransportAdapter;
 use crate::peers::user_agent::vault_gate::{
     self as vault_gate, HandleBootstrapEncryptedKey, HandleHandshake, HandleUnsealEncryptedKey,
-    VaultGate,
 };
 
 fn wrap_vault_response(payload: VaultResponsePayload) -> UserAgentResponsePayload {
@@ -40,112 +43,196 @@ fn wrap_bootstrap_response(result: ProtoBootstrapResult) -> UserAgentResponsePay
     }))
 }
 
-pub(super) async fn dispatch(
-    gate: &ActorRef<VaultGate>,
-    req: proto_vault::Request,
-) -> Result<Option<UserAgentResponsePayload>, Status> {
-    let Some(payload) = req.payload else {
-        return Err(Status::invalid_argument("Missing vault request payload"));
-    };
-
-    match payload {
-        VaultRequestPayload::QueryState(_) => {
-            use arbiter_proto::proto::shared::VaultState as ProtoVaultState;
-            Ok(Some(wrap_vault_response(VaultResponsePayload::State(
-                ProtoVaultState::Sealed.into(),
-            ))))
-        }
-        VaultRequestPayload::Unseal(req) => dispatch_unseal(gate, req).await,
-        VaultRequestPayload::Bootstrap(req) => dispatch_bootstrap(gate, req).await,
+impl AuthTransportAdapter<'_> {
+    async fn send_query_state(&mut self) -> Result<(), TransportError> {
+        use arbiter_proto::proto::shared::VaultState as ProtoVaultState;
+        self.send_response_payload(wrap_vault_response(VaultResponsePayload::State(
+            ProtoVaultState::Sealed.into(),
+        )))
+        .await
     }
 }
 
-async fn dispatch_unseal(
-    gate: &ActorRef<VaultGate>,
-    req: proto_unseal::Request,
-) -> Result<Option<UserAgentResponsePayload>, Status> {
-    let Some(payload) = req.payload else {
-        return Err(Status::invalid_argument("Missing unseal request payload"));
-    };
+#[async_trait]
+impl Receiver<vault_gate::Inbound> for AuthTransportAdapter<'_> {
+    async fn recv(&mut self) -> Option<vault_gate::Inbound> {
+        loop {
+            let request = match self.bi_mut().recv().await? {
+                Ok(request) => request,
+                Err(error) => {
+                    warn!(?error, "Failed to receive user agent request during vault gate");
+                    return None;
+                }
+            };
 
-    match payload {
-        UnsealRequestPayload::Start(req) => handle_unseal_start(gate, req).await,
-        UnsealRequestPayload::EncryptedKey(req) => handle_unseal_encrypted_key(gate, req).await,
+            if let Err(err) = self.tracker_mut().request(request.id) {
+                let _ = self.bi_mut().send(Err(err)).await;
+                return None;
+            }
+
+            let Some(payload) = request.payload else {
+                let _ = self
+                    .bi_mut()
+                    .send(Err(Status::invalid_argument("Missing request payload")))
+                    .await;
+                return None;
+            };
+
+            let vault_req = match payload {
+                UserAgentRequestPayload::Vault(req) => req,
+                _ => {
+                    let _ = self
+                        .bi_mut()
+                        .send(Err(Status::permission_denied(
+                            "Only vault operations are permitted before unsealing",
+                        )))
+                        .await;
+                    return None;
+                }
+            };
+
+            let Some(vault_payload) = vault_req.payload else {
+                let _ = self
+                    .bi_mut()
+                    .send(Err(Status::invalid_argument("Missing vault request payload")))
+                    .await;
+                return None;
+            };
+
+            match vault_payload {
+                VaultRequestPayload::QueryState(_) => {
+                    if self.send_query_state().await.is_err() {
+                        return None;
+                    }
+                    continue;
+                }
+                VaultRequestPayload::Unseal(req) => {
+                    let Some(unseal_payload) = req.payload else {
+                        let _ = self
+                            .bi_mut()
+                            .send(Err(Status::invalid_argument("Missing unseal request payload")))
+                            .await;
+                        return None;
+                    };
+                    match unseal_payload {
+                        UnsealRequestPayload::Start(start) => {
+                            let Ok(bytes) = <[u8; 32]>::try_from(start.client_pubkey) else {
+                                let _ = self
+                                    .bi_mut()
+                                    .send(Err(Status::invalid_argument(
+                                        "Invalid X25519 public key",
+                                    )))
+                                    .await;
+                                return None;
+                            };
+                            return Some(vault_gate::Inbound::HandleHandshake(HandleHandshake {
+                                client_pubkey: x25519_dalek::PublicKey::from(bytes),
+                            }));
+                        }
+                        UnsealRequestPayload::EncryptedKey(key) => {
+                            return Some(vault_gate::Inbound::HandleUnsealEncryptedKey(
+                                HandleUnsealEncryptedKey {
+                                    nonce: key.nonce,
+                                    ciphertext: key.ciphertext,
+                                    associated_data: key.associated_data,
+                                },
+                            ));
+                        }
+                    }
+                }
+                VaultRequestPayload::Bootstrap(req) => {
+                    let Some(encrypted_key) = req.encrypted_key else {
+                        let _ = self
+                            .bi_mut()
+                            .send(Err(Status::invalid_argument(
+                                "Missing bootstrap encrypted key",
+                            )))
+                            .await;
+                        return None;
+                    };
+                    return Some(vault_gate::Inbound::HandleBootstrapEncryptedKey(
+                        HandleBootstrapEncryptedKey {
+                            nonce: encrypted_key.nonce,
+                            ciphertext: encrypted_key.ciphertext,
+                            associated_data: encrypted_key.associated_data,
+                        },
+                    ));
+                }
+            }
+        }
     }
 }
 
-async fn handle_unseal_start(
-    gate: &ActorRef<VaultGate>,
-    req: UnsealStart,
-) -> Result<Option<UserAgentResponsePayload>, Status> {
-    let client_pubkey = <[u8; 32]>::try_from(req.client_pubkey)
-        .map(x25519_dalek::PublicKey::from)
-        .map_err(|_| Status::invalid_argument("Invalid X25519 public key"))?;
+#[async_trait]
+impl Sender<Result<vault_gate::Outbound, vault_gate::Error>> for AuthTransportAdapter<'_> {
+    async fn send(
+        &mut self,
+        item: Result<vault_gate::Outbound, vault_gate::Error>,
+    ) -> Result<(), TransportError> {
+        let outbound = match item {
+            Ok(outbound) => outbound,
+            Err(err) => {
+                warn!(?err, "vault gate produced transport-level error");
+                return self
+                    .bi_mut()
+                    .send(Err(Status::internal(err.to_string())))
+                    .await;
+            }
+        };
 
-    let response = gate
-        .ask(HandleHandshake { client_pubkey })
-        .await
-        .map_err(|err| {
-            warn!(error = ?err, "Failed to handle unseal start");
-            Status::internal("Failed to start unseal flow")
-        })?;
+        let payload = match outbound {
+            vault_gate::Outbound::HandleHandshake(Ok(response)) => {
+                wrap_unseal_response(UnsealResponsePayload::Start(
+                    proto_unseal::UnsealStartResponse {
+                        server_pubkey: response.server_pubkey.as_bytes().to_vec(),
+                    },
+                ))
+            }
+            vault_gate::Outbound::HandleHandshake(Err(err)) => {
+                warn!(?err, "handshake failed");
+                return self
+                    .bi_mut()
+                    .send(Err(Status::internal("Failed to start unseal flow")))
+                    .await;
+            }
+            vault_gate::Outbound::HandleUnsealEncryptedKey(result) => {
+                let proto_result = match result {
+                    Ok(()) => ProtoUnsealResult::Success,
+                    Err(vault_gate::Error::InvalidKey) => ProtoUnsealResult::InvalidKey,
+                    Err(err) => {
+                        warn!(?err, "unseal failed");
+                        return self
+                            .bi_mut()
+                            .send(Err(Status::internal("Failed to unseal vault")))
+                            .await;
+                    }
+                };
+                wrap_unseal_response(UnsealResponsePayload::Result(proto_result.into()))
+            }
+            vault_gate::Outbound::HandleBootstrapEncryptedKey(result) => {
+                let proto_result = match result {
+                    Ok(()) => ProtoBootstrapResult::Success,
+                    Err(vault_gate::Error::InvalidKey) => ProtoBootstrapResult::InvalidKey,
+                    Err(vault_gate::Error::AlreadyBootstrapped) => {
+                        ProtoBootstrapResult::AlreadyBootstrapped
+                    }
+                    Err(err) => {
+                        warn!(?err, "bootstrap failed");
+                        return self
+                            .bi_mut()
+                            .send(Err(Status::internal("Failed to bootstrap vault")))
+                            .await;
+                    }
+                };
+                wrap_bootstrap_response(proto_result)
+            }
+        };
 
-    Ok(Some(wrap_unseal_response(UnsealResponsePayload::Start(
-        proto_unseal::UnsealStartResponse {
-            server_pubkey: response.server_pubkey.as_bytes().to_vec(),
-        },
-    ))))
+        self.send_response_payload(payload).await
+    }
 }
 
-async fn handle_unseal_encrypted_key(
-    gate: &ActorRef<VaultGate>,
-    req: arbiter_proto::proto::user_agent::vault::unseal::UnsealEncryptedKey,
-) -> Result<Option<UserAgentResponsePayload>, Status> {
-    let result = match gate
-        .ask(HandleUnsealEncryptedKey {
-            nonce: req.nonce,
-            ciphertext: req.ciphertext,
-            associated_data: req.associated_data,
-        })
-        .await
-    {
-        Ok(()) => ProtoUnsealResult::Success,
-        Err(SendError::HandlerError(vault_gate::Error::InvalidKey)) => ProtoUnsealResult::InvalidKey,
-        Err(err) => {
-            warn!(error = ?err, "Failed to handle unseal request");
-            return Err(Status::internal("Failed to unseal vault"));
-        }
-    };
-    Ok(Some(wrap_unseal_response(UnsealResponsePayload::Result(
-        result.into(),
-    ))))
-}
-
-async fn dispatch_bootstrap(
-    gate: &ActorRef<VaultGate>,
-    req: proto_bootstrap::Request,
-) -> Result<Option<UserAgentResponsePayload>, Status> {
-    let encrypted_key = req
-        .encrypted_key
-        .ok_or_else(|| Status::invalid_argument("Missing bootstrap encrypted key"))?;
-
-    let result = match gate
-        .ask(HandleBootstrapEncryptedKey {
-            nonce: encrypted_key.nonce,
-            ciphertext: encrypted_key.ciphertext,
-            associated_data: encrypted_key.associated_data,
-        })
-        .await
-    {
-        Ok(()) => ProtoBootstrapResult::Success,
-        Err(SendError::HandlerError(vault_gate::Error::InvalidKey)) => ProtoBootstrapResult::InvalidKey,
-        Err(SendError::HandlerError(vault_gate::Error::AlreadyBootstrapped)) => {
-            ProtoBootstrapResult::AlreadyBootstrapped
-        }
-        Err(err) => {
-            warn!(error = ?err, "Failed to handle bootstrap request");
-            return Err(Status::internal("Failed to bootstrap vault"));
-        }
-    };
-    Ok(Some(wrap_bootstrap_response(result)))
+impl Bi<vault_gate::Inbound, Result<vault_gate::Outbound, vault_gate::Error>>
+    for AuthTransportAdapter<'_>
+{
 }

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

@@ -1,4 +1,3 @@
-#![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, AttestationStatus},
+    crypto::integrity::{self, Nested, Verified},
     db::{
         self,
         models::{ProgramClientMetadata, SqliteTimestamp},
@@ -104,44 +104,6 @@ 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(
@@ -214,7 +176,7 @@ async fn insert_client(
     vault: &ActorRef<Vault>,
     pubkey: &authn::PublicKey,
     metadata: &ClientMetadata,
-) -> Result<i32, Error> {
+) -> Result<Verified<i32, Nested<ClientCredentials>>, Error> {
     use crate::db::schema::{client_metadata, program_client};
     let pubkey = pubkey.clone();
     let metadata = metadata.clone();
@@ -251,7 +213,7 @@ async fn insert_client(
                 .get_result::<i32>(conn)
                 .await?;
 
-            integrity::sign_entity(
+            let verified_id = integrity::sign_entity(
                 conn,
                 &vault,
                 &ClientCredentials {
@@ -266,7 +228,7 @@ async fn insert_client(
                 Error::DatabaseOperationFailed
             })?;
 
-            Ok(client_id)
+            Ok(verified_id)
         })
     })
     .await
@@ -274,7 +236,7 @@ async fn insert_client(
 
 async fn sync_client_metadata(
     db: &db::DatabasePool,
-    client_id: i32,
+    client_id: &Verified<i32, Nested<ClientCredentials>>,
     metadata: &ClientMetadata,
 ) -> Result<(), Error> {
     use crate::db::schema::{client_metadata, client_metadata_history};
@@ -291,7 +253,7 @@ async fn sync_client_metadata(
         Box::pin(async move {
             let (current_metadata_id, current): (i32, ProgramClientMetadata) =
                 program_client::table
-                    .find(client_id)
+                    .find(client_id.as_ref())
                     .inner_join(client_metadata::table)
                     .select((
                         program_client::metadata_id,
@@ -310,7 +272,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),
+                    client_metadata_history::client_id.eq(client_id.as_ref()),
                 ))
                 .execute(conn)
                 .await?;
@@ -325,7 +287,7 @@ async fn sync_client_metadata(
                 .get_result::<i32>(conn)
                 .await?;
 
-            update(program_client::table.find(client_id))
+            update(program_client::table.find(client_id.as_ref()))
                 .set((
                     program_client::metadata_id.eq(metadata_id),
                     program_client::updated_at.eq(now),
@@ -380,7 +342,10 @@ where
     Ok(())
 }
 
-pub async fn authenticate<T>(props: &mut ClientConnection, transport: &mut T) -> Result<i32, Error>
+pub async fn authenticate<T>(
+    props: &mut ClientConnection,
+    transport: &mut T,
+) -> Result<Verified<i32, Nested<ClientCredentials>>, Error>
 where
     T: Bi<Inbound, Result<Outbound, Error>> + Send + ?Sized,
 {
@@ -389,9 +354,27 @@ where
     };
 
     let client_id = match get_current_nonce_and_id(&props.db, &pubkey).await? {
-        Some((id, _)) => {
-            verify_integrity(&props.db, &props.actors.vault, &pubkey).await?;
-            id
+        Some((id, nonce)) => {
+            let mut db_conn = props.db.get().await.map_err(|e| {
+                error!(error = ?e, "Database pool error");
+                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(
@@ -406,7 +389,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,19 +10,24 @@ 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: i32,
+    client_id: Verified<i32, Nested<ClientCredentials>>,
 }
 
 impl ClientSession {
-    pub(crate) fn new(props: ClientConnection, client_id: i32) -> Self {
+    pub(crate) fn new(
+        props: ClientConnection,
+        client_id: Verified<i32, Nested<ClientCredentials>>,
+    ) -> Self {
         Self { props, client_id }
     }
 }
@@ -55,7 +60,7 @@ impl ClientSession {
             .actors
             .evm
             .ask(ClientSignTransaction {
-                client_id: self.client_id,
+                client_id: *self.client_id.as_ref(),
                 wallet_address,
                 transaction,
             })
@@ -93,11 +98,12 @@ 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: 0,
+            client_id: Verified::new(0),
         }
     }
 }

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

@@ -68,10 +68,10 @@ fn parse_auth_event(payload: Inbound) -> AuthEvents {
 
 pub async fn authenticate<T>(
     props: &mut UserAgentConnection,
-    transport: T,
+    transport: &mut T,
 ) -> Result<AuthCredentials, Error>
 where
-    T: Bi<Inbound, Result<Outbound, Error>> + Send,
+    T: Bi<Inbound, Result<Outbound, Error>> + Send + ?Sized,
 {
     let mut state = AuthStateMachine::new(AuthContext::new(props, transport));
 

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::peers::user_agent::auth::Outbound;
+use crate::{crypto::integrity::{Nested, Verified}, 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<(), Error> {
+) -> Result<Verified<i32, Nested<AuthCredentials>>, Error> {
     integrity::sign_entity(
         conn,
         vault,
@@ -174,20 +174,20 @@ async fn register_key(db: &DatabasePool, pubkey: &authn::PublicKey) -> Result<i3
     Ok(id)
 }
 
-pub struct AuthContext<'a, T> {
+pub struct AuthContext<'a, T: ?Sized> {
     pub(super) conn: &'a mut UserAgentConnection,
-    pub(super) transport: T,
+    pub(super) transport: &'a mut T,
 }
 
-impl<'a, T> AuthContext<'a, T> {
-    pub fn new(conn: &'a mut UserAgentConnection, transport: T) -> Self {
+impl<'a, T: ?Sized> AuthContext<'a, T> {
+    pub fn new(conn: &'a mut UserAgentConnection, transport: &'a mut T) -> Self {
         Self { conn, transport }
     }
 }
 
 impl<T> AuthStateMachineContext for AuthContext<'_, T>
 where
-    T: Bi<super::Inbound, Result<super::Outbound, Error>> + Send,
+    T: Bi<super::Inbound, Result<super::Outbound, Error>> + Send + ?Sized,
 {
     type Error = Error;
 

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

@@ -1,13 +1,25 @@
 use crate::{
-    actors::GlobalActors, crypto::integrity::Integrable, db, peers::client::ClientProfile,
+    actors::GlobalActors,
+    crypto::integrity::{self, Integrable},
+    db::{self, DatabaseError},
+    peers::client::ClientProfile,
 };
 use arbiter_crypto::authn;
 
+use arbiter_proto::transport::{Bi, Sender};
+pub use auth::authenticate;
+use kameo::actor::{ActorRef, Spawn as _};
+pub use session::UserAgentSession;
+use tokio::sync::oneshot;
+use tracing::warn;
+use vault_gate::VaultGate;
+
+use crate::crypto::integrity::hashing::Hashable;
+
 pub mod auth;
 pub mod session;
 pub mod vault_gate;
 
-
 #[derive(Debug, Clone, Hash)]
 pub struct Credentials {
     pub id: i32,
@@ -40,7 +52,6 @@ impl Hashable for AuthCredentials {
     }
 }
 
-
 impl Integrable for AuthCredentials {
     const KIND: &'static str = "useragent_credentials";
 }
@@ -52,6 +63,7 @@ pub enum OutOfBand {
     ClientConnectionCancel { pubkey: authn::PublicKey },
 }
 
+#[derive(Clone)]
 pub struct UserAgentConnection {
     pub(crate) db: db::DatabasePool,
     pub(crate) actors: GlobalActors,
@@ -63,9 +75,103 @@ impl UserAgentConnection {
     }
 }
 
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("authentication failed: {0:?}")]
+    Auth(auth::Error),
+    #[error("vault gate failed: {0}")]
+    VaultGate(#[from] vault_gate::Error),
+    #[error("transport closed unexpectedly")]
+    Transport,
+    #[error("database error: {0}")]
+    Database(DatabaseError),
+    #[error("internal: {0}")]
+    Internal(String),
+}
 
+impl From<auth::Error> for Error {
+    fn from(err: auth::Error) -> Self {
+        Self::Auth(err)
+    }
+}
 
-pub use auth::authenticate;
-pub use session::UserAgentSession;
+pub async fn start<T>(
+    props: &mut UserAgentConnection,
+    mut transport: T,
+    oob_sender: Box<dyn Sender<OutOfBand>>,
+) -> Result<ActorRef<UserAgentSession>, Error>
+where
+    T: Bi<auth::Inbound, Result<auth::Outbound, auth::Error>> + Send,
+    T: Bi<vault_gate::Inbound, Result<vault_gate::Outbound, vault_gate::Error>> + Send,
+{
+    let auth_creds = authenticate(props, &mut transport).await?;
 
-use crate::crypto::integrity::hashing::Hashable;
+    let creds = match integrity::is_signing_available(&props.actors.vault)
+        .await
+        .map_err(|_| Error::Internal("Integrity verification failed".into()))?
+    {
+        // credentials were checked by `auth` stage
+        true => auth_creds.creds,
+        false => run_vault_gate(props, &mut transport, auth_creds).await?,
+    };
+
+    Ok(UserAgentSession::spawn(UserAgentSession::new(
+        props.clone(),
+        creds,
+        oob_sender,
+    )))
+}
+
+async fn run_vault_gate<T>(
+    props: &UserAgentConnection,
+    transport: &mut T,
+    auth_creds: AuthCredentials,
+) -> Result<Credentials, Error>
+where
+    T: Bi<vault_gate::Inbound, Result<vault_gate::Outbound, vault_gate::Error>> + Send + ?Sized,
+{
+    let (promotion_tx, mut promotion_rx) = oneshot::channel();
+    let gate = VaultGate::spawn(VaultGate::new(
+        auth_creds,
+        props.actors.clone(),
+        props.db.clone(),
+        promotion_tx,
+    ));
+
+    let result = loop {
+        tokio::select! {
+            promotion = &mut promotion_rx => {
+                break match promotion {
+                    Ok(Ok(creds)) => Ok(creds),
+                    Ok(Err(err)) => Err(Error::VaultGate(err)),
+                    Err(_) => Err(Error::Internal(
+                        "vault gate promotion channel closed".into(),
+                    )),
+                };
+            }
+
+            inbound = transport.recv() => {
+                let Some(inbound) = inbound else {
+                    break Err(Error::Transport);
+                };
+
+                match gate.ask(inbound).await {
+                    Ok(outbound) => {
+                        if transport.send(Ok(outbound)).await.is_err() {
+                            break Err(Error::Transport);
+                        }
+                    }
+                    Err(err) => {
+                        warn!(?err, "VaultGate failed to handle message");
+                        break Err(Error::Internal(format!(
+                            "vault gate ask failed: {err:?}"
+                        )));
+                    }
+                }
+            }
+        }
+    };
+
+    gate.kill();
+    result
+}

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

@@ -119,7 +119,7 @@ impl VaultGate {
     }
 }
 
-#[messages]
+#[messages(messages = Inbound, replies = Outbound)]
 impl VaultGate {
     #[message]
     pub async fn handle_handshake(
@@ -185,7 +185,7 @@ impl VaultGate {
     }
 
     #[message]
-    pub(crate) async fn handle_bootstrap_encrypted_key(
+    pub async fn handle_bootstrap_encrypted_key(
         &mut self,
         nonce: Vec<u8>,
         ciphertext: Vec<u8>,
@@ -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::verify_entity(
+            match integrity::check_entity_attestation(
                 &mut conn,
                 &self.actors.vault,
                 &self.auth_creds,

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

@@ -2,17 +2,16 @@ use std::sync::Mutex;
 
 use x25519_dalek::{EphemeralSecret, PublicKey, SharedSecret};
 
-
-
 pub struct Handshake {
     client_pubkey: PublicKey,
 }
 
-
-
 #[derive(Default)]
 pub enum State {
     #[default]
     Idle,
-    ReadyForExchange { server_key: PublicKey, secret: SharedSecret },
+    ReadyForExchange {
+        server_key: PublicKey,
+        secret: SharedSecret,
+    },
 }

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

@@ -42,11 +42,11 @@ pub async fn test_bootstrap_token_auth() {
         .unwrap();
     let token = actors.bootstrapper.ask(GetToken).await.unwrap().unwrap();
 
-    let (server_transport, mut test_transport) = ChannelTransport::new();
+    let (mut server_transport, mut test_transport) = ChannelTransport::new();
     let db_for_task = db.clone();
     let task = tokio::spawn(async move {
         let mut props = UserAgentConnection::new(db_for_task, actors);
-        auth::authenticate(&mut props, server_transport).await
+        auth::authenticate(&mut props, &mut server_transport).await
     });
 
     let new_key = MlDsa87::key_gen(&mut rand::rng());
@@ -84,11 +84,11 @@ pub async fn test_bootstrap_invalid_token_auth() {
     let db = db::create_test_pool().await;
     let actors = GlobalActors::spawn(db.clone()).await.unwrap();
 
-    let (server_transport, mut test_transport) = ChannelTransport::new();
+    let (mut server_transport, mut test_transport) = ChannelTransport::new();
     let db_for_task = db.clone();
     let task = tokio::spawn(async move {
         let mut props = UserAgentConnection::new(db_for_task, actors);
-        auth::authenticate(&mut props, server_transport).await
+        auth::authenticate(&mut props, &mut server_transport).await
     });
 
     let new_key = MlDsa87::key_gen(&mut rand::rng());
@@ -157,11 +157,11 @@ pub async fn test_challenge_auth() {
         .unwrap();
     }
 
-    let (server_transport, mut test_transport) = ChannelTransport::new();
+    let (mut server_transport, mut test_transport) = ChannelTransport::new();
     let db_for_task = db.clone();
     let task = tokio::spawn(async move {
         let mut props = UserAgentConnection::new(db_for_task, actors);
-        auth::authenticate(&mut props, server_transport).await
+        auth::authenticate(&mut props, &mut server_transport).await
     });
 
     test_transport
@@ -234,11 +234,11 @@ pub async fn test_challenge_auth_rejects_integrity_tag_mismatch_when_unsealed()
             .unwrap();
     }
 
-    let (server_transport, mut test_transport) = ChannelTransport::new();
+    let (mut server_transport, mut test_transport) = ChannelTransport::new();
     let db_for_task = db.clone();
     let task = tokio::spawn(async move {
         let mut props = UserAgentConnection::new(db_for_task, actors);
-        auth::authenticate(&mut props, server_transport).await
+        auth::authenticate(&mut props, &mut server_transport).await
     });
 
     test_transport
@@ -298,11 +298,11 @@ pub async fn test_challenge_auth_rejects_invalid_signature() {
         .unwrap();
     }
 
-    let (server_transport, mut test_transport) = ChannelTransport::new();
+    let (mut server_transport, mut test_transport) = ChannelTransport::new();
     let db_for_task = db.clone();
     let task = tokio::spawn(async move {
         let mut props = UserAgentConnection::new(db_for_task, actors);
-        auth::authenticate(&mut props, server_transport).await
+        auth::authenticate(&mut props, &mut server_transport).await
     });
 
     test_transport