feat(android): apply authenticated RVD1 budgets to RVT1 clients

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+- Bind Android's RVT1 stream client to the payload and pending-request budgets
+  negotiated by an accepted RVD1 session. Reject oversized outgoing RPCs
+  without emitting a frame or poisoning later requests; oversized incoming
+  frames fail closed. Deterministic JVM tests cover both directions and
+  malformed acceptance budgets. This is companion transport policy, not a
+  claim of a complete Android app or physical-device acceptance.
+
 - Add reusable iOS/iPadOS companion orchestration and SwiftUI UI for explicit
   Bonjour selection, pinned reconnect, provisional RVD1 pairing, pairing-code
   confirmation, Keychain-backed credential persistence, saved-pairing removal,

docs/desktop-closeout-mobile-plan.md

@@ -273,7 +273,12 @@ session client, surfaces explicit pairing approval, hands the accepted
 stream to the product's RVT1 client only after the persistence gate, and
 closes failed or cancelled transport attempts. Kotlin/JVM tests cover
 wrong pins, no reconnect-to-pairing fallback, accepted stream ownership,
-and cancellation during an in-flight TLS dial. The Android `NsdManager`
+and cancellation during an in-flight TLS dial. The accepted RVT1 client now
+applies the payload and pending-request budgets negotiated by its RVD1
+session: oversized outgoing RPCs are rejected before any frame is emitted,
+oversized incoming frames fail closed, and malformed acceptance budgets are
+rejected during session establishment, all pinned by deterministic JVM
+tests. The Android `NsdManager`
 driver, product credential integration, and device-to-device evidence
 remain separate unfinished M3 tasks.
 

platform/android/src/main/kotlin/dev/rivet/runtime/Client.kt

@@ -84,6 +84,7 @@ enum class RivetClientFailure {
     UNEXPECTED_MESSAGE,
     BACKEND,
     TOO_MANY_PENDING_REQUESTS,
+    MESSAGE_TOO_LARGE,
 }
 
 class RivetClientException(
@@ -105,9 +106,13 @@ class RivetClient(
     private val output: OutputStream,
     private val maxPendingRequests: Int = 1024,
     dispatcher: CoroutineDispatcher = Dispatchers.IO,
+    private val maxFramePayloadSize: Int = RIVET_MAX_FRAME_PAYLOAD_SIZE,
 ) : AutoCloseable {
     init {
         require(maxPendingRequests > 0) { "Rivet native pending request limit must be positive" }
+        require(maxFramePayloadSize in 1..RIVET_MAX_FRAME_PAYLOAD_SIZE) {
+            "Rivet frame payload budget must be within protocol limits"
+        }
     }
 
     fun interface EventHandler {
@@ -184,7 +189,11 @@ class RivetClient(
                 addAll(arguments)
             }
             val frame = try {
-                RivetFrame(RivetMessageType.REQUEST, id, encodeRivetValue(RivetValue.ListValue(values)))
+                val payload = encodeRivetValue(RivetValue.ListValue(values))
+                if (payload.size > maxFramePayloadSize) {
+                    clientFailure(RivetClientFailure.MESSAGE_TOO_LARGE, "Rivet request exceeds session payload budget")
+                }
+                RivetFrame(RivetMessageType.REQUEST, id, payload)
             } catch (error: Throwable) {
                 removePending(id, request)?.resumeWith(Result.failure(error))
                 return@suspendCancellableCoroutine
@@ -274,7 +283,7 @@ class RivetClient(
             .order(ByteOrder.LITTLE_ENDIAN)
             .int
             .toLong() and 0xffff_ffffL
-        if (length > RIVET_MAX_FRAME_PAYLOAD_SIZE) {
+        if (length > maxFramePayloadSize) {
             throw RivetProtocolException(
                 RivetProtocolFailure.LENGTH_OVERFLOW,
                 "Rivet frame exceeds protocol length limit",

platform/android/src/main/kotlin/dev/rivet/session/DeviceRVT1Client.kt

@@ -0,0 +1,37 @@
+package dev.rivet.session
+
+import dev.rivet.runtime.RivetClient
+import java.io.InputStream
+import java.io.OutputStream
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.Dispatchers
+
+/**
+ * Build the RVT1 client only after RVD1 has returned an authenticated
+ * [SessionAcceptance]. The caller must start the client, register only
+ * product-allowed APIs, and close both the client and companion session.
+ *
+ * Never turn an mDNS advertisement, a TLS connection alone, or a pairing
+ * prompt into an RVT1 client. The remote gateway remains authoritative
+ * for RPC and State scope enforcement.
+ */
+fun createDeviceRVT1Client(
+    acceptance: SessionAcceptance,
+    input: InputStream,
+    output: OutputStream,
+    dispatcher: CoroutineDispatcher = Dispatchers.IO,
+): RivetClient {
+    require(acceptance.maximumPayload in 1..SESSION_DEFAULT_MAXIMUM_PAYLOAD) {
+        "invalid authenticated device payload budget"
+    }
+    require(acceptance.maximumPending in 1..SESSION_DEFAULT_MAXIMUM_PENDING.toLong()) {
+        "invalid authenticated device pending-request budget"
+    }
+    return RivetClient(
+        input = input,
+        output = output,
+        maxPendingRequests = acceptance.maximumPending.toInt(),
+        dispatcher = dispatcher,
+        maxFramePayloadSize = acceptance.maximumPayload.toInt(),
+    )
+}

platform/android/src/test/kotlin/dev/rivet/runtime/ClientTest.kt

@@ -19,13 +19,16 @@ import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
 import kotlin.test.assertTrue
 
-private class ClientHarness(maxPendingRequests: Int = 1024) : AutoCloseable {
+private class ClientHarness(
+    maxPendingRequests: Int = 1024,
+    maxFramePayloadSize: Int = RIVET_MAX_FRAME_PAYLOAD_SIZE,
+) : AutoCloseable {
     private val backendToClient = PipedOutputStream()
     private val clientInput = PipedInputStream(backendToClient, 1024 * 1024)
     private val clientToBackend = PipedOutputStream()
     val backendInput = PipedInputStream(clientToBackend, 1024 * 1024)
     val backendOutput: OutputStream = backendToClient
-    val client = RivetClient(clientInput, clientToBackend, maxPendingRequests)
+    val client = RivetClient(clientInput, clientToBackend, maxPendingRequests, maxFramePayloadSize = maxFramePayloadSize)
 
     suspend fun start(onEvent: RivetClient.EventHandler? = null) {
         backendOutput.writeFrame(
@@ -214,6 +217,58 @@ class ClientTest {
         }
     }
 
+    @Test
+    fun oversizedOutgoingRequestFailsWithoutPoisoningTheSession() = runBlocking {
+        ClientHarness(maxFramePayloadSize = 64).use { harness ->
+            harness.start()
+            val failed = assertFailsWith<RivetClientException> {
+                harness.client.call("large", listOf(RivetValue.StringValue("x".repeat(128))))
+            }
+            assertEquals(RivetClientFailure.MESSAGE_TOO_LARGE, failed.failure)
+
+            val next = async { harness.client.call("small") }
+            val request = withContext(Dispatchers.IO) { harness.backendInput.readFrame() }
+            assertEquals("small", request.requestName())
+            harness.backendOutput.writeFrame(
+                RivetFrame(RivetMessageType.RESPONSE, request.id, encodeRivetValue(RivetValue.Int64(1))),
+            )
+            assertEquals(RivetValue.Int64(1), withTimeout(2_000) { next.await() })
+        }
+    }
+
+    @Test
+    fun oversizedIncomingFrameFailsClosedAndReleasesPendingCalls() = runBlocking {
+        // An expected failure in an async child must not cancel the parent
+        // test scope before assertFailsWith can inspect the terminal error.
+        supervisorScope {
+            ClientHarness(maxFramePayloadSize = 64).use { harness ->
+                harness.start()
+                val pending = async { harness.client.call("waiting") }
+                val request = withContext(Dispatchers.IO) { harness.backendInput.readFrame() }
+                harness.backendOutput.writeFrame(
+                    RivetFrame(RivetMessageType.RESPONSE, request.id, ByteArray(65)),
+                )
+                val failure = assertFailsWith<RivetProtocolException> {
+                    withTimeout(2_000) { pending.await() }
+                }
+                assertEquals(RivetProtocolFailure.LENGTH_OVERFLOW, failure.failure)
+            }
+        }
+    }
+
+    @Test
+    fun invalidRVT1PayloadBudgetsAreRejected() {
+        assertFailsWith<IllegalArgumentException> {
+            RivetClient(InputStream.nullInputStream(), OutputStream.nullOutputStream(), maxFramePayloadSize = 0)
+        }
+        assertFailsWith<IllegalArgumentException> {
+            RivetClient(
+                InputStream.nullInputStream(), OutputStream.nullOutputStream(),
+                maxFramePayloadSize = RIVET_MAX_FRAME_PAYLOAD_SIZE + 1,
+            )
+        }
+    }
+
     @Test
     fun invalidHelloIsTerminalAndRestartIsRejected() = runBlocking {
         val backendOutput = PipedOutputStream()

platform/android/src/test/kotlin/dev/rivet/session/DeviceRVT1ClientTest.kt

@@ -0,0 +1,46 @@
+package dev.rivet.session
+
+import java.io.ByteArrayInputStream
+import java.io.ByteArrayOutputStream
+import kotlin.test.Test
+import kotlin.test.assertFailsWith
+
+class DeviceRVT1ClientTest {
+    private val uuid = SessionUuid(ByteArray(16))
+
+    private fun acceptance(payload: Long, pending: Long) = SessionAcceptance(
+        sessionID = uuid,
+        deviceID = uuid,
+        scopeNames = listOf("reader"),
+        maximumPayload = payload,
+        maximumPending = pending,
+    )
+
+    @Test
+    fun acceptsTheNegotiatedLimitsWithoutOpeningAConnection() {
+        val client = createDeviceRVT1Client(
+            acceptance(512L * 1024, 64),
+            ByteArrayInputStream(byteArrayOf()),
+            ByteArrayOutputStream(),
+        )
+        client.close()
+    }
+
+    @Test
+    fun rejectsInvalidOrUnboundedNegotiatedLimits() {
+        for (payload in listOf(-1L, 0L, 512L * 1024 + 1)) {
+            assertFailsWith<IllegalArgumentException> {
+                createDeviceRVT1Client(
+                    acceptance(payload, 1), ByteArrayInputStream(byteArrayOf()), ByteArrayOutputStream(),
+                )
+            }
+        }
+        for (pending in listOf(-1L, 0L, 65L, Long.MAX_VALUE)) {
+            assertFailsWith<IllegalArgumentException> {
+                createDeviceRVT1Client(
+                    acceptance(1, pending), ByteArrayInputStream(byteArrayOf()), ByteArrayOutputStream(),
+                )
+            }
+        }
+    }
+}