feat(android): native Android client — full app parity build (A1–A36 + S1)

Mirror the iOS P0+P1 client as a Gradle multi-module app. Pure-JVM (Kover ≥80% gated):
:wire-protocol (frozen contract + HostEndpoint CSWSH origin + byte-exact codec),
:session-core (SessionEngine + reconnect/ping/gate/digest reducers), :api-client,
:client-tls (pure PKCS12/keymanager), :transport-okhttp (OkHttp WS+REST). Framework:
:app (Compose M3 Adaptive, Hilt, FCM, 11 screens + NavGraph), :terminal-view (Termux
terminal-emulator/-view via JitPack, Apache-2.0 — the renderer seam proven headless),
:host-registry (DataStore), :client-tls-android (AndroidKeyStore + Tink).

Highlights: single-key-home mTLS (non-exportable AndroidKeyStore key + re-reading
X509KeyManager + connectionPool.evictAll on rotation, ping-pong single-commit so a failed
rotation never clobbers the prior identity); FCM Allow/Deny trust split (Deny=BroadcastReceiver,
Allow=trampoline Activity hosting BiometricPrompt); per-consumer-Channel EventBus (R10);
config-surviving RetainedSessionHolder; byte-exact KeyByteMap; §5.4 pairing warning tiers.

Verified: ~484 JVM tests + Kover ≥80% on the pure modules + :app assembles to an APK; all
framework modules assemble. Device behaviors (rendering/IME/FCM/biometric/camera, E2E A34/A35,
S2 real-handset FCM spike) deferred to android/DEVICE_QA_CHECKLIST.md per plan §7 (no
emulator/Firebase here). Built via multi-agent orchestration (TDD builders → adversarial
cross-review → fix → re-verify → independent gate); progress in android/PROGRESS_ANDROID.md.

android/session-core/src/main/kotlin/wang/yaojia/webterm/session/SessionEngine.kt

@@ -0,0 +1,415 @@
+package wang.yaojia.webterm.session
+
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withTimeoutOrNull
+import wang.yaojia.webterm.wire.ClientMessage
+import wang.yaojia.webterm.wire.ConnectionPinger
+import wang.yaojia.webterm.wire.HostEndpoint
+import wang.yaojia.webterm.wire.MessageCodec
+import wang.yaojia.webterm.wire.PingableTermTransport
+import wang.yaojia.webterm.wire.ServerMessage
+import wang.yaojia.webterm.wire.TermTransport
+import wang.yaojia.webterm.wire.TimelineEvent
+import wang.yaojia.webterm.wire.TransportConnection
+import wang.yaojia.webterm.wire.Tunables
+
+/**
+ * The session lifecycle state machine (A14, plan §5 / §6.6). The Android analogue of the iOS
+ * `SessionCore.SessionEngine`, and the composition root that drives the pure reducers
+ * ([ReconnectMachine], [PingScheduler], [GateTracker], [AwayDigest]) against an injected
+ * [TermTransport].
+ *
+ * PURE / TESTABLE (invariant #4 — the confinement contract): the engine imports NO OkHttp/Android
+ * types. ALL engine state ([reconnect], [gateTracker], [currentConnection], [generation], …) is
+ * touched ONLY on the injected [scope]'s dispatcher — a `Dispatchers.Default.limitedParallelism(1)`
+ * in production, a `StandardTestDispatcher` under `runTest`. The only cross-boundary calls are the
+ * event fan-out ([events], engine→UI) and [send]/[decideGate]/[notifyForegrounded]/[close]
+ * (UI→engine), each of which hops onto [scope] before touching state. Because the dispatcher is
+ * single-threaded the fields need no locks.
+ *
+ * The load-bearing invariants (plan §1):
+ *  - **attach-first (#2):** [ClientMessage.Attach] is the FIRST frame sent on EVERY (re)connect,
+ *    with the `sessionId` key always present (JSON `null` for a new session), sent BEFORE the
+ *    connection is published so a queued UI send can never overtake it.
+ *  - **adopt-server-id:** every `attached` frame adopts the server-issued id ([ReconnectMachine.withSessionId]),
+ *    so the next reconnect re-attaches to the same PTY and the ring buffer replays.
+ *  - **close == detach, NEVER kill (#2):** [close] closes the WS (server PTY survives) and stops
+ *    reconnecting; it never sends a kill. A [close] that lands while a fresh connection is still
+ *    inside the dial/attach window (before it is published to [currentConnection]) still detaches
+ *    that connection — it is never leaked (see [runOneConnection]).
+ *  - **generation-safe cancellation (defensive-only):** [generation] bumps on each (re)connect.
+ *    Under the strictly-sequential supervisor no two connections ever run at once, so the
+ *    `gen != generation` drop in [onFrame] is belt-and-braces — a superseded connection's late
+ *    frame would be dropped rather than emitted onto a newer generation.
+ *  - **oversized replay is terminal:** a frame past [maxFrameBytes] is the non-retryable
+ *    [FailureReason.REPLAY_TOO_LARGE] — distinct from a retryable disconnect (never fed to backoff).
+ */
+public class SessionEngine(
+    private val transport: TermTransport,
+    private val endpoint: HostEndpoint,
+    private val scope: CoroutineScope,
+    initialSessionId: String? = null,
+    private val initialCwd: String? = null,
+    private val awayTimeline: suspend (sinceMs: Long) -> List<TimelineEvent> = { emptyList() },
+    private val nowMs: () -> Long = { System.currentTimeMillis() },
+    private val pingScheduler: PingScheduler = PingScheduler(),
+    private val maxFrameBytes: Long = Tunables.MAX_WS_MESSAGE_BYTES,
+    private val digestLimit: Int = DEFAULT_DIGEST_LIMIT,
+) {
+    /** A retryable end (drop/clean-server-close/ping-loss) vs a non-retryable [Terminal] one. */
+    private sealed interface EndCause {
+        /** The WS dropped while we were connected — reconnect with back-off. */
+        data object Disconnected : EndCause
+
+        /** `exit` or oversized-replay: the terminal event was already emitted; do NOT reconnect. */
+        data object Terminal : EndCause
+    }
+
+    /** One live connection plus its optional ping capability (null = plain [TermTransport]). */
+    private class OpenConn(val connection: TransportConnection, val pinger: ConnectionPinger?)
+
+    private val eventsChannel = Channel<SessionEvent>(Channel.UNLIMITED)
+
+    /** Engine→UI event stream (single consumer; the App's EventBus fans out per R10). */
+    public val events: Flow<SessionEvent> = eventsChannel.receiveAsFlow()
+
+    // ── engine-confined state (touched only on [scope]) ───────────────────────────────────
+    private var reconnect: ReconnectMachine = ReconnectMachine.initial.withSessionId(initialSessionId)
+    private var gateTracker: GateTracker = GateTracker.INITIAL
+    private var currentConnection: TransportConnection? = null
+    private var lastDims: Pair<Int, Int>? = null
+    private var generation: Int = 0
+    private var reconnectAttempt: Int = 0
+    private var pendingDigest: Boolean = false
+    private var disconnectedAtMs: Long? = null
+    private var retryWakeup: CompletableDeferred<Unit>? = null
+    private var started: Boolean = false
+    private var closed: Boolean = false
+
+    /** Begin connecting. Idempotent — a second call is a no-op. */
+    public fun start() {
+        scope.launch {
+            // Idempotency flip happens ON the confined dispatcher (invariant #4): [started] is engine
+            // state and must never be touched on the caller's thread.
+            if (started) return@launch
+            started = true
+            try {
+                supervisor()
+            } finally {
+                eventsChannel.close()
+            }
+        }
+    }
+
+    /**
+     * Send a UI-originated frame (input / resize / approve / reject) on the live connection.
+     * Dropped when disconnected (a `resize` is re-sent from [lastDims] on the next attach; input
+     * while detached is meaningless). Encoding is byte-exact via [MessageCodec.encode].
+     */
+    public fun send(message: ClientMessage) {
+        scope.launch { doSend(message) }
+    }
+
+    /**
+     * Resolve a held gate ONLY if [epoch] still matches the current gate ([GateTracker.canDecide]);
+     * a stale epoch is DROPPED so a slow tap can never approve the NEXT gate (two-line stale-guard).
+     */
+    public fun decideGate(epoch: Int, message: ClientMessage) {
+        scope.launch { handleDecideGate(epoch, message) }
+    }
+
+    /**
+     * The app returned to the foreground (plan §6.4/§6.6). If connected, re-send [lastDims] to
+     * reclaim full-screen (latest-writer-wins). If in back-off, connect NOW without resetting the
+     * ladder (no-reset-on-foreground, [ReconnectMachine]). [cols]/[rows] update the remembered dims.
+     */
+    public fun notifyForegrounded(cols: Int? = null, rows: Int? = null) {
+        scope.launch { handleForegrounded(cols, rows) }
+    }
+
+    /**
+     * Detach: close the WS (the server-side PTY keeps running) and stop reconnecting. NEVER a kill.
+     *
+     * Returns the launched [Job] so a caller can AWAIT the clean detach (the RFC-6455 `1000` close
+     * frame handed to the transport writer) BEFORE it cancels the confinement [scope] (FIX 5 —
+     * `RetainedSessionHolder.teardown()`). Fire-and-forget callers simply ignore the result.
+     */
+    public fun close(): Job = scope.launch { handleClose() }
+
+    // ── supervisor: the connect / reconnect loop ──────────────────────────────────────────
+
+    private suspend fun supervisor() {
+        emit(Connection(ConnectionState.Connecting))
+        while (!closed) {
+            val gen = ++generation
+            val cause = runOneConnection(gen)
+            if (closed || cause == EndCause.Terminal) break
+            scheduleAndWaitRetry()
+        }
+    }
+
+    private suspend fun runOneConnection(gen: Int): EndCause {
+        val open = dialOrNull() ?: return EndCause.Disconnected
+        if (closed) return abandon(open) // close() landed during dial → detach, never publish
+        if (!attachFirst(open)) return EndCause.Disconnected
+        if (closed) return abandon(open) // close() landed during attach → detach, never publish
+        currentConnection = open.connection
+        onConnected()
+        val cause = collectConnection(gen, open)
+        currentConnection = null
+        if (cause == EndCause.Disconnected) {
+            // A live connection dropped → the next reattach owes an away-digest.
+            pendingDigest = true
+            disconnectedAtMs = nowMs()
+        }
+        return cause
+    }
+
+    /**
+     * A [close] arrived while this connection was still opening (currentConnection was null, so
+     * [handleClose] could not detach it). Detach it here and end terminally — emit NOTHING further
+     * (no Connected/Adopted, no reconnect): [handleClose] already emitted [ConnectionState.Closed]
+     * and [closed] short-circuits the supervisor. Upholds close == detach, NEVER kill (#2 / §6.6).
+     */
+    private suspend fun abandon(open: OpenConn): EndCause {
+        ignoringNonCancel { open.connection.close() }
+        return EndCause.Terminal
+    }
+
+    private suspend fun dialOrNull(): OpenConn? =
+        try {
+            if (transport is PingableTermTransport) {
+                val pingable = transport.connectPingable(endpoint)
+                OpenConn(pingable.connection, pingable.pinger)
+            } else {
+                OpenConn(transport.connect(endpoint), null)
+            }
+        } catch (e: CancellationException) {
+            throw e
+        } catch (_: Throwable) {
+            null // connect-time failure → retryable
+        }
+
+    /** attach-first (+ replay [lastDims]) BEFORE publishing the connection. */
+    private suspend fun attachFirst(open: OpenConn): Boolean =
+        try {
+            open.connection.send(MessageCodec.encode(ClientMessage.Attach(reconnect.sessionId, cwdForAttach())))
+            lastDims?.let { (cols, rows) ->
+                open.connection.send(MessageCodec.encode(ClientMessage.Resize(cols, rows)))
+            }
+            true
+        } catch (e: CancellationException) {
+            throw e
+        } catch (_: Throwable) {
+            ignoringNonCancel { open.connection.close() }
+            false
+        }
+
+    /** `cwd` rides the attach only for a brand-new session (server appends it iff sessionId is null). */
+    private fun cwdForAttach(): String? = if (reconnect.sessionId == null) initialCwd else null
+
+    private fun onConnected() {
+        reconnect = reconnect.reduce(ReconnectMachine.Input.Connected).first // reset ladder, keep id
+        reconnectAttempt = 0
+        emit(Connection(ConnectionState.Connected))
+    }
+
+    private suspend fun scheduleAndWaitRetry() {
+        val (next, effect) = reconnect.reduce(ReconnectMachine.Input.Disconnected)
+        reconnect = next
+        reconnectAttempt += 1
+        val after = (effect as ReconnectMachine.Effect.ScheduleRetry).after
+        emit(Connection(ConnectionState.Reconnecting(reconnectAttempt, after)))
+        val wake = CompletableDeferred<Unit>()
+        retryWakeup = wake
+        // Woken early by foreground/close → connect now; else the timer elapses → connect now.
+        withTimeoutOrNull(after) { wake.await() }
+        retryWakeup = null
+    }
+
+    // ── one connection's lifetime: frames + keep-alive ping ────────────────────────────────
+
+    private suspend fun collectConnection(gen: Int, open: OpenConn): EndCause {
+        var terminal: EndCause? = null
+        return try {
+            coroutineScope {
+                val pingJob = open.pinger?.let { launchPing(open.connection, it) }
+                try {
+                    open.connection.frames.collect { frame ->
+                        onFrame(gen, frame)?.let { cause ->
+                            terminal = cause
+                            ignoringNonCancel { open.connection.close() } // ends the frames flow → collect returns
+                        }
+                    }
+                } finally {
+                    pingJob?.cancel()
+                }
+                terminal ?: EndCause.Disconnected // clean finish with no exit frame = server-side close
+            }
+        } catch (e: CancellationException) {
+            throw e
+        } catch (_: Throwable) {
+            EndCause.Disconnected // transport error → retryable
+        }
+    }
+
+    private fun CoroutineScope.launchPing(conn: TransportConnection, pinger: ConnectionPinger): Job =
+        launch {
+            if (pingScheduler.run { pinger.ping() } == PingScheduler.Outcome.CONNECTION_LOST) {
+                ignoringNonCancel { conn.close() } // declare dead → ends frames → collectConnection = Disconnected
+            }
+        }
+
+    /** Handle one server frame; returns a terminal [EndCause] to stop the connection, else null. */
+    private fun onFrame(gen: Int, frame: String): EndCause? {
+        if (gen != generation) return null // defensive-only: the sequential supervisor never overlaps generations
+        if (utf8Size(frame) > maxFrameBytes) {
+            emit(Connection(ConnectionState.Failed(FailureReason.REPLAY_TOO_LARGE)))
+            return EndCause.Terminal
+        }
+        return when (val msg = MessageCodec.decodeServer(frame)) {
+            null -> null // undecodable → drop
+            is ServerMessage.Attached -> { onAttached(msg.sessionId); null }
+            is ServerMessage.Output -> { emit(Output(msg.data)); null }
+            is ServerMessage.Status -> { onStatus(msg); null }
+            is ServerMessage.Telemetry -> { emit(Telemetry(msg.telemetry)); null }
+            is ServerMessage.Exit -> {
+                emit(Exited(msg.code, msg.reason))
+                EndCause.Terminal
+            }
+        }
+    }
+
+    private fun onAttached(sessionId: String) {
+        reconnect = reconnect.withSessionId(sessionId) // always adopt the server-issued id
+        emit(Adopted(sessionId))
+        if (pendingDigest) {
+            pendingDigest = false
+            disconnectedAtMs?.let { since -> scope.launch { emitDigest(since) } }
+        }
+    }
+
+    private fun onStatus(status: ServerMessage.Status) {
+        val previous = gateTracker.current
+        gateTracker = gateTracker.reduce(status.pending, status.gate, status.detail)
+        val next = gateTracker.current
+        if (next != previous) emit(Gate(next))
+    }
+
+    private suspend fun emitDigest(sinceMs: Long) {
+        val timeline = try {
+            awayTimeline(sinceMs)
+        } catch (e: CancellationException) {
+            throw e // never swallow cancellation (FIX 3)
+        } catch (_: Throwable) {
+            emptyList()
+        }
+        emit(Digest(AwayDigest.reduce(timeline, sinceMs, digestLimit))) // EMPTY still delivered; UI suppresses
+    }
+
+    // ── UI→engine command handlers (all on [scope]) ────────────────────────────────────────
+
+    private suspend fun doSend(message: ClientMessage) {
+        if (message is ClientMessage.Resize) lastDims = message.cols to message.rows
+        currentConnection?.let { conn -> ignoringNonCancel { conn.send(MessageCodec.encode(message)) } }
+    }
+
+    private suspend fun handleDecideGate(epoch: Int, message: ClientMessage) {
+        if (!gateTracker.canDecide(epoch)) return // stale epoch or no gate → drop (防误批新 gate)
+        doSend(message)
+        retireHeldGate() // a decision resolves the gate → a second same-epoch tap must NOT double-send
+    }
+
+    /**
+     * Optimistically clear the held gate after a decision was sent (mirrors the web client setting
+     * `pendingApprovalValue=false` inside `approve()`/`reject()`, public/terminal-session.ts). Folds a
+     * synthetic falling edge through [GateTracker] so `canDecide(sameEpoch)` returns false afterwards;
+     * [lastEpoch] is retained so the server's next rising edge still mints a fresh epoch. The server's
+     * own next `pending=false` status is then a no-op (state already null → no duplicate [Gate] event).
+     */
+    private fun retireHeldGate() {
+        if (gateTracker.current == null) return
+        gateTracker = gateTracker.reduce(pending = false, gate = null, detail = null)
+        emit(Gate(gateTracker.current)) // null → the UI hides the approve/reject affordances immediately
+    }
+
+    private suspend fun handleForegrounded(cols: Int?, rows: Int?) {
+        if (closed) return
+        if (cols != null && rows != null) lastDims = cols to rows
+        val conn = currentConnection
+        if (conn != null) {
+            // Connected: reclaim full-screen by re-sending the remembered dims (latest-writer-wins).
+            lastDims?.let { (c, r) -> ignoringNonCancel { conn.send(MessageCodec.encode(ClientMessage.Resize(c, r))) } }
+        } else {
+            // In back-off / connecting: connect NOW without touching the ladder (no-reset-on-foreground).
+            retryWakeup?.complete(Unit)
+        }
+    }
+
+    private suspend fun handleClose() {
+        if (closed) return
+        closed = true
+        retryWakeup?.complete(Unit) // break any back-off wait so the supervisor exits
+        val conn = currentConnection
+        currentConnection = null
+        conn?.let { ignoringNonCancel { it.close() } } // detach — the server PTY keeps running
+        emit(Connection(ConnectionState.Closed))
+    }
+
+    private fun emit(event: SessionEvent) {
+        eventsChannel.trySend(event) // UNLIMITED → never fails while open; no-op after close
+    }
+
+    /**
+     * Run a best-effort suspend [block] (transport teardown / send), swallowing ordinary failures
+     * but RETHROWING [CancellationException] so structured-concurrency cancellation is never
+     * silently absorbed (FIX 3). Inline so the suspend calls run in the caller's coroutine.
+     */
+    private suspend inline fun ignoringNonCancel(block: () -> Unit) {
+        try {
+            block()
+        } catch (e: CancellationException) {
+            throw e
+        } catch (_: Throwable) {
+            // best-effort: a failed close/send is non-fatal to the engine's lifecycle
+        }
+    }
+
+    /**
+     * UTF-8 byte length of [text], short-circuiting the moment it passes [maxFrameBytes] so a
+     * multi-MB oversized replay is not fully scanned. Cheaper than allocating a `ByteArray` for
+     * every normal frame.
+     */
+    private fun utf8Size(text: String): Long {
+        var bytes = 0L
+        var i = 0
+        while (i < text.length) {
+            val code = text[i].code
+            bytes += when {
+                code < 0x80 -> 1
+                code < 0x800 -> 2
+                code in 0xD800..0xDBFF && i + 1 < text.length && text[i + 1].code in 0xDC00..0xDFFF -> {
+                    i++ // consume the low surrogate: a full pair encodes to 4 bytes
+                    4
+                }
+                else -> 3
+            }
+            if (bytes > maxFrameBytes) return bytes // over the cap — exact size no longer matters
+            i++
+        }
+        return bytes
+    }
+
+    public companion object {
+        /** Max [AwayDigest.recent] entries carried in the once-per-reconnect digest. */
+        public const val DEFAULT_DIGEST_LIMIT: Int = 20
+    }
+}

android/session-core/src/test/kotlin/wang/yaojia/webterm/session/SessionEngineTest.kt

@@ -0,0 +1,468 @@
+package wang.yaojia.webterm.session
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import wang.yaojia.webterm.testsupport.FakePingableTermTransport
+import wang.yaojia.webterm.testsupport.FakeTermTransport
+import wang.yaojia.webterm.testsupport.FakeTimeSource
+import wang.yaojia.webterm.wire.ClientMessage
+import wang.yaojia.webterm.wire.HostEndpoint
+import wang.yaojia.webterm.wire.MessageCodec
+import wang.yaojia.webterm.wire.TermTransport
+import wang.yaojia.webterm.wire.TimelineEvent
+import wang.yaojia.webterm.wire.TransportConnection
+import wang.yaojia.webterm.wire.Tunables
+import kotlin.time.ExperimentalTime
+import kotlin.time.Duration.Companion.seconds
+
+/**
+ * A14 verification (plan §5 "A14 Verify"): the [SessionEngine] state machine driven purely through
+ * the [FakeTermTransport]/[FakePingableTermTransport] doubles and the coroutine virtual clock under
+ * `runTest`. No OkHttp, no Android, no wall-clock waits.
+ */
+class SessionEngineTest {
+
+    private companion object {
+        /** A valid lower-case UUID-v4 the server can issue on `attached`. */
+        const val SERVER_ID = "11111111-1111-4111-8111-111111111111"
+        const val ATTACH_NEW = """{"type":"attach","sessionId":null}"""
+        const val APPROVE = """{"type":"approve"}"""
+    }
+
+    private fun endpoint(): HostEndpoint = HostEndpoint.fromBaseUrl("http://localhost:3000")!!
+
+    private fun attachWith(id: String): String = MessageCodec.encode(ClientMessage.Attach(id))
+
+    private fun attachedFrame(id: String): String = """{"type":"attached","sessionId":"$id"}"""
+
+    private fun statusFrame(pending: Boolean, gate: String? = null): String {
+        val gatePart = if (gate != null) ""","gate":"$gate"""" else ""
+        return """{"type":"status","status":"working","pending":$pending$gatePart}"""
+    }
+
+    private fun TestScope.newEngine(
+        transport: TermTransport,
+        initialSessionId: String? = null,
+        initialCwd: String? = null,
+        maxFrameBytes: Long = Tunables.MAX_WS_MESSAGE_BYTES,
+        awayTimeline: suspend (Long) -> List<TimelineEvent> = { emptyList() },
+        nowMs: () -> Long = { 0L },
+    ): SessionEngine = SessionEngine(
+        transport = transport,
+        endpoint = endpoint(),
+        scope = backgroundScope,
+        initialSessionId = initialSessionId,
+        initialCwd = initialCwd,
+        awayTimeline = awayTimeline,
+        nowMs = nowMs,
+        maxFrameBytes = maxFrameBytes,
+    )
+
+    /** Live-updating list of everything the engine emits (single consumer, buffered channel). */
+    private fun TestScope.collectEvents(engine: SessionEngine): List<SessionEvent> {
+        val out = mutableListOf<SessionEvent>()
+        backgroundScope.launch { engine.events.collect { out += it } }
+        return out
+    }
+
+    private fun List<SessionEvent>.connectionStates(): List<ConnectionState> =
+        filterIsInstance<Connection>().map { it.state }
+
+    // ── attach-first ordering ──────────────────────────────────────────────────────────────
+
+    @Test
+    fun `attach is the first frame on connect with an explicit null sessionId for a new session`() = runTest {
+        val transport = FakeTermTransport()
+        val engine = newEngine(transport)
+        collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent()
+
+        assertEquals(listOf(ATTACH_NEW), transport.sentFramesByConnection[0])
+    }
+
+    @Test
+    fun `on reconnect attach is re-sent first carrying the adopted server id`() = runTest {
+        val transport = FakeTermTransport()
+        val engine = newEngine(transport)
+        collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent()
+        transport.emit(attachedFrame(SERVER_ID)) // server confirms + issues the id
+        testScheduler.runCurrent()
+
+        transport.emitError(RuntimeException("ws dropped")) // conn 0 drops → back-off 1s
+        testScheduler.runCurrent()
+        testScheduler.advanceTimeBy(1.seconds)
+        testScheduler.runCurrent()
+
+        assertEquals(attachWith(SERVER_ID), transport.sentFramesByConnection[1].first())
+    }
+
+    // ── adopt-server-id ─────────────────────────────────────────────────────────────────────
+
+    @Test
+    fun `every attached frame is adopted and surfaced as an Adopted event`() = runTest {
+        val transport = FakeTermTransport()
+        val engine = newEngine(transport)
+        val events = collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent()
+        transport.emit(attachedFrame(SERVER_ID))
+        testScheduler.runCurrent()
+
+        assertTrue(events.contains(Adopted(SERVER_ID)), "expected an Adopted($SERVER_ID) event, got $events")
+    }
+
+    // ── connect-now-on-foreground (and no-reset-on-foreground) ──────────────────────────────
+
+    @Test
+    fun `foregrounding connects immediately without resetting the back-off ladder`() = runTest {
+        val transport = FakeTermTransport()
+        val engine = newEngine(transport)
+        val events = collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent() // conn 0 connected
+        transport.emitError(RuntimeException("drop")) // → Reconnecting(1, 1s), waiting
+        testScheduler.runCurrent()
+
+        transport.scriptConnectFailure() // the foreground connect will fail so we can read the ladder
+        engine.notifyForegrounded()
+        testScheduler.runCurrent() // wake → connect NOW (no 1s advance)
+
+        assertEquals(2, transport.connectAttempts.size) // connected immediately, not after 1s
+        val reconnecting = events.connectionStates().filterIsInstance<ConnectionState.Reconnecting>()
+        assertEquals(2.seconds, reconnecting.last().next) // ladder advanced to 2s, NOT reset to 1s
+    }
+
+    // ── close == detach, NEVER kill ─────────────────────────────────────────────────────────
+
+    @Test
+    fun `close detaches the connection without a kill and never reconnects`() = runTest {
+        val transport = FakeTermTransport()
+        val engine = newEngine(transport)
+        val events = collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent()
+        engine.close()
+        testScheduler.runCurrent()
+
+        assertEquals(1, transport.closeCallCount) // exactly one detach
+        assertEquals(listOf(ATTACH_NEW), transport.sentFrames) // only attach was ever sent — no kill frame
+        assertTrue(events.contains(Connection(ConnectionState.Closed)))
+
+        testScheduler.advanceTimeBy(60.seconds) // no back-off reconnect after a client close
+        testScheduler.runCurrent()
+        assertEquals(1, transport.connectAttempts.size)
+    }
+
+    // ── oversized-replay is a terminal, non-retryable failure ───────────────────────────────
+
+    @Test
+    fun `an oversized frame yields replayTooLarge and never reconnects`() = runTest {
+        val transport = FakeTermTransport()
+        val engine = newEngine(transport, maxFrameBytes = 32)
+        val events = collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent()
+        transport.emit("x".repeat(100)) // 100 bytes > 32-byte cap
+        testScheduler.runCurrent()
+
+        assertTrue(
+            events.contains(Connection(ConnectionState.Failed(FailureReason.REPLAY_TOO_LARGE))),
+            "expected a REPLAY_TOO_LARGE failure, got $events",
+        )
+        // Distinct from a retryable disconnect: no Reconnecting, no second connect.
+        assertFalse(events.connectionStates().any { it is ConnectionState.Reconnecting })
+        testScheduler.advanceTimeBy(60.seconds)
+        testScheduler.runCurrent()
+        assertEquals(1, transport.connectAttempts.size)
+    }
+
+    // ── gate-decision epoch drop (two-line stale-guard) ─────────────────────────────────────
+
+    @Test
+    fun `a decision against a stale gate epoch is dropped while a live epoch sends`() = runTest {
+        val transport = FakeTermTransport()
+        val engine = newEngine(transport)
+        val events = collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent()
+
+        transport.emit(statusFrame(pending = true, gate = "tool")) // rising edge → epoch 1
+        testScheduler.runCurrent()
+        val epoch1 = events.filterIsInstance<Gate>().last().gate!!.epoch
+        assertEquals(1, epoch1)
+
+        engine.decideGate(epoch1, ClientMessage.Approve(null)) // live epoch → sends
+        testScheduler.runCurrent()
+
+        transport.emit(statusFrame(pending = false)) // gate lifted
+        testScheduler.runCurrent()
+        transport.emit(statusFrame(pending = true, gate = "tool")) // rising edge → epoch 2
+        testScheduler.runCurrent()
+
+        engine.decideGate(epoch1, ClientMessage.Approve(null)) // stale epoch → dropped
+        testScheduler.runCurrent()
+
+        assertEquals(1, transport.sentFrames.count { it == APPROVE }) // exactly one approve reached the wire
+    }
+
+    @Test
+    fun `a second decision against the same live gate epoch does not double-send`() = runTest {
+        val transport = FakeTermTransport()
+        val engine = newEngine(transport)
+        val events = collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent()
+        transport.emit(statusFrame(pending = true, gate = "tool")) // rising edge → epoch 1, gate held
+        testScheduler.runCurrent()
+        val epoch1 = events.filterIsInstance<Gate>().last().gate!!.epoch
+
+        engine.decideGate(epoch1, ClientMessage.Approve(null)) // first tap → sends, retires the gate
+        testScheduler.runCurrent()
+        engine.decideGate(epoch1, ClientMessage.Approve(null)) // second same-epoch tap BEFORE any status update
+        testScheduler.runCurrent()
+
+        assertEquals(1, transport.sentFrames.count { it == APPROVE }) // exactly ONE approve reached the wire
+    }
+
+    // ── close during the dial/attach window still detaches (close == detach, #2 / §6.6) ─────
+
+    @Test
+    fun `close during the dial window detaches the freshly-opened connection and emits nothing further`() = runTest {
+        val gate = CompletableDeferred<Unit>()
+        val transport = GatedConnectTransport(connectGate = gate)
+        val engine = newEngine(transport)
+        val events = collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent() // supervisor runs; connect() suspends on the gate (currentConnection is null)
+
+        engine.close() // close lands DURING the dial window
+        testScheduler.runCurrent()
+
+        gate.complete(Unit) // the connection now opens, after close() already fired
+        testScheduler.runCurrent()
+
+        assertTrue(transport.connection.isClosed, "the connection opened during close() must be detached")
+        val states = events.connectionStates()
+        assertFalse(states.contains(ConnectionState.Connected), "must not emit Connected after close, got $states")
+        assertFalse(events.any { it is Adopted }, "must not adopt a server id after close, got $events")
+        assertEquals(1, transport.connectCount) // no reconnect scheduled after a client close
+    }
+
+    @Test
+    fun `close during the attach window detaches the freshly-opened connection and emits nothing further`() = runTest {
+        val sendGate = CompletableDeferred<Unit>()
+        val transport = GatedConnectTransport(sendGate = sendGate) // connect succeeds; the first send suspends
+        val engine = newEngine(transport)
+        val events = collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent() // connect returns; attachFirst's send() suspends on the gate
+
+        engine.close() // close lands DURING the attach window
+        testScheduler.runCurrent()
+
+        sendGate.complete(Unit) // the attach send now completes, after close() already fired
+        testScheduler.runCurrent()
+
+        assertTrue(transport.connection.isClosed, "the connection opened during close() must be detached")
+        val states = events.connectionStates()
+        assertFalse(states.contains(ConnectionState.Connected), "must not emit Connected after close, got $states")
+        assertEquals(1, transport.connectCount) // no reconnect scheduled after a client close
+    }
+
+    // ── back-off ladder 1 → 2 → 4 → 8 → 16 → 30 ─────────────────────────────────────────────
+
+    @Test
+    fun `the reconnect back-off ladder climbs 1 2 4 8 16 30`() = runTest {
+        val transport = FakeTermTransport()
+        val engine = newEngine(transport)
+        val events = collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent() // conn 0 connected
+        repeat(6) { transport.scriptConnectFailure() } // every reconnect attempt fails → ladder climbs
+        transport.emitError(RuntimeException("drop")) // → Reconnecting(1, 1s)
+        testScheduler.runCurrent()
+
+        for (rung in listOf(1, 2, 4, 8, 16, 30)) {
+            testScheduler.advanceTimeBy(rung.seconds)
+            testScheduler.runCurrent()
+        }
+
+        val delays = events.connectionStates()
+            .filterIsInstance<ConnectionState.Reconnecting>()
+            .map { it.next }
+        assertEquals(listOf(1, 2, 4, 8, 16, 30).map { it.seconds }, delays.take(6))
+    }
+
+    // ── ping 25s / 2-miss (only through a PingableTermTransport) ─────────────────────────────
+
+    @Test
+    fun `two consecutive missed pongs on a pingable transport trigger a reconnect`() = runTest {
+        val transport = FakePingableTermTransport()
+        transport.pinger.scriptPongs(false, false) // both pings miss
+        val engine = newEngine(transport)
+        val events = collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent()
+        assertEquals(1, transport.connectPingableCount) // took the pingable path
+        assertEquals(0, transport.pinger.pingCallCount)
+
+        testScheduler.advanceTimeBy(Tunables.PING_INTERVAL) // ping #1 (miss)
+        testScheduler.runCurrent()
+        assertEquals(1, transport.pinger.pingCallCount)
+
+        testScheduler.advanceTimeBy(Tunables.PING_INTERVAL) // ping #2 (miss) → CONNECTION_LOST
+        testScheduler.runCurrent()
+        assertEquals(2, transport.pinger.pingCallCount)
+        assertTrue(events.connectionStates().any { it is ConnectionState.Reconnecting })
+
+        engine.close()
+        testScheduler.runCurrent()
+    }
+
+    @Test
+    fun `a plain non-pingable transport is never pinged`() = runTest {
+        val transport = FakeTermTransport() // does NOT implement PingableTermTransport
+        val engine = newEngine(transport)
+        collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent()
+        // Advance well past several ping intervals — a plain transport must not disconnect on its own.
+        testScheduler.advanceTimeBy(Tunables.PING_INTERVAL * 5)
+        testScheduler.runCurrent()
+
+        assertEquals(1, transport.connectAttempts.size) // still the one live connection, no ping-driven drop
+    }
+
+    // ── away-digest once-per-reconnect (all-zero suppressed by the UI, still delivered here) ──
+
+    @OptIn(ExperimentalTime::class)
+    @Test
+    fun `a completed reconnect emits exactly one away digest reduced since the disconnect`() = runTest {
+        val transport = FakeTermTransport()
+        val clock = FakeTimeSource() // epochMillis starts at 0
+        val timeline = listOf(
+            TimelineEvent(at = 500, eventClass = "tool", label = "ran Bash"),
+            TimelineEvent(at = 600, eventClass = "waiting", label = "needs approval"),
+        )
+        val engine = newEngine(transport, awayTimeline = { timeline }, nowMs = clock::epochMillis)
+        val events = collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent()
+        transport.emit(attachedFrame(SERVER_ID)) // first attach → NO digest
+        testScheduler.runCurrent()
+
+        transport.emitError(RuntimeException("drop")) // disconnect at clock=0 → digest owed
+        testScheduler.runCurrent()
+        testScheduler.advanceTimeBy(1.seconds)
+        testScheduler.runCurrent()
+        transport.emit(attachedFrame(SERVER_ID)) // reconnect adopt → digest fires once
+        testScheduler.runCurrent() // deliver the frame → onAttached launches + runs emitDigest
+
+        val digests = events.filterIsInstance<Digest>()
+        assertEquals(1, digests.size)
+        val digest = digests.single().digest
+        assertEquals(1, digest.toolRuns)
+        assertEquals(1, digest.waitingCount)
+        assertNotNull(digest)
+    }
+
+    // ── output + exit pass-through ──────────────────────────────────────────────────────────
+
+    @Test
+    fun `output frames pass through verbatim and an exit is terminal`() = runTest {
+        val transport = FakeTermTransport()
+        val engine = newEngine(transport)
+        val events = collectEvents(engine)
+
+        engine.start()
+        testScheduler.runCurrent()
+        transport.emit("""{"type":"output","data":"hello"}""")
+        testScheduler.runCurrent()
+        transport.emit("""{"type":"exit","code":-1,"reason":"spawn failed"}""")
+        testScheduler.runCurrent()
+
+        assertTrue(events.contains(Output("hello")))
+        assertTrue(events.contains(Exited(-1, "spawn failed")))
+        // Terminal: no reconnect after an exit.
+        testScheduler.advanceTimeBy(60.seconds)
+        testScheduler.runCurrent()
+        assertEquals(1, transport.connectAttempts.size)
+    }
+}
+
+/**
+ * A [TransportConnection] that records its [close]/[send] and can suspend the FIRST send on a gate,
+ * so a test can wedge `close()` into the attach window (before the connection is published).
+ */
+private class RecordingConnection(private val sendGate: CompletableDeferred<Unit>?) : TransportConnection {
+    private val framesChannel = Channel<String>(Channel.UNLIMITED)
+    override val frames: Flow<String> = framesChannel.receiveAsFlow()
+    val sent: MutableList<String> = mutableListOf()
+
+    @Volatile
+    var isClosed: Boolean = false
+        private set
+
+    private var firstSend = true
+
+    override suspend fun send(frame: String) {
+        if (firstSend) {
+            firstSend = false
+            sendGate?.await() // suspend the attach send → the dispatcher is free to run close()
+        }
+        sent += frame
+    }
+
+    override suspend fun close() {
+        isClosed = true
+        framesChannel.close()
+    }
+}
+
+/**
+ * A [TermTransport] whose [connect] (and/or the connection's first [send]) suspends on a gate, so a
+ * test can fire `close()` while the engine is still inside the dial/attach window (FIX 1 regression).
+ */
+private class GatedConnectTransport(
+    private val connectGate: CompletableDeferred<Unit>? = null,
+    sendGate: CompletableDeferred<Unit>? = null,
+) : TermTransport {
+    val connection: RecordingConnection = RecordingConnection(sendGate)
+
+    @Volatile
+    var connectCount: Int = 0
+        private set
+
+    override suspend fun connect(endpoint: HostEndpoint): TransportConnection {
+        connectCount += 1
+        connectGate?.await() // suspend the dial → the dispatcher is free to run close()
+        return connection
+    }
+}