feat(integrtity): introduce zero cost wrapper Verified

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)]
@@ -191,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/src/crypto/integrity/v1.rs

@@ -1,9 +1,9 @@
-use crate::{
+use crate::{actors::vault::{self, GetState}, crypto::integrity::hashing::Hashable};
-    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] {
+impl<T: Integrable> Integrable for &T {
-    let mut hasher = Sha256::new();
+    const KIND: &'static str = T::KIND;
-    payload.hash(&mut hasher);
+    const VERSION: i32 = T::VERSION;
-    hasher.finalize().into()
 }
 
-fn push_len_prefixed(out: &mut Vec<u8>, bytes: &[u8]) {
+#[derive(Debug, Clone)]
-    out.extend_from_slice(&(bytes.len() as u32).to_be_bytes());
+pub struct EntityId(Vec<u8>);
-    out.extend_from_slice(bytes);
-}
 
-fn build_mac_input(
+impl Deref for EntityId {
-    entity_kind: &str,
+    type Target = [u8];
-    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
-}
 
-pub trait IntoId {
+    fn deref(&self) -> &Self::Target {
-    fn into_id(self) -> Vec<u8>;
+        &self.0
-}
-
-impl IntoId for i32 {
-    fn into_id(self) -> Vec<u8> {
-        self.to_be_bytes().to_vec()
     }
 }
 
-impl IntoId for &'_ [u8] {
+impl From<i32> for EntityId {
-    fn into_id(self) -> Vec<u8> {
+    fn from(value: i32) -> Self {
-        self.to_vec()
+        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,
+    as_entity_id: Id,
-) -> Result<(), Error> {
+) -> Result<Verified<Id, Nested<E>>, Error> {
-    let payload_hash = payload_hash(&entity);
+    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 entity_id = entity_id.into();
-    let envelope: IntegrityEnvelope = integrity_envelope::table
+    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)) => {
+        Err(SendError::HandlerError(vault::Error::Sealed)) => Ok(AttestationStatus::Unavailable),
-            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)
+        let err = check_entity_attestation(&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

@@ -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::{
+    peers::user_agent::{OutOfBand, UserAgentConnection, UserAgentSession},
-        Credentials, OutOfBand, UserAgentConnection, UserAgentSession,
-        vault_gate::VaultGate,
-    },
 };
 
 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 = {
-    let actor_for_cleanup = actor.clone();
+        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;
+            }
+        }
+    };
 
-    dispatch_loop(bi, actor, oob_receiver, request_tracker).await;
+    info!("User agent session established");
-    actor_for_cleanup.kill();
+
+    dispatch_loop(bi, actor.clone(), oob_receiver, request_tracker).await;
+    actor.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,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, _)) => {
+        Some((id, nonce)) => {
-            verify_integrity(&props.db, &props.actors.vault, &pubkey).await?;
+            let mut db_conn = props.db.get().await.map_err(|e| {
-            id
+                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/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,

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

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