feat(transport): add domain error type to GrpcTransportActor

refactor: introduce TransportActor abstraction
2026-02-26 15:07:11 +01:00 · 2026-02-25 21:44:01 +01:00
18 changed files with 1210 additions and 1132 deletions
--- a/app/.dart_tool/package_config.json
+++ b/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"
+}
--- a/app/.dart_tool/package_graph.json
+++ b/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
+}
--- a/app/.dart_tool/version
+++ b/app/.dart_tool/version
@@ -0,0 +1 @@
+3.38.9
--- a/app/macos/Flutter/ephemeral/Flutter-Generated.xcconfig
+++ b/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
--- a/app/macos/Flutter/ephemeral/flutter_export_environment.sh
+++ b/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"
--- a/server/Cargo.lock
+++ b/server/Cargo.lock
@@ -47,9 +47,9 @@ dependencies = [

 [[package]]
 name = "anyhow"
-version = "1.0.102"
+version = "1.0.101"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
+checksum = "5f0e0fee31ef5ed1ba1316088939cea399010ed7731dba877ed44aeb407a75ea"

 [[package]]
 name = "arbiter-client"
@@ -121,13 +121,9 @@ version = "0.1.0"
 dependencies = [
 "arbiter-proto",
 "ed25519-dalek",
- "http",
 "kameo",
- "rustls-webpki",
 "smlang",
- "thiserror",
 "tokio",
- "tokio-stream",
 "tonic",
 "tracing",
 "x25519-dalek",
@@ -170,7 +166,7 @@ checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 "synstructure",
 ]

@@ -182,7 +178,7 @@ checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -193,7 +189,7 @@ checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -210,9 +206,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"

 [[package]]
 name = "aws-lc-rs"
-version = "1.16.0"
+version = "1.15.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d9a7b350e3bb1767102698302bc37256cbd48422809984b98d292c40e2579aa9"
+checksum = "7b7b6141e96a8c160799cc2d5adecd5cbbe5054cb8c7c4af53da0f83bb7ad256"
 dependencies = [
 "aws-lc-sys",
 "untrusted 0.7.1",
@@ -348,18 +344,18 @@ dependencies = [

 [[package]]
 name = "block-buffer"
-version = "0.12.0"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cdd35008169921d80bc60d3d0ab416eecb028c4cd653352907921d95084790be"
+checksum = "96eb4cdd6cf1b31d671e9efe75c5d1ec614776856cefbe109ca373554a6d514f"
 dependencies = [
 "hybrid-array",
 ]

 [[package]]
 name = "bumpalo"
-version = "3.20.2"
+version = "3.19.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb"
+checksum = "5dd9dc738b7a8311c7ade152424974d8115f2cdad61e8dab8dac9f2362298510"

 [[package]]
 name = "bytes"
@@ -422,9 +418,9 @@ dependencies = [

 [[package]]
 name = "chrono"
-version = "0.4.44"
+version = "0.4.43"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c673075a2e0e5f4a1dde27ce9dee1ea4558c7ffe648f576438a20ca1d2acc4b0"
+checksum = "fac4744fb15ae8337dc853fee7fb3f4e48c0fbaa23d0afe49c447b4fab126118"
 dependencies = [
 "iana-time-zone",
 "js-sys",
@@ -518,9 +514,9 @@ dependencies = [

 [[package]]
 name = "crypto-common"
-version = "0.2.1"
+version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "77727bb15fa921304124b128af125e7e3b968275d1b108b379190264f4423710"
+checksum = "211f05e03c7d03754740fd9e585de910a095d6b99f8bcfffdef8319fa02a8331"
 dependencies = [
 "hybrid-array",
 ]
@@ -549,7 +545,7 @@ dependencies = [
 "cfg-if",
 "cpufeatures 0.2.17",
 "curve25519-dalek-derive",
- "digest 0.11.1",
+ "digest 0.11.0",
 "fiat-crypto 0.3.0",
 "rustc_version",
 "subtle",
@@ -564,7 +560,7 @@ checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -588,7 +584,7 @@ dependencies = [
 "proc-macro2",
 "quote",
 "strsim",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -599,7 +595,7 @@ checksum = "d38308df82d1080de0afee5d069fa14b0326a88c14f15c5ccda35b4a6c414c81"
 dependencies = [
 "darling_core",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -638,9 +634,9 @@ dependencies = [

 [[package]]
 name = "deranged"
-version = "0.5.8"
+version = "0.5.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c"
+checksum = "cc3dc5ad92c2e2d1c193bbbbdf2ea477cb81331de4f3103f267ca18368b988c4"
 dependencies = [
 "powerfmt",
 ]
@@ -686,7 +682,7 @@ dependencies = [
 "dsl_auto_type",
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -706,7 +702,7 @@ version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fe2444076b48641147115697648dc743c2c00b61adade0f01ce67133c7babe8c"
 dependencies = [
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -722,12 +718,12 @@ dependencies = [

 [[package]]
 name = "digest"
-version = "0.11.1"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "285743a676ccb6b3e116bc14cc69319b957867930ae9c4822f8e0f54509d7243"
+checksum = "f8bf3682cdec91817be507e4aa104314898b95b84d74f3d43882210101a545b6"
 dependencies = [
- "block-buffer 0.12.0",
- "crypto-common 0.2.1",
+ "block-buffer 0.11.0",
+ "crypto-common 0.2.0",
 ]

 [[package]]
@@ -738,7 +734,7 @@ checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -758,7 +754,7 @@ dependencies = [
 "heck",
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -953,7 +949,7 @@ checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -1413,9 +1409,9 @@ dependencies = [

 [[package]]
 name = "js-sys"
-version = "0.3.90"
+version = "0.3.85"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "14dc6f6450b3f6d4ed5b16327f38fed626d375a886159ca555bd7822c0c3a5a6"
+checksum = "8c942ebf8e95485ca0d52d97da7c5a2c387d0e7f0ba4c35e93bfcaee045955b3"
 dependencies = [
 "once_cell",
 "wasm-bindgen",
@@ -1445,7 +1441,7 @@ dependencies = [
 "heck",
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -1478,9 +1474,9 @@ dependencies = [

 [[package]]
 name = "linux-raw-sys"
-version = "0.12.1"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"
+checksum = "df1d3c3b53da64cf5760482273a98e575c651a67eec7f77df96b5b642de8f039"

 [[package]]
 name = "litemap"
@@ -1562,7 +1558,7 @@ checksum = "db5b29714e950dbb20d5e6f74f9dcec4edbcc1067bb7f8ed198c097b8c1a818b"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -1710,9 +1706,9 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"

 [[package]]
 name = "owo-colors"
-version = "4.3.0"
+version = "4.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d211803b9b6b570f68772237e415a029d5a50c65d382910b879fb19d3271f94d"
+checksum = "9c6901729fa79e91a0913333229e9ca5dc725089d1c363b2f4b4760709dc4a52"

 [[package]]
 name = "parking_lot"
@@ -1792,7 +1788,7 @@ checksum = "6e918e4ff8c4549eb882f14b3a4bc8c8bc93de829416eacf579f1207a8fbf861"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -1852,7 +1848,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b"
 dependencies = [
 "proc-macro2",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -1924,7 +1920,7 @@ dependencies = [
 "pulldown-cmark",
 "pulldown-cmark-to-cmark",
 "regex",
- "syn 2.0.117",
+ "syn 2.0.115",
 "tempfile",
 ]

@@ -1938,7 +1934,7 @@ dependencies = [
 "itertools",
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -1952,9 +1948,9 @@ dependencies = [

 [[package]]
 name = "pulldown-cmark"
-version = "0.13.1"
+version = "0.13.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "83c41efbf8f90ac44de7f3a868f0867851d261b56291732d0cbf7cceaaeb55a6"
+checksum = "1e8bbe1a966bd2f362681a44f6edce3c2310ac21e4d5067a6e7ec396297a6ea0"
 dependencies = [
 "bitflags",
 "memchr",
@@ -2060,9 +2056,9 @@ dependencies = [

 [[package]]
 name = "regex-syntax"
-version = "0.8.10"
+version = "0.8.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a"
+checksum = "a96887878f22d7bad8a3b6dc5b7440e0ada9a245242924394987b21cf2210a4c"

 [[package]]
 name = "relative-path"
@@ -2079,7 +2075,7 @@ dependencies = [
 "proc-macro-error",
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -2131,7 +2127,7 @@ dependencies = [
 "regex",
 "relative-path",
 "rustc_version",
- "syn 2.0.117",
+ "syn 2.0.115",
 "unicode-ident",
 ]

@@ -2161,9 +2157,9 @@ dependencies = [

 [[package]]
 name = "rustix"
-version = "1.1.4"
+version = "1.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190"
+checksum = "146c9e247ccc180c1f61615433868c99f3de3ae256a30a43b49f67c2d9171f34"
 dependencies = [
 "bitflags",
 "errno",
@@ -2174,9 +2170,9 @@ dependencies = [

 [[package]]
 name = "rustls"
-version = "0.23.37"
+version = "0.23.36"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4"
+checksum = "c665f33d38cea657d9614f766881e4d510e0eda4239891eea56b4cadcf01801b"
 dependencies = [
 "aws-lc-rs",
 "log",
@@ -2271,7 +2267,7 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -2304,7 +2300,7 @@ checksum = "7c5f3b1e2dc8aad28310d8410bd4d7e180eca65fca176c52ab00d364475d0024"
 dependencies = [
 "cfg-if",
 "cpufeatures 0.2.17",
- "digest 0.11.1",
+ "digest 0.11.0",
 ]

 [[package]]
@@ -2441,7 +2437,7 @@ dependencies = [
 "heck",
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -2484,9 +2480,9 @@ dependencies = [

 [[package]]
 name = "syn"
-version = "2.0.117"
+version = "2.0.115"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e665b8803e7b1d2a727f4023456bbbbe74da67099c585258af0ad9c5013b9b99"
+checksum = "6e614ed320ac28113fa64972c4262d5dbc89deacdfd00c34a3e4cea073243c12"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -2507,14 +2503,14 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
 name = "tempfile"
-version = "3.26.0"
+version = "3.25.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "82a72c767771b47409d2345987fda8628641887d5466101319899796367354a0"
+checksum = "0136791f7c95b1f6dd99f9cc786b91bb81c3800b639b3478e561ddb7be95e5f1"
 dependencies = [
 "fastrand",
 "getrandom 0.4.1",
@@ -2551,7 +2547,7 @@ checksum = "be35209fd0781c5401458ab66e4f98accf63553e8fae7425503e92fdd319783b"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -2581,7 +2577,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -2660,7 +2656,7 @@ checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -2734,18 +2730,18 @@ dependencies = [

 [[package]]
 name = "toml_parser"
-version = "1.0.9+spec-1.1.0"
+version = "1.0.8+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "702d4415e08923e7e1ef96cd5727c0dfed80b4d2fa25db9647fe5eb6f7c5a4c4"
+checksum = "0742ff5ff03ea7e67c8ae6c93cac239e0d9784833362da3f9a9c1da8dfefcbdc"
 dependencies = [
 "winnow",
 ]

 [[package]]
 name = "tonic"
-version = "0.14.5"
+version = "0.14.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fec7c61a0695dc1887c1b53952990f3ad2e3a31453e1f49f10e75424943a93ec"
+checksum = "7f32a6f80051a4111560201420c7885d0082ba9efe2ab61875c587bb6b18b9a0"
 dependencies = [
 "async-trait",
 "axum",
@@ -2775,21 +2771,21 @@ dependencies = [

 [[package]]
 name = "tonic-build"
-version = "0.14.5"
+version = "0.14.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1882ac3bf5ef12877d7ed57aad87e75154c11931c2ba7e6cde5e22d63522c734"
+checksum = "ce6d8958ed3be404120ca43ffa0fb1e1fc7be214e96c8d33bd43a131b6eebc9e"
 dependencies = [
 "prettyplease",
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
 name = "tonic-prost"
-version = "0.14.5"
+version = "0.14.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a55376a0bbaa4975a3f10d009ad763d8f4108f067c7c2e74f3001fb49778d309"
+checksum = "9f86539c0089bfd09b1f8c0ab0239d80392af74c21bc9e0f15e1b4aca4c1647f"
 dependencies = [
 "bytes",
 "prost",
@@ -2798,16 +2794,16 @@ dependencies = [

 [[package]]
 name = "tonic-prost-build"
-version = "0.14.5"
+version = "0.14.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f3144df636917574672e93d0f56d7edec49f90305749c668df5101751bb8f95a"
+checksum = "65873ace111e90344b8973e94a1fc817c924473affff24629281f90daed1cd2e"
 dependencies = [
 "prettyplease",
 "proc-macro2",
 "prost-build",
 "prost-types",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 "tempfile",
 "tonic-build",
 ]
@@ -2862,7 +2858,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -2924,9 +2920,9 @@ checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142"

 [[package]]
 name = "unicode-ident"
-version = "1.0.24"
+version = "1.0.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
+checksum = "537dd038a89878be9b64dd4bd1b260315c1bb94f4d784956b81e27a088d9a09e"

 [[package]]
 name = "unicode-linebreak"
@@ -3055,9 +3051,9 @@ dependencies = [

 [[package]]
 name = "wasm-bindgen"
-version = "0.2.113"
+version = "0.2.108"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "60722a937f594b7fde9adb894d7c092fc1bb6612897c46368d18e7a20208eff2"
+checksum = "64024a30ec1e37399cf85a7ffefebdb72205ca1c972291c51512360d90bd8566"
 dependencies = [
 "cfg-if",
 "once_cell",
@@ -3068,9 +3064,9 @@ dependencies = [

 [[package]]
 name = "wasm-bindgen-macro"
-version = "0.2.113"
+version = "0.2.108"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0fac8c6395094b6b91c4af293f4c79371c163f9a6f56184d2c9a85f5a95f3950"
+checksum = "008b239d9c740232e71bd39e8ef6429d27097518b6b30bdf9086833bd5b6d608"
 dependencies = [
 "quote",
 "wasm-bindgen-macro-support",
@@ -3078,22 +3074,22 @@ dependencies = [

 [[package]]
 name = "wasm-bindgen-macro-support"
-version = "0.2.113"
+version = "0.2.108"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ab3fabce6159dc20728033842636887e4877688ae94382766e00b180abac9d60"
+checksum = "5256bae2d58f54820e6490f9839c49780dff84c65aeab9e772f15d5f0e913a55"
 dependencies = [
 "bumpalo",
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 "wasm-bindgen-shared",
 ]

 [[package]]
 name = "wasm-bindgen-shared"
-version = "0.2.113"
+version = "0.2.108"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "de0e091bdb824da87dc01d967388880d017a0a9bc4f3bdc0d86ee9f9336e3bb5"
+checksum = "1f01b580c9ac74c8d8f0c0e4afb04eeef2acf145458e52c03845ee9cd23e3d12"
 dependencies = [
 "unicode-ident",
 ]
@@ -3175,7 +3171,7 @@ checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -3186,7 +3182,7 @@ checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -3417,7 +3413,7 @@ dependencies = [
 "heck",
 "indexmap",
 "prettyplease",
- "syn 2.0.117",
+ "syn 2.0.115",
 "wasm-metadata",
 "wit-bindgen-core",
 "wit-component",
@@ -3433,7 +3429,7 @@ dependencies = [
 "prettyplease",
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 "wit-bindgen-core",
 "wit-bindgen-rust",
 ]
@@ -3539,7 +3535,7 @@ checksum = "b659052874eb698efe5b9e8cf382204678a0086ebf46982b79d6ca3182927e5d"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 "synstructure",
 ]

@@ -3560,7 +3556,7 @@ checksum = "d71e5d6e06ab090c67b5e44993ec16b72dcbaabc526db883a360057678b48502"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 "synstructure",
 ]

@@ -3581,7 +3577,7 @@ checksum = "85a5b4158499876c763cb03bc4e49185d3cccbabb15b33c627f7884f43db852e"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
@@ -3614,7 +3610,7 @@ checksum = "eadce39539ca5cb3985590102671f2567e659fca9666581ad3411d59207951f3"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.117",
+ "syn 2.0.115",
 ]

 [[package]]
--- a/server/crates/arbiter-proto/Cargo.toml
+++ b/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"

--- a/server/crates/arbiter-proto/src/transport.rs
+++ b/server/crates/arbiter-proto/src/transport.rs
@@ -1,293 +1,371 @@
-//! Transport-facing abstractions for protocol/session code.
+//! Transport abstraction layer for bridging gRPC bidirectional streaming with kameo actors.
 //!
-//! This module separates three concerns:
+//! This module provides a clean separation between the gRPC transport layer and business logic
+//! by modeling the connection as two linked kameo actors:
 //!
-//! - protocol/session logic wants a small duplex interface ([`Bi`])
-//! - transport adapters push concrete stream items to an underlying IO layer
-//! - transport boundaries translate between protocol-facing and transport-facing
-//!   item types via direction-specific converters
+//! - 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.
 //!
-//! [`Bi`] is intentionally minimal and transport-agnostic:
-//! - [`Bi::recv`] yields inbound protocol messages
-//! - [`Bi::send`] accepts outbound protocol/domain items
+//! 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.
 //!
-//! # Generic Ordering Rule
+//! # Terminology
 //!
-//! This module uses a single convention consistently: when a type or trait is
-//! parameterized by protocol message directions, the generic parameters are
-//! declared as `Inbound` first, then `Outbound`.
+//! - **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.
 //!
-//! For [`Bi`], that means `Bi<Inbound, Outbound>`:
-//! - `recv() -> Option<Inbound>`
-//! - `send(Outbound)`
-//!
-//! For adapter types that are parameterized by direction-specific converters,
-//! inbound-related converter parameters are declared before outbound-related
-//! converter parameters.
-//!
-//! [`RecvConverter`] and [`SendConverter`] are infallible conversion traits used
-//! by adapters to map between protocol-facing and transport-facing item types.
-//! The traits themselves are not result-aware; adapters decide how transport
-//! errors are handled before (or instead of) conversion.
-//!
-//! [`grpc::GrpcAdapter`] combines:
-//! - a tonic inbound stream
-//! - a Tokio sender for outbound transport items
-//! - a [`RecvConverter`] for the receive path
-//! - a [`SendConverter`] for the send path
-//!
-//! [`DummyTransport`] is a no-op implementation useful for tests and local actor
-//! execution where no real network stream exists.
-//!
-//! # Component Interaction
+//! # Architecture
 //!
 //! ```text
-//! inbound (network -> protocol)
-//! ============================
-//!
-//! tonic::Streaming<RecvTransport>
-//!     -> grpc::GrpcAdapter::recv()
-//!          |
-//!          +--> on `Ok(item)`: RecvConverter::convert(RecvTransport) -> Inbound
-//!          +--> on `Err(status)`: log error and close stream (`None`)
-//!     -> Bi::recv()
-//!     -> protocol/session actor
-//!
-//! outbound (protocol -> network)
-//! ==============================
-//!
-//! protocol/session actor
-//!     -> Bi::send(Outbound)
-//!     -> grpc::GrpcAdapter::send()
-//!          |
-//!          +--> SendConverter::convert(Outbound) -> SendTransport
-//!     -> Tokio mpsc::Sender<SendTransport>
-//!     -> tonic response stream
+//! gRPC Stream ──InboundMessage──▶ GrpcTransportActor ──tell(InboundMessage)──▶ BusinessActor
+//!                                 ▲                                    │
+//!                                 └─tell(Result<OutboundMessage, _>)────┘
+//!                                 │
+//!                          mpsc::Sender ──▶ Client
 //! ```
 //!
-//! # Design Notes
+//! # Example
 //!
-//! - `send()` returns [`Error`] only for transport delivery failures (for
-//!   example, when the outbound channel is closed).
-//! - [`grpc::GrpcAdapter`] logs tonic receive errors and treats them as stream
-//!   closure (`None`).
-//! - When protocol-facing and transport-facing types are identical, use
-//!   [`IdentityRecvConverter`] / [`IdentitySendConverter`].
+//! ```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 std::marker::PhantomData;
+use futures::{Stream, StreamExt};
+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`].
-pub enum Error {
-    /// The outbound side of the transport is no longer accepting messages.
-    ChannelClosed,
-}
-
-/// Minimal bidirectional transport abstraction used by protocol code.
+/// A bidirectional stream abstraction for sans-io testing.
 ///
-/// `Bi<Inbound, Outbound>` models a duplex channel with:
-/// - inbound items of type `Inbound` read via [`Bi::recv`]
-/// - outbound items of type `Outbound` written via [`Bi::send`]
-pub trait Bi<Inbound, Outbound>: Send + Sync + 'static {
+/// Combines a [`Stream`] of incoming messages with the ability to [`send`](Bi::send)
+/// outgoing responses. This trait allows business logic to be tested without a real
+/// gRPC connection by swapping in an in-memory implementation.
+///
+/// # Type Parameters
+/// - `T`: `InboundMessage` received from the channel/socket (e.g., `UserAgentRequest`)
+/// - `U`: `OutboundMessage` sent to the channel/socket (e.g., `UserAgentResponse`)
+pub trait Bi<T, U>: Stream<Item = Result<T, Status>> + Send + Sync + 'static {
+    type Error;
    fn send(
        &mut self,
-        item: Outbound,
-    ) -> impl std::future::Future<Output = Result<(), Error>> + Send;
-
-    fn recv(&mut self) -> impl std::future::Future<Output = Option<Inbound>> + Send;
+        item: Result<U, Status>,
+    ) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send;
 }

-/// Converts transport-facing inbound items into protocol-facing inbound items.
-pub trait RecvConverter: Send + Sync + 'static {
-    type Input;
-    type Output;
-
-    fn convert(&self, item: Self::Input) -> Self::Output;
+/// Concrete [`Bi`] implementation backed by a tonic gRPC [`Streaming`] and an [`mpsc::Sender`].
+///
+/// This is the production implementation used in gRPC service handlers. The `request_stream`
+/// receives messages from the client, and `response_sender` sends responses back.
+pub struct BiStream<T, U> {
+    pub request_stream: Streaming<T>,
+    pub response_sender: mpsc::Sender<Result<U, Status>>,
 }

-/// Converts protocol/domain outbound items into transport-facing outbound items.
-pub trait SendConverter: Send + Sync + 'static {
-    type Input;
-    type Output;
-
-    fn convert(&self, item: Self::Input) -> Self::Output;
-}
-
-/// A [`RecvConverter`] that forwards values unchanged.
-pub struct IdentityRecvConverter<T> {
-    _marker: PhantomData<T>,
-}
-
-impl<T> IdentityRecvConverter<T> {
-    pub fn new() -> Self {
-        Self {
-            _marker: PhantomData,
-        }
-    }
-}
-
-impl<T> Default for IdentityRecvConverter<T> {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl<T> RecvConverter for IdentityRecvConverter<T>
+impl<T, U> Stream for BiStream<T, U>
 where
-    T: Send + Sync + 'static,
+    T: Send + 'static,
+    U: Send + 'static,
 {
-    type Input = T;
-    type Output = T;
+    type Item = Result<T, Status>;

-    fn convert(&self, item: Self::Input) -> Self::Output {
-        item
+    fn poll_next(
+        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)
    }
 }

-/// A [`SendConverter`] that forwards values unchanged.
-pub struct IdentitySendConverter<T> {
-    _marker: PhantomData<T>,
-}
-
-impl<T> IdentitySendConverter<T> {
-    pub fn new() -> Self {
-        Self {
-            _marker: PhantomData,
-        }
-    }
-}
-
-impl<T> Default for IdentitySendConverter<T> {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl<T> SendConverter for IdentitySendConverter<T>
+impl<T, U> Bi<T, U> for BiStream<T, U>
 where
-    T: Send + Sync + 'static,
+    T: Send + 'static,
+    U: Send + 'static,
 {
-    type Input = T;
-    type Output = T;
+    type Error = SendError<Result<U, Status>>;

-    fn convert(&self, item: Self::Input) -> Self::Output {
-        item
+    async fn send(&mut self, item: Result<U, Status>) -> Result<(), Self::Error> {
+        self.response_sender.send(item).await
    }
 }

-/// gRPC-specific transport adapters and helpers.
-pub mod grpc {
-    use futures::StreamExt;
-    use tokio::sync::mpsc;
-    use tonic::Streaming;
+/// 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>>
+{
+}

-    use super::{Bi, Error, RecvConverter, SendConverter};
+/// 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>,
+}

-    /// [`Bi`] adapter backed by a tonic gRPC bidirectional stream.
-    ///
-
-    /// Tonic receive errors are logged and treated as stream closure (`None`).
-    /// The receive converter is only invoked for successful inbound transport
-    /// items.
-    pub struct GrpcAdapter<InboundConverter, OutboundConverter>
-    where
-        InboundConverter: RecvConverter,
-        OutboundConverter: SendConverter,
-    {
-        sender: mpsc::Sender<OutboundConverter::Output>,
-        receiver: Streaming<InboundConverter::Input>,
-        inbound_converter: InboundConverter,
-        outbound_converter: OutboundConverter,
-    }
-
-
-    impl<InboundTransport, Inbound, InboundConverter, OutboundConverter>
-        GrpcAdapter<InboundConverter, OutboundConverter>
-    where
-        InboundConverter: RecvConverter<Input = InboundTransport, Output = Inbound>,
-        OutboundConverter: SendConverter,
-    {
+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<OutboundConverter::Output>,
-            receiver: Streaming<InboundTransport>,
-            inbound_converter: InboundConverter,
-            outbound_converter: OutboundConverter,
+        sender: mpsc::Sender<Result<Outbound, tonic::Status>>,
+        receiver: tonic::Streaming<Inbound>,
+        business_logic_actor: Recipient<Inbound>,
    ) -> Self {
        Self {
            sender,
            receiver,
-                inbound_converter,
-                outbound_converter,
-            }
-        }
-    }
-
-
-    impl< InboundConverter, OutboundConverter> Bi<InboundConverter::Output, OutboundConverter::Input>
-        for GrpcAdapter<InboundConverter, OutboundConverter>
-    where
-        InboundConverter: RecvConverter,
-        OutboundConverter: SendConverter,
-        OutboundConverter::Input: Send + 'static,
-        OutboundConverter::Output: Send + 'static,
-    {
-        #[tracing::instrument(level = "trace", skip(self, item))]
-        async fn send(&mut self, item: OutboundConverter::Input) -> Result<(), Error> {
-            let outbound = self.outbound_converter.convert(item);
-            self.sender
-                .send(outbound)
-                .await
-                .map_err(|_| Error::ChannelClosed)
-        }
-
-        #[tracing::instrument(level = "trace", skip(self))]
-        async fn recv(&mut self) -> Option<InboundConverter::Output> {
-            match self.receiver.next().await {
-                Some(Ok(item)) => Some(self.inbound_converter.convert(item)),
-                Some(Err(error)) => {
-                    tracing::error!(error = ?error, "grpc transport recv failed; closing stream");
-                    None
-                }
-                None => None,
-            }
-        }
-    }
-}
-
-/// No-op [`Bi`] transport for tests and manual actor usage.
-///
-/// `send` drops all items and succeeds. [`Bi::recv`] never resolves and therefore
-/// does not busy-wait or spuriously close the stream.
-pub struct DummyTransport<Inbound, Outbound> {
-    _marker: PhantomData<(Inbound, Outbound)>,
-}
-
-impl<Inbound, Outbound> DummyTransport<Inbound, Outbound> {
-    pub fn new() -> Self {
-        Self {
-            _marker: PhantomData,
+            business_logic_actor,
+            _error: std::marker::PhantomData,
        }
    }
 }

-impl<Inbound, Outbound> Default for DummyTransport<Inbound, Outbound> {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl<Inbound, Outbound> Bi<Inbound, Outbound> for DummyTransport<Inbound, Outbound>
+impl<Outbound, Inbound, E> Actor for GrpcTransportActor<Outbound, Inbound, E>
 where
-    Inbound: Send + Sync + 'static,
-    Outbound: Send + Sync + 'static,
+    Outbound: Send + 'static,
+    Inbound: Send + 'static,
+    E: Into<tonic::Status> + Send + 'static,
 {
-    async fn send(&mut self, _item: Outbound) -> Result<(), Error> {
-        Ok(())
+    type Args = Self;
+
+    type Error = ();
+
+    async fn on_start(args: Self::Args, _: ActorRef<Self>) -> Result<Self, Self::Error> {
+        Ok(args)
    }

-    fn recv(&mut self) -> impl std::future::Future<Output = Option<Inbound>> + Send {
-        async {
-            std::future::pending::<()>().await;
-            None
+    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,
+    DomainError: Send + 'static,
+    Business: Actor + Message<Inbound> + Send + 'static,
+    BusinessCtor: FnOnce(PreparedActor<Business>, Recipient<Result<Outbound, DomainError>>),
+    TransportCtor:
+        FnOnce(PreparedActor<Transport>, Recipient<Inbound>),
+{
+    let prepared_business: PreparedActor<Business> = Spawn::prepare();
+    let prepared_transport: PreparedActor<Transport> = Spawn::prepare();
+
+    let business_ref = prepared_business.actor_ref().clone();
+    let transport_ref = prepared_transport.actor_ref().clone();
+
+    transport_ref.link(&business_ref).await;
+    business_ref.link(&transport_ref).await;
+
+    let recipient = transport_ref.clone().recipient();
+    business_ctor(prepared_business, recipient);
+    let business_recipient = business_ref.clone().recipient();
+    transport_ctor(prepared_transport, business_recipient);
+
+
+    (transport_ref, business_ref)
+}
--- a/server/crates/arbiter-server/src/actors/user_agent/error.rs
+++ b/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()),
+        }
+    }
+}
--- a/server/crates/arbiter-server/src/actors/user_agent/mod.rs
+++ b/server/crates/arbiter-server/src/actors/user_agent/mod.rs
@@ -1,26 +1,23 @@
 use std::{ops::DerefMut, sync::Mutex};

-use arbiter_proto::{
-    proto::{
+use arbiter_proto::proto::{
    UnsealEncryptedKey, UnsealResult, UnsealStart, UnsealStartResponse, UserAgentRequest,
    UserAgentResponse,
    auth::{
        self, AuthChallengeRequest, AuthOk, ClientMessage as ClientAuthMessage,
-            ServerMessage as AuthServerMessage, client_message::Payload as ClientAuthPayload,
+        ServerMessage as AuthServerMessage,
+        client_message::Payload as ClientAuthPayload,
        server_message::Payload as ServerAuthPayload,
    },
    user_agent_request::Payload as UserAgentRequestPayload,
    user_agent_response::Payload as UserAgentResponsePayload,
-    },
-    transport::{Bi, DummyTransport},
 };
 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 tracing::{error, info};
 use x25519_dalek::{EphemeralSecret, PublicKey};

@@ -38,59 +35,24 @@ use crate::{
    db::{self, schema},
 };

+mod error;
 mod state;

-#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
-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,
-}
+pub use error::UserAgentError;

-pub struct UserAgentActor<Transport>
-where
-    Transport: Bi<UserAgentRequest, Result<UserAgentResponse, UserAgentError>>,
-{
+#[derive(Actor)]
+pub struct UserAgentActor {
    db: db::DatabasePool,
    actors: GlobalActors,
    state: UserAgentStateMachine<DummyContext>,
-    transport: Transport,
+    transport: Recipient<Result<UserAgentResponse, UserAgentError>>,
 }

-impl<Transport> UserAgentActor<Transport>
-where
-    Transport: Bi<UserAgentRequest, Result<UserAgentResponse, UserAgentError>>,
-{
-    pub(crate) fn new(context: ServerContext, transport: Transport) -> Self {
+impl UserAgentActor {
+    pub(crate) fn new(
+        context: ServerContext,
+        transport: Recipient<Result<UserAgentResponse, UserAgentError>>,
+    ) -> Self {
        Self {
            db: context.db.clone(),
            actors: context.actors.clone(),
@@ -99,18 +61,23 @@ where
        }
    }

-    fn transition(&mut self, event: UserAgentEvents) -> Result<(), UserAgentError> {
-        self.state.process_event(event).map_err(|e| {
-            error!(?e, "State transition failed");
-            UserAgentError::StateTransitionFailed
-        })?;
-        Ok(())
+    pub fn new_manual(
+        db: db::DatabasePool,
+        actors: GlobalActors,
+        transport: Recipient<Result<UserAgentResponse, UserAgentError>>,
+    ) -> Self {
+        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 {
@@ -126,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
@@ -142,7 +117,7 @@ where
            .await
            .map_err(|e| {
                error!(?pubkey, "Failed to consume bootstrap token: {e}");
-                UserAgentError::BootstrapperActorUnreachable
+                UserAgentError::ActorUnavailable
            })?;

        if !token_ok {
@@ -151,10 +126,7 @@ where
        }

        {
-            let mut conn = self.db.get().await.map_err(|e| {
-                error!(error = ?e, "Database pool error");
-                UserAgentError::DatabasePoolUnavailable
-            })?;
+            let mut conn = self.db.get().await?;

            diesel::insert_into(schema::useragent_client::table)
                .values((
@@ -162,11 +134,7 @@ where
                    schema::useragent_client::nonce.eq(1),
                ))
                .execute(&mut conn)
-                .await
-                .map_err(|e| {
-                    error!(error = ?e, "Database error");
-                    UserAgentError::DatabaseOperationFailed
-                })?;
+                .await?;
        }

        self.transition(UserAgentEvents::ReceivedBootstrapToken)?;
@@ -176,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| {
-                error!(error = ?e, "Database pool error");
-                UserAgentError::DatabasePoolUnavailable
-            })?;
+            let mut db_conn = self.db.get().await?;
            db_conn
                .exclusive_transaction(|conn| {
                    Box::pin(async move {
@@ -203,16 +168,12 @@ where
                    })
                })
                .await
-                .optional()
-                .map_err(|e| {
-                    error!(error = ?e, "Database error");
-                    UserAgentError::DatabaseOperationFailed
-                })?
+                .optional()?
        };

        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 {
@@ -241,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);

@@ -275,18 +236,17 @@ fn unseal_response(payload: UserAgentResponsePayload) -> UserAgentResponse {
    }
 }

-impl<Transport> UserAgentActor<Transport>
-where
-    Transport: Bi<UserAgentRequest, Result<UserAgentResponse, UserAgentError>>,
-{
-    async fn handle_unseal_request(&mut self, req: UnsealStart) -> Output {
+#[messages]
+impl UserAgentActor {
+    #[message]
+    pub 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);

@@ -302,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();
@@ -369,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)
                    }
                }
            }
@@ -383,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)?;
@@ -401,7 +363,8 @@ where
        }
    }

-    async fn handle_auth_challenge_solution(
+    #[message]
+    pub async fn handle_auth_challenge_solution(
        &mut self,
        solution: auth::AuthChallengeSolution,
    ) -> Output {
@@ -422,67 +385,17 @@ where
    }
 }

+impl Message<UserAgentRequest> for UserAgentActor {
+    type Reply = ();

-impl<Transport> Actor for UserAgentActor<Transport>
-where
-    Transport: Bi<UserAgentRequest, Result<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(
+    async fn handle(
        &mut self,
-        _actor_ref: kameo::prelude::WeakActorRef<Self>,
-        mailbox_rx: &mut kameo::prelude::MailboxReceiver<Self>,
-    ) -> Option<kameo::mailbox::Signal<Self>> {
-        loop {
-            select! {
-                signal = mailbox_rx.recv() => {
-                    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, Result<UserAgentResponse, UserAgentError>>> {
-    pub fn new_manual(db: db::DatabasePool, actors: GlobalActors) -> Self {
-        Self {
-            db,
-            actors,
-            state: UserAgentStateMachine::new(DummyContext),
-            transport: DummyTransport::new(),
+        msg: UserAgentRequest,
+        _ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
+    ) -> Self::Reply {
+        let result = self.process_request(msg).await;
+        if let Err(e) = self.transport.tell(result).await {
+            error!(actor = "useragent", "Failed to send response to transport: {}", e);
        }
    }
 }
--- a/server/crates/arbiter-server/src/lib.rs
+++ b/server/crates/arbiter-server/src/lib.rs
@@ -1,18 +1,20 @@
 #![forbid(unsafe_code)]
 use arbiter_proto::{
    proto::{ClientRequest, ClientResponse, UserAgentRequest, UserAgentResponse},
-    transport::{IdentityRecvConverter, SendConverter, grpc},
+    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, UserAgentError},
+    actors::{
+        client::handle_client,
+        user_agent::UserAgentActor,
+    },
    context::ServerContext,
 };

@@ -22,69 +24,6 @@ pub mod db;

 const DEFAULT_CHANNEL_SIZE: usize = 1000;

-/// Converts User Agent domain outbounds into the tonic stream item emitted by
-/// the server.§
-///
-/// The conversion is defined at the server boundary so the actor module remains
-/// focused on domain semantics and does not depend on tonic status encoding.
-struct UserAgentGrpcSender;
-
-impl SendConverter for UserAgentGrpcSender {
-    type Input = Result<UserAgentResponse, UserAgentError>;
-    type Output = Result<UserAgentResponse, Status>;
-
-    fn convert(&self, item: Self::Input) -> Self::Output {
-        match item {
-            Ok(message) => Ok(message),
-            Err(err) => Err(user_agent_error_status(err)),
-        }
-    }
-}
-
-/// Maps User Agent domain errors to public gRPC transport errors for the
-/// `user_agent` streaming endpoint.
-fn user_agent_error_status(value: UserAgentError) -> Status {
-    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 Server {
    context: ServerContext,
 }
@@ -102,28 +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 transport = grpc::GrpcAdapter::new(
+        wire(
+            |prepared: PreparedActor<UserAgentActor>, recipient| {
+                prepared.spawn(UserAgentActor::new(context, recipient));
+            },
+            |prepared: PreparedActor<GrpcTransportActor<_, _, _>>, business_recipient| {
+                prepared.spawn(GrpcTransportActor::new(
                    tx,
                    req_stream,
-            IdentityRecvConverter::<UserAgentRequest>::new(),
-            UserAgentGrpcSender,
-        );
-        UserAgentActor::spawn(UserAgentActor::new(self.context.clone(), transport));
-
-        info!(event = "connection established", "grpc.user_agent");
+                    business_recipient,
+                ));
+            },
+        )
+        .await;

        Ok(Response::new(ReceiverStream::new(rx)))
    }
--- a/server/crates/arbiter-server/tests/user_agent.rs
+++ b/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"]
--- a/server/crates/arbiter-server/tests/user_agent/auth.rs
+++ b/server/crates/arbiter-server/tests/user_agent/auth.rs
@@ -1,29 +1,20 @@
 use arbiter_proto::proto::{
    UserAgentResponse,
-    UserAgentRequest,
-    auth::{self, AuthChallengeRequest, AuthOk, ClientMessage, client_message::Payload as ClientAuthPayload},
-    user_agent_request::Payload as UserAgentRequestPayload,
+    auth::{self, AuthChallengeRequest, AuthOk},
    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 _;
-
-fn auth_request(payload: ClientAuthPayload) -> UserAgentRequest {
-    UserAgentRequest {
-        payload: Some(UserAgentRequestPayload::AuthMessage(ClientMessage {
-            payload: Some(payload),
-        })),
-    }
-}
+use kameo::actor::Spawn;

 #[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
-        .process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
-            AuthChallengeRequest {
+    let result = user_agent_ref
+        .ask(HandleAuthChallengeRequest {
+            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,23 +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
-        .process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
-            AuthChallengeRequest {
+    let result = user_agent_ref
+        .ask(HandleAuthChallengeRequest {
+            req: AuthChallengeRequest {
                pubkey: pubkey_bytes,
                bootstrap_token: Some("invalid_token".to_string()),
            },
-        )))
+        })
        .await;

    match result {
-        Err(err) => {
-            assert_eq!(err, UserAgentError::InvalidBootstrapToken);
+        Err(kameo::error::SendError::HandlerError(err)) => {
+            assert!(
+                matches!(err, arbiter_server::actors::user_agent::UserAgentError::InvalidBootstrapToken),
+                "Expected InvalidBootstrapToken, got {err:?}"
+            );
+        }
+        Err(other) => {
+            panic!("Expected SendError::HandlerError, got {other:?}");
        }
        Ok(_) => {
            panic!("Expected error due to invalid bootstrap token, but got success");
@@ -105,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();
@@ -119,15 +122,15 @@ pub async fn test_challenge_auth() {
            .unwrap();
    }

-    let result = user_agent
-        .process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
-            AuthChallengeRequest {
+    let result = user_agent_ref
+        .ask(HandleAuthChallengeRequest {
+            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:
@@ -144,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
-        .process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeSolution(
-            auth::AuthChallengeSolution {
+    let result = user_agent_ref
+        .ask(HandleAuthChallengeSolution {
+            solution: auth::AuthChallengeSolution {
                signature: serialized_signature,
            },
-        )))
+        })
        .await
-        .expect("Shouldn't fail to process message");
+        .expect("Shouldn't fail to send message");

    assert_eq!(
        result,
--- a/server/crates/arbiter-server/tests/user_agent/unseal.rs
+++ b/server/crates/arbiter-server/tests/user_agent/unseal.rs
@@ -1,52 +1,27 @@
 use arbiter_proto::proto::{
-    UnsealEncryptedKey, UnsealResult, UnsealStart, UserAgentRequest, UserAgentResponse,
-    auth::{AuthChallengeRequest, ClientMessage, client_message::Payload as ClientAuthPayload},
-    user_agent_request::Payload as UserAgentRequestPayload,
+    UnsealEncryptedKey, UnsealResult, UnsealStart, auth::AuthChallengeRequest,
    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, Result<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,
-    TestUserAgent,
-) {
+) -> (arbiter_server::db::DatabasePool, ActorRef<UserAgentActor>) {
    let db = db::create_test_pool().await;

    let actors = GlobalActors::spawn(db.clone()).await.unwrap();
@@ -59,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
-        .process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
-            AuthChallengeRequest {
+    user_agent_ref
+        .ask(HandleAuthChallengeRequest {
+            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
-        .process_transport_inbound(unseal_start_request(UnsealStart {
+    let response = user_agent_ref
+        .ask(HandleUnsealRequest {
+            req: UnsealStart {
                client_pubkey: client_public.as_bytes().to_vec(),
-        }))
+            },
+        })
        .await
        .unwrap();

@@ -116,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
-        .process_transport_inbound(unseal_key_request(encrypted_key))
+    let response = user_agent_ref
+        .ask(HandleUnsealEncryptedKey { req: encrypted_key })
        .await
        .unwrap();

@@ -134,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
-        .process_transport_inbound(unseal_key_request(encrypted_key))
+    let response = user_agent_ref
+        .ask(HandleUnsealEncryptedKey { req: encrypted_key })
        .await
        .unwrap();

@@ -152,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
-        .process_transport_inbound(unseal_start_request(UnsealStart {
+    user_agent_ref
+        .ask(HandleUnsealRequest {
+            req: UnsealStart {
                client_pubkey: client_public.as_bytes().to_vec(),
-        }))
+            },
+        })
        .await
        .unwrap();

-    let response = user_agent
-        .process_transport_inbound(unseal_key_request(UnsealEncryptedKey {
+    let response = user_agent_ref
+        .ask(HandleUnsealEncryptedKey {
+            req: UnsealEncryptedKey {
                nonce: vec![0u8; 24],
                ciphertext: vec![0u8; 32],
                associated_data: vec![],
-        }))
+            },
+        })
        .await
        .unwrap();

@@ -185,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
-        .process_transport_inbound(unseal_start_request(UnsealStart {
+    let result = user_agent_ref
+        .ask(HandleUnsealRequest {
+            req: UnsealStart {
                client_pubkey: client_public.as_bytes().to_vec(),
-        }))
+            },
+        })
        .await;

    match result {
-        Err(err) => {
-            assert_eq!(err, UserAgentError::StateTransitionFailed);
+        Err(kameo::error::SendError::HandlerError(err)) => {
+            assert!(
+                matches!(err, arbiter_server::actors::user_agent::UserAgentError::InvalidState),
+                "Expected InvalidState, got {err:?}"
+            );
        }
        other => panic!("Expected state machine error, got {other:?}"),
    }
@@ -208,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
-            .process_transport_inbound(unseal_key_request(encrypted_key))
+        let response = user_agent_ref
+            .ask(HandleUnsealEncryptedKey { req: encrypted_key })
            .await
            .unwrap();

@@ -225,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
-            .process_transport_inbound(unseal_key_request(encrypted_key))
+        let response = user_agent_ref
+            .ask(HandleUnsealEncryptedKey { req: encrypted_key })
            .await
            .unwrap();

--- a/server/crates/arbiter-useragent/Cargo.toml
+++ b/server/crates/arbiter-useragent/Cargo.toml
@@ -9,12 +9,7 @@ arbiter-proto.path = "../arbiter-proto"
 kameo.workspace = true
 tokio = {workspace = true, features = ["net"]}
 tonic.workspace = true
-tonic.features = ["tls-aws-lc"]
 tracing.workspace = true
 ed25519-dalek.workspace = true
 smlang.workspace = true
 x25519-dalek.workspace = true
-thiserror.workspace = true
-tokio-stream.workspace = true
-http = "1.4.0"
-rustls-webpki = { version = "0.103.9", features = ["aws-lc-rs"] }
--- a/server/crates/arbiter-useragent/src/grpc.rs
+++ b/server/crates/arbiter-useragent/src/grpc.rs
@@ -1,71 +0,0 @@
-use arbiter_proto::{
-    proto::{
-        UserAgentRequest, UserAgentResponse, arbiter_service_client::ArbiterServiceClient,
-    },
-    transport::{IdentityRecvConverter, IdentitySendConverter, RecvConverter, grpc},
-    url::ArbiterUrl,
-};
-use ed25519_dalek::SigningKey;
-use kameo::actor::{ActorRef, Spawn};
-
-use tokio::sync::mpsc;
-use tokio_stream::wrappers::ReceiverStream;
-
-use tonic::transport::ClientTlsConfig;
-
-
-#[derive(Debug, thiserror::Error)]
-pub enum ConnectError {
-    #[error("Could establish connection")]
-    Connection(#[from] tonic::transport::Error),
-
-    #[error("Invalid server URI")]
-    InvalidUri(#[from] http::uri::InvalidUri),
-
-    #[error("Invalid CA certificate")]
-    InvalidCaCert(#[from] webpki::Error),
-
-    #[error("gRPC error")]
-    Grpc(#[from] tonic::Status),
-}
-
-    use super::UserAgentActor;
-
-pub type UserAgentGrpc = ActorRef<
-    UserAgentActor<
-        grpc::GrpcAdapter<
-            IdentityRecvConverter<UserAgentResponse>,
-            IdentitySendConverter<UserAgentRequest>,
-        >,
-    >,
->;
-pub async fn connect_grpc(
-    url: ArbiterUrl,
-    key: SigningKey,
-) -> Result<UserAgentGrpc, ConnectError> {
-    let bootstrap_token = url.bootstrap_token.clone();
-    let anchor = webpki::anchor_from_trusted_cert(&url.ca_cert)?.to_owned();
-    let tls = ClientTlsConfig::new().trust_anchor(anchor);
-
-    // TODO: if `host` is localhost, we need to verify server's process authenticity
-    let channel = tonic::transport::Channel::from_shared(format!("{}:{}", url.host, url.port))?
-        .tls_config(tls)?
-        .connect()
-        .await?;
-
-    let mut client = ArbiterServiceClient::new(channel);
-    let (tx, rx) = mpsc::channel(16);
-    let bistream = client.user_agent(ReceiverStream::new(rx)).await?;
-    let bistream = bistream.into_inner();
-
-    let adapter = grpc::GrpcAdapter::new(
-        tx,
-        bistream,
-        IdentityRecvConverter::new(),
-        IdentitySendConverter::new(),
-    );
-
-    let actor = UserAgentActor::spawn(UserAgentActor::new(key, bootstrap_token, adapter));
-
-    Ok(actor)
-}
--- a/server/crates/arbiter-useragent/src/lib.rs
+++ b/server/crates/arbiter-useragent/src/lib.rs
@@ -1,209 +1,66 @@
-use arbiter_proto::{
-    format_challenge,
-    proto::{
-        UserAgentRequest, UserAgentResponse,
-        auth::{
-            self, AuthChallengeRequest, AuthOk, ClientMessage as AuthClientMessage,
-            ServerMessage as AuthServerMessage, client_message::Payload as ClientAuthPayload,
-            server_message::Payload as ServerAuthPayload,
-        },
-        user_agent_request::Payload as UserAgentRequestPayload,
-        user_agent_response::Payload as UserAgentResponsePayload,
-    },
-    transport::Bi,
+use arbiter_proto::{proto::UserAgentRequest, transport::TransportActor};
+use ed25519_dalek::SigningKey;
+use kameo::{
+    Actor, Reply,
+    actor::{ActorRef, WeakActorRef},
+    prelude::Message,
 };
-use ed25519_dalek::{Signer, SigningKey};
-use kameo::{Actor, actor::ActorRef};
 use smlang::statemachine;
-use tokio::select;
-use tracing::{error, info};
+use tonic::transport::CertificateDer;
+use tracing::{debug, error};
+
+struct Storage {
+    pub identity: SigningKey,
+    pub server_ca_cert: CertificateDer<'static>,
+}
+
+#[derive(Debug)]
+pub enum InitError {
+    StorageError,
+    Other(String),
+}

 statemachine! {
-    name: UserAgent,
+    name: UserAgentStateMachine,
    custom_error: false,
    transitions: {
-        *Init + SentAuthChallengeRequest = WaitingForServerAuth,
-        WaitingForServerAuth + ReceivedAuthChallenge = WaitingForAuthOk,
-        WaitingForServerAuth + ReceivedAuthOk = Authenticated,
-        WaitingForAuthOk + ReceivedAuthOk = Authenticated,
+        *Init + SendAuthChallenge = WaitingForAuthSolution
    }
 }

-pub struct DummyContext;
-impl UserAgentStateMachineContext for DummyContext {}

-#[derive(Debug, thiserror::Error)]
-pub enum InboundError {
-    #[error("Invalid user agent response")]
-    InvalidResponse,
-    #[error("Expected response payload")]
-    MissingResponsePayload,
-    #[error("Unexpected response payload")]
-    UnexpectedResponsePayload,
-    #[error("Invalid state for auth challenge")]
-    InvalidStateForAuthChallenge,
-    #[error("Invalid state for auth ok")]
-    InvalidStateForAuthOk,
-    #[error("State machine error")]
-    StateTransitionFailed,
-    #[error("Transport send failed")]
-    TransportSendFailed,
-}

-pub struct UserAgentActor<Transport>
-where
-    Transport: Bi<UserAgentResponse, UserAgentRequest>,
-{
+pub struct UserAgentActor<A: TransportActor<UserAgentRequest>> {
    key: SigningKey,
-    bootstrap_token: Option<String>,
-    state: UserAgentStateMachine<DummyContext>,
-    transport: Transport,
+    server_ca_cert: CertificateDer<'static>,
+    sender: ActorRef<A>,
 }
-
-impl<Transport> UserAgentActor<Transport>
-where
-    Transport: Bi<UserAgentResponse, UserAgentRequest>,
-{
-    pub fn new(key: SigningKey, bootstrap_token: Option<String>, transport: Transport) -> Self {
-        Self {
-            key,
-            bootstrap_token,
-            state: UserAgentStateMachine::new(DummyContext),
-            transport,
-        }
-    }
-
-    fn transition(&mut self, event: UserAgentEvents) -> Result<(), InboundError> {
-        self.state.process_event(event).map_err(|e| {
-            error!(?e, "useragent state transition failed");
-            InboundError::StateTransitionFailed
-        })?;
-        Ok(())
-    }
-
-    fn auth_request(payload: ClientAuthPayload) -> UserAgentRequest {
-        UserAgentRequest {
-            payload: Some(UserAgentRequestPayload::AuthMessage(AuthClientMessage {
-                payload: Some(payload),
-            })),
-        }
-    }
-
-    async fn send_auth_challenge_request(&mut self) -> Result<(), InboundError> {
-        let req = AuthChallengeRequest {
-            pubkey: self.key.verifying_key().to_bytes().to_vec(),
-            bootstrap_token: self.bootstrap_token.take(),
-        };
-
-        self.transition(UserAgentEvents::SentAuthChallengeRequest)?;
-
-        self.transport
-            .send(Self::auth_request(ClientAuthPayload::AuthChallengeRequest(
-                req,
-            )))
-            .await
-            .map_err(|_| InboundError::TransportSendFailed)?;
-
-        info!(actor = "useragent", "auth.request.sent");
-        Ok(())
-    }
-
-    async fn handle_auth_challenge(
-        &mut self,
-        challenge: auth::AuthChallenge,
-    ) -> Result<(), InboundError> {
-        self.transition(UserAgentEvents::ReceivedAuthChallenge)?;
-
-        let formatted = format_challenge(&challenge);
-        let signature = self.key.sign(&formatted);
-        let solution = auth::AuthChallengeSolution {
-            signature: signature.to_bytes().to_vec(),
-        };
-
-        self.transport
-            .send(Self::auth_request(
-                ClientAuthPayload::AuthChallengeSolution(solution),
-            ))
-            .await
-            .map_err(|_| InboundError::TransportSendFailed)?;
-
-        info!(actor = "useragent", "auth.solution.sent");
-        Ok(())
-    }
-
-    fn handle_auth_ok(&mut self, _ok: AuthOk) -> Result<(), InboundError> {
-        self.transition(UserAgentEvents::ReceivedAuthOk)?;
-        info!(actor = "useragent", "auth.ok");
-        Ok(())
-    }
-
-    pub async fn process_inbound_transport(
-        &mut self,
-        inbound: UserAgentResponse
-    ) -> Result<(), InboundError> {
-        let payload = inbound   
-            .payload
-            .ok_or(InboundError::MissingResponsePayload)?;
-
-        match payload {
-            UserAgentResponsePayload::AuthMessage(AuthServerMessage {
-                payload: Some(ServerAuthPayload::AuthChallenge(challenge)),
-            }) => self.handle_auth_challenge(challenge).await,
-            UserAgentResponsePayload::AuthMessage(AuthServerMessage {
-                payload: Some(ServerAuthPayload::AuthOk(ok)),
-            }) => self.handle_auth_ok(ok),
-            _ => Err(InboundError::UnexpectedResponsePayload),
-        }
-    }
-}
-
-impl<Transport> Actor for UserAgentActor<Transport>
-where
-    Transport: Bi<UserAgentResponse, UserAgentRequest>,
-{
+impl<A: TransportActor<UserAgentRequest>> Actor for UserAgentActor<A> {
    type Args = Self;

-    type Error = ();
+    type Error = InitError;

-    async fn on_start(
-        mut args: Self::Args,
-        _actor_ref: ActorRef<Self>,
-    ) -> Result<Self, Self::Error> {
-        if let Err(err) = args.send_auth_challenge_request().await {
-            error!(?err, actor = "useragent", "auth.start.failed");
-            return Err(());
-        }
-        Ok(args)
+    async fn on_start(args: Self::Args, actor_ref: ActorRef<Self>) -> Result<Self, Self::Error> {
+        todo!()
    }

-    async fn next(
+    async fn on_link_died(
        &mut self,
-        _actor_ref: kameo::prelude::WeakActorRef<Self>,
-        mailbox_rx: &mut kameo::prelude::MailboxReceiver<Self>,
-    ) -> Option<kameo::mailbox::Signal<Self>> {
-        loop {
-            select! {
-                signal = mailbox_rx.recv() => {
-                    return signal;
-                }
-                inbound = self.transport.recv() => {
-                    match inbound {
-                        Some(inbound) => {
-                            if let Err(err) = self.process_inbound_transport(inbound).await {
-                                error!(?err, actor = "useragent", "transport.inbound.failed");
-                                return Some(kameo::mailbox::Signal::Stop);
-                            }
-                        }
-                        None => {
-                            info!(actor = "useragent", "transport.closed");
-                            return Some(kameo::mailbox::Signal::Stop);
-                        }
-                    }
-                }
-            }
+        _: 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(()))
        }
    }
 }
-
-mod grpc;
-pub use grpc::{connect_grpc, ConnectError};
--- a/server/crates/arbiter-useragent/tests/auth.rs
+++ b/server/crates/arbiter-useragent/tests/auth.rs
@@ -1,151 +0,0 @@
-use arbiter_proto::{
-    format_challenge,
-    proto::{
-        UserAgentRequest, UserAgentResponse,
-        auth::{
-            AuthChallenge, AuthOk, ClientMessage as AuthClientMessage,
-            ServerMessage as AuthServerMessage, client_message::Payload as ClientAuthPayload,
-            server_message::Payload as ServerAuthPayload,
-        },
-        user_agent_request::Payload as UserAgentRequestPayload,
-        user_agent_response::Payload as UserAgentResponsePayload,
-    },
-    transport::Bi,
-};
-use arbiter_useragent::{InboundError, UserAgentActor};
-use ed25519_dalek::SigningKey;
-use kameo::actor::Spawn;
-use tokio::sync::mpsc;
-use tokio::time::{Duration, timeout};
-
-struct TestTransport {
-    inbound_rx: mpsc::Receiver<UserAgentResponse>,
-    outbound_tx: mpsc::Sender<UserAgentRequest>,
-}
-
-impl Bi<UserAgentResponse, UserAgentRequest> for TestTransport {
-    async fn send(&mut self, item: UserAgentRequest) -> Result<(), arbiter_proto::transport::Error> {
-        self.outbound_tx
-            .send(item)
-            .await
-            .map_err(|_| arbiter_proto::transport::Error::ChannelClosed)
-    }
-
-    async fn recv(&mut self) -> Option<UserAgentResponse> {
-        self.inbound_rx.recv().await
-    }
-}
-
-fn make_transport() -> (
-    TestTransport,
-    mpsc::Sender<UserAgentResponse>,
-    mpsc::Receiver<UserAgentRequest>,
-) {
-    let (inbound_tx, inbound_rx) = mpsc::channel(8);
-    let (outbound_tx, outbound_rx) = mpsc::channel(8);
-    (
-        TestTransport {
-            inbound_rx,
-            outbound_tx,
-        },
-        inbound_tx,
-        outbound_rx,
-    )
-}
-
-fn test_key() -> SigningKey {
-    SigningKey::from_bytes(&[7u8; 32])
-}
-
-fn auth_response(payload: ServerAuthPayload) -> UserAgentResponse {
-    UserAgentResponse {
-        payload: Some(UserAgentResponsePayload::AuthMessage(AuthServerMessage {
-            payload: Some(payload),
-        })),
-    }
-}
-
-#[tokio::test]
-async fn sends_auth_request_on_start_with_bootstrap_token() {
-    let key = test_key();
-    let pubkey = key.verifying_key().to_bytes().to_vec();
-    let bootstrap_token = Some("bootstrap-123".to_string());
-    let (transport, inbound_tx, mut outbound_rx) = make_transport();
-
-    let actor = UserAgentActor::spawn(UserAgentActor::new(key, bootstrap_token.clone(), transport));
-
-    let outbound = timeout(Duration::from_secs(1), outbound_rx.recv())
-        .await
-        .expect("timed out waiting for auth request")
-        .expect("channel closed before auth request");
-
-    let UserAgentRequest {
-        payload: Some(UserAgentRequestPayload::AuthMessage(AuthClientMessage {
-            payload: Some(ClientAuthPayload::AuthChallengeRequest(req)),
-        })),
-    } = outbound
-    else {
-        panic!("expected auth challenge request");
-    };
-
-    assert_eq!(req.pubkey, pubkey);
-    assert_eq!(req.bootstrap_token, bootstrap_token);
-
-    drop(inbound_tx);
-    drop(actor);
-}
-
-#[tokio::test]
-async fn challenge_flow_sends_solution_from_transport_inbound() {
-    let key = test_key();
-    let verify_key = key.verifying_key();
-    let (transport, inbound_tx, mut outbound_rx) = make_transport();
-
-    let actor = UserAgentActor::spawn(UserAgentActor::new(key, None, transport));
-
-    let _initial_auth_request = timeout(Duration::from_secs(1), outbound_rx.recv())
-        .await
-        .expect("timed out waiting for initial auth request")
-        .expect("missing initial auth request");
-
-    let challenge = AuthChallenge {
-        pubkey: verify_key.to_bytes().to_vec(),
-        nonce: 42,
-    };
-    inbound_tx
-        .send(auth_response(ServerAuthPayload::AuthChallenge(challenge.clone())))
-        .await
-        .unwrap();
-
-    let outbound = timeout(Duration::from_secs(1), outbound_rx.recv())
-        .await
-        .expect("timed out waiting for challenge solution")
-        .expect("missing challenge solution");
-
-    let UserAgentRequest {
-        payload: Some(UserAgentRequestPayload::AuthMessage(AuthClientMessage {
-            payload: Some(ClientAuthPayload::AuthChallengeSolution(solution)),
-        })),
-    } = outbound
-    else {
-        panic!("expected auth challenge solution");
-    };
-
-    let formatted = format_challenge(&challenge);
-    let sig: ed25519_dalek::Signature = solution
-        .signature
-        .as_slice()
-        .try_into()
-        .expect("signature bytes length");
-    verify_key
-        .verify_strict(&formatted, &sig)
-        .expect("solution signature should verify");
-
-    inbound_tx
-        .send(auth_response(ServerAuthPayload::AuthOk(AuthOk {})))
-        .await
-        .unwrap();
-
-    drop(inbound_tx);
-    drop(actor);
-}
Author	SHA1	Message	Date
hdbg	1b4369b1cb	feat(transport): add domain error type to GrpcTransportActor Some checks failed ci/woodpecker/pr/server-audit Pipeline was successful Details ci/woodpecker/pr/server-lint Pipeline failed Details ci/woodpecker/pr/server-vet Pipeline failed Details ci/woodpecker/pr/server-test Pipeline failed Details	2026-02-26 15:07:11 +01:00
hdbg	7bd37b3c4a	refactor: introduce `TransportActor` abstraction	2026-02-25 21:44:01 +01:00