diff --git a/.gitignore b/.gitignore
index e5f7bd8..7d49af0 100644
--- a/.gitignore
+++ b/.gitignore
@@ -47,3 +47,7 @@ out/
 /.gigaide/gigaide.properties
 local.properties
 /kotlin-js-store/yarn.lock
+
+# JVM crash diagnostics must never be published
+*.hprof
+hs_err_pid*.log
diff --git a/PRIORITY.md b/PRIORITY.md
new file mode 100644
index 0000000..6b18bc8
--- /dev/null
+++ b/PRIORITY.md
@@ -0,0 +1,69 @@
+# Numeric transport priority — 0.7.2-SNAPSHOT
+
+```kotlin
+import net.sergeych.kiloparsec.call
+import net.sergeych.kiloparsec.push
+
+remote.call(command, arguments, priority = 100)
+remote.push(notification, arguments, priority = 0)
+```
+
+Larger signed `Int` values go first. Equal priorities preserve FIFO. Existing
+call/push entry points remain unchanged and use zero unless invoked inside an
+explicit priority call context. The overloads are extension functions, preserving
+existing RemoteInterface implementers. A coroutine context carries the selected
+value through the encrypted RPC wrappers; it becomes serialized outer-block data.
+
+The adapter owns one outgoing writer. Waiting for a response does not occupy the
+writer: calls retain independent request IDs and response waiters. Both transport
+responses and errors copy the request's priority, including encrypted RPC errors
+returned inside an outer response. Pushes have no normal response.
+
+## Wire compatibility
+
+The legacy block tag and body are unchanged. Nonzero priority is appended as four
+big-endian two's-complement bytes. Zero omits the trailer entirely. The new reader
+defaults to zero only when the trailer is absent; a partial trailer is malformed.
+
+This relies on each Transport.Block having its own finite byte-array frame. TCP's
+existing length prefix, WebSocket messages and UDP data blocks provide that
+boundary. The old decoder consumes the legacy body and ignores trailing bytes.
+Do not use this EOF-terminated extension to concatenate unframed Block objects or
+embed them directly in a positional collection; wrap each block in its own bytes.
+The handshake remains version 1 because the original wire format is retained.
+
+Old peers interoperate but do not honor/inherit priority. New peers can decode
+both formats. BiPack @Extendable is deliberately not added: it would insert a new
+field-count prefix before existing fields and break old readers.
+
+## Adapter behavior
+
+TCP, WebSocket client/server, and UDP use a per-connection priority writer.
+The writer reads the legacy header lengths and skips payloads to locate priority;
+it does not deserialize/copy/decrypt the payload or guess from the last four bytes.
+`Transport.Device.send(block)` defaults to the existing output channel for custom
+or legacy devices. Built-in ProxyDevice-based adapters override it to enter their
+writer directly. The old output channel remains available, bridged into that writer;
+new transports should call send so pending senders are visible to priority ordering.
+Custom adapters retain their existing FIFO behavior until they implement priority.
+
+Queue admission is bounded to 128 waiting blocks and 32 MiB (plus the block
+currently being written). Overflow fails explicitly rather than allocating an
+unbounded backlog. Send awaits its adapter write callback, not a remote RPC reply.
+Queued cancellation removes the block; cancellation cannot recall a write already
+started. Closing the writer releases pending senders. There is no rate limiter or
+batch-collection delay. WebSocket writes flush after each block to avoid excessive
+look-ahead in the engine's FIFO. Socket/TLS buffers and network loss can still delay
+control behind bytes already submitted; priority is not a preemption guarantee.
+
+## Verification
+
+Focused JVM tests cover signed numeric ordering, stable ties, queued cancellation,
+bounded admission, close, absent/partial trailers, unchanged legacy bodies, and
+priority inheritance through encrypted concurrent calls and encrypted errors.
+An encrypted simulated 16 KiB/s link saturated in both directions maintains control
+round trips below 400 ms and aggregate utilization above 80% of full-duplex capacity.
+This simulation does not model kernel/TLS buffering or packet loss.
+JVM, Android and JS compilation and socket/resource lifecycle tests pass.
+Caravan also checked call/error/response fixtures with the actual packaged
+0.7.1-SNAPSHOT binary and the new 0.7.2-SNAPSHOT binary in separate JVM processes.
diff --git a/README.md b/README.md
index d580743..73f7c3d 100644
--- a/README.md
+++ b/README.md
@@ -309,4 +309,42 @@ export restrictions are lifted. We do not support such practices here at 8-rays.
 
 [Sergey Chernov]: https://t.me/real_sergeych
 [^1]: On some new Xiaomi phones we found problems with websocket binary frames, probably in ktor; use text frames
-otherwise. 
\ No newline at end of file
+otherwise.
+
+## Numeric transport priority (0.7.2-SNAPSHOT)
+
+`call(command, args, priority = 100)` and `push(command, args, priority = 0)`
+select numeric outgoing priority. Responses inherit request priority, calls remain
+concurrent, and the framed wire extension interoperates with legacy peers.
+See [priority API, compatibility and limits](PRIORITY.md).
+
+
+## Connection health and WebSocket keepalive (0.7.2-SNAPSHOT)
+
+Encrypted application heartbeat is opt-in: `KiloClientConnection.run()` does not
+start it. Pass `ConnectionHealth` explicitly only when periodic probes are needed.
+`ReconnectionOptions` configures connection deadlines, retry delay, and an optional
+finite attempt budget; `reconnectExhausted` signals when that budget ends.
+
+The server WebSocket helper retains its existing defaults and signature. Its new
+configuration overload permits applications to disable the transport pinger:
+
+```kotlin
+setupWebsocketServer(api, path = "/rpc", serverKey = key,
+    configureWebSockets = { pingPeriodMillis = 0 }) { Session() }
+```
+
+Without heartbeat, applications must handle transport closure and bound real
+request response times. A successful socket write does not prove peer reachability.
+Proxy/NAT inactivity timeouts remain independent of these settings.
+
+## Publishing
+
+`./gradlew publishAllPublicationsToMavenRepository` publishes the configured
+version to the private Gitea Maven registry using the existing local credential
+file. Never commit that credential. Apple target publication requires a supported
+macOS builder; check task output for skipped targets on other hosts.
+
+`bin/pubdocs` generates Dokka and uploads it to the existing Kiloparsec documentation
+site. For development-only Maven Local publication without configuring the remote
+repository, pass `-PlocalPublishingOnly=true`.
diff --git a/build.gradle.kts b/build.gradle.kts
index 41398f7..2c6fb79 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -19,7 +19,7 @@ plugins {
 }
 
 group = "net.sergeych"
-version = "0.7.1-SNAPSHOT"
+version = "0.7.2-SNAPSHOT"
 
 repositories {
     google()
@@ -48,7 +48,9 @@ kotlin {
 
     mingwX64()
 
-    androidTarget()
+    androidTarget {
+        publishLibraryVariants("release")
+    }
     @OptIn(ExperimentalWasmDsl::class)
     wasmJs {
         browser()
@@ -84,6 +86,7 @@ kotlin {
                 implementation("io.ktor:ktor-network:$ktor_version")
             }
         }
+        androidMain.dependsOn(ktorSocketMain)
         val commonTest by getting {
             dependencies {
                 implementation(kotlin("test"))
@@ -155,14 +158,16 @@ publishing {
         File("${System.getProperty("user.home")}/.gitea_token").readText()
     }
     repositories {
-        maven {
-            credentials(HttpHeaderCredentials::class) {
-                name = "Authorization"
-                value = mavenToken
-            }
-            url = uri("https://gitea.sergeych.net/api/packages/SergeychWorks/maven")
-            authentication {
-                create("Authorization", HttpHeaderAuthentication::class)
+        if (!providers.gradleProperty("localPublishingOnly").map(String::toBoolean).getOrElse(false)) {
+            maven {
+                credentials(HttpHeaderCredentials::class) {
+                    name = "Authorization"
+                    value = mavenToken
+                }
+                url = uri("https://gitea.sergeych.net/api/packages/SergeychWorks/maven")
+                authentication {
+                    create("Authorization", HttpHeaderAuthentication::class)
+                }
             }
         }
     }
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/CallPriority.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/CallPriority.kt
new file mode 100644
index 0000000..5d63f37
--- /dev/null
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/CallPriority.kt
@@ -0,0 +1,18 @@
+package net.sergeych.kiloparsec
+
+import kotlinx.coroutines.withContext
+import kotlin.coroutines.AbstractCoroutineContextElement
+import kotlin.coroutines.CoroutineContext
+
+/** Local propagation through the encrypted RPC layers; serialized on the outer transport block. */
+internal class CallPriority(val value: Int) : AbstractCoroutineContextElement(Key) {
+    companion object Key : CoroutineContext.Key<CallPriority>
+}
+
+/** Larger values are sent first. Existing two-argument calls remain priority zero. */
+suspend fun <A, R> RemoteInterface.call(cmd: Command<A, R>, args: A, priority: Int): R =
+    withContext(CallPriority(priority)) { call(cmd, args) }
+suspend fun <R> RemoteInterface.call(cmd: Command<Unit, R>, priority: Int): R = call(cmd, Unit, priority)
+suspend fun <A> RemoteInterface.push(cmd: Command<A, Unit>, args: A, priority: Int) =
+    withContext(CallPriority(priority)) { push(cmd, args) }
+suspend fun RemoteInterface.push(cmd: Command<Unit, Unit>, priority: Int) = push(cmd, Unit, priority)
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/ConnectionHealth.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/ConnectionHealth.kt
new file mode 100644
index 0000000..5c8da18
--- /dev/null
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/ConnectionHealth.kt
@@ -0,0 +1,27 @@
+package net.sergeych.kiloparsec
+
+import kotlinx.serialization.builtins.serializer
+
+/** Opt-in end-to-end encrypted heartbeat, independent of the carrier's keepalive. */
+data class ConnectionHealth(
+    val intervalMillis: Long = 15_000,
+    val timeoutMillis: Long = 5_000,
+) {
+    init { require(intervalMillis > 0 && timeoutMillis > 0) }
+}
+
+/** Null attempt limit preserves legacy unlimited reconnect behavior. */
+data class ReconnectionOptions(
+    val health: ConnectionHealth? = null,
+    val maxAttempts: Int? = null,
+    val connectTimeoutMillis: Long = 15_000,
+    val retryDelayMillis: Long = 700,
+) {
+    init {
+        require(maxAttempts == null || maxAttempts > 0)
+        require(connectTimeoutMillis > 0 && retryDelayMillis >= 0)
+    }
+}
+
+// Reserved inner command: encrypted with the current session and handled before application dispatch.
+internal val connectionPing = Command("@kiloparsec/ping", Long.serializer(), Long.serializer())
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClient.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClient.kt
index c6a712d..40c541c 100644
--- a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClient.kt
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClient.kt
@@ -13,6 +13,7 @@ package net.sergeych.kiloparsec
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.isActive
@@ -24,14 +25,13 @@ import net.sergeych.crypto2.VerifyingPublicKey
 import net.sergeych.mp_logger.LogTag
 import net.sergeych.mp_logger.Loggable
 import net.sergeych.mp_logger.debug
-import net.sergeych.mp_logger.exception
 import net.sergeych.mp_tools.globalLaunch
 
 /**
  * The auto-connecting client that reconnects to the kiloparsec server,
  * [KiloServer],
  * and maintain connection state flow. Client factory launches a disconnected
- * set of coroutines to support automatic reconnection, so you _must_ [close]
+ * set of coroutines to support automatic reconnection (unlimited by default; see [options]), so you _must_ [close]
  * it manually when it is unnecessary, otherwise it will continue to reconnect.
  *
  * ## Usage
@@ -59,9 +59,18 @@ import net.sergeych.mp_tools.globalLaunch
 class KiloClient<S>(
     val localInterface: KiloInterface<S>,
     secretKey: SigningKey? = null,
+    val options: ReconnectionOptions = ReconnectionOptions(),
     connectionDataFactory: ConnectionDataFactory<S>,
 ) : RemoteInterface,
     Loggable by LogTag("CLIF") {
+    constructor(localInterface: KiloInterface<S>, secretKey: SigningKey? = null, connectionDataFactory: ConnectionDataFactory<S>)
+        : this(localInterface, secretKey, ReconnectionOptions(), connectionDataFactory)
+
+    private val stopped = MutableStateFlow(false)
+    private val exhausted = MutableStateFlow(false)
+    /** True when the finite attempt budget has ended. Create a new client for another carrier. */
+    val reconnectExhausted = exhausted.asStateFlow()
+
     val _state = MutableStateFlow(false)
 
     /**
@@ -81,49 +90,66 @@ class KiloClient<S>(
 
     private var deferredClient = CompletableDeferred<KiloClientConnection<S>>()
 
+    private val defMutex = Mutex()
+
     private val job =
         globalLaunch {
             debug { "starting connector" }
-            while (isActive) {
-                try {
-                    debug { "getting connection" }
-                    val kc = connectionDataFactory()
-                    debug { "get device and session" }
-                    val client = KiloClientConnection(localInterface, kc, secretKey)
-                    deferredClient.complete(client)
-                    debug { "starting client run"}
-                    val r = runCatching {  client.run {
-                        _state.value = it
-                    } }
-                    debug { "----------- client run finished: $r" }
-                    resetDeferredClient()
-                    debug { "client run finished" }
-                } catch (_: RemoteInterface.ClosedException) {
-                    debug { "remote closed" }
-                    delay(1000)
-                } catch (_: CancellationException) {
-                    debug { "cancelled" }
-                    break
-                } catch (t: Throwable) {
-                    exception { "unexpected exception" to t }
-                    delay(1000)
+            var attempts = 0
+            try {
+                while (isActive && (options.maxAttempts == null || attempts < options.maxAttempts)) {
+                    attempts++
+                    var connectedAt: kotlin.time.TimeMark? = null
+                    try {
+                        val client = kotlinx.coroutines.withTimeout(options.connectTimeoutMillis) {
+                            KiloClientConnection(localInterface, connectionDataFactory(), secretKey)
+                        }
+                        deferredClient.complete(client)
+                        kotlinx.coroutines.coroutineScope {
+                            val setupDeadline = launch {
+                                delay(options.connectTimeoutMillis)
+                                throw RemoteInterface.ClosedException()
+                            }
+                            val onState: (Boolean) -> Unit = {
+                                _state.value = it
+                                if (it) {
+                                    connectedAt = kotlin.time.TimeSource.Monotonic.markNow()
+                                    setupDeadline.cancel()
+                                }
+                            }
+                            if (options.health == null) client.run(onState)
+                            else client.run(options.health, onState)
+                            setupDeadline.cancel()
+                        }
+                    } catch (t: kotlinx.coroutines.TimeoutCancellationException) {
+                        debug { "connection setup timed out" }
+                    } catch (t: CancellationException) { throw t }
+                    catch (t: Throwable) { debug { "connection ended: $t" } }
+                    finally {
+                        _state.value = false
+                        resetDeferredClient()
+                    }
+                    // Short authenticated-but-dead sessions must still exhaust the budget.
+                    val stableMillis = options.health?.let { 3 * (it.intervalMillis + it.timeoutMillis) } ?: 30_000L
+                    if ((connectedAt?.elapsedNow()?.inWholeMilliseconds ?: 0) > stableMillis) attempts = 0
+                    if (options.maxAttempts == null || attempts < options.maxAttempts) delay(options.retryDelayMillis)
                 }
+                exhausted.value = true
+            } finally {
                 _state.value = false
-                resetDeferredClient()
-                // reconnection timeout
-                delay(700)
+                deferredClient.completeExceptionally(RemoteInterface.ClosedException())
             }
         }
 
     fun close() {
+        stopped.value = true
         job.cancel()
         debug { "client is closed" }
     }
 
-    private val defMutex = Mutex()
     private suspend fun resetDeferredClient() {
         defMutex.withLock {
-            if (!deferredClient.isActive) {
+            if (!deferredClient.isActive && !exhausted.value && !stopped.value) {
                 deferredClient = CompletableDeferred()
             }
 
@@ -147,6 +173,9 @@ class KiloClient<S>(
         }
     }
 
+    /** Perform an encrypted liveness round trip on the active connection. */
+    suspend fun ping() = deferredClient.await().ping()
+
     /**
      * Current session token. This is a per-connection unique random value same on the client and server part so
      * it could be used as a nonce to pair MITM and like attacks, be sure that the server is actually
@@ -174,6 +203,8 @@ class KiloClient<S>(
             }
             private var connectionBuilder: (suspend () -> Transport.Device)? = null
 
+            var reconnectionOptions: ReconnectionOptions = ReconnectionOptions()
+
             var secretIdKey: SigningKey? = null
 
             /**
@@ -228,7 +259,7 @@ class KiloClient<S>(
                 for (ep in errorProviders) i.addErrorProvider(ep)
                 interfaceBuilder?.let { i.it() }
                 val connector = connectionBuilder ?: throw IllegalArgumentException("connect handler was not set")
-                return KiloClient(i, secretIdKey) {
+                return KiloClient(i, secretIdKey, reconnectionOptions) {
                     KiloConnectionData(connector(), sessionBuilder())
                 }
             }
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt
index 816e881..fa5827a 100644
--- a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt
@@ -43,7 +43,12 @@ class KiloClientConnection<S>(
      * will stop the transport, or the calling scope will be canceled.
      * Cancelling the scope where server is running is a preferred way to stop the client.
      */
-    suspend fun run(onConnectedStateChanged: ((Boolean) -> Unit)? = null) {
+    suspend fun run(onConnectedStateChanged: ((Boolean) -> Unit)? = null) = runConfigured(null, onConnectedStateChanged)
+
+    suspend fun run(health: ConnectionHealth, onConnectedStateChanged: ((Boolean) -> Unit)? = null) =
+        runConfigured(health, onConnectedStateChanged)
+
+    private suspend fun runConfigured(health: ConnectionHealth?, onConnectedStateChanged: ((Boolean) -> Unit)?) {
         coroutineScope {
             var job: Job? = null
             var connectedScope: KiloScope<S>? = null
@@ -94,21 +99,43 @@ class KiloClientConnection<S>(
                 connectedScope = params.scope
                 clientInterface.onConnectHandlers.invokeAll(connectedScope)
                 onConnectedStateChanged?.invoke(true)
-                job.join()
+                val heartbeat = health?.let { options ->
+                    launch {
+                        while (isActive) {
+                            delay(options.intervalMillis)
+                            if (withTimeoutOrNull(options.timeoutMillis) {
+                                withContext(CallPriority(100)) { ping() }; true
+                            } != true) throw RemoteInterface.ClosedException()
+                        }
+                    }
+                }
+                try { job.join() }
+                finally { heartbeat?.cancelAndJoin() }
 
             } catch (x: CancellationException) {
                 info { "client is cancelled" }
             } catch (x: RemoteInterface.ClosedException) {
                 debug { "connection closed/refused by remote" }
             } finally {
+                if (!deferredParams.isCompleted) deferredParams.completeExceptionally(RemoteInterface.ClosedException())
+                if (!kiloRemoteInterface.isCompleted) kiloRemoteInterface.completeExceptionally(RemoteInterface.ClosedException())
                 connectedScope?.let { clientInterface.onDisconnectHandlers.invokeAll(it) }
                 onConnectedStateChanged?.invoke(false)
-                job?.cancel()
-                device.apply { runCatching { close() } }
+                withContext(NonCancellable) {
+                    job?.cancelAndJoin()
+                    device.apply { runCatching { close() } }
+                }
             }
         }
     }
 
+    /** Encrypted round trip. Older peers' encrypted UnknownCommand also proves liveness. */
+    suspend fun ping() {
+        val nonce = kotlin.random.Random.nextLong()
+        try { check(call(connectionPing, nonce) == nonce) { "Invalid heartbeat response" } }
+        catch (_: RemoteInterface.UnknownCommand) { /* Compatible with peers predating native ping. */ }
+    }
+
     suspend fun token() = deferredParams.await().token
     override suspend fun <A, R> call(cmd: Command<A, R>, args: A): R =
         kiloRemoteInterface.await().call(cmd, args)
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloL0Interface.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloL0Interface.kt
index 2076b4a..c8e747b 100644
--- a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloL0Interface.kt
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloL0Interface.kt
@@ -41,7 +41,8 @@ internal class KiloL0Interface<T>(
             val result: Transport.Block = try {
                 Transport.Block.Response(
                     0u,
-                    clientInterface.execute(params.scope, call.name, call.serializedArgs)
+                    if (call.name == connectionPing.name) connectionPing.exec(call.serializedArgs) { it }
+                    else clientInterface.execute(params.scope, call.name, call.serializedArgs)
                 )
             } catch(t: BreakConnectionException) {
                 throw t
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloRemoteInterface.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloRemoteInterface.kt
index 77a8ef0..04ca22e 100644
--- a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloRemoteInterface.kt
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloRemoteInterface.kt
@@ -11,12 +11,17 @@
 package net.sergeych.kiloparsec
 
 import kotlinx.coroutines.CompletableDeferred
+import kotlinx.serialization.builtins.serializer
 import net.sergeych.mp_logger.LogTag
 import net.sergeych.mp_logger.Loggable
 import net.sergeych.utools.unpack
 
 private var L1IdCounter = 0
 
+// Same encrypted L0Call wire command, sent as an existing id=0 notification.
+// The result serializer is unused: neither endpoint sends/decodes a reply to a push.
+private val l0Notification = Command(L0Call.name, L0Call.argsSerializer, Unit.serializer())
+
 /**
  * Represents a remote interface for interacting with a Kiloparsec service.
  *
@@ -48,7 +53,6 @@ class KiloRemoteInterface<S>(
 
     override suspend fun <A> push(cmd: Command<A, Unit>, args: A) {
         val params = deferredParams.await()
-        params.transport.call(L0Call, params.encrypt(cmd.packCall(args)))
+        params.transport.push(l0Notification, params.encrypt(cmd.packCall(args)))
     }
 }
-
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/LocalInterface.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/LocalInterface.kt
index 91052d1..2f60aac 100644
--- a/src/commonMain/kotlin/net/sergeych/kiloparsec/LocalInterface.kt
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/LocalInterface.kt
@@ -72,7 +72,8 @@ open class LocalInterface<S> : Loggable by LogTag("LocalInterface${idCounter.inc
         }
     }
 
-    suspend fun execute(
+    /** Override to gate dispatch before command lookup and argument deserialization. */
+    open suspend fun execute(
         scope: S,
         name: String,
         packedArgs: UByteArray,
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/Transport.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/Transport.kt
index 646ce40..56bcd16 100644
--- a/src/commonMain/kotlin/net/sergeych/kiloparsec/Transport.kt
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/Transport.kt
@@ -68,10 +68,18 @@ class Transport<S>(
         suspend fun close()
 
         suspend fun flush() {}
+
+        /** Adapter admission; legacy devices retain their existing output-channel behavior. */
+        suspend fun send(block: UByteArray) { output.send(block) }
     }
 
     @Serializable(TransportBlockSerializer::class)
     sealed class Block {
+        // Kept outside the legacy generated body serializer. Only the outer custom
+        // serializer appends it, so old field positions and constructors stay intact.
+        @kotlinx.serialization.Transient
+        var priority: Int = 0
+
         @Serializable
         data class Call(
             @Unsigned
@@ -121,37 +129,27 @@ class Transport<S>(
     private suspend fun sendCallBlock(name: String, packedArgs: UByteArray): UByteArray {
         if (isClosed) throw RemoteInterface.ClosedException()
 
-        val b: Block
+        val b: Block.Call
         val deferred = CompletableDeferred<UByteArray>()
 
         // We need to shield calls and lastID with mutex, but nothing more:
         access.withLock {
             if (isClosed) throw RemoteInterface.ClosedException()
             // the order is important: first id in use MUST BE >= 1, not zero:
-            b = Block.Call(++lastId, name, packedArgs)
+            b = Block.Call(++lastId, name, packedArgs).also { it.priority = currentCoroutineContext()[CallPriority]?.value ?: 0 }
             calls[b.id] = deferred
         }
 
-        // now we have mutex freed so we can call:
-        val r = runCatching {
-            do {
-                val cr = device.output.trySend(pack(b))
-                if( cr.isClosed ) throw ClosedSendChannelException("can't send block: channel is closed")
-                delay(100)
-            } while(!cr.isSuccess)
+        // Suspend on actual channel backpressure; never poll or delay a successful send.
+        try {
+            device.send(pack<Block>(b))
+            return deferred.await()
+        } catch (_: ClosedSendChannelException) {
+            throw RemoteInterface.ClosedException()
+        } finally {
+            // Cancellation must not retain a response waiter for the connection lifetime.
+            withContext(NonCancellable) { access.withLock { calls.remove(b.id) } }
         }
-        if (!r.isSuccess) {
-            r.exceptionOrNull()?.let {
-                exception { "failed to send output block" to it }
-            } ?: run {
-                error { "It should not happen: empty exception on block send failure" }
-                throw RuntimeException("unexpected failure in sending block")
-            }
-            deferred.completeExceptionally(RemoteInterface.ClosedException())
-        }
-
-        // it returns packed result or throws a proper error:
-        return deferred.await()
     }
 
     /**
@@ -164,9 +162,10 @@ class Transport<S>(
         if (isClosed) throw RemoteInterface.ClosedException()
 
         // All push blocks have the same id == 0:
-        val b = Block.Call(0u, name, packedArgs)
-        val r = runCatching { device.output.send(pack(b).also { debug { ">>$\n${it.toDump()}" } }) }
+        val b = Block.Call(0u, name, packedArgs).also { it.priority = currentCoroutineContext()[CallPriority]?.value ?: 0 }
+        val r = runCatching { device.send(pack<Block>(b).also { debug { ">>$\n${it.toDump()}" } }) }
         when(val e = r.exceptionOrNull()) {
+            null -> return
             is RemoteInterface.ClosedException, is CancellationException, is RemoteInterface.RemoteException
                 -> throw e
             else -> throw RemoteInterface.ClosedException()
@@ -224,7 +223,7 @@ class Transport<S>(
                                             Block.Response(
                                                 b.id,
                                                 localInterface.execute(commandContext, b.name, b.packedArgs)
-                                            )
+                                            ).also { it.priority = b.priority }
                                         )
                                 } catch (x: LocalInterface.BreakConnectionException) {
                                     // handler forced close
@@ -233,9 +232,9 @@ class Transport<S>(
                                     if (x.flushSendQueue) device.flush()
                                     device.close()
                                 } catch (x: RemoteInterface.RemoteException) {
-                                    send(Block.Error(b.id, x.code, x.text, x.extra))
+                                    send(Block.Error(b.id, x.code, x.text, x.extra).also { it.priority = b.priority })
                                 } catch (t: Throwable) {
-                                    send(Block.Error(b.id, "UnknownError", t.message))
+                                    send(Block.Error(b.id, "UnknownError", t.message).also { it.priority = b.priority })
                                 }
                             }
                         }
@@ -281,7 +280,7 @@ class Transport<S>(
 
     private suspend fun send(block: Block) {
         try {
-            device.output.send(pack(block))
+            device.send(pack(block))
         } catch (_: ClosedSendChannelException) {
             throw RemoteInterface.ClosedException()
         }
@@ -309,14 +308,26 @@ object TransportBlockSerializer : KSerializer<Transport.Block> {
                 encoder.encodeSerializableValue(serializer<Transport.Block.Response>(), value)
             }
         }
+        // Fixed-size trailer is omitted for zero, preserving exact legacy bytes.
+        if (value.priority != 0) {
+            for (shift in 24 downTo 0 step 8) encoder.encodeByte((value.priority ushr shift).toByte())
+        }
     }
 
 
-    override fun deserialize(decoder: Decoder): Transport.Block =
-        when (val id = decoder.decodeByte().toInt()) {
+    override fun deserialize(decoder: Decoder): Transport.Block {
+        val block = when (val id = decoder.decodeByte().toInt()) {
             0 -> decoder.decodeSerializableValue(serializer<Transport.Block.Call>())
             1 -> decoder.decodeSerializableValue(serializer<Transport.Block.Error>())
             2 -> decoder.decodeSerializableValue(serializer<Transport.Block.Response>())
             else -> throw RemoteInterface.InvalidDataException("wrong block type: $id")
         }
-}
\ No newline at end of file
+        val first = try { decoder.decodeByte().toInt() and 255 }
+        catch (_: net.sergeych.bintools.DataSource.EndOfData) { return block }
+        // A present but truncated trailer is malformed, not a legacy priority-zero block.
+        var priority = first
+        repeat(3) { priority = (priority shl 8) or (decoder.decodeByte().toInt() and 255) }
+        block.priority = priority
+        return block
+    }
+}
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/PriorityBlockWriter.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/PriorityBlockWriter.kt
new file mode 100644
index 0000000..040cd3f
--- /dev/null
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/PriorityBlockWriter.kt
@@ -0,0 +1,113 @@
+package net.sergeych.kiloparsec.adapter
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import net.sergeych.bintools.DataSource
+import net.sergeych.kiloparsec.RemoteInterface
+
+/** Reads only headers and skips length-delimited payloads; never copies or decrypts payload data. */
+internal fun blockPriority(bytes: UByteArray): Int {
+    val source = object : DataSource {
+        var position = 0
+        override fun readByte(): Byte {
+            if (position >= bytes.size) throw DataSource.EndOfData()
+            return bytes[position++].toByte()
+        }
+        fun skipBlob() {
+            val size = readSmartUInt().toLong()
+            require(size <= bytes.size - position)
+            position += size.toInt()
+        }
+    }
+    return try {
+        val tag = source.readUByte().toInt()
+        source.readSmartUInt() // call ID or response ID
+        when (tag) {
+            0 -> { source.skipBlob(); source.skipBlob() }
+            1 -> {
+                source.skipBlob()
+                if (source.readByte() != 0.toByte()) source.skipBlob()
+                if (source.readByte() != 0.toByte()) source.skipBlob()
+            }
+            2 -> source.skipBlob()
+            else -> return 0 // Opaque legacy device users need not send RPC blocks.
+        }
+        if (bytes.size - source.position != 4) return 0
+        var result = 0
+        repeat(4) { result = (result shl 8) or source.readUByte().toInt() }
+        result
+    } catch (_: Exception) { 0 }
+}
+
+/**
+ * One adapter writer. Send completes after its write callback, not after an RPC reply.
+ * Queue ordering is numeric priority then FIFO. Cancellation removes queued work;
+ * bytes already being written cannot be recalled. No rate limit or idle batching delay.
+ */
+internal class PriorityBlockWriter(
+    scope: CoroutineScope,
+    private val maxQueuedItems: Int = 128,
+    private val maxQueuedBytes: Int = 32 * 1024 * 1024,
+    private val write: suspend (UByteArray) -> Unit,
+) {
+    private class Pending(val bytes: UByteArray, val priority: Int, val done: CompletableDeferred<Unit>)
+    private val mutex = Mutex()
+    private val wake = Channel<Unit>(Channel.CONFLATED)
+    private val pending = mutableListOf<Pending>()
+    private var closed = false
+    private var writing: Pending? = null
+    private val job = scope.launch {
+        var current: Pending? = null
+        try {
+            while (isActive) {
+                current = mutex.withLock {
+                    pending.removeAll { !it.done.isActive }
+                    // maxByOrNull retains the first equal value: stable FIFO ties.
+                    pending.maxByOrNull { it.priority }?.also { pending.remove(it) }.also { writing = it }
+                }
+                val entry = current
+                if (entry == null) { wake.receive(); continue }
+                if (entry.done.isActive) write(entry.bytes)
+                entry.done.complete(Unit)
+                mutex.withLock { writing = null }
+                current = null
+            }
+        } catch (failure: Throwable) {
+            current?.done?.completeExceptionally(failure)
+            throw failure
+        } finally {
+            withContext(NonCancellable) {
+                mutex.withLock {
+                    closed = true
+                    pending.forEach { it.done.completeExceptionally(RemoteInterface.ClosedException()) }
+                    pending.clear()
+                }
+            }
+        }
+    }
+
+    suspend fun send(bytes: UByteArray) {
+        currentCoroutineContext().ensureActive()
+        val entry = Pending(bytes, blockPriority(bytes), CompletableDeferred())
+        mutex.withLock {
+            if (closed || !job.isActive) throw RemoteInterface.ClosedException()
+            pending.removeAll { !it.done.isActive }
+            if (pending.size >= maxQueuedItems || bytes.size > maxQueuedBytes - pending.sumOf { it.bytes.size })
+                throw RemoteInterface.Exception("Outgoing block queue is full")
+            pending.add(entry)
+        }
+        wake.trySend(Unit)
+        try { entry.done.await() }
+        finally {
+            entry.done.cancel()
+            withContext(NonCancellable) { mutex.withLock { pending.remove(entry) } }
+        }
+    }
+    suspend fun flush() {
+        val queued = mutex.withLock { (pending + listOfNotNull(writing)).map { it.done } }
+        queued.forEach { it.await() }
+    }
+    fun close() { job.cancel() }
+}
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/ProxyDevice.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/ProxyDevice.kt
index d201088..4e7f766 100644
--- a/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/ProxyDevice.kt
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/ProxyDevice.kt
@@ -25,16 +25,26 @@ open class ProxyDevice(
     private val doFlush: (suspend ()->Unit)? = null,
 ): Transport.Device {
 
+    internal var prioritizedSend: (suspend (UByteArray) -> Unit)? = null
+    internal var priorityWriter: PriorityBlockWriter? = null
+    override suspend fun send(block: UByteArray) {
+        prioritizedSend?.let { it(block); return }
+        val writer = priorityWriter
+        if (writer == null) output.send(block) else writer.send(block)
+    }
+
     override val input: ReceiveChannel<UByteArray> = inputChannel
     override val output: SendChannel<UByteArray> = outputChannel
 
     override suspend fun close() {
+        priorityWriter?.close()
         kotlin.runCatching {  doClose?.invoke() }
         runCatching { inputChannel.close() }
         runCatching { outputChannel.close() }
     }
 
     override suspend fun flush() {
+        priorityWriter?.flush()
         doFlush?.invoke()
         var cnt = 10
         while(!outputChannel.isEmpty) {
diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/websocketClient.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/websocketClient.kt
index 6ca2fcb..b485b20 100644
--- a/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/websocketClient.kt
+++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/websocketClient.kt
@@ -78,10 +78,10 @@ fun websocketTransportDevice(
 
     val input = Channel<UByteArray>()
     val output = Channel<UByteArray>()
+    val writerReady = CompletableDeferred<PriorityBlockWriter>()
     val closeHandle = CompletableDeferred<Boolean>()
-    val readyHandle = CompletableDeferred<Unit>()
 
-    globalLaunch {
+    val connectionJob = globalLaunch {
         val log = LogTag("KC:${counter.incrementAndGet()}")
         try {
             actualClient.webSocket({
@@ -95,17 +95,15 @@ fun websocketTransportDevice(
                 log.info { "connected to the server" }
 //                println("SENDING!!!")
 //                send("Helluva")
-                readyHandle.complete(Unit)
+                val writer = PriorityBlockWriter(this) { block ->
+                    if (useTextFrames) send(Frame.Text(block.asByteArray().encodeToBase64Compact()))
+                    else send(block.asByteArray())
+                    flush() // Do not turn the WebSocket outgoing FIFO into an unbounded look-ahead.
+                }
+                writerReady.complete(writer)
                 launch {
                     try {
-                        for (block in output) {
-                            if (useTextFrames)
-                                send(
-                                    Frame.Text(block.asByteArray().encodeToBase64Compact())
-                                )
-                            else
-                                send(block.toByteArray())
-                        }
+                        for (block in output) writer.send(block)
                         log.info { "input is closed, closing the websocket" }
                         if (closeHandle.isActive) closeHandle.complete(true)
                     } catch (_: ClosedSendChannelException) {
@@ -115,6 +113,7 @@ fun websocketTransportDevice(
                         log.info { "unexpected exception in websock sender: ${t.stackTraceToString()}" }
                         closeHandle.completeExceptionally(t)
                     }
+                    writer.close()
                     if (closeHandle.isActive) closeHandle.complete(false)
                 }
                 launch {
@@ -151,6 +150,7 @@ fun websocketTransportDevice(
             runCatching { output.close() }
             runCatching { input.close() }
         } finally {
+            writerReady.completeExceptionally(RemoteInterface.ClosedException())
             if (ownsClient) actualClient.close()
         }
         log.info { "closing connection" }
@@ -162,7 +162,13 @@ fun websocketTransportDevice(
         runCatching { output.close() }
         runCatching { input.close() }
         closeHandle.complete(true)
+        connectionJob.cancel()
 //            job.cancel()
     })
+    device.prioritizedSend = { block ->
+        val writer = writerReady.await()
+        device.priorityWriter = writer
+        writer.send(block)
+    }
     return device
 }
diff --git a/src/commonTest/kotlin/PriorityRpcTest.kt b/src/commonTest/kotlin/PriorityRpcTest.kt
new file mode 100644
index 0000000..6ff6e84
--- /dev/null
+++ b/src/commonTest/kotlin/PriorityRpcTest.kt
@@ -0,0 +1,49 @@
+@file:OptIn(ExperimentalUnsignedTypes::class, kotlinx.coroutines.ExperimentalCoroutinesApi::class)
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.*
+import net.sergeych.crypto2.initCrypto
+import net.sergeych.kiloparsec.*
+import net.sergeych.utools.unpack
+import kotlin.test.*
+
+class PriorityRpcTest {
+    @Test fun encryptedConcurrentRepliesAndErrorsInheritPriority() = runTest {
+        initCrypto()
+        val echo by command<Int, Int>()
+        val wait by command<Int, Int>()
+        val fail by command<Int, Int>()
+        val release = CompletableDeferred<Unit>()
+        val entered = CompletableDeferred<Unit>()
+        val ready = CompletableDeferred<Unit>()
+        val requests = mutableListOf<Transport.Block>()
+        val replies = mutableListOf<Transport.Block>()
+        val (a, b) = createTestDevice()
+        fun recording(device: Transport.Device, blocks: MutableList<Transport.Block>) = object : Transport.Device by device {
+            override suspend fun send(block: UByteArray) {
+                blocks += unpack<Transport.Block>(block)
+                device.send(block)
+            }
+        }
+        val server = KiloServerConnection(KiloInterface<Unit>().apply {
+            on(echo) { it }
+            on(wait) { entered.complete(Unit); release.await(); it }
+            on(fail) { throw RemoteInterface.RemoteException("test-error") }
+        }, recording(b, replies), Unit)
+        val client = KiloClientConnection(KiloInterface<Unit>(), recording(a, requests), Unit)
+        val jobs = listOf(launch { server.run() }, launch { client.run { if (it) ready.complete(Unit) } })
+        try {
+            ready.await()
+            val slow = async { client.call(wait, 1, priority = 7) }
+            entered.await()
+            assertEquals(2, client.call(echo, 2, priority = 123))
+            assertFalse(slow.isCompleted, "A pending reply must not prevent another concurrent call")
+            assertFailsWith<RemoteInterface.RemoteException> { client.call(fail, 3, priority = 81) }
+            release.complete(Unit); assertEquals(1, slow.await())
+            for (priority in listOf(7, 123, 81)) {
+                val request = requests.filterIsInstance<Transport.Block.Call>().single { it.priority == priority }
+                val response = replies.filterIsInstance<Transport.Block.Response>().single { it.forId == request.id }
+                assertEquals(priority, response.priority)
+            }
+        } finally { release.complete(Unit); jobs.forEach { it.cancelAndJoin() } }
+    }
+}
diff --git a/src/commonTest/kotlin/TransportAdmissionTest.kt b/src/commonTest/kotlin/TransportAdmissionTest.kt
new file mode 100644
index 0000000..ba7b187
--- /dev/null
+++ b/src/commonTest/kotlin/TransportAdmissionTest.kt
@@ -0,0 +1,65 @@
+@file:OptIn(kotlinx.coroutines.ExperimentalCoroutinesApi::class, ExperimentalUnsignedTypes::class)
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.*
+import net.sergeych.crypto2.initCrypto
+import net.sergeych.kiloparsec.*
+import kotlin.test.*
+import kotlin.time.Duration.Companion.seconds
+
+class TransportAdmissionTest {
+    @Test fun readyCallsDoNotAddPollingDelay() = runTest(timeout = 10.seconds) {
+        val ping by command<Int, Int>()
+        val (a, b) = createTestDevice()
+        val local = LocalInterface<Unit>().apply { on(ping) { it } }
+        val client = Transport(a, local, Unit)
+        val server = Transport(b, local, Unit)
+        val jobs = listOf(launch { client.run() }, launch { server.run() })
+        try {
+            val start = currentTime
+            repeat(20) { assertEquals(it, client.call(ping, it)) }
+            assertEquals(0L, currentTime - start, "Ready transports must not impose an artificial per-call delay")
+        } finally { jobs.forEach { it.cancelAndJoin() } }
+    }
+
+    @Test fun successfulPlainPushReturnsNormally() = runTest(timeout = 10.seconds) {
+        val notify by command<Int, Unit>()
+        val received = CompletableDeferred<Int>()
+        val (a, b) = createTestDevice()
+        val client = Transport(a, LocalInterface<Unit>(), Unit)
+        val server = Transport(b, LocalInterface<Unit>().apply { on(notify) { received.complete(it); Unit } }, Unit)
+        val jobs = listOf(launch { client.run() }, launch { server.run() })
+        try {
+            client.push(notify, 7)
+            assertEquals(7, received.await())
+        } finally { jobs.forEach { it.cancelAndJoin() } }
+    }
+
+    @Test fun encryptedPushDoesNotWaitForRemoteHandler() = runTest(timeout = 10.seconds) {
+        initCrypto()
+        val notify by command<Int, Unit>()
+        val ping by command<Int, Int>()
+        val entered = CompletableDeferred<Unit>()
+        val release = CompletableDeferred<Unit>()
+        val ready = CompletableDeferred<Unit>()
+        val (a, b) = createTestDevice()
+        val server = KiloServerConnection(KiloInterface<Unit>().apply {
+            on(notify) { entered.complete(Unit); release.await() }
+            on(ping) { it }
+        }, b, Unit)
+        val client = KiloClientConnection(KiloInterface<Unit>(), a, Unit)
+        val jobs = listOf(launch { server.run() }, launch { client.run { if (it) ready.complete(Unit) } })
+        try {
+            ready.await()
+            val notification = async { client.push(notify, 7) }
+            entered.await()
+            runCurrent()
+            assertTrue(notification.isCompleted, "Push must return before the remote handler finishes")
+            notification.await()
+            assertEquals(9, client.call(ping, 9), "A pending notification must not block ordinary RPC")
+        } finally {
+            release.complete(Unit)
+            jobs.forEach { it.cancelAndJoin() }
+        }
+    }
+}
diff --git a/src/commonTest/kotlin/net/sergeych/kiloparsec/adapter/PriorityLinkTest.kt b/src/commonTest/kotlin/net/sergeych/kiloparsec/adapter/PriorityLinkTest.kt
new file mode 100644
index 0000000..2b00e22
--- /dev/null
+++ b/src/commonTest/kotlin/net/sergeych/kiloparsec/adapter/PriorityLinkTest.kt
@@ -0,0 +1,59 @@
+@file:OptIn(ExperimentalUnsignedTypes::class, kotlinx.coroutines.ExperimentalCoroutinesApi::class)
+package net.sergeych.kiloparsec.adapter
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.test.*
+import net.sergeych.crypto2.initCrypto
+import net.sergeych.kiloparsec.*
+import kotlin.math.ceil
+import kotlin.test.*
+
+class PriorityLinkTest {
+    @Test fun saturatedEncryptedLinkKeepsConcurrentControlResponsiveWithoutRateCap() = runTest {
+        initCrypto()
+        val bulk by command<UByteArray, Unit>()
+        val ping by command<Int, Int>()
+        val received = Channel<UByteArray>(Channel.UNLIMITED)
+        val aInput = Channel<UByteArray>(Channel.UNLIMITED)
+        val bInput = Channel<UByteArray>(Channel.UNLIMITED)
+        var sentBytes = 0L
+        val rate = 16_384
+        fun endpoint(input: Channel<UByteArray>, destination: Channel<UByteArray>): Transport.Device {
+            val writer = PriorityBlockWriter(backgroundScope) { block ->
+                delay(ceil(block.size * 1000.0 / rate).toLong())
+                sentBytes += block.size
+                destination.send(block)
+            }
+            return ProxyDevice(input, Channel(), doClose = { writer.close(); input.close() }).also { it.priorityWriter = writer }
+        }
+        val a = endpoint(aInput, bInput)
+        val b = endpoint(bInput, aInput)
+        val ready = CompletableDeferred<Unit>()
+        val api = KiloInterface<Unit>().apply {
+            on(bulk) { received.send(it) }
+            on(ping) { it }
+        }
+        val server = KiloServerConnection(api, b, Unit)
+        val client = KiloClientConnection(api, a, Unit)
+        val jobs = listOf(backgroundScope.launch { server.run() }, backgroundScope.launch { client.run { if (it) ready.complete(Unit) } })
+        try {
+            ready.await()
+            val start = currentTime
+            val initialBytes = sentBytes
+            val producers = listOf<RemoteInterface>(client, server).flatMap { sender ->
+                List(8) { backgroundScope.launch { while (isActive) sender.push(bulk, UByteArray(2000), priority = 0) } }
+            }
+            received.receive()
+            repeat(10) {
+                val began = currentTime
+                assertEquals(it, client.call(ping, it, priority = 100))
+                assertTrue(currentTime - began < 400, "control latency=${currentTime - began}")
+                delay(80)
+            }
+            val elapsed = currentTime - start
+            assertTrue((sentBytes - initialBytes) * 1000 / elapsed > rate * 1.6, "Writer must use available capacity")
+            producers.forEach { it.cancelAndJoin() }
+        } finally { jobs.forEach { it.cancelAndJoin() }; a.close(); b.close() }
+    }
+}
diff --git a/src/commonTest/kotlin/net/sergeych/kiloparsec/adapter/PriorityTest.kt b/src/commonTest/kotlin/net/sergeych/kiloparsec/adapter/PriorityTest.kt
new file mode 100644
index 0000000..7320a99
--- /dev/null
+++ b/src/commonTest/kotlin/net/sergeych/kiloparsec/adapter/PriorityTest.kt
@@ -0,0 +1,85 @@
+@file:OptIn(ExperimentalUnsignedTypes::class, kotlinx.coroutines.ExperimentalCoroutinesApi::class)
+package net.sergeych.kiloparsec.adapter
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.*
+import kotlinx.serialization.serializer
+import net.sergeych.bintools.DataSource
+import net.sergeych.bipack.BipackDecoder
+import net.sergeych.bipack.BipackEncoder
+import net.sergeych.kiloparsec.*
+import net.sergeych.utools.pack
+import net.sergeych.utools.unpack
+import kotlin.test.*
+
+class PriorityTest {
+    private fun block(priority: Int, id: Int = 1): UByteArray = pack<Transport.Block>(
+        Transport.Block.Call(id.toUInt(), "example", ubyteArrayOf(1u, 2u)).also { it.priority = priority })
+
+    @Test fun legacyBodyAndTrailerCompatibility() {
+        for (original in listOf(
+            Transport.Block.Call(123456u, "test", UByteArray(3500) { it.toUByte() }),
+            Transport.Block.Response(123456u, UByteArray(256)),
+            Transport.Block.Error(123456u, "code", "message", UByteArray(180)),
+            Transport.Block.Error(0u, "code", null, null),
+        )) {
+            val legacy = when (original) {
+                is Transport.Block.Call -> ubyteArrayOf(0u) + BipackEncoder.encode(original).asUByteArray()
+                is Transport.Block.Error -> ubyteArrayOf(1u) + BipackEncoder.encode(original).asUByteArray()
+                is Transport.Block.Response -> ubyteArrayOf(2u) + BipackEncoder.encode(original).asUByteArray()
+            }
+            assertContentEquals(legacy, pack<Transport.Block>(original))
+            assertEquals(0, unpack<Transport.Block>(legacy).priority)
+            for (priority in listOf(Int.MIN_VALUE, -1, 0, 1, 100, Int.MAX_VALUE)) {
+                original.priority = priority
+                val packed = pack<Transport.Block>(original)
+                assertEquals(priority, unpack<Transport.Block>(packed).priority)
+                assertEquals(priority, blockPriority(packed))
+                assertContentEquals(legacy, packed.copyOf(legacy.size))
+                // Legacy decoding invokes the original body serializer and ignores trailing bytes.
+                val oldRead = when (original) {
+                    is Transport.Block.Call -> BipackDecoder.decode<Transport.Block.Call>(packed.drop(1).toUByteArray())
+                    is Transport.Block.Response -> BipackDecoder.decode<Transport.Block.Response>(packed.drop(1).toUByteArray())
+                    is Transport.Block.Error -> BipackDecoder.decode<Transport.Block.Error>(packed.drop(1).toUByteArray())
+                }
+                assertEquals(0, oldRead.priority)
+            }
+            assertFailsWith<DataSource.EndOfData> { unpack<Transport.Block>(legacy + ubyteArrayOf(1u)) }
+        }
+    }
+
+    @Test fun numericOrderingStableTiesAndCancellation() = runTest {
+        val release = CompletableDeferred<Unit>()
+        val started = CompletableDeferred<Unit>()
+        val sent = mutableListOf<Pair<Int, UInt>>()
+        val writer = PriorityBlockWriter(backgroundScope) { bytes ->
+            val b = unpack<Transport.Block>(bytes) as Transport.Block.Call
+            if (b.id == 1u) { started.complete(Unit); release.await() }
+            sent += b.priority to b.id
+        }
+        val first = async { writer.send(block(0, 1)) }
+        started.await()
+        val queued = listOf(0 to 2, 100 to 3, 2 to 4, 100 to 5, -10 to 6).map { (p, id) ->
+            async { writer.send(block(p, id)) }
+        }
+        val cancelled = async { writer.send(block(Int.MAX_VALUE, 7)) }
+        runCurrent(); cancelled.cancelAndJoin()
+        release.complete(Unit)
+        first.await(); queued.awaitAll()
+        assertEquals(listOf(0 to 1u, 100 to 3u, 100 to 5u, 2 to 4u, 0 to 2u, -10 to 6u), sent)
+        writer.close()
+    }
+
+    @Test fun boundedAdmissionAndCloseReleaseWaiters() = runTest {
+        val started = CompletableDeferred<Unit>()
+        val writer = PriorityBlockWriter(backgroundScope, maxQueuedItems = 1) { started.complete(Unit); awaitCancellation() }
+        val sending = async { runCatching { writer.send(block(0)) } }
+        started.await()
+        val queued = async { runCatching { writer.send(block(2)) } }
+        runCurrent()
+        assertFailsWith<RemoteInterface.Exception> { writer.send(block(100)) }
+        writer.close()
+        assertIs<CancellationException>(sending.await().exceptionOrNull())
+        assertIs<RemoteInterface.ClosedException>(queued.await().exceptionOrNull())
+    }
+}
diff --git a/src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/WebsocketServer.kt b/src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/WebsocketServer.kt
index 5a6961f..aa98dd0 100644
--- a/src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/WebsocketServer.kt
+++ b/src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/WebsocketServer.kt
@@ -54,12 +54,22 @@ fun <S> Application.setupWebsocketServer(
     path: String = "/kp",
     serverKey: SigningKey? = null,
     createSession: () -> S,
+) = setupWebsocketServer(localInterface, path, serverKey, {}, createSession)
+
+/** Configure transport keepalive without changing the legacy defaults for other applications. */
+fun <S> Application.setupWebsocketServer(
+    localInterface: KiloInterface<S>,
+    path: String = "/kp",
+    serverKey: SigningKey? = null,
+    configureWebSockets: WebSockets.WebSocketOptions.() -> Unit,
+    createSession: () -> S,
 ) {
     install(WebSockets) {
         pingPeriod = 60.seconds //Duration.ofSeconds(15)
         timeout = 45.seconds
         maxFrameSize = Long.MAX_VALUE
         masking = false
+        configureWebSockets()
     }
     val counter = AtomicCounter()
     routing {
@@ -67,28 +77,20 @@ fun <S> Application.setupWebsocketServer(
             val log = LogTag("KWS:${counter.incrementAndGet()}")
             log.debug { "opening the connection" }
             val input = Channel<UByteArray>(256)
-            val output = Channel<UByteArray>(256)
+            val output = Channel<UByteArray>()
             var useBinary: Boolean? = null
+            val writer = PriorityBlockWriter(this) { block ->
+                if (useBinary == false) send(block.asByteArray().encodeToBase64Compact())
+                else send(block.asByteArray())
+                flush()
+            }
             launch {
-                log.debug { "starting output pump" }
-                while (isActive) {
-                    try {
-                        val block = output.receive()
-                        if (useBinary == false)
-                            send(block.asByteArray().encodeToBase64Compact())
-                        else
-                            send(block.toByteArray())
-
-                    } catch (_: ClosedReceiveChannelException) {
-                        log.debug { "closing output pump as output channel is closed" }
-                        break
-                    }
-                }
-                log.debug { "closing output pump" }
+                try { for (block in output) writer.send(block) }
+                finally { writer.close() }
             }
             val server = KiloServerConnection(
                 localInterface,
-                ProxyDevice(input, output),
+                ProxyDevice(input, output).also { it.priorityWriter = writer },
                 createSession(),
                 serverKey
             )
diff --git a/src/jvmTest/kotlin/net/sergeych/kiloparsec/ConnectionHealthTest.kt b/src/jvmTest/kotlin/net/sergeych/kiloparsec/ConnectionHealthTest.kt
new file mode 100644
index 0000000..ae5c5d1
--- /dev/null
+++ b/src/jvmTest/kotlin/net/sergeych/kiloparsec/ConnectionHealthTest.kt
@@ -0,0 +1,95 @@
+package net.sergeych.kiloparsec
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.first
+import net.sergeych.crypto2.initCrypto
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.*
+
+class ConnectionHealthTest {
+    @Test(timeout = 15_000)
+    fun silentHandshakesExhaustBudgetAndReleaseWaitingCalls() = runBlocking {
+        initCrypto()
+        val attempts = AtomicInteger()
+        val closes = AtomicInteger()
+        val client = KiloClient(KiloInterface<Unit>(), options = ReconnectionOptions(
+            health = ConnectionHealth(50, 50), maxAttempts = 3,
+            connectTimeoutMillis = 100, retryDelayMillis = 10,
+        )) {
+            attempts.incrementAndGet()
+            KiloConnectionData(object : Transport.Device {
+                val closed = java.util.concurrent.atomic.AtomicBoolean()
+                override val input = Channel<UByteArray>(16)
+                override val output = Channel<UByteArray>(16)
+                override suspend fun close() { if (closed.compareAndSet(false, true)) closes.incrementAndGet(); input.cancel(); output.cancel() }
+            }, Unit)
+        }
+        try {
+            val waiter = async { runCatching { client.token() } }
+            withTimeout(10_000) { client.reconnectExhausted.first { it } }
+            assertEquals(3, attempts.get())
+            assertEquals(3, closes.get())
+            assertFalse(client.connectedStateFlow.value)
+            assertTrue(withTimeout(1_000) { waiter.await() }.isFailure)
+            val missing by command<Unit, Unit>()
+            repeat(2) {
+                withTimeout(1_000) {
+                    assertFailsWith<RemoteInterface.ClosedException> { client.call(missing, Unit) }
+                }
+            }
+            delay(200)
+            assertEquals(3, attempts.get(), "An exhausted client must not reconnect forever")
+        } finally { client.close() }
+    }
+    @Test(timeout = 15_000)
+    fun establishedSilentLossTriggersHeartbeatAndBoundedRecovery() = runBlocking {
+        initCrypto()
+        val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+        val blocked = java.util.concurrent.atomic.AtomicBoolean()
+        val attempts = AtomicInteger()
+        val api = object : KiloInterface<Unit>() {
+            override suspend fun execute(scope: KiloScope<Unit>, name: String, packedArgs: UByteArray): UByteArray {
+                error("Heartbeat must bypass application command dispatch")
+            }
+        }
+        val client = KiloClient(KiloInterface<Unit>(), options = ReconnectionOptions(
+            health = ConnectionHealth(100, 100), maxAttempts = 3,
+            connectTimeoutMillis = 300, retryDelayMillis = 10,
+        )) {
+            attempts.incrementAndGet()
+            val requests = Channel<UByteArray>(16)
+            val forwarded = Channel<UByteArray>(16)
+            val responses = Channel<UByteArray>(16)
+            val pump = scope.launch { for (block in requests) if (!blocked.get()) forwarded.send(block) }
+            suspend fun retire() { pump.cancel(); requests.cancel(); forwarded.cancel(); responses.cancel() }
+            val serverDevice = object : Transport.Device {
+                override val input = forwarded
+                override val output = responses
+                override suspend fun close() { retire() }
+            }
+            scope.launch { runCatching { KiloServerConnection(api, serverDevice, Unit).run() } }
+            KiloConnectionData(object : Transport.Device {
+                override val input = responses
+                override val output = requests
+                override suspend fun close() { retire() }
+            }, Unit)
+        }
+        try {
+            withTimeout(5_000) { client.connectedStateFlow.first { it } }
+            // Several successful native encrypted heartbeats; no application RPC is exported.
+            client.ping()
+            delay(800)
+            assertTrue(client.connectedStateFlow.value)
+            assertEquals(1, attempts.get())
+            blocked.set(true)
+            withTimeout(5_000) { client.reconnectExhausted.first { it } }
+            // The healthy session earns a new budget: three failed recovery connections.
+            assertEquals(4, attempts.get())
+            assertFalse(client.connectedStateFlow.value)
+            delay(400)
+            assertEquals(4, attempts.get())
+        } finally { client.close(); scope.coroutineContext[Job]!!.cancelAndJoin() }
+    }
+
+}
diff --git a/src/jvmTest/kotlin/net/sergeych/kiloparsec/adapter/SocketLifecycleTest.kt b/src/jvmTest/kotlin/net/sergeych/kiloparsec/adapter/SocketLifecycleTest.kt
new file mode 100644
index 0000000..42a2d66
--- /dev/null
+++ b/src/jvmTest/kotlin/net/sergeych/kiloparsec/adapter/SocketLifecycleTest.kt
@@ -0,0 +1,95 @@
+package net.sergeych.kiloparsec.adapter
+
+import java.net.DatagramPacket
+import java.net.DatagramSocket
+import java.net.InetAddress
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
+import kotlin.test.*
+
+class SocketLifecycleTest {
+    @Test fun tcpRoundTripAndRebind() = runBlocking {
+        withTimeout(10_000) {
+            val server = TcpServer(0, "127.0.0.1")
+            val port = server.boundPort()
+            val accepted = Channel<InetTransportDevice>(1)
+            val collector = launch { runCatching { server.transportFlow.collect { accepted.send(it) } } }
+            val client = connectTcpDevice(NetworkAddress("127.0.0.1", port))
+            val peer = accepted.receive()
+            try {
+                val payload = ubyteArrayOf(2u, 3u, 4u)
+                client.output.send(payload)
+                assertContentEquals(payload, peer.input.receive())
+                peer.output.send(payload)
+                assertContentEquals(payload, client.input.receive())
+            } finally { client.close(); server.close(); collector.cancelAndJoin() }
+            val replacement = TcpServer(port, "127.0.0.1")
+            try { assertEquals(port, replacement.boundPort()) } finally { replacement.close() }
+        }
+    }
+
+    @Test fun udpMalformedPacketDoesNotStopListenerAndRebind() = runBlocking {
+        withTimeout(10_000) {
+            val server = UdpServer(0, "127.0.0.1")
+            val port = server.boundPort()
+            val accepted = Channel<InetTransportDevice>(1)
+            val collector = launch { server.transportFlow.collect { accepted.send(it) } }
+            DatagramSocket().use { socket ->
+                socket.send(DatagramPacket(byteArrayOf(1), 1, InetAddress.getLoopbackAddress(), port))
+            }
+            val client = connectUdpDevice(NetworkAddress("127.0.0.1", port))
+            try {
+                val payload = ubyteArrayOf(2u, 3u, 4u)
+                client.output.send(payload)
+                val peer = accepted.receive()
+                assertContentEquals(payload, peer.input.receive())
+                peer.output.send(payload)
+                assertContentEquals(payload, client.input.receive())
+            } finally { client.close(); server.close(); collector.cancelAndJoin() }
+            val replacement = UdpServer(port, "127.0.0.1")
+            try { assertEquals(port, replacement.boundPort()) } finally { replacement.close() }
+        }
+    }
+
+    @Test fun deviceCloseWorksInCancelledContext() = runBlocking {
+        withTimeout(10_000) {
+            val tcp = TcpServer(0, "127.0.0.1")
+            val udp = UdpServer(0, "127.0.0.1")
+            val peers = Channel<InetTransportDevice>(2)
+            val tcpCollector = launch { runCatching { tcp.transportFlow.collect { peers.send(it) } } }
+            val udpCollector = launch { udp.transportFlow.collect { peers.send(it) } }
+            val tcpClient = connectTcpDevice(NetworkAddress("127.0.0.1", tcp.boundPort()))
+            val udpClient = connectUdpDevice(NetworkAddress("127.0.0.1", udp.boundPort()))
+            udpClient.output.send(ubyteArrayOf(2u, 3u))
+            val firstPeer = peers.receive()
+            val secondPeer = peers.receive()
+            try {
+                val closer = launch(start = CoroutineStart.UNDISPATCHED) {
+                    currentCoroutineContext().cancel()
+                    tcpClient.close()
+                    udpClient.close()
+                }
+                closer.join()
+                // EOF reaches both peers even when caller cancellation precedes close.
+                for (peer in listOf(firstPeer, secondPeer)) {
+                    while (peer.input.receiveCatching().isSuccess) { }
+                }
+            } finally {
+                tcpClient.close(); udpClient.close(); tcp.close(); udp.close()
+                tcpCollector.cancelAndJoin(); udpCollector.cancelAndJoin()
+            }
+        }
+    }
+
+    @Test fun bindFailuresAreReported() = runBlocking {
+        val tcp = TcpServer(0, "127.0.0.1")
+        val udp = UdpServer(0, "127.0.0.1")
+        try {
+            val duplicateTcp = TcpServer(tcp.boundPort(), "127.0.0.1")
+            try { assertFails { duplicateTcp.boundPort() } } finally { duplicateTcp.close() }
+            val duplicateUdp = UdpServer(udp.boundPort(), "127.0.0.1")
+            try { assertFails { duplicateUdp.boundPort() } } finally { duplicateUdp.close() }
+        } finally { tcp.close(); udp.close() }
+        Unit
+    }
+}
diff --git a/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/TcpProvider.kt b/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/TcpProvider.kt
index bed37be..acb278a 100644
--- a/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/TcpProvider.kt
+++ b/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/TcpProvider.kt
@@ -40,19 +40,12 @@ internal val PING_INACTIVITY_TIME = 30.seconds
  * Listen for incoming TCP/IP connections on all local interfaces and the specified [port]
  * anc create flow of [InetTransportDevice] suitable for [KiloClient].
  */
-fun acceptTcpDevice(port: Int, localInterface: String = "0.0.0.0"): Flow<InetTransportDevice> {
-    val selectorManager = SelectorManager(Dispatchers.IO)
-    return flow {
-        val serverSocket = aSocket(selectorManager).tcp().bind(localInterface, port)
-        while (true) {
-            serverSocket.accept().let { sock ->
-                emit(inetTransportDevice(sock, "srv"))
-            }
-        }
-    }
+fun acceptTcpDevice(port: Int, localInterface: String = "0.0.0.0"): Flow<InetTransportDevice> = flow {
+    val server = TcpServer(port, localInterface)
+    try { server.transportFlow.collect { emit(it) } }
+    finally { withContext(NonCancellable) { server.close() } }
 }
 
-
 /**
  * Connect to the TCP/IP server (see [KiloServer]) at the specified address and provide th compatible
  * [InetTransportDevice] to use with [KiloClient]. [hostPortAddress] is a string in the form of `host:port`.
@@ -64,10 +57,49 @@ suspend fun connectTcpDevice(hostPortAddress: String): InetTransportDevice =
  * Connect to the TCP/IP server (see [KiloServer]) at the specified address and provide th compatible
  * [InetTransportDevice] to use with [KiloClient].
  */
-suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
+suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice = withContext(Dispatchers.IO) {
     val selectorManager = SelectorManager(Dispatchers.IO)
-    val socket = aSocket(selectorManager).tcp().connect(address.host, address.port)
-    return inetTransportDevice(socket)
+    try {
+        val socket = aSocket(selectorManager).tcp().connect(address.host, address.port)
+        return@withContext inetTransportDevice(socket, onClose = { selectorManager.close() })
+    } catch (failure: Throwable) {
+        selectorManager.close()
+        throw failure
+    }
+}
+
+/** Establish the existing TCP carrier through a caller-selected HTTP CONNECT relay. */
+suspend fun connectTcpDeviceThroughHttpProxy(
+    address: NetworkAddress,
+    proxy: NetworkAddress,
+    authorization: String,
+): InetTransportDevice = withContext(Dispatchers.IO) {
+    require(!authorization.contains('\r') && !authorization.contains('\n'))
+    require(!address.host.any { it.isWhitespace() || it == '\r' || it == '\n' })
+    val selector = SelectorManager(Dispatchers.IO)
+    var socket: Socket? = null
+    try {
+        socket = aSocket(selector).tcp().connect(proxy.host, proxy.port)
+        val input = socket.openReadChannel()
+        val output = socket.openWriteChannel()
+        val target = "${address.host}:${address.port}"
+        output.writeStringUtf8("CONNECT $target HTTP/1.1\r\nHost: $target\r\nProxy-Authorization: $authorization\r\n\r\n")
+        output.flush()
+        val status = input.readUTF8Line(4096) ?: error("Proxy closed before response")
+        require(status.startsWith("HTTP/1.1 200 ") || status.startsWith("HTTP/1.0 200 ")) { "CONNECT refused" }
+        var headerBytes = status.length
+        while (true) {
+            val line = input.readUTF8Line(4096) ?: error("Incomplete proxy response")
+            headerBytes += line.length
+            require(headerBytes <= 16384) { "Proxy headers too large" }
+            if (line.isEmpty()) break
+        }
+        return@withContext inetTransportDevice(socket, onClose = { selector.close() }, providedInput = input, providedOutput = output)
+    } catch (failure: Throwable) {
+        socket?.close()
+        selector.close()
+        throw failure
+    }
 }
 
 /**
@@ -78,147 +110,84 @@ fun String.toNetworkAddress(): NetworkAddress {
     return NetworkAddress(host, port.toInt())
 }
 
-private fun inetTransportDevice(
+internal fun inetTransportDevice(
     sock: Socket,
     suffix: String = "cli",
+    providedInput: ByteReadChannel? = null,
+    providedOutput: ByteWriteChannel? = null,
+    onClose: suspend () -> Unit = {},
 ): InetTransportDevice {
     val networkAddress = (sock.remoteAddress as InetSocketAddress).let { NetworkAddress(it.hostname, it.port) }
     val inputBlocks = Channel<UByteArray>(4096)
-    val outputBlocks = Channel<UByteArray>(4096)
-
-    val log = LogTag("TCPT${logCounter.incrementAndGet()}:$suffix:$networkAddress")
-    val job = AtomicAsyncValue<Job?>(null)
-
-    val sockOutput = sock.openWriteChannel()
-    val sockInput = runCatching { sock.openReadChannel() }.getOrElse {
-        log.warning { "failed to open read channel $it" }
-        throw IllegalStateException("failed to open read channel")
-    }
-
-    suspend fun stop() {
-        job.mutate {
-            if (it != null) {
-                log.debug { "stopping" }
-                runCatching { inputBlocks.close() }
-                runCatching { outputBlocks.close() }
-                // The problem: on mac platofrms closing the socket does not close its input
-                // and output channels!
-                runCatching { sockInput.cancel() }
-                runCatching { sockOutput.flushAndClose() }
-                if (!sock.isClosed)
-                    runCatching {
-                        log.debug { "closing socket by stop" }
-                        sock.close()
-                    }
-                else
-                    log.debug { "socket is already closed when stop is called" }
-                it.cancel()
-                log.debug { "implementation job cancel called" }
-            } else
-                log.debug { "already stopped" }
-            null
+    // A push must backpressure on the writer, not build thousands of blocks ahead
+    // of latency-sensitive calls sharing this carrier.
+    val outputBlocks = Channel<UByteArray>(Channel.RENDEZVOUS)
+    val sockOutput = providedOutput ?: sock.openWriteChannel()
+    val sockInput = providedInput ?: sock.openReadChannel()
+    val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+    val closeAccess = Mutex()
+    var closed = false
+    suspend fun stop() = withContext(NonCancellable) {
+        closeAccess.withLock {
+            if (!closed) {
+                closed = true
+                inputBlocks.close()
+                outputBlocks.close()
+                sockInput.cancel()
+                sock.close()
+                scope.cancel()
+                onClose()
+            }
         }
     }
-
     var lastActiveAt = Clock.System.now()
-    globalLaunch {
-        job.reset(globalLaunch {
-            launch {
-
-                log.debug { "opening read channel" }
-
-                while (isActive && sock.isActive) {
-                    try {
-                        val size = AsyncVarint.decodeUnsigned(sockInput).toInt()
-                        if (size > MAX_TCP_BLOCK_SIZE) // 16M is a max command block
-                            throw ProtocolException("Illegal block size: $size should be < $MAX_TCP_BLOCK_SIZE")
-                        val data = ByteArray(size)
-                        if (size == 0) {
-                            log.debug { "ping received" }
-                            lastActiveAt = Clock.System.now()
-                        } else {
-                            sockInput.readFully(data, 0, size)
-                            inputBlocks.send(data.toUByteArray())
-                        }
-                    } catch (_: ClosedReceiveChannelException) {
-                        log.error { "closed receive channel " }
-                        stop()
-                        break
-                    } catch (_: CancellationException) {
-                        log.error { "cancellation exception " }
-                        break
-                    } catch (e: Exception) {
-                        log.exception { "unexpected exception in TCP socket read" to e }
-                        stop()
-                        break
-                    }
+    val writeAccess = Mutex()
+    scope.launch {
+        try {
+            while (isActive) {
+                val size = AsyncVarint.decodeUnsigned(sockInput)
+                if (size > MAX_TCP_BLOCK_SIZE.toULong())
+                    throw ProtocolException("Illegal block size: $size")
+                lastActiveAt = Clock.System.now()
+                if (size != 0uL) {
+                    val data = ByteArray(size.toInt())
+                    sockInput.readFully(data, 0, data.size)
+                    inputBlocks.send(data.toUByteArray())
                 }
             }
-        })
-        launch {
-            val outAccess = Mutex()
-            var lastSentAt = Clock.System.now()
-            launch {
-                while (isActive && sock.isActive) {
-                    delay(500)
-                    val activityTime = if (lastSentAt > lastActiveAt) lastSentAt else lastActiveAt
-                    if (Clock.System.now() - activityTime > PING_INACTIVITY_TIME) {
-                        log.debug { "pinging for inactivity" }
-                        val repeat = outAccess.withLock {
-                            try {
-                                sockOutput.writeByte(0)
-                                sockOutput.flush()
-                                lastSentAt = Clock.System.now()
-                                true
-                            } catch (e: ClosedReceiveChannelException) {
-                                e.printStackTrace()
-                                stop()
-                                false
-                            } catch (_: CancellationException) {
-                                false
-                            } catch (e: Throwable) {
-                                e.printStackTrace()
-                                stop()
-                                false
-                            }
-                        }
-                        if (!repeat) break
-                    }
-                }
-            }
-
-            while (isActive && sock.isActive) {
-                try {
-                    val block = outputBlocks.receive()
-                    outAccess.withLock {
-                        AsyncVarint.encodeUnsigned(block.size.toULong(), sockOutput)
-                        sockOutput.writeFully(block.toByteArray(), 0, block.size)
-                        sockOutput.flush()
-                        lastSentAt = Clock.System.now()
-                    }
-                } catch (_: CancellationException) {
-                    log.debug { "cancellation exception on output" }
-                    stop()
-                    break
-                } catch (_: LocalInterface.BreakConnectionException) {
-                    log.debug { "requested connection break" }
-                    stop()
-                    break
-                } catch (_: ClosedReceiveChannelException) {
-                    log.debug { "receive block channel closed, closing the socket" }
-                    stop()
-                    break
-                } catch (e: Exception) {
-                    log.exception { "unexpected exception. closing." to e }
-                    stop()
-                    break
-                }
-            }
-        }
+        } catch (_: Exception) {
+            // EOF, cancellation, or invalid framing ends only this connection.
+        } finally { stop() }
     }
-    val device = InetTransportDevice(inputBlocks, outputBlocks, networkAddress, {
-        stop()
-    })
-    log.debug { "Transport ready" }
-    return device
+    val priorityWriter = PriorityBlockWriter(scope) { block ->
+        try {
+            writeAccess.withLock {
+                AsyncVarint.encodeUnsigned(block.size.toULong(), sockOutput)
+                sockOutput.writeFully(block.asByteArray(), 0, block.size)
+                sockOutput.flush()
+                lastActiveAt = Clock.System.now()
+            }
+        } catch (failure: Throwable) { stop(); throw failure }
+    }
+    // Preserve the public legacy output channel; current transports use send directly.
+    scope.launch {
+        try { for (block in outputBlocks) priorityWriter.send(block) }
+        finally { priorityWriter.close(); stop() }
+    }
+    scope.launch {
+        try {
+            while (isActive) {
+                delay(500)
+                writeAccess.withLock {
+                    if (Clock.System.now() - lastActiveAt > PING_INACTIVITY_TIME) {
+                        sockOutput.writeByte(0)
+                        sockOutput.flush()
+                        lastActiveAt = Clock.System.now()
+                    }
+                }
+            }
+        } catch (_: Exception) {
+        } finally { stop() }
+    }
+    return InetTransportDevice(inputBlocks, outputBlocks, networkAddress, { stop() }).also { it.priorityWriter = priorityWriter }
 }
diff --git a/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/TcpServer.kt b/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/TcpServer.kt
new file mode 100644
index 0000000..4dab9ae
--- /dev/null
+++ b/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/TcpServer.kt
@@ -0,0 +1,54 @@
+package net.sergeych.kiloparsec.adapter
+
+import io.ktor.network.selector.SelectorManager
+import io.ktor.network.sockets.*
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+
+/** Managed listener using the same framing as [connectTcpDevice]. Collect once. */
+class TcpServer(val port: Int, localInterface: String = "0.0.0.0") {
+    private val selector = SelectorManager(Dispatchers.IO)
+    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+    private val socket = scope.async { aSocket(selector).tcp().bind(localInterface, port) }
+    private val access = Mutex()
+    private val connections = mutableSetOf<InetTransportDevice>()
+    private var closed = false
+
+    /** Wait for binding, propagating bind errors; supports port zero. */
+    suspend fun boundPort(): Int = (socket.await().localAddress as InetSocketAddress).port
+
+    val transportFlow = flow {
+        try {
+            val listener = socket.await()
+            while (currentCoroutineContext().isActive) {
+                val accepted = listener.accept()
+                val device = access.withLock {
+                    if (closed) { accepted.close(); null }
+                    else {
+                        lateinit var device: InetTransportDevice
+                        device = inetTransportDevice(accepted, "srv") {
+                            access.withLock { connections.remove(device) }
+                        }
+                        connections.add(device)
+                        device
+                    }
+                }
+                if (device != null) emit(device)
+            }
+        } finally { withContext(NonCancellable) { close() } }
+    }
+
+    suspend fun close() = withContext(NonCancellable) {
+        val pending = access.withLock {
+            if (closed) return@withContext
+            closed = true
+            connections.toList().also { connections.clear() }
+        }
+        runCatching { socket.await().close() }
+        pending.forEach { runCatching { it.close() } }
+        scope.cancel()
+        selector.close()
+    }
+}
diff --git a/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpProvider.kt b/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpProvider.kt
index 742e049..6b71e4b 100644
--- a/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpProvider.kt
+++ b/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpProvider.kt
@@ -122,17 +122,20 @@ suspend fun connectUdpDevice(
     maxInactivityTimeout: Duration = 2.minutes,
 ): InetTransportDevice {
     val selectorManager = SelectorManager(Dispatchers.IO)
-    val remoteAddress = InetSocketAddress(addr.host, addr.port)
+    val remoteAddress = withContext(Dispatchers.IO) { InetSocketAddress(addr.host, addr.port) }
 
     val done = CompletableDeferred<Unit>()
 
-    val socket = aSocket(selectorManager).udp().connect(remoteAddress)
+    val socket = try { aSocket(selectorManager).udp().connect(remoteAddress) }
+    catch (failure: Throwable) { selectorManager.close(); throw failure }
     val transport = UdpSocketTransport(object : UdpConnector {
         override suspend fun sendBlock(block: UdpBlock, toAddress: SocketAddress) {
             socket.send(block.toDatagram(remoteAddress))
         }
 
         override suspend fun disconnectClient(address: SocketAddress) {
+            socket.close()
+            selectorManager.close()
             done.complete(Unit)
         }
 
@@ -151,8 +154,9 @@ suspend fun connectUdpDevice(
                 }
             }
         }
-        done.await()
+        try { done.await() }
+        finally { withContext(NonCancellable) { transport.close() }; coroutineContext.cancelChildren() }
     }
 
     return transport.transportDevice
-}
\ No newline at end of file
+}
diff --git a/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpServer.kt b/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpServer.kt
index c3abd88..29b2033 100644
--- a/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpServer.kt
+++ b/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpServer.kt
@@ -12,7 +12,7 @@ package net.sergeych.kiloparsec.adapter
 
 import io.ktor.network.selector.*
 import io.ktor.network.sockets.*
-import io.ktor.utils.io.*
+import kotlinx.coroutines.*
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.IO
 import kotlinx.coroutines.channels.ClosedReceiveChannelException
@@ -60,12 +60,16 @@ class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivity
 
     private val sessions = mutableMapOf<SocketAddress, UdpSocketTransport>()
     private val access = Mutex()
+    private var closed = false
 
     private val selectorManager = SelectorManager(Dispatchers.IO)
-    private val serverSocket = globalDefer {
+    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+    private val serverSocket = scope.async {
         aSocket(selectorManager).udp().bind(InetSocketAddress(localInterface, port))
     }
 
+    suspend fun boundPort(): Int = (serverSocket.await().localAddress as InetSocketAddress).port
+
     override suspend fun disconnectClient(address: SocketAddress) {
         access.withLock { sessions.remove(address) }
     }
@@ -76,13 +80,16 @@ class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivity
      */
     val transportFlow by lazy {
         flow {
+            try {
             while (true) {
                 try {
                     val datagram = serverSocket.await().receive()
-                    val block = UdpBlock.decode(datagram)
+                    val block = try { UdpBlock.decode(datagram) }
+                    catch (_: Exception) { continue } // Malformed traffic must not stop the listener.
                     val remoteAddress = datagram.address
 
                     access.withLock {
+                        if (closed) return@withLock
                         if (block == UdpBlock.Cancel) {
                             // if the cancel comes to already closed transport, do nothing
                             sessions.remove(remoteAddress)?.processIncoming(block)
@@ -106,6 +113,7 @@ class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivity
                     break
                 }
             }
+            } finally { withContext(NonCancellable) { close() } }
         }
     }
 
@@ -124,17 +132,14 @@ class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivity
      *
      * Call suspends until socket and all sessions are closed. Later calls do nothing.
      */
-    suspend fun close() {
-        access.withLock {
-            if (!isClosed()) {
-                runCatching { serverSocket.await().close() }
-            }
-        }
-        while (sessions.isNotEmpty()) {
-            runCatching {
-                access.withLock { sessions.values.firstOrNull() }
-                    ?.close()
-            }
+    suspend fun close() = withContext(NonCancellable) {
+        runCatching { serverSocket.await().close() }
+        val pending = access.withLock {
+            closed = true
+            sessions.values.toList().also { sessions.clear() }
         }
+        pending.forEach { runCatching { it.close() } }
+        scope.cancel()
+        selectorManager.close()
     }
-}
\ No newline at end of file
+}
diff --git a/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpSocketTransport.kt b/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpSocketTransport.kt
index 7750061..d984d34 100644
--- a/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpSocketTransport.kt
+++ b/src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpSocketTransport.kt
@@ -16,6 +16,9 @@ import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ClosedReceiveChannelException
 import kotlinx.coroutines.channels.ClosedSendChannelException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.NonCancellable
+import kotlinx.coroutines.withContext
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
@@ -54,10 +57,13 @@ internal class UdpSocketTransport(
     private val pingMinTimeout = pingTimeout * 2 / 3
 
     val inputDataBlocks = Channel<UByteArray>(256, onBufferOverflow = BufferOverflow.DROP_OLDEST)
-    val outputDataBlocks = Channel<UByteArray>(256, onBufferOverflow = BufferOverflow.DROP_OLDEST)
+    val outputDataBlocks = Channel<UByteArray>()
+    private val writerReady = CompletableDeferred<PriorityBlockWriter>()
 
     val inputUdpBlocks = Channel<UdpBlock>(256, onBufferOverflow = BufferOverflow.DROP_OLDEST)
 
+    private val closedFlag = SyncValue(false)
+
     private val job = globalLaunch {
         coroutineScope {
             launch { convertOutput() }
@@ -75,20 +81,22 @@ internal class UdpSocketTransport(
 
 
     val transportDevice: InetTransportDevice by lazy {
-        InetTransportDevice(inputDataBlocks, outputDataBlocks, address, { close() }, {})
+        InetTransportDevice(inputDataBlocks, outputDataBlocks, address, { close() }, {}).also { device ->
+            device.prioritizedSend = { writerReady.await().send(it) }
+        }
     }
 
-    private val closedFlag = SyncValue(false)
+    val isClosed: Boolean get() = closedFlag.value
 
-    val isClosed: Boolean = closedFlag.value
-
-    suspend fun close() {
+    suspend fun close(): Unit = withContext(NonCancellable) {
         closedFlag.mutate {
             if (!it) {
                 runCatching { server.sendBlock(UdpBlock.Cancel, socketAddress) }
                 server.disconnectClient(socketAddress)
                 runCatching { inputDataBlocks.close() }
                 runCatching { outputDataBlocks.close() }
+                runCatching { inputUdpBlocks.close() }
+                writerReady.completeExceptionally(net.sergeych.kiloparsec.RemoteInterface.ClosedException())
                 job.cancel()
             }
             true
@@ -167,23 +175,27 @@ internal class UdpSocketTransport(
         }
     }
 
-    private suspend fun convertOutput() {
-        while (!isClosed) {
-            try {
-                server.sendBlock(UdpBlock.Data(outputDataBlocks.receive()), socketAddress)
-            } catch (_: CancellationException) {
-                // this is ok
-                break
-            } catch (_: ClosedReceiveChannelException) {
-                debug { "input channel is closed, closing" }
-                close()
-                break
-            } catch (e: Exception) {
-                exception { "unexpected exception in convertOutput" to e }
-                close()
-                break
+    private suspend fun convertOutput(): Unit = coroutineScope {
+        val writer = PriorityBlockWriter(this) { server.sendBlock(UdpBlock.Data(it), socketAddress) }
+        writerReady.complete(writer)
+        try {
+            while (!isClosed) {
+                try {
+                    writer.send(outputDataBlocks.receive())
+                } catch (_: CancellationException) {
+                    // this is ok
+                    break
+                } catch (_: ClosedReceiveChannelException) {
+                    debug { "input channel is closed, closing" }
+                    close()
+                    break
+                } catch (e: Exception) {
+                    exception { "unexpected exception in convertOutput" to e }
+                    close()
+                    break
+                }
             }
-        }
+        } finally { writer.close() }
         debug { "exiting convertOutput" }
     }
 }
\ No newline at end of file