feat(transport): add domain error type to GrpcTransportActor

app/.dart_tool/package_config.json

@@ -0,0 +1,178 @@
+{
+  "configVersion": 2,
+  "packages": [
+    {
+      "name": "async",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/async-2.13.0",
+      "packageUri": "lib/",
+      "languageVersion": "3.4"
+    },
+    {
+      "name": "boolean_selector",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/boolean_selector-2.1.2",
+      "packageUri": "lib/",
+      "languageVersion": "3.1"
+    },
+    {
+      "name": "characters",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/characters-1.4.0",
+      "packageUri": "lib/",
+      "languageVersion": "3.4"
+    },
+    {
+      "name": "clock",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/clock-1.1.2",
+      "packageUri": "lib/",
+      "languageVersion": "3.4"
+    },
+    {
+      "name": "collection",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/collection-1.19.1",
+      "packageUri": "lib/",
+      "languageVersion": "3.4"
+    },
+    {
+      "name": "cupertino_icons",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/cupertino_icons-1.0.8",
+      "packageUri": "lib/",
+      "languageVersion": "3.1"
+    },
+    {
+      "name": "fake_async",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/fake_async-1.3.3",
+      "packageUri": "lib/",
+      "languageVersion": "3.3"
+    },
+    {
+      "name": "flutter",
+      "rootUri": "file:///Users/kaska/.local/share/mise/installs/flutter/3.38.9-stable/packages/flutter",
+      "packageUri": "lib/",
+      "languageVersion": "3.8"
+    },
+    {
+      "name": "flutter_lints",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/flutter_lints-6.0.0",
+      "packageUri": "lib/",
+      "languageVersion": "3.8"
+    },
+    {
+      "name": "flutter_test",
+      "rootUri": "file:///Users/kaska/.local/share/mise/installs/flutter/3.38.9-stable/packages/flutter_test",
+      "packageUri": "lib/",
+      "languageVersion": "3.8"
+    },
+    {
+      "name": "leak_tracker",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/leak_tracker-11.0.2",
+      "packageUri": "lib/",
+      "languageVersion": "3.2"
+    },
+    {
+      "name": "leak_tracker_flutter_testing",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/leak_tracker_flutter_testing-3.0.10",
+      "packageUri": "lib/",
+      "languageVersion": "3.2"
+    },
+    {
+      "name": "leak_tracker_testing",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/leak_tracker_testing-3.0.2",
+      "packageUri": "lib/",
+      "languageVersion": "3.2"
+    },
+    {
+      "name": "lints",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/lints-6.1.0",
+      "packageUri": "lib/",
+      "languageVersion": "3.8"
+    },
+    {
+      "name": "matcher",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/matcher-0.12.17",
+      "packageUri": "lib/",
+      "languageVersion": "3.4"
+    },
+    {
+      "name": "material_color_utilities",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/material_color_utilities-0.11.1",
+      "packageUri": "lib/",
+      "languageVersion": "2.17"
+    },
+    {
+      "name": "meta",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/meta-1.17.0",
+      "packageUri": "lib/",
+      "languageVersion": "3.5"
+    },
+    {
+      "name": "path",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/path-1.9.1",
+      "packageUri": "lib/",
+      "languageVersion": "3.4"
+    },
+    {
+      "name": "sky_engine",
+      "rootUri": "file:///Users/kaska/.local/share/mise/installs/flutter/3.38.9-stable/bin/cache/pkg/sky_engine",
+      "packageUri": "lib/",
+      "languageVersion": "3.8"
+    },
+    {
+      "name": "source_span",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/source_span-1.10.2",
+      "packageUri": "lib/",
+      "languageVersion": "3.1"
+    },
+    {
+      "name": "stack_trace",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/stack_trace-1.12.1",
+      "packageUri": "lib/",
+      "languageVersion": "3.4"
+    },
+    {
+      "name": "stream_channel",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/stream_channel-2.1.4",
+      "packageUri": "lib/",
+      "languageVersion": "3.3"
+    },
+    {
+      "name": "string_scanner",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/string_scanner-1.4.1",
+      "packageUri": "lib/",
+      "languageVersion": "3.1"
+    },
+    {
+      "name": "term_glyph",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/term_glyph-1.2.2",
+      "packageUri": "lib/",
+      "languageVersion": "3.1"
+    },
+    {
+      "name": "test_api",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/test_api-0.7.7",
+      "packageUri": "lib/",
+      "languageVersion": "3.5"
+    },
+    {
+      "name": "vector_math",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/vector_math-2.2.0",
+      "packageUri": "lib/",
+      "languageVersion": "3.1"
+    },
+    {
+      "name": "vm_service",
+      "rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/vm_service-15.0.2",
+      "packageUri": "lib/",
+      "languageVersion": "3.5"
+    },
+    {
+      "name": "app",
+      "rootUri": "../",
+      "packageUri": "lib/",
+      "languageVersion": "3.10"
+    }
+  ],
+  "generator": "pub",
+  "generatorVersion": "3.10.8",
+  "flutterRoot": "file:///Users/kaska/.local/share/mise/installs/flutter/3.38.9-stable",
+  "flutterVersion": "3.38.9",
+  "pubCache": "file:///Users/kaska/.pub-cache"
+}

app/.dart_tool/package_graph.json

@@ -0,0 +1,230 @@
+{
+  "roots": [
+    "app"
+  ],
+  "packages": [
+    {
+      "name": "app",
+      "version": "1.0.0+1",
+      "dependencies": [
+        "cupertino_icons",
+        "flutter"
+      ],
+      "devDependencies": [
+        "flutter_lints",
+        "flutter_test"
+      ]
+    },
+    {
+      "name": "flutter_lints",
+      "version": "6.0.0",
+      "dependencies": [
+        "lints"
+      ]
+    },
+    {
+      "name": "flutter_test",
+      "version": "0.0.0",
+      "dependencies": [
+        "clock",
+        "collection",
+        "fake_async",
+        "flutter",
+        "leak_tracker_flutter_testing",
+        "matcher",
+        "meta",
+        "path",
+        "stack_trace",
+        "stream_channel",
+        "test_api",
+        "vector_math"
+      ]
+    },
+    {
+      "name": "cupertino_icons",
+      "version": "1.0.8",
+      "dependencies": []
+    },
+    {
+      "name": "flutter",
+      "version": "0.0.0",
+      "dependencies": [
+        "characters",
+        "collection",
+        "material_color_utilities",
+        "meta",
+        "sky_engine",
+        "vector_math"
+      ]
+    },
+    {
+      "name": "lints",
+      "version": "6.1.0",
+      "dependencies": []
+    },
+    {
+      "name": "stream_channel",
+      "version": "2.1.4",
+      "dependencies": [
+        "async"
+      ]
+    },
+    {
+      "name": "meta",
+      "version": "1.17.0",
+      "dependencies": []
+    },
+    {
+      "name": "collection",
+      "version": "1.19.1",
+      "dependencies": []
+    },
+    {
+      "name": "leak_tracker_flutter_testing",
+      "version": "3.0.10",
+      "dependencies": [
+        "flutter",
+        "leak_tracker",
+        "leak_tracker_testing",
+        "matcher",
+        "meta"
+      ]
+    },
+    {
+      "name": "vector_math",
+      "version": "2.2.0",
+      "dependencies": []
+    },
+    {
+      "name": "stack_trace",
+      "version": "1.12.1",
+      "dependencies": [
+        "path"
+      ]
+    },
+    {
+      "name": "clock",
+      "version": "1.1.2",
+      "dependencies": []
+    },
+    {
+      "name": "fake_async",
+      "version": "1.3.3",
+      "dependencies": [
+        "clock",
+        "collection"
+      ]
+    },
+    {
+      "name": "path",
+      "version": "1.9.1",
+      "dependencies": []
+    },
+    {
+      "name": "matcher",
+      "version": "0.12.17",
+      "dependencies": [
+        "async",
+        "meta",
+        "stack_trace",
+        "term_glyph",
+        "test_api"
+      ]
+    },
+    {
+      "name": "test_api",
+      "version": "0.7.7",
+      "dependencies": [
+        "async",
+        "boolean_selector",
+        "collection",
+        "meta",
+        "source_span",
+        "stack_trace",
+        "stream_channel",
+        "string_scanner",
+        "term_glyph"
+      ]
+    },
+    {
+      "name": "sky_engine",
+      "version": "0.0.0",
+      "dependencies": []
+    },
+    {
+      "name": "material_color_utilities",
+      "version": "0.11.1",
+      "dependencies": [
+        "collection"
+      ]
+    },
+    {
+      "name": "characters",
+      "version": "1.4.0",
+      "dependencies": []
+    },
+    {
+      "name": "async",
+      "version": "2.13.0",
+      "dependencies": [
+        "collection",
+        "meta"
+      ]
+    },
+    {
+      "name": "leak_tracker_testing",
+      "version": "3.0.2",
+      "dependencies": [
+        "leak_tracker",
+        "matcher",
+        "meta"
+      ]
+    },
+    {
+      "name": "leak_tracker",
+      "version": "11.0.2",
+      "dependencies": [
+        "clock",
+        "collection",
+        "meta",
+        "path",
+        "vm_service"
+      ]
+    },
+    {
+      "name": "term_glyph",
+      "version": "1.2.2",
+      "dependencies": []
+    },
+    {
+      "name": "string_scanner",
+      "version": "1.4.1",
+      "dependencies": [
+        "source_span"
+      ]
+    },
+    {
+      "name": "source_span",
+      "version": "1.10.2",
+      "dependencies": [
+        "collection",
+        "path",
+        "term_glyph"
+      ]
+    },
+    {
+      "name": "boolean_selector",
+      "version": "2.1.2",
+      "dependencies": [
+        "source_span",
+        "string_scanner"
+      ]
+    },
+    {
+      "name": "vm_service",
+      "version": "15.0.2",
+      "dependencies": []
+    }
+  ],
+  "configVersion": 1
+}

app/.dart_tool/version

@@ -0,0 +1 @@
+3.38.9

app/macos/Flutter/ephemeral/Flutter-Generated.xcconfig

@@ -0,0 +1,11 @@
+// This is a generated file; do not edit or check into version control.
+FLUTTER_ROOT=/Users/kaska/.local/share/mise/installs/flutter/3.38.9-stable
+FLUTTER_APPLICATION_PATH=/Users/kaska/Documents/Projects/Major/arbiter/app
+COCOAPODS_PARALLEL_CODE_SIGN=true
+FLUTTER_BUILD_DIR=build
+FLUTTER_BUILD_NAME=1.0.0
+FLUTTER_BUILD_NUMBER=1
+DART_OBFUSCATION=false
+TRACK_WIDGET_CREATION=true
+TREE_SHAKE_ICONS=false
+PACKAGE_CONFIG=.dart_tool/package_config.json

app/macos/Flutter/ephemeral/flutter_export_environment.sh

@@ -0,0 +1,12 @@
+#!/bin/sh
+# This is a generated file; do not edit or check into version control.
+export "FLUTTER_ROOT=/Users/kaska/.local/share/mise/installs/flutter/3.38.9-stable"
+export "FLUTTER_APPLICATION_PATH=/Users/kaska/Documents/Projects/Major/arbiter/app"
+export "COCOAPODS_PARALLEL_CODE_SIGN=true"
+export "FLUTTER_BUILD_DIR=build"
+export "FLUTTER_BUILD_NAME=1.0.0"
+export "FLUTTER_BUILD_NUMBER=1"
+export "DART_OBFUSCATION=false"
+export "TRACK_WIDGET_CREATION=true"
+export "TREE_SHAKE_ICONS=false"
+export "PACKAGE_CONFIG=.dart_tool/package_config.json"

server/crates/arbiter-proto/Cargo.toml

@@ -19,6 +19,7 @@ rustls-pki-types.workspace = true
 base64 = "0.22.1"
 tracing.workspace = true
 
+
 [build-dependencies]
 tonic-prost-build = "0.14.3"
 

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

@@ -1,125 +1,371 @@
-use std::marker::PhantomData;
+//! Transport abstraction layer for bridging gRPC bidirectional streaming with kameo actors.
+//!
+//! This module provides a clean separation between the gRPC transport layer and business logic
+//! by modeling the connection as two linked kameo actors:
+//!
+//! - A **transport actor** ([`GrpcTransportActor`]) that owns the gRPC stream and channel,
+//!   forwarding inbound messages to the business actor and outbound messages to the client.
+//! - A **business logic actor** that receives inbound messages from the transport actor and
+//!   sends outbound messages back through the transport actor.
+//!
+//! The [`wire()`] function sets up bidirectional linking between the two actors, ensuring
+//! that if either actor dies, the other is notified and can shut down gracefully.
+//!
+//! # Terminology
+//!
+//! - **InboundMessage**: a message received by the transport actor from the channel/socket
+//!   and forwarded to the business actor.
+//! - **OutboundMessage**: a message produced by the business actor and sent to the transport
+//!   actor to be forwarded to the channel/socket.
+//!
+//! # Architecture
+//!
+//! ```text
+//! gRPC Stream ──InboundMessage──▶ GrpcTransportActor ──tell(InboundMessage)──▶ BusinessActor
+//!                                 ▲                                    │
+//!                                 └─tell(Result<OutboundMessage, _>)────┘
+//!                                 │
+//!                          mpsc::Sender ──▶ Client
+//! ```
+//!
+//! # Example
+//!
+//! ```rust,ignore
+//! let (tx, rx) = mpsc::channel(1000);
+//! let context = server_context.clone();
+//!
+//! wire(
+//!     |transport_ref| MyBusinessActor::new(context, transport_ref),
+//!     |business_recipient, business_id| GrpcTransportActor {
+//!         sender: tx,
+//!         receiver: grpc_stream,
+//!         business_logic_actor: business_recipient,
+//!         business_logic_actor_id: business_id,
+//!     },
+//! ).await;
+//!
+//! Ok(Response::new(ReceiverStream::new(rx)))
+//! ```
 
-use futures::StreamExt;
+use futures::{Stream, StreamExt};
-use tokio::sync::mpsc;
+use kameo::{
+    Actor,
+    actor::{ActorRef, PreparedActor, Recipient, Spawn, WeakActorRef},
+    mailbox::Signal,
+    prelude::Message,
+};
+use tokio::{
+    select,
+    sync::mpsc::{self, error::SendError},
+};
 use tonic::{Status, Streaming};
+use tracing::{debug, error};
 
-/// Errors returned by transport adapters implementing [`Bi`].
+/// A bidirectional stream abstraction for sans-io testing.
-pub enum Error {
-    /// The outbound side of the transport is no longer accepting messages.
-    ChannelClosed,
-}
-
-/// Minimal bidirectional transport abstraction used by protocol code.
 ///
-/// `Bi<T, U, E>` models a duplex channel with:
+/// Combines a [`Stream`] of incoming messages with the ability to [`send`](Bi::send)
-/// - inbound items of type `T` read via [`Bi::recv`]
+/// outgoing responses. This trait allows business logic to be tested without a real
-/// - outbound success items of type `U` or domain errors of type `E` written via [`Bi::send`]
+/// gRPC connection by swapping in an in-memory implementation.
 ///
-/// The trait intentionally exposes only the operations the protocol layer needs,
+/// # Type Parameters
-/// allowing it to work with gRPC streams and other transport implementations.
+/// - `T`: `InboundMessage` received from the channel/socket (e.g., `UserAgentRequest`)
-///
+/// - `U`: `OutboundMessage` sent to the channel/socket (e.g., `UserAgentResponse`)
-/// # Stream termination and errors
+pub trait Bi<T, U>: Stream<Item = Result<T, Status>> + Send + Sync + 'static {
-///
+    type Error;
-/// [`Bi::recv`] returns:
-/// - `Some(item)` when a new inbound message is available
-/// - `None` when the inbound stream ends or the underlying transport reports an error
-///
-/// Implementations may collapse transport-specific receive errors into `None`
-/// when the protocol does not need to distinguish them from normal stream
-/// termination.
-pub trait Bi<T, U, E>: Send + Sync + 'static {
-    /// Sends one outbound result to the peer.
     fn send(
         &mut self,
-        item: Result<U, E>,
+        item: Result<U, Status>,
-    ) -> impl std::future::Future<Output = Result<(), Error>> + Send;
+    ) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send;
-
-    /// Receives the next inbound item.
-    ///
-    /// Returns `None` when the inbound stream is finished or can no longer
-    /// produce items.
-    fn recv(&mut self) -> impl std::future::Future<Output = Option<T>> + Send;
 }
 
-/// [`Bi`] adapter backed by a tonic gRPC bidirectional stream.
+/// Concrete [`Bi`] implementation backed by a tonic gRPC [`Streaming`] and an [`mpsc::Sender`].
 ///
-/// Outbound items are sent through a Tokio MPSC sender, while inbound items are
+/// This is the production implementation used in gRPC service handlers. The `request_stream`
-/// read from tonic [`Streaming`].
+/// receives messages from the client, and `response_sender` sends responses back.
-pub struct GrpcAdapter<Inbound, Outbound, E> {
+pub struct BiStream<T, U> {
-    sender: mpsc::Sender<Result<Outbound, Status>>,
+    pub request_stream: Streaming<T>,
-    receiver: Streaming<Inbound>,
+    pub response_sender: mpsc::Sender<Result<U, Status>>,
-    _error: PhantomData<E>,
 }
 
-impl<Inbound, Outbound, E> GrpcAdapter<Inbound, Outbound, E> {
+impl<T, U> Stream for BiStream<T, U>
+where
+    T: Send + 'static,
+    U: Send + 'static,
+{
+    type Item = Result<T, Status>;
 
-    /// Creates a new gRPC-backed [`Bi`] adapter.
+    fn poll_next(
-    pub fn new(sender: mpsc::Sender<Result<Outbound, Status>>, receiver: Streaming<Inbound>) -> Self {
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        self.request_stream.poll_next_unpin(cx)
+    }
+}
+
+impl<T, U> Bi<T, U> for BiStream<T, U>
+where
+    T: Send + 'static,
+    U: Send + 'static,
+{
+    type Error = SendError<Result<U, Status>>;
+
+    async fn send(&mut self, item: Result<U, Status>) -> Result<(), Self::Error> {
+        self.response_sender.send(item).await
+    }
+}
+
+/// Marker trait for transport actors that can receive outbound messages of type `T`.
+///
+/// Implement this on your transport actor to indicate it can handle outbound messages
+/// produced by the business actor. Requires the actor to implement [`Message<Result<T, E>>`]
+/// so business logic can forward responses via [`tell()`](ActorRef::tell).
+///
+/// # Example
+///
+/// ```rust,ignore
+/// #[derive(Actor)]
+/// struct MyTransportActor { /* ... */ }
+///
+/// impl Message<Result<MyResponse, MyError>> for MyTransportActor {
+///     type Reply = ();
+///     async fn handle(&mut self, msg: Result<MyResponse, MyError>, _ctx: &mut Context<Self, Self::Reply>) -> Self::Reply {
+///         // forward outbound message to channel/socket
+///     }
+/// }
+///
+/// impl TransportActor<MyResponse, MyError> for MyTransportActor {}
+/// ```
+pub trait TransportActor<Outbound: Send + 'static, DomainError: Send + 'static>:
+    Actor + Send + Message<Result<Outbound, DomainError>>
+{
+}
+
+/// A kameo actor that bridges a gRPC bidirectional stream with a business logic actor.
+///
+/// This actor owns the gRPC [`Streaming`] receiver and an [`mpsc::Sender`] for responses.
+/// It multiplexes between its own mailbox (for outbound messages from the business actor)
+/// and the gRPC stream (for inbound client messages) using [`tokio::select!`].
+///
+/// # Message Flow
+///
+/// - **Inbound**: Messages from the gRPC stream are forwarded to `business_logic_actor`
+///   via [`tell()`](Recipient::tell).
+/// - **Outbound**: The business actor sends `Result<Outbound, DomainError>` messages to this
+///   actor, which forwards them through the `sender` channel to the gRPC response stream.
+///
+/// # Lifecycle
+///
+/// - If the business logic actor dies (detected via actor linking), this actor stops,
+///   which closes the gRPC stream.
+/// - If the gRPC stream closes or errors, this actor stops, which (via linking) notifies
+///   the business actor.
+/// - Error responses (`Err(DomainError)`) are forwarded to the client and then the actor stops,
+///   closing the connection.
+///
+/// # Type Parameters
+/// - `Outbound`: `OutboundMessage` sent to the client (e.g., `UserAgentResponse`)
+/// - `Inbound`: `InboundMessage` received from the client (e.g., `UserAgentRequest`)
+/// - `E`: The domain error type, must implement `Into<tonic::Status>` for gRPC conversion
+pub struct GrpcTransportActor<Outbound, Inbound, DomainError>
+where
+    Outbound: Send + 'static,
+    Inbound: Send + 'static,
+    DomainError: Into<tonic::Status> + Send + 'static,
+{
+    sender: mpsc::Sender<Result<Outbound, tonic::Status>>,
+    receiver: tonic::Streaming<Inbound>,
+    business_logic_actor: Recipient<Inbound>,
+    _error: std::marker::PhantomData<DomainError>,
+}
+
+impl<Outbound, Inbound, DomainError> GrpcTransportActor<Outbound, Inbound, DomainError>
+where
+    Outbound: Send + 'static,
+    Inbound: Send + 'static,
+    DomainError: Into<tonic::Status> + Send + 'static,
+{
+    pub fn new(
+        sender: mpsc::Sender<Result<Outbound, tonic::Status>>,
+        receiver: tonic::Streaming<Inbound>,
+        business_logic_actor: Recipient<Inbound>,
+    ) -> Self {
         Self {
             sender,
             receiver,
-            _error: PhantomData,
+            business_logic_actor,
+            _error: std::marker::PhantomData,
         }
     }
 }
 
-impl<Inbound, Outbound, E> Bi<Inbound, Outbound, E> for GrpcAdapter<Inbound, Outbound, E>
+impl<Outbound, Inbound, E> Actor for GrpcTransportActor<Outbound, Inbound, E>
 where
+    Outbound: Send + 'static,
+    Inbound: Send + 'static,
+    E: Into<tonic::Status> + Send + 'static,
+{
+    type Args = Self;
+
+    type Error = ();
+
+    async fn on_start(args: Self::Args, _: ActorRef<Self>) -> Result<Self, Self::Error> {
+        Ok(args)
+    }
+
+    fn on_link_died(
+        &mut self,
+        _: WeakActorRef<Self>,
+        id: kameo::prelude::ActorId,
+        _: kameo::prelude::ActorStopReason,
+    ) -> impl Future<
+        Output = Result<std::ops::ControlFlow<kameo::prelude::ActorStopReason>, Self::Error>,
+    > + Send {
+        async move {
+            if id == self.business_logic_actor.id() {
+                error!("Business logic actor died, stopping GrpcTransportActor");
+                Ok(std::ops::ControlFlow::Break(
+                    kameo::prelude::ActorStopReason::Normal,
+                ))
+            } else {
+                debug!(
+                    "Linked actor {} died, but it's not the business logic actor, ignoring",
+                    id
+                );
+                Ok(std::ops::ControlFlow::Continue(()))
+            }
+        }
+    }
+
+    async fn next(
+        &mut self,
+        _: WeakActorRef<Self>,
+        mailbox_rx: &mut kameo::prelude::MailboxReceiver<Self>,
+    ) -> Option<kameo::mailbox::Signal<Self>> {
+        select! {
+            msg = mailbox_rx.recv() => {
+                msg
+            }
+            recv_msg = self.receiver.next() => {
+                match recv_msg {
+                    Some(Ok(msg)) => {
+                        match self.business_logic_actor.tell(msg).await {
+                            Ok(_) => None,
+                            Err(e) => {
+                                // TODO: this would probably require better error handling - or resending if backpressure is the issue
+                                error!("Failed to send message to business logic actor: {}", e);
+                                Some(Signal::Stop)
+                            }
+                        }
+                    }
+                    Some(Err(e)) => {
+                        error!("Received error from stream: {}, stopping GrpcTransportActor", e);
+                        Some(Signal::Stop)
+                    }
+                    None => {
+                        error!("Receiver channel closed, stopping GrpcTransportActor");
+                        Some(Signal::Stop)
+                    }
+                }
+            }
+        }
+    }
+}
+
+impl<Outbound, Inbound, E> Message<Result<Outbound, E>> for GrpcTransportActor<Outbound, Inbound, E>
+where
+    Outbound: Send + 'static,
+    Inbound: Send + 'static,
+    E: Into<tonic::Status> + Send + 'static,
+{
+    type Reply = ();
+
+    async fn handle(
+        &mut self,
+        msg: Result<Outbound, E>,
+        ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        let is_err = msg.is_err();
+        let grpc_msg = msg.map_err(Into::into);
+        match self.sender.send(grpc_msg).await {
+            Ok(_) => {
+                if is_err {
+                    ctx.stop();
+                }
+            }
+            Err(e) => {
+                error!("Failed to send message: {}", e);
+                ctx.stop();
+            }
+        }
+    }
+}
+
+impl<Outbound, Inbound, E> TransportActor<Outbound, E> for GrpcTransportActor<Outbound, Inbound, E>
+where
+    Outbound: Send + 'static,
+    Inbound: Send + 'static,
+    E: Into<tonic::Status> + Send + 'static,
+{
+}
+
+/// Wires together a transport actor and a business logic actor with bidirectional linking.
+///
+/// This function handles the chicken-and-egg problem of two actors that need references
+/// to each other at construction time. It uses kameo's [`PreparedActor`] to obtain
+/// [`ActorRef`]s before spawning, then links both actors so that if either dies,
+/// the other is notified via [`on_link_died`](Actor::on_link_died).
+///
+/// The business actor receives a type-erased [`Recipient<Result<Outbound, DomainError>>`] instead of an
+/// `ActorRef<Transport>`, keeping it decoupled from the concrete transport implementation.
+///
+/// # Type Parameters
+/// - `Transport`: The transport actor type (e.g., [`GrpcTransportActor`])
+/// - `Inbound`: `InboundMessage` received by the business actor from the transport
+/// - `Outbound`: `OutboundMessage` sent by the business actor back to the transport
+/// - `Business`: The business logic actor
+/// - `BusinessCtor`: Closure that receives a prepared business actor and transport recipient,
+///   spawns the business actor, and returns its [`ActorRef`]
+/// - `TransportCtor`: Closure that receives a prepared transport actor, a recipient for
+///   inbound messages, and the business actor id, then spawns the transport actor
+///
+/// # Returns
+/// A tuple of `(transport_ref, business_ref)` — actor references for both spawned actors.
+pub async fn wire<
+    Transport,
+    Inbound,
+    Outbound,
+    DomainError,
+    Business,
+    BusinessCtor,
+    TransportCtor,
+>(
+    business_ctor: BusinessCtor,
+    transport_ctor: TransportCtor,
+) -> (ActorRef<Transport>, ActorRef<Business>)
+where
+    Transport: TransportActor<Outbound, DomainError>,
     Inbound: Send + 'static,
     Outbound: Send + 'static,
-    E: Into<Status> + Send + Sync + 'static,
+    DomainError: Send + 'static,
+    Business: Actor + Message<Inbound> + Send + 'static,
+    BusinessCtor: FnOnce(PreparedActor<Business>, Recipient<Result<Outbound, DomainError>>),
+    TransportCtor:
+        FnOnce(PreparedActor<Transport>, Recipient<Inbound>),
 {
-    #[tracing::instrument(level = "trace", skip(self, item))]
+    let prepared_business: PreparedActor<Business> = Spawn::prepare();
-    async fn send(&mut self, item: Result<Outbound, E>) -> Result<(), Error> {
+    let prepared_transport: PreparedActor<Transport> = Spawn::prepare();
-        self.sender
-            .send(item.map_err(Into::into))
-            .await
-            .map_err(|_| Error::ChannelClosed)
-    }
 
-    #[tracing::instrument(level = "trace", skip(self))]
+    let business_ref = prepared_business.actor_ref().clone();
-    async fn recv(&mut self) -> Option<Inbound> {
+    let transport_ref = prepared_transport.actor_ref().clone();
-        self.receiver.next().await.transpose().ok().flatten()
+
-    }
+    transport_ref.link(&business_ref).await;
-}
+    business_ref.link(&transport_ref).await;
-
+
-/// No-op [`Bi`] transport for tests and manual actor usage.
+    let recipient = transport_ref.clone().recipient();
-///
+    business_ctor(prepared_business, recipient);
-/// `send` drops all items and succeeds. [`Bi::recv`] never resolves and therefore
+    let business_recipient = business_ref.clone().recipient();
-/// does not busy-wait or spuriously close the stream.
+    transport_ctor(prepared_transport, business_recipient);
-pub struct DummyTransport<T, U, E> {
+
-    _marker: PhantomData<(T, U, E)>,
+
-}
+    (transport_ref, business_ref)
-
-impl<T, U, E> DummyTransport<T, U, E> {
-    pub fn new() -> Self {
-        Self {
-            _marker: PhantomData,
-        }
-    }
-}
-
-impl<T, U, E> Default for DummyTransport<T, U, E> {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl<T, U, E> Bi<T, U, E> for DummyTransport<T, U, E>
-where
-    T: Send + Sync + 'static,
-    U: Send + Sync + 'static,
-    E: Send + Sync + 'static,
-{
-    async fn send(&mut self, _item: Result<U, E>) -> Result<(), Error> {
-        Ok(())
-    }
-
-    fn recv(&mut self) -> impl std::future::Future<Output = Option<T>> + Send {
-        async {
-            std::future::pending::<()>().await;
-            None
-        }
-    }
 }

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

@@ -7,6 +7,6 @@ use crate::ServerContext;
 
 pub(crate) async fn handle_client(
     _context: ServerContext,
-    _bistream: impl Bi<ClientRequest, ClientResponse, tonic::Status>,
+    _bistream: impl Bi<ClientRequest, ClientResponse>,
 ) {
 }

server/crates/arbiter-server/src/actors/user_agent/error.rs

@@ -0,0 +1,57 @@
+use tonic::Status;
+
+use crate::db;
+
+#[derive(Debug, thiserror::Error)]
+pub enum UserAgentError {
+    #[error("Missing payload in request")]
+    MissingPayload,
+
+    #[error("Invalid bootstrap token")]
+    InvalidBootstrapToken,
+
+    #[error("Public key not registered")]
+    PubkeyNotRegistered,
+
+    #[error("Invalid public key format")]
+    InvalidPubkey,
+
+    #[error("Invalid signature length")]
+    InvalidSignatureLength,
+
+    #[error("Invalid challenge solution")]
+    InvalidChallengeSolution,
+
+    #[error("Invalid state for operation")]
+    InvalidState,
+
+    #[error("Actor unavailable")]
+    ActorUnavailable,
+
+    #[error("Database error")]
+    Database(#[from] diesel::result::Error),
+
+    #[error("Database pool error")]
+    DatabasePool(#[from] db::PoolError),
+}
+
+impl From<UserAgentError> for Status {
+    fn from(err: UserAgentError) -> Self {
+        match err {
+            UserAgentError::MissingPayload
+            | UserAgentError::InvalidBootstrapToken
+            | UserAgentError::InvalidPubkey
+            | UserAgentError::InvalidSignatureLength => Status::invalid_argument(err.to_string()),
+
+            UserAgentError::PubkeyNotRegistered | UserAgentError::InvalidChallengeSolution => {
+                Status::unauthenticated(err.to_string())
+            }
+
+            UserAgentError::InvalidState => Status::failed_precondition(err.to_string()),
+
+            UserAgentError::ActorUnavailable
+            | UserAgentError::Database(_)
+            | UserAgentError::DatabasePool(_) => Status::internal(err.to_string()),
+        }
+    }
+}

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

@@ -1,27 +1,23 @@
 use std::{ops::DerefMut, sync::Mutex};
 
-use arbiter_proto::{
+use arbiter_proto::proto::{
-    proto::{
+    UnsealEncryptedKey, UnsealResult, UnsealStart, UnsealStartResponse, UserAgentRequest,
-        UnsealEncryptedKey, UnsealResult, UnsealStart, UnsealStartResponse, UserAgentRequest,
+    UserAgentResponse,
-        UserAgentResponse,
+    auth::{
-        auth::{
+        self, AuthChallengeRequest, AuthOk, ClientMessage as ClientAuthMessage,
-            self, AuthChallengeRequest, AuthOk, ClientMessage as ClientAuthMessage,
+        ServerMessage as AuthServerMessage,
-            ServerMessage as AuthServerMessage, client_message::Payload as ClientAuthPayload,
+        client_message::Payload as ClientAuthPayload,
-            server_message::Payload as ServerAuthPayload,
+        server_message::Payload as ServerAuthPayload,
-        },
-        user_agent_request::Payload as UserAgentRequestPayload,
-        user_agent_response::Payload as UserAgentResponsePayload,
     },
-    transport::{Bi, DummyTransport},
+    user_agent_request::Payload as UserAgentRequestPayload,
+    user_agent_response::Payload as UserAgentResponsePayload,
 };
 use chacha20poly1305::{AeadInPlace, XChaCha20Poly1305, XNonce, aead::KeyInit};
 use diesel::{ExpressionMethods as _, OptionalExtension as _, QueryDsl, dsl::update};
 use diesel_async::RunQueryDsl;
 use ed25519_dalek::VerifyingKey;
-use kameo::{Actor, error::SendError};
+use kameo::{Actor, actor::Recipient, error::SendError, messages, prelude::Message};
 use memsafe::MemSafe;
-use tokio::select;
-use tonic::Status;
 use tracing::{error, info};
 use x25519_dalek::{EphemeralSecret, PublicKey};
 
@@ -39,103 +35,24 @@ use crate::{
     db::{self, schema},
 };
 
+mod error;
 mod state;
 
-#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+pub use error::UserAgentError;
-pub enum UserAgentError {
-    #[error("Expected message with payload")]
-    MissingRequestPayload,
-    #[error("Expected message with payload")]
-    UnexpectedRequestPayload,
-    #[error("Invalid state for challenge solution")]
-    InvalidStateForChallengeSolution,
-    #[error("Invalid state for unseal encrypted key")]
-    InvalidStateForUnsealEncryptedKey,
-    #[error("client_pubkey must be 32 bytes")]
-    InvalidClientPubkeyLength,
-    #[error("Expected pubkey to have specific length")]
-    InvalidAuthPubkeyLength,
-    #[error("Failed to convert pubkey to VerifyingKey")]
-    InvalidAuthPubkeyEncoding,
-    #[error("Invalid signature length")]
-    InvalidSignatureLength,
-    #[error("Invalid bootstrap token")]
-    InvalidBootstrapToken,
-    #[error("Public key not registered")]
-    PublicKeyNotRegistered,
-    #[error("Invalid challenge solution")]
-    InvalidChallengeSolution,
-    #[error("State machine error")]
-    StateTransitionFailed,
-    #[error("Bootstrap token consumption failed")]
-    BootstrapperActorUnreachable,
-    #[error("Vault is not available")]
-    KeyHolderActorUnreachable,
-    #[error("Database pool error")]
-    DatabasePoolUnavailable,
-    #[error("Database error")]
-    DatabaseOperationFailed,
-}
 
-impl From<UserAgentError> for Status {
+#[derive(Actor)]
-    fn from(value: UserAgentError) -> Self {
+pub struct UserAgentActor {
-        match value {
-            UserAgentError::MissingRequestPayload | UserAgentError::UnexpectedRequestPayload => {
-                Status::invalid_argument("Expected message with payload")
-            }
-            UserAgentError::InvalidStateForChallengeSolution => {
-                Status::invalid_argument("Invalid state for challenge solution")
-            }
-            UserAgentError::InvalidStateForUnsealEncryptedKey => {
-                Status::failed_precondition("Invalid state for unseal encrypted key")
-            }
-            UserAgentError::InvalidClientPubkeyLength => {
-                Status::invalid_argument("client_pubkey must be 32 bytes")
-            }
-            UserAgentError::InvalidAuthPubkeyLength => {
-                Status::invalid_argument("Expected pubkey to have specific length")
-            }
-            UserAgentError::InvalidAuthPubkeyEncoding => {
-                Status::invalid_argument("Failed to convert pubkey to VerifyingKey")
-            }
-            UserAgentError::InvalidSignatureLength => {
-                Status::invalid_argument("Invalid signature length")
-            }
-            UserAgentError::InvalidBootstrapToken => {
-                Status::invalid_argument("Invalid bootstrap token")
-            }
-            UserAgentError::PublicKeyNotRegistered => {
-                Status::unauthenticated("Public key not registered")
-            }
-            UserAgentError::InvalidChallengeSolution => {
-                Status::unauthenticated("Invalid challenge solution")
-            }
-            UserAgentError::StateTransitionFailed => Status::internal("State machine error"),
-            UserAgentError::BootstrapperActorUnreachable => {
-                Status::internal("Bootstrap token consumption failed")
-            }
-            UserAgentError::KeyHolderActorUnreachable => Status::internal("Vault is not available"),
-            UserAgentError::DatabasePoolUnavailable => Status::internal("Database pool error"),
-            UserAgentError::DatabaseOperationFailed => Status::internal("Database error"),
-        }
-    }
-}
-
-pub struct UserAgentActor<Transport>
-where
-    Transport: Bi<UserAgentRequest, UserAgentResponse, UserAgentError>,
-{
     db: db::DatabasePool,
     actors: GlobalActors,
     state: UserAgentStateMachine<DummyContext>,
-    transport: Transport,
+    transport: Recipient<Result<UserAgentResponse, UserAgentError>>,
 }
 
-impl<Transport> UserAgentActor<Transport>
+impl UserAgentActor {
-where
+    pub(crate) fn new(
-    Transport: Bi<UserAgentRequest, UserAgentResponse, UserAgentError>,
+        context: ServerContext,
-{
+        transport: Recipient<Result<UserAgentResponse, UserAgentError>>,
-    pub(crate) fn new(context: ServerContext, transport: Transport) -> Self {
+    ) -> Self {
         Self {
             db: context.db.clone(),
             actors: context.actors.clone(),
@@ -144,18 +61,23 @@ where
         }
     }
 
-    fn transition(&mut self, event: UserAgentEvents) -> Result<(), UserAgentError> {
+    pub fn new_manual(
-        self.state.process_event(event).map_err(|e| {
+        db: db::DatabasePool,
-            error!(?e, "State transition failed");
+        actors: GlobalActors,
-            UserAgentError::StateTransitionFailed
+        transport: Recipient<Result<UserAgentResponse, UserAgentError>>,
-        })?;
+    ) -> Self {
-        Ok(())
+        Self {
+            db,
+            actors,
+            state: UserAgentStateMachine::new(DummyContext),
+            transport,
+        }
     }
 
-    pub async fn process_transport_inbound(&mut self, req: UserAgentRequest) -> Output {
+    async fn process_request(&mut self, req: UserAgentRequest) -> Output {
         let msg = req.payload.ok_or_else(|| {
             error!(actor = "useragent", "Received message with no payload");
-            UserAgentError::MissingRequestPayload
+            UserAgentError::MissingPayload
         })?;
 
         match msg {
@@ -171,15 +93,23 @@ where
             UserAgentRequestPayload::UnsealEncryptedKey(unseal_encrypted_key) => {
                 self.handle_unseal_encrypted_key(unseal_encrypted_key).await
             }
-            _ => Err(UserAgentError::UnexpectedRequestPayload),
+            _ => Err(UserAgentError::MissingPayload),
         }
     }
 
+    fn transition(&mut self, event: UserAgentEvents) -> Result<(), UserAgentError> {
+        self.state.process_event(event).map_err(|e| {
+            error!(?e, "State transition failed");
+            UserAgentError::InvalidState
+        })?;
+        Ok(())
+    }
+
     async fn auth_with_bootstrap_token(
         &mut self,
         pubkey: ed25519_dalek::VerifyingKey,
         token: String,
-    ) -> Result<UserAgentResponse, UserAgentError> {
+    ) -> Output {
         let token_ok: bool = self
             .actors
             .bootstrapper
@@ -187,7 +117,7 @@ where
             .await
             .map_err(|e| {
                 error!(?pubkey, "Failed to consume bootstrap token: {e}");
-                UserAgentError::BootstrapperActorUnreachable
+                UserAgentError::ActorUnavailable
             })?;
 
         if !token_ok {
@@ -196,10 +126,7 @@ where
         }
 
         {
-            let mut conn = self.db.get().await.map_err(|e| {
+            let mut conn = self.db.get().await?;
-                error!(error = ?e, "Database pool error");
-                UserAgentError::DatabasePoolUnavailable
-            })?;
 
             diesel::insert_into(schema::useragent_client::table)
                 .values((
@@ -207,11 +134,7 @@ where
                     schema::useragent_client::nonce.eq(1),
                 ))
                 .execute(&mut conn)
-                .await
+                .await?;
-                .map_err(|e| {
-                    error!(error = ?e, "Database error");
-                    UserAgentError::DatabaseOperationFailed
-                })?;
         }
 
         self.transition(UserAgentEvents::ReceivedBootstrapToken)?;
@@ -221,10 +144,7 @@ where
 
     async fn auth_with_challenge(&mut self, pubkey: VerifyingKey, pubkey_bytes: Vec<u8>) -> Output {
         let nonce: Option<i32> = {
-            let mut db_conn = self.db.get().await.map_err(|e| {
+            let mut db_conn = self.db.get().await?;
-                error!(error = ?e, "Database pool error");
-                UserAgentError::DatabasePoolUnavailable
-            })?;
             db_conn
                 .exclusive_transaction(|conn| {
                     Box::pin(async move {
@@ -248,16 +168,12 @@ where
                     })
                 })
                 .await
-                .optional()
+                .optional()?
-                .map_err(|e| {
-                    error!(error = ?e, "Database error");
-                    UserAgentError::DatabaseOperationFailed
-                })?
         };
 
         let Some(nonce) = nonce else {
             error!(?pubkey, "Public key not found in database");
-            return Err(UserAgentError::PublicKeyNotRegistered);
+            return Err(UserAgentError::PubkeyNotRegistered);
         };
 
         let challenge = auth::AuthChallenge {
@@ -286,7 +202,7 @@ where
         let UserAgentStates::WaitingForChallengeSolution(challenge_context) = self.state.state()
         else {
             error!("Received challenge solution in invalid state");
-            return Err(UserAgentError::InvalidStateForChallengeSolution);
+            return Err(UserAgentError::InvalidState);
         };
         let formatted_challenge = arbiter_proto::format_challenge(&challenge_context.challenge);
 
@@ -320,18 +236,17 @@ fn unseal_response(payload: UserAgentResponsePayload) -> UserAgentResponse {
     }
 }
 
-impl<Transport> UserAgentActor<Transport>
+#[messages]
-where
+impl UserAgentActor {
-    Transport: Bi<UserAgentRequest, UserAgentResponse, UserAgentError>,
+    #[message]
-{
+    pub async fn handle_unseal_request(&mut self, req: UnsealStart) -> Output {
-    async fn handle_unseal_request(&mut self, req: UnsealStart) -> Output {
         let secret = EphemeralSecret::random();
         let public_key = PublicKey::from(&secret);
 
         let client_pubkey_bytes: [u8; 32] = req
             .client_pubkey
             .try_into()
-            .map_err(|_| UserAgentError::InvalidClientPubkeyLength)?;
+            .map_err(|_| UserAgentError::InvalidPubkey)?;
 
         let client_public_key = PublicKey::from(client_pubkey_bytes);
 
@@ -347,10 +262,11 @@ where
         ))
     }
 
-    async fn handle_unseal_encrypted_key(&mut self, req: UnsealEncryptedKey) -> Output {
+    #[message]
+    pub async fn handle_unseal_encrypted_key(&mut self, req: UnsealEncryptedKey) -> Output {
         let UserAgentStates::WaitingForUnsealKey(unseal_context) = self.state.state() else {
             error!("Received unseal encrypted key in invalid state");
-            return Err(UserAgentError::InvalidStateForUnsealEncryptedKey);
+            return Err(UserAgentError::InvalidState);
         };
         let ephemeral_secret = {
             let mut secret_lock = unseal_context.secret.lock().unwrap();
@@ -414,7 +330,7 @@ where
                     Err(err) => {
                         error!(?err, "Failed to send unseal request to keyholder");
                         self.transition(UserAgentEvents::ReceivedInvalidKey)?;
-                        Err(UserAgentError::KeyHolderActorUnreachable)
+                        Err(UserAgentError::ActorUnavailable)
                     }
                 }
             }
@@ -428,14 +344,15 @@ where
         }
     }
 
-    async fn handle_auth_challenge_request(&mut self, req: AuthChallengeRequest) -> Output {
+    #[message]
+    pub async fn handle_auth_challenge_request(&mut self, req: AuthChallengeRequest) -> Output {
         let pubkey = req
             .pubkey
             .as_array()
-            .ok_or(UserAgentError::InvalidAuthPubkeyLength)?;
+            .ok_or(UserAgentError::InvalidPubkey)?;
         let pubkey = VerifyingKey::from_bytes(pubkey).map_err(|_err| {
             error!(?pubkey, "Failed to convert to VerifyingKey");
-            UserAgentError::InvalidAuthPubkeyEncoding
+            UserAgentError::InvalidPubkey
         })?;
 
         self.transition(UserAgentEvents::AuthRequest)?;
@@ -446,7 +363,8 @@ where
         }
     }
 
-    async fn handle_auth_challenge_solution(
+    #[message]
+    pub async fn handle_auth_challenge_solution(
         &mut self,
         solution: auth::AuthChallengeSolution,
     ) -> Output {
@@ -467,67 +385,17 @@ where
     }
 }
 
+impl Message<UserAgentRequest> for UserAgentActor {
+    type Reply = ();
 
-impl<Transport> Actor for UserAgentActor<Transport>
+    async fn handle(
-where
-    Transport: Bi<UserAgentRequest, UserAgentResponse, UserAgentError>,
-{
-    type Args = Self;
-
-    type Error = ();
-
-    async fn on_start(
-        args: Self::Args,
-        _: kameo::prelude::ActorRef<Self>,
-    ) -> Result<Self, Self::Error> {
-        Ok(args)
-    }
-
-    async fn next(
         &mut self,
-        _actor_ref: kameo::prelude::WeakActorRef<Self>,
+        msg: UserAgentRequest,
-        mailbox_rx: &mut kameo::prelude::MailboxReceiver<Self>,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
-    ) -> Option<kameo::mailbox::Signal<Self>> {
+    ) -> Self::Reply {
-        loop {
+        let result = self.process_request(msg).await;
-            select! {
+        if let Err(e) = self.transport.tell(result).await {
-                signal = mailbox_rx.recv() => {
+            error!(actor = "useragent", "Failed to send response to transport: {}", e);
-                    return signal;
-                }
-                msg = self.transport.recv() => {
-                    match msg {
-                        Some(request) => {
-                            match self.process_transport_inbound(request).await {
-                                Ok(response) => {
-                                    if self.transport.send(Ok(response)).await.is_err() {
-                                        error!(actor = "useragent", reason = "channel closed", "send.failed");
-                                        return Some(kameo::mailbox::Signal::Stop);
-                                    }
-                                }
-                                Err(err) => {
-                                    let _ = self.transport.send(Err(err)).await;
-                                    return Some(kameo::mailbox::Signal::Stop);
-                                }
-                            }
-                        }
-                        None => {
-                            info!(actor = "useragent", "transport.closed");
-                            return Some(kameo::mailbox::Signal::Stop);
-                        }
-                    }
-                }
-            }
         }
     }
 }
-
-
-impl UserAgentActor<DummyTransport<UserAgentRequest, UserAgentResponse, UserAgentError>> {
-    pub fn new_manual(db: db::DatabasePool, actors: GlobalActors) -> Self {
-        Self {
-            db,
-            actors,
-            state: UserAgentStateMachine::new(DummyContext),
-            transport: DummyTransport::new(),
-        }
-    }
-}

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

@@ -1,24 +0,0 @@
-use tonic::Status;
-use tracing::error;
-
-pub trait GrpcStatusExt<T> {
-    fn to_status(self) -> Result<T, Status>;
-}
-
-impl<T> GrpcStatusExt<T> for Result<T, diesel::result::Error> {
-    fn to_status(self) -> Result<T, Status> {
-        self.map_err(|e| {
-            error!(error = ?e, "Database error");
-            Status::internal("Database error")
-        })
-    }
-}
-
-impl<T> GrpcStatusExt<T> for Result<T, crate::db::PoolError> {
-    fn to_status(self) -> Result<T, Status> {
-        self.map_err(|e| {
-            error!(error = ?e, "Database pool error");
-            Status::internal("Database pool error")
-        })
-    }
-}

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

@@ -1,25 +1,26 @@
 #![forbid(unsafe_code)]
 use arbiter_proto::{
     proto::{ClientRequest, ClientResponse, UserAgentRequest, UserAgentResponse},
-    transport::GrpcAdapter,
+    transport::{BiStream, GrpcTransportActor, wire},
 };
 use async_trait::async_trait;
-use kameo::actor::Spawn;
+use kameo::actor::PreparedActor;
 use tokio_stream::wrappers::ReceiverStream;
 
 use tokio::sync::mpsc;
 use tonic::{Request, Response, Status};
-use tracing::info;
 
 use crate::{
-    actors::user_agent::UserAgentActor,
+    actors::{
+        client::handle_client,
+        user_agent::UserAgentActor,
+    },
     context::ServerContext,
 };
 
 pub mod actors;
 pub mod context;
 pub mod db;
-mod errors;
 
 const DEFAULT_CHANNEL_SIZE: usize = 1000;
 
@@ -40,24 +41,42 @@ impl arbiter_proto::proto::arbiter_service_server::ArbiterService for Server {
 
     async fn client(
         &self,
-        _request: Request<tonic::Streaming<ClientRequest>>,
+        request: Request<tonic::Streaming<ClientRequest>>,
     ) -> Result<Response<Self::ClientStream>, Status> {
-        todo!()
+        let req_stream = request.into_inner();
+        let (tx, rx) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
+        tokio::spawn(handle_client(
+            self.context.clone(),
+            BiStream {
+                request_stream: req_stream,
+                response_sender: tx,
+            },
+        ));
+
+        Ok(Response::new(ReceiverStream::new(rx)))
     }
 
-    #[tracing::instrument(level = "debug", skip(self))]
     async fn user_agent(
         &self,
         request: Request<tonic::Streaming<UserAgentRequest>>,
     ) -> Result<Response<Self::UserAgentStream>, Status> {
         let req_stream = request.into_inner();
         let (tx, rx) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
+        let context = self.context.clone();
 
-        let adapter = GrpcAdapter::new(tx, req_stream);
+        wire(
-
+            |prepared: PreparedActor<UserAgentActor>, recipient| {
-        UserAgentActor::spawn(UserAgentActor::new(self.context.clone(), adapter));
+                prepared.spawn(UserAgentActor::new(context, recipient));
-
+            },
-        info!(event = "connection established", "grpc.user_agent");
+            |prepared: PreparedActor<GrpcTransportActor<_, _, _>>, business_recipient| {
+                prepared.spawn(GrpcTransportActor::new(
+                    tx,
+                    req_stream,
+                    business_recipient,
+                ));
+            },
+        )
+        .await;
 
         Ok(Response::new(ReceiverStream::new(rx)))
     }

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

@@ -1,5 +1,30 @@
 mod common;
 
+use arbiter_proto::proto::UserAgentResponse;
+use arbiter_server::actors::user_agent::UserAgentError;
+use kameo::{Actor, actor::Recipient, actor::Spawn, prelude::Message};
+
+/// A no-op actor that discards any messages it receives.
+#[derive(Actor)]
+struct NullSink;
+
+impl Message<Result<UserAgentResponse, UserAgentError>> for NullSink {
+    type Reply = ();
+
+    async fn handle(
+        &mut self,
+        _msg: Result<UserAgentResponse, UserAgentError>,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+    }
+}
+
+/// Creates a `Recipient` that silently discards all messages.
+fn null_recipient() -> Recipient<Result<UserAgentResponse, UserAgentError>> {
+    let actor_ref = NullSink::spawn(NullSink);
+    actor_ref.recipient()
+}
+
 #[path = "user_agent/auth.rs"]
 mod auth;
 #[path = "user_agent/unseal.rs"]

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

@@ -1,29 +1,20 @@
 use arbiter_proto::proto::{
     UserAgentResponse,
-    UserAgentRequest,
+    auth::{self, AuthChallengeRequest, AuthOk},
-    auth::{self, AuthChallengeRequest, AuthOk, ClientMessage, client_message::Payload as ClientAuthPayload},
-    user_agent_request::Payload as UserAgentRequestPayload,
     user_agent_response::Payload as UserAgentResponsePayload,
 };
 use arbiter_server::{
     actors::{
         GlobalActors,
         bootstrap::GetToken,
-        user_agent::{UserAgentActor, UserAgentError},
+        user_agent::{HandleAuthChallengeRequest, HandleAuthChallengeSolution, UserAgentActor},
     },
     db::{self, schema},
 };
 use diesel::{ExpressionMethods as _, QueryDsl, insert_into};
 use diesel_async::RunQueryDsl;
 use ed25519_dalek::Signer as _;
-
+use kameo::actor::Spawn;
-fn auth_request(payload: ClientAuthPayload) -> UserAgentRequest {
-    UserAgentRequest {
-        payload: Some(UserAgentRequestPayload::AuthMessage(ClientMessage {
-            payload: Some(payload),
-        })),
-    }
-}
 
 #[tokio::test]
 #[test_log::test]
@@ -32,20 +23,22 @@ pub async fn test_bootstrap_token_auth() {
 
     let actors = GlobalActors::spawn(db.clone()).await.unwrap();
     let token = actors.bootstrapper.ask(GetToken).await.unwrap().unwrap();
-    let mut user_agent = UserAgentActor::new_manual(db.clone(), actors);
+    let user_agent =
+        UserAgentActor::new_manual(db.clone(), actors, super::null_recipient());
+    let user_agent_ref = UserAgentActor::spawn(user_agent);
 
     let new_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
     let pubkey_bytes = new_key.verifying_key().to_bytes().to_vec();
 
-    let result = user_agent
+    let result = user_agent_ref
-        .process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
+        .ask(HandleAuthChallengeRequest {
-            AuthChallengeRequest {
+            req: AuthChallengeRequest {
                 pubkey: pubkey_bytes,
                 bootstrap_token: Some(token),
             },
-        )))
+        })
         .await
-        .expect("Shouldn't fail to process message");
+        .expect("Shouldn't fail to send message");
 
     assert_eq!(
         result,
@@ -75,25 +68,31 @@ pub async fn test_bootstrap_invalid_token_auth() {
     let db = db::create_test_pool().await;
 
     let actors = GlobalActors::spawn(db.clone()).await.unwrap();
-    let mut user_agent = UserAgentActor::new_manual(db.clone(), actors);
+    let user_agent =
+        UserAgentActor::new_manual(db.clone(), actors, super::null_recipient());
+    let user_agent_ref = UserAgentActor::spawn(user_agent);
 
     let new_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
     let pubkey_bytes = new_key.verifying_key().to_bytes().to_vec();
 
-    let result = user_agent
+    let result = user_agent_ref
-        .process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
+        .ask(HandleAuthChallengeRequest {
-            AuthChallengeRequest {
+            req: AuthChallengeRequest {
                 pubkey: pubkey_bytes,
                 bootstrap_token: Some("invalid_token".to_string()),
             },
-        )))
+        })
         .await;
 
     match result {
-        Err(err) => {
+        Err(kameo::error::SendError::HandlerError(err)) => {
-            assert_eq!(err, UserAgentError::InvalidBootstrapToken);
+            assert!(
-            let status: tonic::Status = err.into();
+                matches!(err, arbiter_server::actors::user_agent::UserAgentError::InvalidBootstrapToken),
-            assert_eq!(status.code(), tonic::Code::InvalidArgument);
+                "Expected InvalidBootstrapToken, got {err:?}"
+            );
+        }
+        Err(other) => {
+            panic!("Expected SendError::HandlerError, got {other:?}");
         }
         Ok(_) => {
             panic!("Expected error due to invalid bootstrap token, but got success");
@@ -107,7 +106,9 @@ pub async fn test_challenge_auth() {
     let db = db::create_test_pool().await;
 
     let actors = GlobalActors::spawn(db.clone()).await.unwrap();
-    let mut user_agent = UserAgentActor::new_manual(db.clone(), actors);
+    let user_agent =
+        UserAgentActor::new_manual(db.clone(), actors, super::null_recipient());
+    let user_agent_ref = UserAgentActor::spawn(user_agent);
 
     let new_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
     let pubkey_bytes = new_key.verifying_key().to_bytes().to_vec();
@@ -121,15 +122,15 @@ pub async fn test_challenge_auth() {
             .unwrap();
     }
 
-    let result = user_agent
+    let result = user_agent_ref
-        .process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
+        .ask(HandleAuthChallengeRequest {
-            AuthChallengeRequest {
+            req: AuthChallengeRequest {
                 pubkey: pubkey_bytes,
                 bootstrap_token: None,
             },
-        )))
+        })
         .await
-        .expect("Shouldn't fail to process message");
+        .expect("Shouldn't fail to send message");
 
     let UserAgentResponse {
         payload:
@@ -146,14 +147,14 @@ pub async fn test_challenge_auth() {
     let signature = new_key.sign(&formatted_challenge);
     let serialized_signature = signature.to_bytes().to_vec();
 
-    let result = user_agent
+    let result = user_agent_ref
-        .process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeSolution(
+        .ask(HandleAuthChallengeSolution {
-            auth::AuthChallengeSolution {
+            solution: auth::AuthChallengeSolution {
                 signature: serialized_signature,
             },
-        )))
+        })
         .await
-        .expect("Shouldn't fail to process message");
+        .expect("Shouldn't fail to send message");
 
     assert_eq!(
         result,

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

@@ -1,51 +1,27 @@
 use arbiter_proto::proto::{
-    UnsealEncryptedKey, UnsealResult, UnsealStart, UserAgentRequest, UserAgentResponse,
+    UnsealEncryptedKey, UnsealResult, UnsealStart, auth::AuthChallengeRequest,
-    auth::{AuthChallengeRequest, ClientMessage, client_message::Payload as ClientAuthPayload},
-    user_agent_request::Payload as UserAgentRequestPayload,
     user_agent_response::Payload as UserAgentResponsePayload,
 };
-use arbiter_proto::transport::DummyTransport;
 use arbiter_server::{
     actors::{
         GlobalActors,
         bootstrap::GetToken,
         keyholder::{Bootstrap, Seal},
-        user_agent::{UserAgentActor, UserAgentError},
+        user_agent::{
+            HandleAuthChallengeRequest, HandleUnsealEncryptedKey, HandleUnsealRequest,
+            UserAgentActor,
+        },
     },
     db,
 };
 use chacha20poly1305::{AeadInPlace, XChaCha20Poly1305, XNonce, aead::KeyInit};
+use kameo::actor::{ActorRef, Spawn};
 use memsafe::MemSafe;
 use x25519_dalek::{EphemeralSecret, PublicKey};
 
-type TestUserAgent = UserAgentActor<DummyTransport<UserAgentRequest, UserAgentResponse, UserAgentError>>;
-
-fn auth_request(payload: ClientAuthPayload) -> UserAgentRequest {
-    UserAgentRequest {
-        payload: Some(UserAgentRequestPayload::AuthMessage(ClientMessage {
-            payload: Some(payload),
-        })),
-    }
-}
-
-fn unseal_start_request(req: UnsealStart) -> UserAgentRequest {
-    UserAgentRequest {
-        payload: Some(UserAgentRequestPayload::UnsealStart(req)),
-    }
-}
-
-fn unseal_key_request(req: UnsealEncryptedKey) -> UserAgentRequest {
-    UserAgentRequest {
-        payload: Some(UserAgentRequestPayload::UnsealEncryptedKey(req)),
-    }
-}
-
 async fn setup_authenticated_user_agent(
     seal_key: &[u8],
-) -> (
+) -> (arbiter_server::db::DatabasePool, ActorRef<UserAgentActor>) {
-    arbiter_server::db::DatabasePool,
-    TestUserAgent,
-) {
     let db = db::create_test_pool().await;
 
     let actors = GlobalActors::spawn(db.clone()).await.unwrap();
@@ -58,34 +34,38 @@ async fn setup_authenticated_user_agent(
         .unwrap();
     actors.key_holder.ask(Seal).await.unwrap();
 
-    let mut user_agent = UserAgentActor::new_manual(db.clone(), actors.clone());
+    let user_agent =
+        UserAgentActor::new_manual(db.clone(), actors.clone(), super::null_recipient());
+    let user_agent_ref = UserAgentActor::spawn(user_agent);
 
     let token = actors.bootstrapper.ask(GetToken).await.unwrap().unwrap();
     let auth_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
-    user_agent
+    user_agent_ref
-        .process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
+        .ask(HandleAuthChallengeRequest {
-            AuthChallengeRequest {
+            req: AuthChallengeRequest {
                 pubkey: auth_key.verifying_key().to_bytes().to_vec(),
                 bootstrap_token: Some(token),
             },
-        )))
+        })
         .await
         .unwrap();
 
-    (db, user_agent)
+    (db, user_agent_ref)
 }
 
 async fn client_dh_encrypt(
-    user_agent: &mut TestUserAgent,
+    user_agent_ref: &ActorRef<UserAgentActor>,
     key_to_send: &[u8],
 ) -> UnsealEncryptedKey {
     let client_secret = EphemeralSecret::random();
     let client_public = PublicKey::from(&client_secret);
 
-    let response = user_agent
+    let response = user_agent_ref
-        .process_transport_inbound(unseal_start_request(UnsealStart {
+        .ask(HandleUnsealRequest {
-            client_pubkey: client_public.as_bytes().to_vec(),
+            req: UnsealStart {
-        }))
+                client_pubkey: client_public.as_bytes().to_vec(),
+            },
+        })
         .await
         .unwrap();
 
@@ -115,12 +95,12 @@ async fn client_dh_encrypt(
 #[test_log::test]
 pub async fn test_unseal_success() {
     let seal_key = b"test-seal-key";
-    let (_db, mut user_agent) = setup_authenticated_user_agent(seal_key).await;
+    let (_db, user_agent_ref) = setup_authenticated_user_agent(seal_key).await;
 
-    let encrypted_key = client_dh_encrypt(&mut user_agent, seal_key).await;
+    let encrypted_key = client_dh_encrypt(&user_agent_ref, seal_key).await;
 
-    let response = user_agent
+    let response = user_agent_ref
-        .process_transport_inbound(unseal_key_request(encrypted_key))
+        .ask(HandleUnsealEncryptedKey { req: encrypted_key })
         .await
         .unwrap();
 
@@ -133,12 +113,12 @@ pub async fn test_unseal_success() {
 #[tokio::test]
 #[test_log::test]
 pub async fn test_unseal_wrong_seal_key() {
-    let (_db, mut user_agent) = setup_authenticated_user_agent(b"correct-key").await;
+    let (_db, user_agent_ref) = setup_authenticated_user_agent(b"correct-key").await;
 
-    let encrypted_key = client_dh_encrypt(&mut user_agent, b"wrong-key").await;
+    let encrypted_key = client_dh_encrypt(&user_agent_ref, b"wrong-key").await;
 
-    let response = user_agent
+    let response = user_agent_ref
-        .process_transport_inbound(unseal_key_request(encrypted_key))
+        .ask(HandleUnsealEncryptedKey { req: encrypted_key })
         .await
         .unwrap();
 
@@ -151,24 +131,28 @@ pub async fn test_unseal_wrong_seal_key() {
 #[tokio::test]
 #[test_log::test]
 pub async fn test_unseal_corrupted_ciphertext() {
-    let (_db, mut user_agent) = setup_authenticated_user_agent(b"test-key").await;
+    let (_db, user_agent_ref) = setup_authenticated_user_agent(b"test-key").await;
 
     let client_secret = EphemeralSecret::random();
     let client_public = PublicKey::from(&client_secret);
 
-    user_agent
+    user_agent_ref
-        .process_transport_inbound(unseal_start_request(UnsealStart {
+        .ask(HandleUnsealRequest {
-            client_pubkey: client_public.as_bytes().to_vec(),
+            req: UnsealStart {
-        }))
+                client_pubkey: client_public.as_bytes().to_vec(),
+            },
+        })
         .await
         .unwrap();
 
-    let response = user_agent
+    let response = user_agent_ref
-        .process_transport_inbound(unseal_key_request(UnsealEncryptedKey {
+        .ask(HandleUnsealEncryptedKey {
-            nonce: vec![0u8; 24],
+            req: UnsealEncryptedKey {
-            ciphertext: vec![0u8; 32],
+                nonce: vec![0u8; 24],
-            associated_data: vec![],
+                ciphertext: vec![0u8; 32],
-        }))
+                associated_data: vec![],
+            },
+        })
         .await
         .unwrap();
 
@@ -184,20 +168,27 @@ pub async fn test_unseal_start_without_auth_fails() {
     let db = db::create_test_pool().await;
 
     let actors = GlobalActors::spawn(db.clone()).await.unwrap();
-    let mut user_agent = UserAgentActor::new_manual(db.clone(), actors);
+    let user_agent =
+        UserAgentActor::new_manual(db.clone(), actors, super::null_recipient());
+    let user_agent_ref = UserAgentActor::spawn(user_agent);
 
     let client_secret = EphemeralSecret::random();
     let client_public = PublicKey::from(&client_secret);
 
-    let result = user_agent
+    let result = user_agent_ref
-        .process_transport_inbound(unseal_start_request(UnsealStart {
+        .ask(HandleUnsealRequest {
-            client_pubkey: client_public.as_bytes().to_vec(),
+            req: UnsealStart {
-        }))
+                client_pubkey: client_public.as_bytes().to_vec(),
+            },
+        })
         .await;
 
     match result {
-        Err(err) => {
+        Err(kameo::error::SendError::HandlerError(err)) => {
-            assert_eq!(err, UserAgentError::StateTransitionFailed);
+            assert!(
+                matches!(err, arbiter_server::actors::user_agent::UserAgentError::InvalidState),
+                "Expected InvalidState, got {err:?}"
+            );
         }
         other => panic!("Expected state machine error, got {other:?}"),
     }
@@ -207,13 +198,13 @@ pub async fn test_unseal_start_without_auth_fails() {
 #[test_log::test]
 pub async fn test_unseal_retry_after_invalid_key() {
     let seal_key = b"real-seal-key";
-    let (_db, mut user_agent) = setup_authenticated_user_agent(seal_key).await;
+    let (_db, user_agent_ref) = setup_authenticated_user_agent(seal_key).await;
 
     {
-        let encrypted_key = client_dh_encrypt(&mut user_agent, b"wrong-key").await;
+        let encrypted_key = client_dh_encrypt(&user_agent_ref, b"wrong-key").await;
 
-        let response = user_agent
+        let response = user_agent_ref
-            .process_transport_inbound(unseal_key_request(encrypted_key))
+            .ask(HandleUnsealEncryptedKey { req: encrypted_key })
             .await
             .unwrap();
 
@@ -224,10 +215,10 @@ pub async fn test_unseal_retry_after_invalid_key() {
     }
 
     {
-        let encrypted_key = client_dh_encrypt(&mut user_agent, seal_key).await;
+        let encrypted_key = client_dh_encrypt(&user_agent_ref, seal_key).await;
 
-        let response = user_agent
+        let response = user_agent_ref
-            .process_transport_inbound(unseal_key_request(encrypted_key))
+            .ask(HandleUnsealEncryptedKey { req: encrypted_key })
             .await
             .unwrap();
 

server/crates/arbiter-useragent/Cargo.toml

@@ -12,4 +12,4 @@ tonic.workspace = true
 tracing.workspace = true
 ed25519-dalek.workspace = true
 smlang.workspace = true
-x25519-dalek.workspace = true
+x25519-dalek.workspace = true

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

@@ -1,13 +1,66 @@
+use arbiter_proto::{proto::UserAgentRequest, transport::TransportActor};
 use ed25519_dalek::SigningKey;
-use kameo::Actor;
+use kameo::{
+    Actor, Reply,
+    actor::{ActorRef, WeakActorRef},
+    prelude::Message,
+};
+use smlang::statemachine;
 use tonic::transport::CertificateDer;
+use tracing::{debug, error};
 
 struct Storage {
     pub identity: SigningKey,
     pub server_ca_cert: CertificateDer<'static>,
 }
 
-#[derive(Actor)]
+#[derive(Debug)]
-pub struct UserAgent {
+pub enum InitError {
+    StorageError,
+    Other(String),
+}
 
-}
+statemachine! {
+    name: UserAgentStateMachine,
+    custom_error: false,
+    transitions: {
+        *Init + SendAuthChallenge = WaitingForAuthSolution
+    }
+}
+
+
+
+pub struct UserAgentActor<A: TransportActor<UserAgentRequest>> {
+    key: SigningKey,
+    server_ca_cert: CertificateDer<'static>,
+    sender: ActorRef<A>,
+}
+impl<A: TransportActor<UserAgentRequest>> Actor for UserAgentActor<A> {
+    type Args = Self;
+
+    type Error = InitError;
+
+    async fn on_start(args: Self::Args, actor_ref: ActorRef<Self>) -> Result<Self, Self::Error> {
+        todo!()
+    }
+
+    async fn on_link_died(
+        &mut self,
+        _: WeakActorRef<Self>,
+        id: kameo::prelude::ActorId,
+        _: kameo::prelude::ActorStopReason,
+    ) -> Result<std::ops::ControlFlow<kameo::prelude::ActorStopReason>, Self::Error> {
+        if id == self.sender.id() {
+            error!("Transport actor died, stopping UserAgentActor");
+            Ok(std::ops::ControlFlow::Break(
+                kameo::prelude::ActorStopReason::Normal,
+            ))
+        } else {
+            debug!(
+                "Linked actor {} died, but it's not the transport actor, ignoring",
+                id
+            );
+            Ok(std::ops::ControlFlow::Continue(()))
+        }
+    }
+}