fix(core/rumble): the Deck's keepalive stops being swallowed by its own renewals

Three faults in the shared rumble policy engine, all answered by one change of shape: the free-running jitter phase becomes `last_emit` — the exact value last handed to an embedder — and every emit routes through one helper. That single field answers all three live questions: would re-sending this be a no-op device write, is this stop redundant, and would the nudge invent a stop. The Steam Deck declares a 40 ms keepalive with a 1-LSB nudge, because an SDL-class layer discards a write identical to the last one. But the nudge lived only in the keepalive branch, so every host renewal re-emitted the raw level, collided with the last jittered write, was discarded, AND re-anchored the keepalive timer. The gap between distinct device writes stretched to 80 ms at the 400 ms default TTL and 100 ms at the hatch floor — two to two and a half times the cadence the quirk exists to guarantee. Nudging on any repeat closes it: 40 ms throughout. Level (1, 0) turned that nudge into (0, 0) — the value the engine reserves for "stop now" — and handed it out with a non-zero backstop, under a live lease. It is the only such level: high must already be zero, and low ^ 1 == 0 implies low == 1. The nudge now steps the LSB up instead, so the phase still alternates and no stop is ever invented. A zero for a pad the engine already believes silent is now dropped. Under the legacy hatch the host re-sends zeros for every latched pad every 500 ms for the rest of the session, which cost Android an unconditional log line and a binder cancel() at 2 Hz per pad. The deliberate stop-burst heal is untouched, because a stop that was LOST leaves the pad buzzing, and that is exactly the guard's pass condition. The client also now bounds the lease it will honour. RUMBLE_TTL_CEIL_MS is sender-side only, so a modified or third-party host could stamp a long TTL and wedge its pump, leaving Apple — whose renderer deliberately keeps no staleness policy of its own — and a Deck slot buzzing for all of it. Every new test was proven to fail with its own fix reverted, including the two that guard against over-reach: a default-quirks pad must still get the level verbatim, or an off-by-one amplitude would land in Apple's identical-target comparison and Android's one-shots. One suspicion from the audit did NOT survive: a v2 envelope carrying ttl_ms 0 cannot take the legacy backstop, because the expiry check preempts the relay branch. No fix; pinned with a test so that ordering stays load-bearing. Verified: 17/17 rumble tests, clippy --all-targets --features quic -D warnings = 0, fmt clean, generated C header unchanged. (`c_abi_harness_round_trips` fails on this Mac with a linker error, identically on an unmodified tree.) From the 2026-08-03 force-feedback sweep (B12, B22, R9, T1).
2026-08-04 07:40:19 +02:00
8 changed files with 275 additions and 389 deletions
@@ -160,14 +160,6 @@ jobs:
          key: gradle-${{ hashFiles('clients/android/**/*.gradle.kts', 'clients/android/gradle/wrapper/gradle-wrapper.properties') }}
          restore-keys: gradle-
      # The kit's JVM unit tests — the pure parsers, migrations and feedback policies. They were
      # running nowhere: this workflow only assembled, and android-screenshots.yml runs the :app
      # module's tests, so nothing enforced :kit's. Cheap (a couple of seconds against an already
      # built module) and it is the only automated cover those behaviours have.
      - name: kit unit tests
        working-directory: clients/android
        run: ./gradlew :kit:testDebugUnitTest --stacktrace
      - name: assembleDebug (cargo-ndk → jniLibs → APK)
        working-directory: clients/android
        env:
@@ -116,9 +116,7 @@ class DsCapture(
            // The interfaces are about to release with the kernel driver still detached — a
            // mid-rumble teardown would leave the motors running with nobody to stop them.
            // EP0-direct (the reader thread is stopping; the queue would never drain).
-            // Nothing can retry after this point, so a failure is worth saying out loud: it is
+            usb.writeControl(stopReport(m))
            // the difference between a quiet pad and one that buzzes until it is unplugged.
            if (!usb.writeControl(stopReport(m))) Log.w(TAG, "teardown rumble stop was not written")
        }
        disarmBackstop()
        usb.stop()
@@ -147,9 +145,6 @@ class DsCapture(
        val wasActive = model != null
        model = null
        releaseSlot()
        // Release the transport too: the link only *signals* the drop, so without this an unplug
        // left its connection open, its interfaces claimed and its detach receiver registered.
        usb.stop()
        if (wasActive) onActiveChanged?.invoke(false)
    }
@@ -221,20 +216,17 @@ class DsCapture(
    override fun rumble(pad: Int, low: Int, high: Int, backstopMs: Long) {
        val m = model ?: return
-        val stop = low == 0 && high == 0
+        if (low == 0 && high == 0) {
-        if (!stop) armBackstop(backstopMs)
+            disarmBackstop()
-        val sent = if (m == DsDevice.Model.DUALSHOCK4) {
+        } else {
            armBackstop(backstopMs)
        }
        if (m == DsDevice.Model.DUALSHOCK4) {
            ds4Low = low
            ds4High = high
            writeDs4()
        } else {
-            usb.writeRaw(0, DsDevice.ds5RumbleReport(m, low, high), OutReportQueue.KEY_RUMBLE)
+            usb.writeRaw(0, DsDevice.ds5RumbleReport(m, low, high))
        }
        if (stop) {
            // Disarm only once the stop is actually on its way. Dropping the net *before* the
            // write — as this used to — meant a discarded stop left the motors running with
            // nothing scheduled to try again; a USB pad holds its last level until told zero.
            if (sent) disarmBackstop() else armBackstop(STOP_RETRY_MS)
        }
    }
@@ -260,9 +252,6 @@ class DsCapture(
        usb.writeRaw(0, DsDevice.ds5TriggerReport(m, which, effect))
    }
    // Coalescable: the DS4's write is full-state (motors AND lightbar, rebuilt from the current
    // fields on every call), so a newer one supersedes an older one wholesale — nothing is lost by
    // collapsing a backlog of them down to the last.
    private fun writeDs4() = usb.writeRaw(
        0,
        DsDevice.ds4Report(
@@ -272,7 +261,6 @@ class DsCapture(
            (ds4Rgb shr 8) and 0xFF,
            ds4Rgb and 0xFF,
        ),
        OutReportQueue.KEY_RUMBLE,
    )
    /** The report that stops the motors. The DS4's is a full-state write, so it zeroes the
@@ -296,12 +284,7 @@ class DsCapture(
        backstop?.let { mainHandler.removeCallbacks(it) }
        val r = Runnable {
            backstop = null
-            val m = model ?: return@Runnable
+            model?.let { usb.writeRaw(0, stopReport(it)) }
            // The net itself can be refused (a full queue, a connection going away). Re-arm rather
            // than give up: this is the last thing between a stalled poll thread and a pad that
            // buzzes until it is unplugged. It stops re-arming as soon as the link closes, which
            // clears `model` and disarms.
            if (!usb.writeRaw(0, stopReport(m), OutReportQueue.KEY_RUMBLE)) armBackstop(STOP_RETRY_MS)
        }
        backstop = r
        mainHandler.postDelayed(r, ms.coerceAtLeast(1))
@@ -314,9 +297,5 @@ class DsCapture(
    private companion object {
        const val TAG = "DsCapture"
        /** How soon to retry a rumble stop whose write was rejected. Short: the motors are running
         *  and the host has already moved on, so nothing else is coming to silence them. */
        const val STOP_RETRY_MS = 100L
    }
 }
@@ -88,9 +88,6 @@ class GamepadFeedback(
        const val TAG_PLAYER_LEDS: Byte = 0x02
        const val TAG_TRIGGER: Byte = 0x03
        const val TAG_HID_RAW: Byte = 0x05
        /** Sparse-log cadence for swallowed render failures — see [noteRenderFailure]. */
        const val LOG_EVERY = 128L
    }
    /** One controller's rumble binding — VibratorManager (API 31+) OR the legacy single Vibrator (API 28–30). */
@@ -128,7 +125,6 @@ class GamepadFeedback(
    fun start() {
        running = true
        rumbleThread = Thread({
            var failures = 0L
            while (running) {
                val ev = NativeBridge.nativeNextRumble(handle)
                if (ev < 0L) continue // timeout / closed
@@ -140,50 +136,26 @@ class GamepadFeedback(
                // the backstop (the hardware net under a stalled poll thread).
                val pad = ((ev ushr 49) and 0xFL).toInt()
                val backstopMs = ((ev ushr 32) and 0xFFFF)
-                // Rendering is binder calls into the vibrator service, and every one of them can
+                renderRumble(
-                // throw unchecked — DeadSystemRuntimeException when system_server goes down, and
+                    pad,
-                // the ordinary RuntimeException a dying service wraps its RemoteException in.
+                    ((ev ushr 16) and 0xFFFF).toInt(),
-                // Unguarded, ONE of those killed this thread outright: `running` stayed true, so
+                    (ev and 0xFFFF).toInt(),
-                // nothing noticed and nothing restarted it, and rumble was gone for the rest of
+                    backstopMs,
-                // the session. Losing a single command is recoverable; losing the loop is not.
+                )
                runCatching {
                    renderRumble(
                        pad,
                        ((ev ushr 16) and 0xFFFF).toInt(),
                        (ev and 0xFFFF).toInt(),
                        backstopMs,
                    )
                }.onFailure { failures = noteRenderFailure("rumble", it, failures) }
            }
        }, "pf-rumble").apply { isDaemon = true; start() }
        hidoutThread = Thread({
            // 128: the raw as-is passthrough events are [pad][kind tag][report kind][≤64 bytes].
            val buf = ByteBuffer.allocateDirect(128)
            var failures = 0L
            while (running) {
                val n = NativeBridge.nativeNextHidout(handle, buf)
                if (n < 0) continue // timeout / closed
-                // Same hazard as the rumble loop above: lights/trigger rendering is binder and USB
+                dispatchHidout(buf, n)
                // calls, and an unchecked throw here would silently end the rich-feedback plane.
                runCatching { dispatchHidout(buf, n) }
                    .onFailure { failures = noteRenderFailure("hidout", it, failures) }
            }
        }, "pf-hidout").apply { isDaemon = true; start() }
    }
    /**
     * Record a render failure the poll loop swallowed, and return the updated count. Logged on the
     * first occurrence and sparsely after: a genuinely dead vibrator service fails on *every*
     * command, which at a rumble plane's rate would bury the log.
     */
    private fun noteRenderFailure(plane: String, t: Throwable, seen: Long): Long {
        if (seen == 0L || seen % LOG_EVERY == 0L) {
            Log.w(TAG, "$plane render failed (#${seen + 1}) — command dropped, poll loop alive", t)
        }
        return seen + 1
    }
    /** Idempotent. Stops + joins the poll threads (must complete before the router is released / handle freed). */
    fun stop() {
        running = false
@@ -297,7 +269,7 @@ class GamepadFeedback(
        val m = bind.vm
        if (m != null) {
            if (lo == 0 && hi == 0) {
-                runCatching { m.cancel() } // (0,0) = stop
+                m.cancel() // (0,0) = stop
                return
            }
            val combo = CombinedVibration.startParallel()
@@ -322,7 +294,7 @@ class GamepadFeedback(
        // API 28–30 legacy single-motor path: blend both motors into one effect.
        val lv = bind.legacy ?: return
        if (lo == 0 && hi == 0) {
-            runCatching { lv.cancel() } // (0,0) = stop
+            lv.cancel() // (0,0) = stop
            return
        }
        val a = (lo * 0.8 + hi * 0.33).toInt().coerceIn(1, 255)
@@ -14,8 +14,8 @@ import android.hardware.usb.UsbRequest
 import android.os.Build
 import android.util.Log
 import java.nio.ByteBuffer
 import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.TimeoutException
 import java.util.concurrent.atomic.AtomicBoolean
 /**
 * Generic USB transport for a client-captured HID controller — the device-agnostic half of what
@@ -81,20 +81,14 @@ class HidUsbLink(
    /** Pending OUT reports, submitted by the reader thread — only one thread may drive a
     *  connection's [UsbRequest]s ([UsbDeviceConnection.requestWait] returns ANY completed
-     *  request; a second waiter would steal the reader's completions). See [OutReportQueue] for
+     *  request; a second waiter would steal the reader's completions). */
-     *  what gets discarded when it fills, and why that is not simply "the oldest". */
+    private val outQueue = ConcurrentLinkedQueue<ByteArray>()
    private val outQueue = OutReportQueue()
    private var reader: Thread? = null
    private var detachReceiver: BroadcastReceiver? = null
    @Volatile private var running = false
    /** Latches on the first "this link is down" signal so [onClosed] fires exactly once, however
     *  many of the racing detectors (detach broadcast, reader error streak, failed re-queue) see
     *  it. Reset by [start]. */
    private val down = AtomicBoolean(false)
    /** First attached matching device, or null. Does not need USB permission to enumerate. */
    fun findDevice(): UsbDevice? = usb.deviceList.values.firstOrNull(config.deviceMatch)
@@ -120,7 +114,6 @@ class HidUsbLink(
        connection = conn
        device = dev
        claims = claimed
        down.set(false)
        running = true
        Log.i(
            config.tag,
@@ -141,7 +134,10 @@ class HidUsbLink(
                val gone: UsbDevice? = intent.getParcelableExtra(UsbManager.EXTRA_DEVICE)
                if (gone?.deviceName == dev.deviceName) {
                    Log.i(config.tag, "USB detached (${dev.deviceName})")
-                    linkDown()
+                    if (running) {
                        running = false
                        onClosed()
                    }
                }
            }
        }
@@ -225,9 +221,6 @@ class HidUsbLink(
        if (live.isEmpty()) {
            Log.e(config.tag, "no IN request could be queued")
            finishReader(claims)
            // `start` already returned true, so without this the owner would sit waiting on a
            // capture that never streams and never reports itself dead.
            linkDown()
            return
        }
        val scratch = ByteArray(64)
@@ -302,23 +295,10 @@ class HidUsbLink(
        } finally {
            finishReader(claims)
        }
-        linkDown()
+        if (running) {
-    }
+            running = false
-
+            onClosed()
-    /**
+        }
     * Report the link down, exactly once, from whichever detector noticed first — the detach
     * broadcast (main thread) or the reader thread on its way out.
     *
     * This only *signals*; releasing the connection and the interfaces stays the owner's job, via
     * the [stop] its `onClosed` handler calls. Previously nothing released them on this path: the
     * detach receiver flipped a flag and fired the callback, so an unplug left the connection open,
     * the interfaces claimed (the pad could not return to Android's own input stack) and the
     * receiver still registered — and a re-plug overwrote the field holding it, leaking a receiver
     * that stayed live for the process's lifetime.
     */
    private fun linkDown() {
        running = false
        if (down.compareAndSet(false, true)) onClosed()
    }
    private fun finishReader(claims: List<Claim>) {
@@ -334,35 +314,28 @@ class HidUsbLink(
     * Write one raw report to the device: kind 0 = output report (the active interface's
     * interrupt-OUT, else a `SET_REPORT(Output)` control transfer), kind 1 = feature report
     * (`SET_REPORT(Feature)`). [data] is the full report, id byte first, hidapi framing.
     *
     * [coalesce] tells the pending-OUT queue whether a newer report of the same kind may replace
     * this one — [OutReportQueue.KEY_RUMBLE] for motor levels, the default [OutReportQueue.NO_COALESCE]
     * for one-shots (lightbar, player LEDs, trigger effects) the sender will not repeat.
     *
     * Returns whether the report reached the device or is queued for it. A caller that is writing
     * a **stop** needs this: a discarded stop has nothing behind it, so it must not be mistaken
     * for one that landed.
     */
-    fun writeRaw(kind: Int, data: ByteArray, coalesce: Int = OutReportQueue.NO_COALESCE): Boolean {
+    fun writeRaw(kind: Int, data: ByteArray) {
-        if (data.isEmpty()) return false
+        if (data.isEmpty()) return
-        return when (kind) {
+        when (kind) {
            0 -> {
                if ((activeClaim ?: claims.firstOrNull())?.outReq != null) {
-                    // Interrupt-OUT rides UsbRequests submitted by the reader thread.
+                    // Interrupt-OUT rides UsbRequests submitted by the reader thread. Bounded,
-                    outQueue.offer(data, coalesce)
+                    // newest-wins: these are level-styled commands the sender re-sends anyway.
                    while (outQueue.size >= 32) outQueue.poll()
                    outQueue.offer(data)
                } else {
                    setReport(REPORT_TYPE_OUTPUT, data)
                }
            }
            1 -> setReport(REPORT_TYPE_FEATURE, data)
            else -> false
        }
    }
-    private fun setReport(type: Int, data: ByteArray): Boolean {
+    private fun setReport(type: Int, data: ByteArray) {
-        val conn = connection ?: return false
+        val conn = connection ?: return
-        val ifId = (activeClaim ?: claims.firstOrNull())?.iface?.id ?: return false
+        val ifId = (activeClaim ?: claims.firstOrNull())?.iface?.id ?: return
-        return sendReport(conn, ifId, type, data)
+        sendReport(conn, ifId, type, data)
    }
    /**
@@ -371,8 +344,9 @@ class HidUsbLink(
     * queue would never drain (e.g. a rumble stop before the interfaces release). Safe from any
     * thread: EP0 control transfers are independent of the reader's `requestWait`.
     */
-    fun writeControl(data: ByteArray): Boolean =
+    fun writeControl(data: ByteArray) {
-        data.isNotEmpty() && setReport(REPORT_TYPE_OUTPUT, data)
+        if (data.isNotEmpty()) setReport(REPORT_TYPE_OUTPUT, data)
    }
    private fun sendKeepAlive(conn: UsbDeviceConnection, ifaceId: Int) {
        for (f in config.keepAliveFeatures) sendReport(conn, ifaceId, REPORT_TYPE_FEATURE, f)
@@ -384,48 +358,27 @@ class HidUsbLink(
     * "unnumbered" (id 0 in wValue, id byte stripped from the payload). EP0 is independent of
     * the interrupt endpoints, so this is safe alongside the reader thread's requestWait.
     */
-    private fun sendReport(
+    private fun sendReport(conn: UsbDeviceConnection, ifaceId: Int, type: Int, data: ByteArray) {
        conn: UsbDeviceConnection,
        ifaceId: Int,
        type: Int,
        data: ByteArray,
    ): Boolean {
        val id = data[0].toInt() and 0xFF
        val payload = if (id == 0) data.copyOfRange(1, data.size) else data
-        // controlTransfer returns the byte count, or a negative value on failure — a failed write
+        conn.controlTransfer(
-        // must be reported as such, not swallowed (a dropped rumble stop has nothing behind it).
+            0x21, // host→device, class, interface
-        val n = runCatching {
+            0x09, // SET_REPORT
-            conn.controlTransfer(
+            (type shl 8) or id,
-                0x21, // host→device, class, interface
+            ifaceId,
-                0x09, // SET_REPORT
+            payload,
-                (type shl 8) or id,
+            payload.size,
-                ifaceId,
+            WRITE_TIMEOUT_MS,
-                payload,
+        )
                payload.size,
                WRITE_TIMEOUT_MS,
            )
        }.getOrDefault(-1)
        return n >= 0
    }
-    /**
+    /** Stop the read loop and release the interfaces. Idempotent; does not fire [onClosed]. */
     * Stop the read loop and release the interfaces. Idempotent; does not fire [onClosed].
     *
     * Safe to call from the `onClosed` handler itself — that is how an unplug now gets cleaned up,
     * and it arrives on the reader thread, which must not try to join itself.
     */
    fun stop() {
        running = false
        // Claim the down-latch so the reader's own exit does not report a close the owner asked for.
        down.set(true)
        detachReceiver?.let { runCatching { context.unregisterReceiver(it) } }
        detachReceiver = null
-        if (reader !== Thread.currentThread()) {
+        runCatching { reader?.join(1000) }
-            runCatching { reader?.join(1000) }
+        reader = null
            // Only forget the thread once it is actually gone: clearing it while it still runs
            // would let a later stop() skip the join and free the connection under it.
            reader = null
        }
        outQueue.clear()
        activeClaim = null
        for (c in claims) runCatching { connection?.releaseInterface(c.iface) }
@@ -1,89 +0,0 @@
 package io.unom.punktfunk.kit
 /**
 * The pending interrupt-OUT reports for a captured controller: a bounded FIFO whose overflow
 * policy knows which reports may be thrown away and which may not.
 *
 * The queue exists because only one thread may drive a connection's `UsbRequest`s, so writes from
 * the feedback threads are handed to the reader thread rather than submitted directly. It has to
 * be bounded — a stalled or unplugged device would otherwise grow it without limit — and the
 * question is what to discard when it fills.
 *
 * The old policy was "newest wins": drop from the head until there is room. That is right for
 * rumble, which is *level-styled* — the host re-sends it continuously, so a dropped frame is
 * replaced milliseconds later and nothing is permanently lost. It is wrong for everything else.
 * A lightbar colour, a player-LED mask and an adaptive-trigger effect are **one-shots**: the host
 * sends them on change and never repeats them. Dropping one leaves the pad wrong until the next
 * time that value happens to change, which may be never.
 *
 * So eviction is driven by an explicit [key] supplied by the caller, not by inspecting the bytes.
 * That distinction cannot be recovered from the report itself: every DualSense output report
 * carries the *same* report id and differs only in its `valid_flag` bytes, so an id-keyed policy
 * would happily let a rumble supersede a lightbar — the very bug this replaces, relocated.
 *
 * Two rules:
 *  - A report offered with a coalescing key **replaces** the pending report with that key, in
 *    place. A burst of rumble collapses to its latest value and never displaces anything else.
 *  - Only when the queue is full does anything get dropped, and then the oldest *coalescable*
 *    report goes first. A one-shot is discarded only if the queue is full of nothing but
 *    one-shots — which needs [cap] distinct one-shots outstanding, far beyond what a real pad
 *    produces.
 *
 * Thread-safe: offered by the feedback threads, drained by the reader thread.
 */
 internal class OutReportQueue(private val cap: Int = CAP) {
    private class Entry(val key: Int, val data: ByteArray)
    private val items = ArrayDeque<Entry>()
    /**
     * Queue [data] for submission. [key] is [NO_COALESCE] for a one-shot, or a caller-chosen
     * constant identifying a level-styled stream whose newer values supersede older ones.
     *
     * Returns false only if the report had to be dropped outright — the caller can then treat the
     * write as failed rather than assuming it is on its way.
     */
    fun offer(data: ByteArray, key: Int = NO_COALESCE): Boolean = synchronized(items) {
        if (key != NO_COALESCE) {
            val at = items.indexOfFirst { it.key == key }
            if (at >= 0) {
                // Supersede in place: keeping the queue position stops a fast rumble stream from
                // repeatedly jumping the one-shots queued ahead of it.
                items[at] = Entry(key, data)
                return true
            }
        }
        if (items.size >= cap) {
            val victim = items.indexOfFirst { it.key != NO_COALESCE }
            if (victim >= 0) {
                items.removeAt(victim)
            } else if (key != NO_COALESCE) {
                // Nothing coalescable to sacrifice and this report is itself replaceable — drop it
                // rather than a one-shot that will never come again.
                return false
            } else {
                items.removeFirst()
            }
        }
        items.addLast(Entry(key, data))
        return true
    }
    /** The next report to submit, or null when nothing is pending. */
    fun poll(): ByteArray? = synchronized(items) { items.removeFirstOrNull()?.data }
    fun clear() = synchronized(items) { items.clear() }
    val size: Int get() = synchronized(items) { items.size }
    companion object {
        /** This report is a one-shot: never superseded, evicted only as a last resort. */
        const val NO_COALESCE = 0
        /** Motor levels — re-sent continuously, so only the newest is worth keeping. */
        const val KEY_RUMBLE = 1
        /** Deep enough to absorb a burst, small enough that a stalled device cannot bloat us. */
        const val CAP = 32
    }
 }
@@ -273,20 +273,10 @@ class Sc2Capture(
    private fun onLinkClosed() {
        Log.i(TAG, "SC2 link closed (unplug / power-off)")
        // Both transports share this callback, so read which one was live BEFORE clearing it —
        // releasing the other would tear down a link that never dropped.
        val dropped = activeLink
        activeLink = LINK_NONE
        dongleLink = false
        releaseSlot()
        releaseUiKeys()
        // Release the transport too — see the note in DsCapture.onLinkClosed. The Puck makes this
        // worse than a single leak: it is the pad that gets power-cycled, so the same process can
        // round-trip a link many times in one session.
        when (dropped) {
            LINK_USB -> usb.stop()
            LINK_BLE -> ble.stop()
        }
        onActiveChanged?.invoke(false)
    }
@@ -1,102 +0,0 @@
 package io.unom.punktfunk.kit
 import org.junit.Assert.assertArrayEquals
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
 import org.junit.Assert.assertNull
 import org.junit.Assert.assertTrue
 import org.junit.Test
 /**
 * The pending-OUT queue's overflow policy. What is being pinned here is the distinction the old
 * "drop from the head until there is room" policy did not make: rumble is re-sent continuously and
 * may be thrown away, while a lightbar/player-LED/trigger report is sent once and never repeated.
 */
 class OutReportQueueTest {
    /** A report carrying a 0..255 marker so a test can tell which one came back out. */
    private fun report(marker: Int) = byteArrayOf(0x02, marker.toByte())
    // Masked: the marker rides in a Byte, and Byte.toInt() sign-extends.
    private fun drain(q: OutReportQueue): List<Int> =
        generateSequence { q.poll() }.map { it[1].toInt() and 0xFF }.toList()
    @Test
    fun `rumble supersedes the pending rumble instead of queueing another`() {
        val q = OutReportQueue()
        assertTrue(q.offer(report(1), OutReportQueue.KEY_RUMBLE))
        assertTrue(q.offer(report(2), OutReportQueue.KEY_RUMBLE))
        assertTrue(q.offer(report(3), OutReportQueue.KEY_RUMBLE))
        assertEquals("a rumble burst must collapse to one entry", 1, q.size)
        assertArrayEquals(report(3), q.poll())
        assertNull(q.poll())
    }
    @Test
    fun `superseding keeps the queue position so a rumble stream cannot jump one-shots`() {
        val q = OutReportQueue()
        q.offer(report(1), OutReportQueue.KEY_RUMBLE)
        q.offer(report(10)) // a one-shot queued behind it
        q.offer(report(2), OutReportQueue.KEY_RUMBLE)
        // The newer rumble takes the OLD rumble's slot, so the one-shot does not get starved
        // behind an endlessly-renewed entry.
        assertEquals(listOf(2, 10), drain(q))
    }
    @Test
    fun `a full queue sacrifices rumble, never a one-shot`() {
        val q = OutReportQueue(cap = 4)
        q.offer(report(1), OutReportQueue.KEY_RUMBLE)
        q.offer(report(10))
        q.offer(report(11))
        q.offer(report(12))
        assertEquals(4, q.size)
        // Full. The old policy dropped the head — here that is a rumble, but only by luck of
        // ordering; what matters is that the one-shots all survive.
        assertTrue(q.offer(report(13)))
        assertEquals(listOf(10, 11, 12, 13), drain(q))
    }
    @Test
    fun `the one-shot the host never repeats survives a rumble storm`() {
        val q = OutReportQueue(cap = 4)
        // The exact regression: a lightbar colour queued once, then a flood of rumble. Under the
        // old newest-wins eviction the colour was dropped from the head and never came back,
        // leaving the pad lit wrong until the value next happened to change.
        q.offer(report(200)) // lightbar
        repeat(50) { q.offer(report(it), OutReportQueue.KEY_RUMBLE) }
        val out = drain(q)
        assertTrue("the lightbar report must still be queued, got $out", out.contains(200))
        assertEquals("rumble must not have accumulated", listOf(200, 49), out)
    }
    @Test
    fun `a queue full of one-shots refuses a rumble rather than dropping one`() {
        val q = OutReportQueue(cap = 2)
        q.offer(report(10))
        q.offer(report(11))
        assertFalse(
            "with nothing coalescable to sacrifice, the replaceable report yields",
            q.offer(report(1), OutReportQueue.KEY_RUMBLE),
        )
        assertEquals(listOf(10, 11), drain(q))
    }
    @Test
    fun `only a queue of nothing but one-shots drops one, and it is the oldest`() {
        val q = OutReportQueue(cap = 2)
        q.offer(report(10))
        q.offer(report(11))
        assertTrue(q.offer(report(12)))
        assertEquals(listOf(11, 12), drain(q))
    }
    @Test
    fun `clear empties the queue`() {
        val q = OutReportQueue()
        q.offer(report(1), OutReportQueue.KEY_RUMBLE)
        q.offer(report(10))
        q.clear()
        assertEquals(0, q.size)
        assertNull(q.poll())
    }
 }
@@ -36,6 +36,22 @@ pub const LEGACY_STALE_MS: u64 = 1000;
 /// engine's staleness zero lands at 1 s; this is the hardware-level net under an engine stall).
 const BACKSTOP_LEGACY_MS: u32 = 2000;
 /// The longest lease the engine honours, whatever the envelope claims — the receiver-side mirror of
 /// the host's own `RUMBLE_TTL_CEIL_MS`.
 ///
 /// No host built from this tree can exceed it (the `PUNKTFUNK_RUMBLE_TTL_MS` hatch is clamped to
 /// `[150, 5000]` before it reaches the wire), so this is defence in depth against a third-party or
 /// modified sender that stamps a long TTL and then wedges its renewal pump while the connection
 /// stays up. It matters on exactly the platforms that sustain a level for the whole lease: Apple,
 /// whose renderer deliberately keeps no staleness policy of its own, and a Deck slot, whose
 /// keepalive re-kicks the actuator until the lease ends. Duration-parameterized embedders (SDL,
 /// Android) already self-terminate at the clamped backstop.
 ///
 /// Deliberately NOT `pub`: an embedder has no use for it, and every `pub` const in this crate is
 /// emitted into `include/punktfunk_core.h` as an UNPREFIXED `#define` — a collision hazard the
 /// header already has ~170 instances of, and one this has no reason to add to.
 const MAX_LEASE_MS: u16 = 5_000;
 /// One effective actuator command. `(0, 0)` means stop now. `backstop_ms` is a safety-net
 /// duration for platform APIs that take one (SDL rumble, Android one-shots): the engine emits
 /// explicit zeros at every policy stop, so the backstop only matters if the embedder thread itself
@@ -75,8 +91,11 @@ struct PadState {
    /// A wire update landed since the last emit (level change OR renewal — renewals re-emit).
    dirty: bool,
    next_keepalive: Option<Instant>,
-    /// Current jitter phase (see [`ActuatorQuirks::dedup_jitter`]).
+    /// The exact value last handed to an embedder. `(0, 0)` ⇔ the engine believes this actuator is
-    jitter: bool,
+    /// silent. It replaces a free-running jitter phase because one field answers all three live
    /// questions: would re-sending this be a no-op device write (the dedupe nudge), is a stop
    /// redundant, and would the nudge synthesize the reserved stop.
    last_emit: (u16, u16),
    quirks: ActuatorQuirks,
 }
@@ -88,7 +107,7 @@ impl PadState {
        legacy_wire: None,
        dirty: false,
        next_keepalive: None,
-        jitter: false,
+        last_emit: (0, 0),
        quirks: ActuatorQuirks {
            keepalive_ms: 0,
            min_pulse_ms: 0,
@@ -112,6 +131,7 @@ impl PadState {
        self.legacy_wire = None;
        self.next_keepalive = None;
        self.dirty = false;
        self.last_emit = (0, 0);
        RumbleCommand {
            pad,
            low: 0,
@@ -119,6 +139,40 @@ impl PadState {
            backstop_ms: 0,
        }
    }
    /// Build the command for the pad's current level, and record what we handed out.
    ///
    /// On a `dedup_jitter` actuator, re-emitting the value the device last took is a no-op write on
    /// an SDL-class layer, so the low motor's LSB is nudged. Keying that on `last_emit` rather than
    /// on a free-running phase is what makes it work on EVERY emit path. Previously the nudge lived
    /// only in the keepalive branch, so a host renewal — which arrives every `ttl*3/10` ms, 120 ms
    /// at the 400 ms default and 60 ms at the hatch floor — re-emitted the raw level, collided with
    /// the last jittered write, was swallowed, AND re-anchored the keepalive. That stretched the
    /// gap between *distinct* device writes to 80 ms at the default cadence and 100 ms at the
    /// floor, on an actuator whose quirk declares 40.
    ///
    /// The nudge is refused when it would synthesize the reserved `(0, 0)` stop. That is level
    /// `(1, 0)` and only that: `high` must already be 0, and `low ^ 1 == 0` implies `low == 1`.
    /// There the LSB steps up instead, so the phase still alternates (1 ↔ 3, two parts in 65535)
    /// and the pad never receives a stop the policy did not order.
    fn emit(&mut self, pad: u16) -> RumbleCommand {
        let (mut low, high) = self.level;
        if self.quirks.dedup_jitter && (low, high) == self.last_emit {
            let alt = low ^ 1;
            low = if (alt, high) == (0, 0) {
                low | 0b10
            } else {
                alt
            };
        }
        self.last_emit = (low, high);
        RumbleCommand {
            pad,
            low,
            high,
            backstop_ms: self.backstop(),
        }
    }
 }
 /// The pure per-connection policy state machine. Time is always passed in (`now`) so the policy
@@ -156,6 +210,8 @@ impl RumbleEngine {
        p.dirty = true;
        match ttl_ms {
            Some(t) => {
                // Never honour a lease longer than [`MAX_LEASE_MS`], whatever the sender claims.
                let t = t.min(MAX_LEASE_MS);
                p.ttl_ms = t;
                p.legacy_wire = None;
                p.deadline = if (low, high) != (0, 0) {
@@ -214,22 +270,25 @@ impl RumbleEngine {
            if p.dirty {
                p.dirty = false;
                if p.level == (0, 0) {
-                    return (Some(p.silence(pad)), None);
+                    // Relay a stop only if the actuator is, as far as the engine knows, still
                    // buzzing. A zero on an already-silent pad heals nothing and costs every
                    // embedder a command — Android an unconditional log line plus a binder
                    // `cancel()`. Two senders produce them: the host's deliberate
                    // `RUMBLE_STOP_BURST` re-sends after the first stop already landed, and (behind
                    // `PUNKTFUNK_RUMBLE_ENVELOPE=0`) the legacy flat 500 ms refresh, which re-sends
                    // zeros for every latched pad for the rest of the session. The burst still
                    // heals the case it exists for: a LOST first stop leaves the pad buzzing, so
                    // `last_emit != (0, 0)` and the re-send does emit.
                    if p.last_emit != (0, 0) {
                        return (Some(p.silence(pad)), None);
                    }
                    continue;
                }
                if p.quirks.keepalive_ms > 0 {
                    p.next_keepalive =
                        Some(now + Duration::from_millis(p.quirks.keepalive_ms as u64));
                }
-                let (low, high) = p.level;
+                return (Some(p.emit(pad)), None);
                return (
                    Some(RumbleCommand {
                        pad,
                        low,
                        high,
                        backstop_ms: p.backstop(),
                    }),
                    None,
                );
            }
            // 4) actuator-decay keepalive, bounded by (1)/(2) above by construction: an expired
            // or stale pad was silenced before reaching here, so a keepalive can never sustain a
@@ -239,20 +298,7 @@ impl RumbleEngine {
                let due = *p.next_keepalive.get_or_insert(now + ka);
                if now >= due {
                    p.next_keepalive = Some(now + ka);
-                    let (mut low, high) = p.level;
+                    return (Some(p.emit(pad)), None);
                    if p.quirks.dedup_jitter {
                        p.jitter = !p.jitter;
                        low ^= p.jitter as u16;
                    }
                    return (
                        Some(RumbleCommand {
                            pad,
                            low,
                            high,
                            backstop_ms: p.backstop(),
                        }),
                        None,
                    );
                }
                merge_wake(&mut wake, due);
            }
@@ -357,6 +403,22 @@ pub(crate) struct Closed;
 mod tests {
    use super::*;
    /// The Steam Deck's declared quirks — the only shipping actuator with `dedup_jitter`.
    const DECK: ActuatorQuirks = ActuatorQuirks {
        keepalive_ms: 40,
        min_pulse_ms: 0,
        dedup_jitter: true,
    };
    /// Drain the engine the way an embedder does: poll until nothing is due.
    fn drain(e: &mut RumbleEngine, t: Instant) -> Vec<(u16, u16)> {
        let mut out = Vec::new();
        while let (Some(c), _) = e.poll(t) {
            out.push((c.low, c.high));
        }
        out
    }
    fn ms(v: u64) -> Duration {
        Duration::from_millis(v)
    }
@@ -527,4 +589,133 @@ mod tests {
        );
        assert_eq!(shared.next_command(ms(10)), Err(Closed));
    }
    /// A host renewal must not repeat the value the device last took, or an SDL-class layer
    /// swallows the write. Before the jitter moved onto every emit path it lived only in the
    /// keepalive branch, so each renewal collided with the last jittered write and was deduped.
    #[test]
    fn renewal_keeps_the_dedupe_jitter_alternating() {
        let mut e = RumbleEngine::new();
        e.set_quirks(0, DECK);
        let t0 = Instant::now();
        e.wire_update(t0, 0, 100, 200, Some(400));
        assert_eq!(drain(&mut e, t0), vec![(100, 200)]);
        assert_eq!(drain(&mut e, t0 + ms(40)), vec![(101, 200)]);
        assert_eq!(drain(&mut e, t0 + ms(80)), vec![(100, 200)]);
        // The renewal at the 120 ms default cadence: same level, must still be a distinct write.
        e.wire_update(t0 + ms(120), 0, 100, 200, Some(400));
        assert_eq!(drain(&mut e, t0 + ms(120)), vec![(101, 200)]);
        assert_eq!(drain(&mut e, t0 + ms(160)), vec![(100, 200)]);
    }
    /// Phase-robust version of the same property, at the TTL hatch's 60 ms renewal floor: no two
    /// consecutive DISTINCT device writes may be further apart than the declared 40 ms cadence.
    #[test]
    fn renewal_never_gaps_distinct_writes_at_the_60ms_floor() {
        let mut e = RumbleEngine::new();
        e.set_quirks(0, DECK);
        let t0 = Instant::now();
        let (mut last, mut last_write, mut worst) = ((0u16, 0u16), 0u64, 0u64);
        for tick in 0..=360u64 {
            let t = t0 + ms(tick);
            if tick % 60 == 0 {
                e.wire_update(t, 0, 100, 200, Some(400));
            }
            for v in drain(&mut e, t) {
                assert_ne!(v, (0, 0), "a live lease must never emit the stop sentinel");
                if v != last {
                    worst = worst.max(tick - last_write);
                    last_write = tick;
                    last = v;
                }
            }
        }
        assert!(
            worst <= 41,
            "worst distinct-write gap {worst} ms exceeds the 40 ms declared cadence"
        );
    }
    /// The nudge must stay behind `dedup_jitter`: an off-by-one amplitude on a default-quirks pad
    /// would land in Apple's identical-target comparison and Android's one-shot amplitudes.
    #[test]
    fn default_quirks_pads_get_the_level_verbatim_on_every_renewal() {
        let mut e = RumbleEngine::new(); // Apple / Android / plain SDL
        let t0 = Instant::now();
        e.wire_update(t0, 0, 100, 200, Some(400));
        assert_eq!(e.poll(t0).0, Some(cmd(0, 100, 200, 800)));
        e.wire_update(t0 + ms(120), 0, 100, 200, Some(400));
        assert_eq!(e.poll(t0 + ms(120)).0, Some(cmd(0, 100, 200, 800)));
    }
    /// Level `(1, 0)` is the one value whose LSB flip is the reserved stop. The nudge steps up
    /// instead, so the phase still alternates and no stop is invented under a live lease.
    #[test]
    fn jitter_never_synthesizes_the_stop_sentinel() {
        let mut e = RumbleEngine::new();
        e.set_quirks(0, DECK);
        let t0 = Instant::now();
        e.wire_update(t0, 0, 1, 0, Some(400));
        assert_eq!(e.poll(t0).0, Some(cmd(0, 1, 0, 800)));
        assert_eq!(e.poll(t0 + ms(40)).0, Some(cmd(0, 3, 0, 800)));
        assert_eq!(e.poll(t0 + ms(80)).0, Some(cmd(0, 1, 0, 800)));
    }
    /// A zero for a pad the engine already believes is silent is dropped: it heals nothing and
    /// costs every embedder a command. The deliberate stop-burst heal is unaffected, because a
    /// LOST stop leaves the pad buzzing and the re-send therefore does emit.
    #[test]
    fn a_redundant_stop_is_dropped_but_the_burst_still_heals_a_lost_one() {
        let mut e = RumbleEngine::new();
        let t0 = Instant::now();
        e.wire_update(t0, 0, 100, 200, Some(400));
        assert_eq!(drain(&mut e, t0), vec![(100, 200)]);
        // First stop reaches the embedder…
        e.wire_update(t0 + ms(10), 0, 0, 0, Some(0));
        assert_eq!(drain(&mut e, t0 + ms(10)), vec![(0, 0)]);
        // …and the burst re-sends behind it are now silent.
        e.wire_update(t0 + ms(20), 0, 0, 0, Some(0));
        e.wire_update(t0 + ms(30), 0, 0, 0, Some(0));
        assert_eq!(drain(&mut e, t0 + ms(30)), Vec::new());
        // But if the pad is buzzing (the stop that mattered was lost), a re-send still emits.
        e.wire_update(t0 + ms(40), 0, 100, 200, Some(400));
        assert_eq!(drain(&mut e, t0 + ms(40)), vec![(100, 200)]);
        e.wire_update(t0 + ms(50), 0, 0, 0, Some(0));
        assert_eq!(drain(&mut e, t0 + ms(50)), vec![(0, 0)]);
    }
    /// The client bounds the host's lease. `RUMBLE_TTL_CEIL_MS` is sender-side only, so a modified
    /// or third-party host could otherwise stamp a huge TTL and wedge its pump, leaving Apple and
    /// the Deck buzzing for the whole of it.
    #[test]
    fn an_overlong_lease_is_clamped_to_the_ceiling() {
        let mut e = RumbleEngine::new();
        let t0 = Instant::now();
        e.wire_update(t0, 0, 100, 200, Some(u16::MAX));
        assert_eq!(e.poll(t0).0, Some(cmd(0, 100, 200, 5000)));
        // Silenced at the ceiling, not at the 65 s the sender asked for.
        assert!(e.poll(t0 + ms(MAX_LEASE_MS as u64 - 1)).0.is_none());
        assert_eq!(
            e.poll(t0 + ms(MAX_LEASE_MS as u64)).0,
            Some(cmd(0, 0, 0, 0)),
            "the lease must end at the ceiling"
        );
    }
    /// A v2 envelope carrying `ttl_ms == 0` on a LIVE level. The audit suspected the zero would be
    /// mistaken for the legacy sentinel in `backstop()`; it cannot, because the expiry check
    /// preempts the relay branch — the pad silences on the same poll and never reaches a backstop.
    /// Pinned so that ordering stays load-bearing rather than incidental.
    #[test]
    fn a_zero_ttl_envelope_silences_rather_than_taking_the_legacy_backstop() {
        let mut e = RumbleEngine::new();
        let t0 = Instant::now();
        e.wire_update(t0, 0, 100, 200, Some(0));
        assert_eq!(
            e.poll(t0).0,
            Some(cmd(0, 0, 0, 0)),
            "a zero-length lease must expire immediately, not emit with a legacy backstop"
        );
    }
 }