diff --git a/CHANGELOG.md b/CHANGELOG.md
index 05b582114..0d8ae6ab2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -25,6 +25,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/), and this
 - Android keeps the selected Appearance palette after the startup loading screen instead of leaving parts of the UI dark. (#678)
 
 - Android keeps user-started Dashboard browser sign-in connected through its callback and session verification, then ends the foreground service. Callback pages identify the selected sign-in provider.
+- Android keeps quiet tools and provider waits alive, including turns submitted by another client, and recovers missing completion events from authoritative Gateway state without sending Stop.
 
 ## [0.4.0-beta.10] - 2026-10-08
 
diff --git a/app/src/androidTest/kotlin/com/hermesandroid/relay/viewmodel/GatewayExternalFixtureInstrumentedTest.kt b/app/src/androidTest/kotlin/com/hermesandroid/relay/viewmodel/GatewayExternalFixtureInstrumentedTest.kt
index 2943b90b1..5cda92412 100644
--- a/app/src/androidTest/kotlin/com/hermesandroid/relay/viewmodel/GatewayExternalFixtureInstrumentedTest.kt
+++ b/app/src/androidTest/kotlin/com/hermesandroid/relay/viewmodel/GatewayExternalFixtureInstrumentedTest.kt
@@ -175,6 +175,117 @@ class GatewayExternalFixtureInstrumentedTest {
         assertEquals("2", state["history_rows"].toString())
     }
 
+    @Test
+    fun quietTurns_surviveToolAndProviderSilenceAndRecoverUnsolicitedHistory() {
+        val answer = "Background work finished quietly."
+        val fixtureBaseUrl = InstrumentationRegistry.getArguments()
+            .getString(ARG_FIXTURE_BASE_URL)
+            ?.trim()
+            ?.trimEnd('/')
+        assumeTrue(
+            "Pass -e $ARG_FIXTURE_BASE_URL <url> to run the external fixture lane",
+            !fixtureBaseUrl.isNullOrBlank(),
+        )
+        requireNotNull(fixtureBaseUrl)
+
+        val okHttp = OkHttpClient.Builder()
+            .callTimeout(10, TimeUnit.SECONDS)
+            .build()
+        val initialState = readFixtureJson(okHttp, "$fixtureBaseUrl/__fixture__/state")
+        assertEquals("quiet_turn_recovery", initialState["scenario"]?.jsonString())
+        assertEquals("1", initialState["remaining_turns"].toString())
+        val dashboard = DashboardApiClient(fixtureBaseUrl, okHttp)
+        val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO).also { gatewayScope = it }
+        val gateway = GatewayChatClient(
+            initialDashboardClient = dashboard,
+            okHttpClient = okHttp,
+            callbackDispatcher = { block -> Handler(Looper.getMainLooper()).post(block) },
+            scope = scope,
+            reconnectJitterUnit = { 0.0 },
+            turnIdleTimeoutMs = 200L,
+        ).also { gatewayClient = it }
+        val handler = ChatHandler().also { it.setSessionId(STORED_SESSION_ID) }
+        val vm = ChatViewModel().also {
+            // Deliberately omit HermesApiClient: this lane has no API-server
+            // fallback surface, so a passing turn proves Gateway ownership.
+            it.initialize(null, handler)
+            it.streamingEndpoint = "gateway"
+            it.setProfileMessageLoaderWithMode { profile, sessionId, mode ->
+                dashboard.getSessionMessages(sessionId, profile, mode)
+            }
+            it.updateGatewayClient(gateway)
+            it.setChatVisible(true)
+        }.also { viewModel = it }
+
+        compose.setContent {
+            val messages by vm.messages.collectAsStateWithLifecycle()
+            val streaming by vm.isStreaming.collectAsStateWithLifecycle()
+            MaterialTheme {
+                Column(Modifier.testTag("external-contract-transcript")) {
+                    Text(
+                        text = if (streaming) "STREAMING" else "IDLE",
+                        modifier = Modifier.testTag("external-stream-state"),
+                    )
+                    messages.forEach { message ->
+                        Text(
+                            text = "${message.role.name}:${message.content}",
+                            modifier = Modifier.testTag("external-message-${message.id}"),
+                        )
+                    }
+                }
+            }
+        }
+
+        assertTrue(runBlocking { gateway.prewarmAwait(STORED_SESSION_ID) })
+        vm.sendMessage("Exercise quiet turns.")
+        compose.waitUntil(5_000) { gateway.hasActiveTurn() && handler.isStreaming.value }
+        Thread.sleep(600)
+        compose.onNodeWithTag("external-stream-state").assertTextEquals("STREAMING")
+        compose.waitUntil(5_000) {
+            handler.messages.value.any { it.content == "Quiet tool finished." }
+        }
+        compose.waitUntil(5_000) {
+            handler.messages.value.any { it.content == "Background work" }
+        }
+        Thread.sleep(600)
+        compose.onNodeWithTag("external-stream-state").assertTextEquals("STREAMING")
+
+        compose.waitUntil(10_000) {
+            !handler.isStreaming.value &&
+                !gateway.hasActiveTurn() &&
+                handler.messages.value.any {
+                    it.role == MessageRole.ASSISTANT && it.content == answer
+                }
+        }
+
+        compose.onNodeWithTag("external-contract-transcript").assertIsDisplayed()
+        compose.onNodeWithTag("external-stream-state").assertTextEquals("IDLE")
+        compose.onAllNodesWithText("${MessageRole.ASSISTANT.name}:$answer")
+            .assertCountEquals(1)
+
+        val messages = handler.messages.value
+        assertEquals(
+            1,
+            messages.count {
+                it.role == MessageRole.ASSISTANT && it.content == answer
+            },
+        )
+        assertEquals(1, messages.count { it.role == MessageRole.USER })
+        assertFalse(messages.any { it.isStreaming || it.isThinkingStreaming })
+        assertEquals("gateway", vm.streamingEndpoint)
+
+        val evidence = readFixtureJson(okHttp, "$fixtureBaseUrl/__fixture__/evidence")
+        assertEquals("quiet_turn_recovery", evidence["scenario"]?.jsonString())
+        val entries = evidence["entries"] as? JsonArray ?: JsonArray(emptyList())
+        assertEquals(1, entries.rpcCount("prompt.submit"))
+        assertEquals(0, entries.rpcCount("session.activate"))
+        assertEquals(0, entries.rpcCount("session.interrupt"))
+
+        val state = readFixtureJson(okHttp, "$fixtureBaseUrl/__fixture__/state")
+        assertEquals("quiet_turn_recovery", state["scenario"]?.jsonString())
+        assertEquals("3", state["history_rows"].toString())
+    }
+
     private fun readFixtureJson(client: OkHttpClient, url: String): JsonObject {
         val request = Request.Builder().url(url).get().build()
         return client.newCall(request).execute().use { response ->
diff --git a/app/src/androidTest/kotlin/com/hermesandroid/relay/viewmodel/GatewayForegroundRecoveryInstrumentedTest.kt b/app/src/androidTest/kotlin/com/hermesandroid/relay/viewmodel/GatewayForegroundRecoveryInstrumentedTest.kt
index 48a7760e8..5cea17a28 100644
--- a/app/src/androidTest/kotlin/com/hermesandroid/relay/viewmodel/GatewayForegroundRecoveryInstrumentedTest.kt
+++ b/app/src/androidTest/kotlin/com/hermesandroid/relay/viewmodel/GatewayForegroundRecoveryInstrumentedTest.kt
@@ -437,6 +437,38 @@ class GatewayForegroundRecoveryInstrumentedTest {
         assertEquals(0, fixture.requestsTo("/v1/chat/completions"))
     }
 
+    @Test
+    fun lostStartRunningActivation_recoversEndedTurnFromHistory() {
+        fixture.recoveryRunning = true
+        fixture.activeSessionStatus = "working"
+        viewModel.sendMessage("Recover before the first frame")
+        fixture.awaitRpc("prompt.submit")
+        // No start, delta or completion reaches Android on the original socket.
+        serverSocket.close(1011, "fixture lost start")
+        serverSocket = fixture.awaitServerSocket()
+        fixture.awaitRpc("session.activate")
+
+        persistedHistory = listOf(
+            MessageItem(id = "lost-start-user", sessionId = STORED_SESSION_ID,
+                role = "user", content = JsonPrimitive("Recover before the first frame")),
+            MessageItem(id = PERSISTED_ANSWER_ID, sessionId = STORED_SESSION_ID,
+                role = "assistant", content = JsonPrimitive(AUTHORITATIVE_ANSWER)),
+        )
+        fixture.activeSessionStatus = "idle"
+        compose.waitUntil(5_000) {
+            runBlocking { gatewayClient.listActiveSessions() }
+            !handler.isStreaming.value && !gatewayClient.hasActiveTurn() &&
+                handler.messages.value.any { it.id == PERSISTED_ANSWER_ID }
+        }
+        compose.onNodeWithTag("stream-state").assertTextEquals("IDLE")
+        compose.onNodeWithTag("message-$PERSISTED_ANSWER_ID")
+            .assertTextEquals("${MessageRole.ASSISTANT.name}:$AUTHORITATIVE_ANSWER")
+        assertEquals(1, fixture.rpcCount("prompt.submit"))
+        assertEquals(1, fixture.rpcCount("session.activate"))
+        assertEquals(0, fixture.rpcCount("session.interrupt"))
+        assertEquals(0, fixture.requestsTo("/v1/chat/completions"))
+    }
+
     @Test
     fun terminalGapActiveList_settlesExactOwnedTurnAndRendersAuthoritativeHistory() {
         viewModel.sendMessage("Run an Android-owned task")
@@ -449,7 +481,13 @@ class GatewayForegroundRecoveryInstrumentedTest {
                 LIVE_SESSION_ID,
             ),
         )
-        compose.waitUntil(5_000) { handler.isStreaming.value }
+        // Sending marks the UI as streaming before Gateway frames arrive.
+        // Wait for the delta to be consumed so the idle probe captures the
+        // current progress generation instead of racing that live progress.
+        compose.waitUntil(5_000) {
+            handler.isStreaming.value &&
+                handler.messages.value.any { it.content.contains(PARTIAL_ANSWER) }
+        }
         compose.onNodeWithTag("stream-state").assertTextEquals("STREAMING")
 
         persistedHistory = listOf(
diff --git a/app/src/main/kotlin/com/hermesandroid/relay/network/upstream/GatewayChatClient.kt b/app/src/main/kotlin/com/hermesandroid/relay/network/upstream/GatewayChatClient.kt
index eb3299cbb..83485cca0 100644
--- a/app/src/main/kotlin/com/hermesandroid/relay/network/upstream/GatewayChatClient.kt
+++ b/app/src/main/kotlin/com/hermesandroid/relay/network/upstream/GatewayChatClient.kt
@@ -121,8 +121,6 @@ class GatewayChatClient(
     private val turnIdleTimeoutMs: Long = TURN_TIMEOUT_MS,
     /** Test seam — lazy session.create/resume readiness barrier. */
     private val sessionReadyTimeoutMs: Long = SESSION_READY_TIMEOUT_MS,
-    /** Test seam — compaction idle lease. Production keeps [COMPACTING_TIMEOUT_MS]. */
-    private val compactingTimeoutMs: Long = COMPACTING_TIMEOUT_MS,
     /** Random source for ordinary reconnect full-jitter. */
     private val reconnectJitterUnit: () -> Double = { kotlin.random.Random.nextDouble() },
 ) : GatewayProfileEditorClient {
@@ -145,37 +143,12 @@ class GatewayChatClient(
         private const val BOT_CHAT_TITLE = "Bot Chat"
 
         /**
-         * Idle-progress turn watchdog — reset on EVERY received gateway event
-         * (deltas, tool events, status lines), so it only fires after this
-         * long with no events at all. It is NOT a hard turn cap: a turn that
-         * keeps streaming lives indefinitely, and a slow `prompt.submit` ack
-         * is bounded separately by [PROMPT_SUBMIT_REQUEST_TIMEOUT_MS].
+         * Quiet-period interval for a read-only liveness check. Silence never
+         * interrupts or fails a turn; exact server Idle can recover a lost final
+         * frame through the existing authoritative history reconciliation.
          */
         private const val TURN_TIMEOUT_MS = 180_000L
 
-        /**
-         * Ask requests block the agent server-side with NO events flowing
-         * until answered — arm with headroom over each kind's upstream block
-         * timeout (clarify/secret 300s, sudo 120s; approval/terminal-read
-         * unbounded). The next regular event rearms [TURN_TIMEOUT_MS].
-         */
-        private const val ASK_CLARIFY_SECRET_TIMEOUT_MS = 330_000L
-        private const val ASK_SUDO_TIMEOUT_MS = 150_000L
-        private const val ASK_UNBOUNDED_TIMEOUT_MS = 600_000L
-
-        /**
-         * Server-side context compaction summarizes the transcript through a
-         * (possibly slow) model with NO deltas or tool events flowing until it
-         * finishes — near the context ceiling that silence routinely exceeds
-         * [TURN_TIMEOUT_MS], so the idle watchdog would `session.interrupt` a
-         * healthy compression, roll back its work, and retrigger on the next
-         * prompt forever. A `status.update` event with kind `compacting`
-         * (emitted at compaction start, and periodically by newer gateways)
-         * arms this longer leash instead; any regular event rearms
-         * [TURN_TIMEOUT_MS].
-         */
-        private const val COMPACTING_TIMEOUT_MS = 600_000L
-
         private const val RPC_TIMEOUT_MS = 15_000L
         const val PROFILE_AVATAR_MAX_BYTES = 2_000_000
 
@@ -251,16 +224,6 @@ class GatewayChatClient(
          */
         private const val MAX_MIDTURN_REJOIN_MS = 20_000L
 
-        /**
-         * After a route RETARGET (LAN⇄Tailscale mid-turn), the fresh socket
-         * can't pick up the in-flight turn's events — upstream `session.resume`
-         * doesn't reattach to a running turn. So arm a SHORT settle on the
-         * reconnect: if nothing flows we fail fast and the post-turn reconcile
-         * recovers the server's answer, instead of waiting the full turn
-         * watchdog. Any live event resets it back to the normal timeout.
-         */
-        private const val POST_RETARGET_SETTLE_MS = 30_000L
-
         private const val DEFAULT_COLS = 80
 
         private object MainThreadDispatcher : (() -> Unit) -> Unit {
@@ -628,6 +591,8 @@ class GatewayChatClient(
         val storedSessionId: String,
         val liveSessionId: String,
         val progressGeneration: Long,
+        val socket: WebSocket,
+        val profile: String?,
     )
 
     /**
@@ -1027,15 +992,10 @@ class GatewayChatClient(
         Log.i(TAG, "Gateway retargeting to a new route (turn active=${hasActiveTurn()})")
         routeTransport = RouteTransport(newDashboardClient, socketClient(newDashboardClient.okHttpClient))
         if (hasActiveTurn()) {
-            retargetedThisTurn = activeTurn?.ended == false
             webSocket?.cancel()
         }
     }
 
-    /** Set by [retarget] so the rejoin arms the short post-retarget settle once. */
-    @Volatile
-    private var retargetedThisTurn = false
-
     /**
      * Toggle the opt-in background keep-alive (sideload). While on, the socket
      * is not auto-closed when the app backgrounds. Turning it off while already
@@ -1248,7 +1208,7 @@ class GatewayChatClient(
                 // this resume. Unknown gateway sessions captured during the
                 // narrow ack race remain foreign and are discarded.
                 val replay = pending.closeAndTake(liveChildId)
-                if (replay.truncated) dispatchedCallbacks.onReconcileRequired()
+                if (replay.truncated) dispatchedCallbacks.onReconcileRequired(false)
                 replay.events.forEach { event ->
                     if (childWatches[liveChildId] === registration) {
                         registration.mapper.onEvent(event.type, event.payload)
@@ -2758,16 +2718,16 @@ class GatewayChatClient(
     }
 
     /**
-     * Capture only a locally submitted/recovered turn. Merely observing an
-     * exact session through the shared Gateway socket never grants Android
-     * authority to settle Desktop/TUI work.
+     * Capture a locally admitted stream, including an exact unsolicited turn.
+     * Settling its UI from read-only state does not attach or stop the runtime.
      */
     private fun captureActiveTurnLivenessProbe(): ActiveTurnLivenessProbe? {
         val turn = activeTurn ?: return null
         val generation = turn.captureLivenessGeneration() ?: return null
         val storedId = storedSessionId ?: return null
         val liveId = liveSessionId ?: return null
-        return ActiveTurnLivenessProbe(turn, storedId, liveId, generation)
+        val socket = webSocket ?: return null
+        return ActiveTurnLivenessProbe(turn, storedId, liveId, generation, socket, liveSessionProfile)
     }
 
     /**
@@ -2782,6 +2742,8 @@ class GatewayChatClient(
     ) {
         probe ?: return
         if (activeTurn !== probe.turn ||
+            webSocket !== probe.socket ||
+            liveSessionProfile != probe.profile ||
             storedSessionId != probe.storedSessionId ||
             liveSessionId != probe.liveSessionId
         ) return
@@ -3189,7 +3151,9 @@ class GatewayChatClient(
     }
 
     fun shutdown() {
-        activeTurn?.cancel()
+        activeTurn?.let { turn ->
+            if (turn.androidOwned) turn.cancel() else turn.detach()
+        }
         activeTurn = null
         backgroundTurns.clear()
         cancelledTurnDrain = null
@@ -4553,17 +4517,50 @@ class GatewayChatClient(
                 }
                 if (preservedLiveId == null) {
                     true
+                } else if (!turn.androidOwned) {
+                    // Restore/retire asks without attaching the producer's runtime.
+                    liveSessionId = preservedLiveId
+                    val ready = readySignal ?: throw GatewayRpcException("gateway connection changed")
+                    val socket = webSocket ?: throw GatewayRpcException("gateway socket changed")
+                    val snapshot = rpc("session.events.since", buildJsonObject {
+                        put("session_id", preservedLiveId)
+                        put("last_seen", Long.MAX_VALUE)
+                    }, expectedConnection = ready, expectedSocket = socket)
+                    if (readySignal !== ready || webSocket !== socket || activeTurn !== turn ||
+                        liveSessionId != preservedLiveId
+                    ) {
+                        false
+                    } else if (snapshot.isSuccess) {
+                        replayServerRequests(snapshot.getOrThrow(), preservedLiveId)
+                        true
+                    } else {
+                        snapshot.exceptionOrNull().isMethodNotFound()
+                    }
                 } else {
                     // Bind before activation so a tail event racing the RPC ack
                     // still matches the original active turn.
                     liveSessionId = preservedLiveId
+                    val ready = readySignal ?: throw GatewayRpcException("gateway connection changed")
+                    val socket = webSocket ?: throw GatewayRpcException("gateway socket changed")
                     val activated = rpc(
                         "session.activate",
                         buildJsonObject { put("session_id", preservedLiveId) },
+                        expectedConnection = ready,
+                        expectedSocket = socket,
                     )
                     when {
+                        readySignal !== ready || webSocket !== socket || activeTurn !== turn ||
+                            liveSessionId != preservedLiveId -> false
                         activated.isSuccess -> {
                             activated.getOrNull()?.let { result ->
+                                // The start frame may have been lost on the old
+                                // socket. Exact running recovery is equivalent
+                                // proof of activity; Idle alone is not. A queued
+                                // prompt may still belong to a later turn.
+                                if (result.stringField("session_id") == preservedLiveId &&
+                                    result.booleanField("running") == true &&
+                                    (result["queued"] == null || result["queued"] == JsonNull)
+                                ) turn.markRecoveredStarted()
                                 applySessionResultInfo(result)
                                 turn.restorePendingClarify(result)
                                 replayServerRequests(result)
@@ -4606,18 +4603,9 @@ class GatewayChatClient(
                     "Gateway socket rejoined mid-turn (session=$storedSessionId) — " +
                         "rebound live session, awaiting tail",
                 )
-                // A reconnect that followed a route RETARGET gets a short settle
-                // (the fresh socket won't replay the in-flight turn); a normal
-                // blip-rejoin keeps the full turn watchdog. A live event resets
-                // either back to the per-event timeout.
-                turn.armWatchdog(
-                    if (retargetedThisTurn) {
-                        retargetedThisTurn = false
-                        POST_RETARGET_SETTLE_MS
-                    } else {
-                        turnIdleTimeoutMs
-                    },
-                )
+                // A quiet tail is checked through read-only state. Route changes
+                // do not shorten a healthy turn's lifetime.
+                turn.armWatchdog()
                 return
             }
             delay(fullJitterDelayMs(backoffMs, reconnectJitterUnit()))
@@ -4848,21 +4836,11 @@ class GatewayChatClient(
     // Turn handle
     // ------------------------------------------------------------------
 
-    /** Per-event idle-watchdog duration — asks block server-side with no events, so they arm longer. */
-    private fun watchdogTimeoutFor(eventType: String, payload: JsonObject? = null): Long = when {
-        eventType == "clarify.request" || eventType == "secret.request" -> ASK_CLARIFY_SECRET_TIMEOUT_MS
-        eventType == "sudo.request" -> ASK_SUDO_TIMEOUT_MS
-        eventType == "approval.request" -> ASK_UNBOUNDED_TIMEOUT_MS
-        eventType == "status.update" &&
-            payload?.stringField("kind") == "compacting" -> compactingTimeoutMs
-        else -> turnIdleTimeoutMs
-    }
-
     private inner class GatewayTurn(
         val callbacks: GatewayTurnCallbacks,
         dedupeAdjacentMessageStarts: Boolean = false,
         deferEvents: Boolean = false,
-        private val androidOwned: Boolean = false,
+        val androidOwned: Boolean = false,
         private val onTransportAccepted: () -> Unit = { },
     ) : ActiveTurnHandle {
         private val mapper = GatewayEventMapper(callbacks, dedupeAdjacentMessageStarts)
@@ -4958,8 +4936,8 @@ class GatewayChatClient(
         val ended: Boolean get() = mapper.turnEnded || cancelled || settledWithoutTerminalFrame
 
         /**
-         * True once any turn-scoped event has arrived — proof the server
-         * received the submit and is running the turn. `session.info` doesn't
+         * True once a turn-scoped event or exact running recovery proves the
+         * server received the submit and is running the turn. `session.info` doesn't
          * count: it's connection-level (resume/config echoes) and can arrive
          * independent of this turn, so it must not suppress a legitimate
          * preflight fallback.
@@ -4998,7 +4976,7 @@ class GatewayChatClient(
             // marshalled through the same dispatcher, preserving callback order
             // even when the WebSocket reader and reconnect coroutine differ.
             if (type == "message.complete" && reconcileRequired) {
-                callbacks.onReconcileRequired()
+                callbacks.onReconcileRequired(false)
             }
             mapper.onEvent(type, payload)
             if (mapper.turnEnded) {
@@ -5007,7 +4985,8 @@ class GatewayChatClient(
                 handoffQueuedSuccessor()
             } else {
                 // Map first: native asks own their deadline, including across unrelated events.
-                armWatchdog(watchdogTimeoutFor(type, payload))
+                // Rearm the read-only check after progress.
+                armWatchdog()
             }
         }
 
@@ -5017,7 +4996,7 @@ class GatewayChatClient(
          * heartbeat can race `prompt.submit` and is not a completion boundary.
          */
         fun captureLivenessGeneration(): Long? =
-            if (androidOwned && started && !ended) progressGeneration.get() else null
+            if (started && !ended) progressGeneration.get() else null
 
         fun settleFromAuthoritativeSessionState(
             running: Boolean?,
@@ -5042,7 +5021,7 @@ class GatewayChatClient(
             disarmWatchdog()
             armSettledTurnDrain()
             Log.i(TAG, "Gateway turn settled from $source after missing terminal frame")
-            callbacks.onReconcileRequired()
+            callbacks.onReconcileRequired(true)
             callbacks.onComplete()
             tracer.done("history-reconcile")
             handoffQueuedSuccessor()
@@ -5117,15 +5096,17 @@ class GatewayChatClient(
             }
         }
 
-        fun armWatchdog(timeoutMs: Long = turnIdleTimeoutMs) {
+        fun armWatchdog() {
             disarmWatchdog()
             if (mapper.currentInteraction?.serverRequest == true) return
             watchdog = scope.launch {
-                delay(timeoutMs)
-                if (!ended) {
-                    Log.w(TAG, "Gateway turn timed out after ${timeoutMs}ms")
-                    interruptServerSide()
-                    failFromTransport("Gateway turn timed out")
+                while (true) {
+                    delay(turnIdleTimeoutMs)
+                    if (ended || activeTurn !== this@GatewayTurn) break
+                    // Missing/unsupported/failed snapshots remain unknown. Only
+                    // exact Idle with no intervening progress settles the stream.
+                    Log.i(TAG, "Gateway turn quiet; checking authoritative state (no interrupt)")
+                    listActiveSessions()
                 }
             }
         }
@@ -5171,6 +5152,9 @@ class GatewayChatClient(
 
         private fun interruptServerSide() {
             val sid = liveSessionId ?: return
+            // This path is explicit handle cancellation, never idle expiry.
+            // Upstream rejects arbitrary reason/source keys on the wire.
+            Log.i(TAG, "session.interrupt requested reason=client_cancel")
             scope.launch {
                 // Best-effort: unblocks the server (also releases blocked
                 // interactive asks). Failure is fine — socket may be gone.
@@ -5315,7 +5299,9 @@ class GatewayChatClient(
         onToolCallFailed = { a, b -> dispatchIfCurrent(stillCurrent) { callbacks.onToolCallFailed(a, b) } },
         onToolOutputRisk = { v -> dispatchIfCurrent(stillCurrent) { callbacks.onToolOutputRisk(v) } },
         onTurnComplete = { dispatchIfCurrent(stillCurrent) { callbacks.onTurnComplete() } },
-        onReconcileRequired = { dispatchIfCurrent(stillCurrent) { callbacks.onReconcileRequired() } },
+        onReconcileRequired = { authoritative ->
+            dispatchIfCurrent(stillCurrent) { callbacks.onReconcileRequired(authoritative) }
+        },
         onComplete = { dispatchIfCurrent(stillCurrent) { callbacks.onComplete() } },
         onUsage = { v -> dispatchIfCurrent(stillCurrent) { callbacks.onUsage(v) } },
         onError = { v -> dispatchIfCurrent(stillCurrent) { callbacks.onError(v) } },
diff --git a/app/src/main/kotlin/com/hermesandroid/relay/network/upstream/GatewayModels.kt b/app/src/main/kotlin/com/hermesandroid/relay/network/upstream/GatewayModels.kt
index 6417a9e84..f71ae6733 100644
--- a/app/src/main/kotlin/com/hermesandroid/relay/network/upstream/GatewayModels.kt
+++ b/app/src/main/kotlin/com/hermesandroid/relay/network/upstream/GatewayModels.kt
@@ -734,8 +734,10 @@ class GatewayTurnCallbacks(
      * Fired before [onComplete] when this turn rejoined after a socket gap.
      * Events emitted while the socket was unavailable are not replayed, so
      * the caller must reconcile the durable transcript after completion.
+     * [authoritativeSettlement] is true when server state replaced a missing
+     * terminal frame: the live text is partial, and no final-text match is known.
      */
-    val onReconcileRequired: () -> Unit,
+    val onReconcileRequired: (authoritativeSettlement: Boolean) -> Unit,
     val onComplete: () -> Unit,
     val onUsage: (UsageInfo?) -> Unit,
     val onError: (String) -> Unit,
diff --git a/app/src/main/kotlin/com/hermesandroid/relay/viewmodel/ChatViewModel.kt b/app/src/main/kotlin/com/hermesandroid/relay/viewmodel/ChatViewModel.kt
index 0c5016e78..48d54cefe 100644
--- a/app/src/main/kotlin/com/hermesandroid/relay/viewmodel/ChatViewModel.kt
+++ b/app/src/main/kotlin/com/hermesandroid/relay/viewmodel/ChatViewModel.kt
@@ -3142,6 +3142,7 @@ class ChatViewModel : ViewModel() {
         val messageId = "gateway-inbound-${UUID.randomUUID()}"
         val queuedUserMessageId = "gateway-queued-user-${UUID.randomUUID()}"
         var baselineAssistantCount = 0
+        var authoritativeHistoryRequired = false
         var started = false
         var accepted = false
         var boundHandle: ActiveTurnHandle? = null
@@ -3228,7 +3229,10 @@ class ChatViewModel : ViewModel() {
             },
             // Recovery can settle a partial live bubble before durable history
             // arrives. That history repairs Chat, but is not a speech receipt.
-            onReconcileRequired = { speechReceiver = null },
+            onReconcileRequired = { authoritative ->
+                speechReceiver = null
+                authoritativeHistoryRequired = authoritative
+            },
             onComplete = {
                 val canWriteTranscript = acceptsEvent()
                 val expectedText = handler.messages.value
@@ -3251,7 +3255,9 @@ class ChatViewModel : ViewModel() {
                     speechReceiver = null
                     scheduleGatewayHistoryReconcile(
                         storedSessionId = storedSessionId,
-                        expectedAssistantText = expectedText,
+                        // A missing terminal leaves only partial live text. It
+                        // cannot be an exact-match condition for the saved answer.
+                        expectedAssistantText = expectedText.takeUnless { authoritativeHistoryRequired },
                     )
                     drainQueue()
                 } else {
diff --git a/app/src/test/kotlin/com/hermesandroid/relay/network/upstream/GatewayChatClientTest.kt b/app/src/test/kotlin/com/hermesandroid/relay/network/upstream/GatewayChatClientTest.kt
index 7ec69b57b..415493c73 100644
--- a/app/src/test/kotlin/com/hermesandroid/relay/network/upstream/GatewayChatClientTest.kt
+++ b/app/src/test/kotlin/com/hermesandroid/relay/network/upstream/GatewayChatClientTest.kt
@@ -931,6 +931,10 @@ class GatewayChatClientTest {
         delay(800)
         assertTrue(client.hasActiveTurn())
         assertTrue(harness.rpcLog.none { it.first == "session.interrupt" })
+        harness.activeSessionListPayload = exactActiveSessionPayload("idle")
+        client.listActiveSessions()
+        assertTrue("pending request must prevent premature settlement", client.hasActiveTurn())
+        assertEquals(0, r.completions.get())
         client.respondAsk(r.interactions.single(), "answer").getOrThrow()
         ws.send(harness.eventFrame("message.complete", buildJsonObject { put("text", "Done") }, "live-1"))
     }
@@ -1107,7 +1111,6 @@ class GatewayChatClientTest {
         promptSubmitTimeoutMs: Long = 1_800_000L,
         turnIdleTimeoutMs: Long = 180_000L,
         sessionReadyTimeoutMs: Long = 300_000L,
-        compactingTimeoutMs: Long = 600_000L,
         callbackDispatcher: (block: () -> Unit) -> Unit = { it() },
         ticketTimeoutMs: Long = 8_000L,
     ) = GatewayChatClient(
@@ -1130,7 +1133,6 @@ class GatewayChatClientTest {
         promptSubmitTimeoutMs = promptSubmitTimeoutMs,
         turnIdleTimeoutMs = turnIdleTimeoutMs,
         sessionReadyTimeoutMs = sessionReadyTimeoutMs,
-        compactingTimeoutMs = compactingTimeoutMs,
     )
 
     private fun awaitCondition(
@@ -1169,7 +1171,6 @@ class GatewayChatClientTest {
         promptSubmitTimeoutMs: Long = 1_800_000L,
         turnIdleTimeoutMs: Long = 180_000L,
         sessionReadyTimeoutMs: Long = 300_000L,
-        compactingTimeoutMs: Long = 600_000L,
         ticketTimeoutMs: Long = 8_000L,
     ) {
         client.shutdown()
@@ -1179,7 +1180,6 @@ class GatewayChatClientTest {
             promptSubmitTimeoutMs = promptSubmitTimeoutMs,
             turnIdleTimeoutMs = turnIdleTimeoutMs,
             sessionReadyTimeoutMs = sessionReadyTimeoutMs,
-            compactingTimeoutMs = compactingTimeoutMs,
             ticketTimeoutMs = ticketTimeoutMs,
         )
     }
@@ -1722,6 +1722,161 @@ class GatewayChatClientTest {
         assertTrue(second.errors.isEmpty())
     }
 
+    @Test
+    fun `quiet unsolicited turn survives silence and settles from exact idle state`() = runBlocking {
+        rebuildClient(turnIdleTimeoutMs = 150L)
+        val r = Recorder()
+        client.setUnsolicitedTurnProvider { GatewayInboundTurnRegistration(r.callbacks) { true } }
+        assertTrue(client.prewarmAwait("stored-session"))
+        val ws = harness.awaitServerSocket()
+        harness.activeSessionListPayload = exactActiveSessionPayload("working", "live-resumed", "stored-session")
+        ws.send(harness.eventFrame("message.start", null, "live-resumed"))
+        awaitCondition { r.starts.get() == 1 }
+        Thread.sleep(650)
+        assertTrue("silence failed observer: ${r.errors}", r.errors.isEmpty())
+        assertEquals(0, r.completions.get())
+        assertTrue(harness.rpcLog.none { it.first == "session.interrupt" })
+        harness.activeSessionListPayload = exactActiveSessionPayload("idle", "live-resumed", "stored-session")
+        assertTrue("quiet probe did not settle observer", r.completeLatch.await(5, TimeUnit.SECONDS))
+        assertEquals(1, r.reconcileRequests.get())
+        assertEquals(1, r.completions.get())
+        assertTrue(harness.rpcLog.none { it.first in setOf("session.interrupt", "session.activate", "prompt.submit") })
+    }
+
+    @Test
+    fun `quiet owned tool and provider wait accepts late completion`() {
+        rebuildClient(turnIdleTimeoutMs = 150L)
+        val r = Recorder()
+        client.sendTurn(null, "quiet tools", null, r.callbacks) { r.preflightFailures += it }
+        val ws = harness.awaitServerSocket()
+        harness.awaitRpc("prompt.submit")
+        harness.activeSessionListPayload = exactActiveSessionPayload("working")
+        ws.send(harness.eventFrame("tool.start", buildJsonObject { put("tool_id", "t1"); put("name", "terminal") }, "live-1"))
+        awaitCondition { r.toolStarts.isNotEmpty() }
+        Thread.sleep(500)
+        ws.send(harness.eventFrame("tool.complete", buildJsonObject { put("tool_id", "t1"); put("name", "terminal") }, "live-1"))
+        Thread.sleep(500)
+        assertTrue("silence failed owner: ${r.errors}", r.errors.isEmpty())
+        assertTrue(harness.rpcLog.none { it.first == "session.interrupt" })
+        ws.send(harness.eventFrame("message.complete", buildJsonObject { put("text", "late answer") }, "live-1"))
+        assertTrue(r.completeLatch.await(5, TimeUnit.SECONDS))
+        assertEquals(listOf("late answer"), r.textDeltas.toList())
+        assertEquals(1, r.completions.get())
+    }
+
+    @Test
+    fun `quiet unknown state never implies failure and explicit Stop still interrupts`() {
+        rebuildClient(turnIdleTimeoutMs = 150L)
+        val r = Recorder()
+        val handle = client.sendTurn(null, "wait", null, r.callbacks) { r.preflightFailures += it }
+        val ws = harness.awaitServerSocket()
+        harness.awaitRpc("prompt.submit")
+        ws.send(harness.eventFrame("message.start", null, "live-1"))
+        for (code in listOf(5000, -32601)) {
+            harness.rpcErrors["session.active_list"] = code to "unavailable"
+            Thread.sleep(500)
+            assertTrue(r.errors.isEmpty())
+            assertTrue(client.hasActiveTurn())
+            assertEquals(0, r.completions.get())
+        }
+        assertTrue(harness.rpcLog.none { it.first == "session.interrupt" })
+        handle.cancel()
+        harness.awaitRpc("session.interrupt")
+        assertEquals(1, harness.rpcLog.count { it.first == "session.interrupt" })
+    }
+
+    @Test
+    fun `quiet unsolicited reconnect observes without activating producer and recovers idle`() = runBlocking {
+        rebuildClient(turnIdleTimeoutMs = 200L)
+        val r = Recorder()
+        client.setUnsolicitedTurnProvider { GatewayInboundTurnRegistration(r.callbacks) { true } }
+        assertTrue(client.prewarmAwait("stored-session"))
+        val ws = harness.awaitServerSocket()
+        harness.activeSessionListPayload = exactActiveSessionPayload("working", "live-resumed", "stored-session")
+        ws.send(harness.eventFrame("message.start", null, "live-resumed"))
+        awaitCondition { r.starts.get() == 1 }
+        ws.close(1012, "fixture reconnect")
+        harness.awaitServerSocket()
+        Thread.sleep(650)
+        assertTrue(r.errors.isEmpty())
+        assertEquals(0, r.completions.get())
+        assertTrue(harness.rpcLog.none { it.first in setOf("session.activate", "session.interrupt", "prompt.submit") })
+        assertEquals(1, harness.rpcLog.count { it.first == "session.resume" })
+        harness.activeSessionListPayload = exactActiveSessionPayload("idle", "live-resumed", "stored-session")
+        assertTrue(r.completeLatch.await(5, TimeUnit.SECONDS))
+        assertEquals(1, r.reconcileRequests.get())
+    }
+
+    @Test
+    fun `unsolicited reconnect restores pending native request without activation`() =
+        verifyUnsolicitedRequestRejoin(stillPending = true)
+
+    @Test
+    fun `unsolicited reconnect retires request answered elsewhere without activation`() =
+        verifyUnsolicitedRequestRejoin(stillPending = false)
+
+    private fun verifyUnsolicitedRequestRejoin(stillPending: Boolean): Unit = runBlocking {
+        rebuildClient(turnIdleTimeoutMs = 200L)
+        val r = Recorder()
+        client.setUnsolicitedTurnProvider { GatewayInboundTurnRegistration(r.callbacks) { true } }
+        assertTrue(client.prewarmAwait("stored-session"))
+        val ws = harness.awaitServerSocket()
+        harness.activeSessionListPayload = exactActiveSessionPayload("working", "live-resumed", "stored-session")
+        ws.send(harness.eventFrame("message.start", null, "live-resumed"))
+        awaitCondition { r.starts.get() == 1 }
+        nativeRequest(ws, JsonPrimitive("srq-inbound"), "clarify", """{"session_id":"live-resumed","question":"Choose?"}""")
+        awaitCondition { r.interactions.size == 1 }
+        harness.recoveryOpenRequests = buildJsonArray {
+            if (stillPending) add(harness.json.parseToJsonElement(
+                """{"id":"srq-inbound","method":"clarify","params":{"session_id":"live-resumed","question":"Choose?"}}""",
+            ))
+        }
+        ws.close(1012, "fixture reconnect")
+        val replacement = harness.awaitServerSocket()
+        harness.awaitRpc("session.events.since")
+        if (stillPending) {
+            // Replay refreshes socket ownership without duplicating the same card.
+            var lastReplyFailure: String? = null
+            var replyAccepted = false
+            val answered = runCatching {
+                awaitCondition {
+                    if (!replyAccepted) {
+                        val result = runBlocking { client.respondAsk(r.interactions.last(), "answer") }
+                        lastReplyFailure = result.exceptionOrNull()?.message
+                        replyAccepted = result.isSuccess
+                    }
+                    replyAccepted
+                }
+            }.isSuccess
+            assertTrue("Reply rejected: $lastReplyFailure; expiries=${r.interactionExpiries}; errors=${r.errors}", answered)
+            assertEquals(JsonPrimitive("srq-inbound"), harness.serverResponses.poll(5, TimeUnit.SECONDS)?.get("id"))
+            assertEquals(1, r.interactions.size)
+        } else {
+            awaitCondition { r.interactionExpiries.size == 1 }
+            assertTrue(client.respondAsk(r.interactions.single(), "late").isFailure)
+        }
+        replacement.send(harness.eventFrame("message.complete", buildJsonObject { put("text", "done") }, "live-resumed"))
+        assertTrue(r.completeLatch.await(5, TimeUnit.SECONDS))
+        assertTrue(r.errors.isEmpty())
+        assertTrue(harness.rpcLog.none { it.first in setOf("session.activate", "session.interrupt", "prompt.submit") })
+    }
+
+    @Test
+    fun `unsolicited client shutdown detaches without sending Stop`() = runBlocking {
+        client.shutdown()
+        scope = CoroutineScope(SupervisorJob() + Dispatchers.Unconfined)
+        client = buildClient()
+        val r = Recorder()
+        client.setUnsolicitedTurnProvider { GatewayInboundTurnRegistration(r.callbacks) { true } }
+        assertTrue(client.prewarmAwait("stored-session"))
+        val ws = harness.awaitServerSocket()
+        ws.send(harness.eventFrame("message.start", null, "live-resumed"))
+        awaitCondition { r.starts.get() == 1 }
+        client.shutdown()
+        Thread.sleep(200)
+        assertTrue(harness.rpcLog.none { it.first == "session.interrupt" })
+    }
+
     @Test
     fun `unsolicited assistant turn for resumed session streams without sendTurn`() = runBlocking {
         val r = Recorder()
@@ -2374,6 +2529,8 @@ class GatewayChatClientTest {
     @Test
     fun `active route retarget replaces socket policy and preserves the live turn`() {
         val replacement = GatewayClientHarness()
+        replacement.recoveryRunning = true
+        replacement.activeSessionListPayload = exactActiveSessionPayload("working")
         fun dashboardFor(target: GatewayClientHarness): DashboardApiClient = DashboardApiClient(
             baseUrl = target.server.url("/").toString(),
             okHttpClient = OkHttpClient.Builder().addInterceptor { chain ->
@@ -2390,18 +2547,25 @@ class GatewayChatClientTest {
             scope = scope,
             callbackDispatcher = { it() },
             midTurnRejoinWindowMs = 3_000L,
+            turnIdleTimeoutMs = 200L,
         )
         try {
             val recorder = Recorder()
             client.sendTurn(null, "follow the route", null, recorder.callbacks) {
                 recorder.preflightFailures += it
             }
-            harness.awaitServerSocket()
+            val original = harness.awaitServerSocket()
             harness.awaitRpc("prompt.submit")
+            original.send(harness.eventFrame("message.start", null, "live-1"))
+            awaitCondition { recorder.starts.get() == 1 }
             client.retarget(dashboardFor(replacement))
             val moved = replacement.awaitServerSocket()
             val activation = replacement.awaitRpc("session.activate")
             assertEquals("live-1", activation["session_id"]?.jsonPrimitive?.content)
+            Thread.sleep(650)
+            assertEquals(0, recorder.completions.get())
+            assertTrue(recorder.errors.isEmpty())
+            assertFalse(replacement.rpcLog.any { it.first == "session.interrupt" })
             moved.send(replacement.eventFrame("message.complete", buildJsonObject {
                 put("text", "Finished on the new route")
             }, "live-1"))
@@ -2670,6 +2834,103 @@ class GatewayChatClientTest {
         assertTrue(harness.rpcLog.none { it.first == "session.resume" })
     }
 
+    @Test
+    fun `lost start recovers from scoped idle after running activation`() =
+        verifyLostStartRecovery(snapshotIdle = false)
+
+    @Test
+    fun `lost start recovers from active list idle after running activation`() =
+        verifyLostStartRecovery(snapshotIdle = true)
+
+    private fun verifyLostStartRecovery(snapshotIdle: Boolean) {
+        client.shutdown()
+        // Apply the activation response inline before reading the following
+        // session.info frame, so this test does not depend on dispatcher timing.
+        scope = CoroutineScope(SupervisorJob() + Dispatchers.Unconfined)
+        client = buildClient(turnIdleTimeoutMs = 150L)
+        harness.suppressAckMethods += "session.activate"
+        harness.activeSessionListPayload = exactActiveSessionPayload("working")
+        val r = Recorder()
+        val accepted = CountDownLatch(1)
+        client.sendTurn(null, "lost start", null, r.callbacks,
+            onTransportAccepted = { accepted.countDown() },
+        ) { r.preflightFailures += it }
+        val original = harness.awaitServerSocket()
+        assertTrue(accepted.await(5, TimeUnit.SECONDS))
+        assertEquals(0, r.starts.get())
+        original.close(1011, "start frame lost")
+        val replacement = harness.awaitServerSocket()
+        val activation = harness.awaitPendingAck()
+        assertEquals("session.activate", activation.method)
+        harness.releaseAck(activation, buildJsonObject {
+            put("session_id", "live-1")
+            put("running", true)
+        })
+        if (snapshotIdle) {
+            harness.activeSessionListPayload = exactActiveSessionPayload("idle")
+        } else {
+            replacement.send(harness.eventFrame("session.info", buildJsonObject {
+                put("running", false)
+            }, "live-1"))
+        }
+        assertTrue("authoritative completion was ignored after the lost start", r.completeLatch.await(3, TimeUnit.SECONDS))
+        assertEquals(1, r.completions.get())
+        assertEquals(1, r.reconcileRequests.get())
+        assertEquals(0, r.starts.get())
+        assertTrue(r.errors.isEmpty())
+        assertTrue(r.preflightFailures.isEmpty())
+        assertEquals(1, harness.rpcLog.count { it.first == "prompt.submit" })
+        assertTrue(harness.rpcLog.none { it.first in setOf("session.resume", "session.interrupt") })
+    }
+
+    @Test
+    fun `prestart idle activation does not invent proof of a running turn`() =
+        verifyUnprovenStart(running = false)
+
+    @Test
+    fun `foreign running activation does not invent proof of a running turn`() =
+        verifyUnprovenStart(running = true, activationSession = "foreign")
+
+    @Test
+    fun `queued running activation does not invent proof of a running turn`() =
+        verifyUnprovenStart(running = true, queued = true)
+
+    private fun verifyUnprovenStart(
+        running: Boolean,
+        activationSession: String = "live-1",
+        queued: Boolean = false,
+    ) {
+        client.shutdown()
+        scope = CoroutineScope(SupervisorJob() + Dispatchers.Unconfined)
+        client = buildClient(turnIdleTimeoutMs = 150L)
+        harness.suppressAckMethods += "session.activate"
+        harness.activeSessionListPayload = exactActiveSessionPayload("idle")
+        val r = Recorder()
+        val accepted = CountDownLatch(1)
+        client.sendTurn(null, "not started yet", null, r.callbacks,
+            onTransportAccepted = { accepted.countDown() },
+        ) { r.preflightFailures += it }
+        val original = harness.awaitServerSocket()
+        assertTrue(accepted.await(5, TimeUnit.SECONDS))
+        original.close(1011, "prestart gap")
+        val replacement = harness.awaitServerSocket()
+        val activation = harness.awaitPendingAck()
+        harness.releaseAck(activation, buildJsonObject {
+            put("session_id", activationSession)
+            put("running", running)
+            if (queued) put("queued", buildJsonObject { put("user", "not started yet") })
+        })
+        replacement.send(harness.eventFrame("session.info", buildJsonObject { put("running", false) }, "live-1"))
+        assertFalse(r.completeLatch.await(650, TimeUnit.MILLISECONDS))
+        assertTrue(client.hasActiveTurn())
+        assertEquals(0, r.reconcileRequests.get())
+        replacement.send(harness.eventFrame("message.start", null, "live-1"))
+        replacement.send(harness.eventFrame("message.complete", buildJsonObject { put("text", "eventually started") }, "live-1"))
+        assertTrue(r.completeLatch.await(5, TimeUnit.SECONDS))
+        assertEquals(1, r.reconcileRequests.get())
+        assertTrue(harness.rpcLog.none { it.first == "session.interrupt" })
+    }
+
     @Test
     fun `failed rejoin surfaces stream error`() {
         val r = Recorder()
@@ -5386,7 +5647,7 @@ class GatewayChatClientTest {
     }
 
     @Test
-    fun `active session idle never settles passively observed turn`() = runBlocking {
+    fun `active session idle settles admitted unsolicited stream without runtime control`() = runBlocking {
         val recorder = Recorder()
         client.setUnsolicitedTurnProvider {
             GatewayInboundTurnRegistration(recorder.callbacks) { true }
@@ -5409,7 +5670,8 @@ class GatewayChatClientTest {
         )
 
         client.listActiveSessions()
-        assertFalse(recorder.completeLatch.await(250, TimeUnit.MILLISECONDS))
+        assertTrue(recorder.completeLatch.await(5, TimeUnit.SECONDS))
+        assertEquals(1, recorder.reconcileRequests.get())
         assertTrue(harness.rpcLog.none { it.first == "session.interrupt" })
 
         serverWs.send(
@@ -5420,6 +5682,29 @@ class GatewayChatClientTest {
             ),
         )
         assertTrue(recorder.completeLatch.await(5, TimeUnit.SECONDS))
+        Thread.sleep(150)
+        assertEquals(1, recorder.completions.get())
+    }
+
+    @Test
+    fun `unsolicited progress invalidates delayed idle snapshot`() = runBlocking {
+        val r = Recorder()
+        client.setUnsolicitedTurnProvider { GatewayInboundTurnRegistration(r.callbacks) { true } }
+        assertTrue(client.prewarmAwait("stored-session"))
+        val ws = harness.awaitServerSocket()
+        ws.send(harness.eventFrame("message.start", null, "live-resumed"))
+        awaitCondition { r.starts.get() == 1 }
+        harness.suppressAckMethods += "session.active_list"
+        val snapshot = scope.async { client.listActiveSessions() }
+        val ack = harness.awaitPendingAck()
+        ws.send(harness.eventFrame("message.delta", buildJsonObject { put("text", "new progress") }, "live-resumed"))
+        awaitCondition { r.textDeltas.isNotEmpty() }
+        harness.releaseAck(ack, exactActiveSessionPayload("idle", "live-resumed", "stored-session"))
+        assertTrue(snapshot.await() is GatewayActiveSessionsResult.Success)
+        assertEquals(0, r.completions.get())
+        assertEquals(0, r.reconcileRequests.get())
+        ws.send(harness.eventFrame("message.complete", buildJsonObject { put("text", "new progress") }, "live-resumed"))
+        assertTrue(r.completeLatch.await(5, TimeUnit.SECONDS))
     }
 
     @Test
@@ -5561,98 +5846,53 @@ class GatewayChatClientTest {
     }
 
     @Test
-    fun `compacting status extends watchdog until a later completion`() {
-        rebuildClient(turnIdleTimeoutMs = 250L, compactingTimeoutMs = 1_000L)
+    fun `status updates never interrupt a quiet turn`() {
+        rebuildClient(turnIdleTimeoutMs = 250L)
         val r = Recorder()
-        client.sendTurn(null, "compact once", null, r.callbacks) { r.preflightFailures += it }
-        val serverWs = harness.awaitServerSocket()
-        harness.awaitRpc("prompt.submit")
-
-        serverWs.send(
-            harness.eventFrame(
-                "status.update",
-                buildJsonObject { put("kind", "compacting") },
-                "live-1",
-            ),
-        )
-        Thread.sleep(500)
-
-        assertTrue("normal idle watchdog fired during compaction: ${r.errors}", r.errors.isEmpty())
-        assertTrue(harness.rpcLog.none { it.first == "session.interrupt" })
-
-        serverWs.send(
-            harness.eventFrame("message.complete", buildJsonObject { put("text", "done") }, "live-1"),
-        )
-        assertTrue("turn never completed", r.completeLatch.await(5, TimeUnit.SECONDS))
-        assertTrue(r.errors.isEmpty())
-        assertTrue(r.preflightFailures.isEmpty())
-    }
-
-    @Test
-    fun `compacting heartbeats rearm watchdog beyond one compaction lease`() {
-        rebuildClient(turnIdleTimeoutMs = 200L, compactingTimeoutMs = 500L)
-        val r = Recorder()
-        client.sendTurn(null, "compact with heartbeats", null, r.callbacks) { r.preflightFailures += it }
+        client.sendTurn(null, "ordinary status", null, r.callbacks) { r.preflightFailures += it }
         val serverWs = harness.awaitServerSocket()
         harness.awaitRpc("prompt.submit")
-
-        repeat(3) {
+        // Compaction is the longest healthy silence; nothing kills on silence
+        // now, so it needs no special lease.
+        listOf("compacting", "process").forEach { kind ->
             serverWs.send(
                 harness.eventFrame(
                     "status.update",
-                    buildJsonObject { put("kind", "compacting") },
+                    buildJsonObject { put("kind", kind) },
                     "live-1",
                 ),
             )
-            Thread.sleep(300)
+            Thread.sleep(400)
         }
 
-        assertTrue("compaction lease was not rearmed: ${r.errors}", r.errors.isEmpty())
+        assertTrue("quiet turn was failed: ${r.errors}", r.errors.isEmpty())
         assertTrue(harness.rpcLog.none { it.first == "session.interrupt" })
 
-        serverWs.send(
-            harness.eventFrame("message.complete", buildJsonObject { put("text", "done") }, "live-1"),
-        )
+        serverWs.send(harness.eventFrame("message.complete", buildJsonObject { put("text", "done") }, "live-1"))
         assertTrue("turn never completed", r.completeLatch.await(5, TimeUnit.SECONDS))
         assertTrue(r.errors.isEmpty())
         assertTrue(r.preflightFailures.isEmpty())
     }
 
     @Test
-    fun `non compacting status keeps the ordinary watchdog`() {
-        rebuildClient(turnIdleTimeoutMs = 250L, compactingTimeoutMs = 2_000L)
-        val r = Recorder()
-        client.sendTurn(null, "ordinary status", null, r.callbacks) { r.preflightFailures += it }
-        val serverWs = harness.awaitServerSocket()
-        harness.awaitRpc("prompt.submit")
-        serverWs.send(
-            harness.eventFrame(
-                "status.update",
-                buildJsonObject { put("kind", "process") },
-                "live-1",
-            ),
-        )
-
-        assertTrue("ordinary watchdog never fired", r.completeLatch.await(5, TimeUnit.SECONDS))
-        assertTrue("expected a stream error from the watchdog", r.errors.isNotEmpty())
-        assertTrue(r.preflightFailures.isEmpty())
-        harness.awaitRpc("session.interrupt")
-    }
-
-    @Test
-    fun `idle watchdog fires when events stop flowing`() {
+    fun `idle watchdog never interrupts a quiet turn`() {
         rebuildClient(turnIdleTimeoutMs = 500L)
         val r = Recorder()
         client.sendTurn(null, "stalls", null, r.callbacks) { r.preflightFailures += it }
         val serverWs = harness.awaitServerSocket()
         harness.awaitRpc("prompt.submit")
         serverWs.send(harness.eventFrame("message.delta", buildJsonObject { put("text", "partial") }, "live-1"))
-        // …then silence: the idle watchdog must fail the turn as a STREAM
-        // error (never a preflight fallback — the turn started server-side)
-        // and interrupt the server so it stops generating.
-        assertTrue("watchdog never fired", r.completeLatch.await(5, TimeUnit.SECONDS))
-        assertTrue("expected a stream error from the watchdog", r.errors.isNotEmpty())
+        // Then silence across several idle windows: no error, no interrupt,
+        // and a late completion still lands.
+        Thread.sleep(2_000)
+        assertTrue("quiet turn was failed: ${r.errors}", r.errors.isEmpty())
+        assertTrue(harness.rpcLog.none { it.first == "session.interrupt" })
+
+        serverWs.send(
+            harness.eventFrame("message.complete", buildJsonObject { put("text", "late but done") }, "live-1"),
+        )
+        assertTrue("late completion was dropped", r.completeLatch.await(5, TimeUnit.SECONDS))
+        assertTrue(r.errors.isEmpty())
         assertTrue(r.preflightFailures.isEmpty())
-        harness.awaitRpc("session.interrupt")
     }
 }
diff --git a/app/src/test/kotlin/com/hermesandroid/relay/viewmodel/ChatViewModelGatewayInboundTurnTest.kt b/app/src/test/kotlin/com/hermesandroid/relay/viewmodel/ChatViewModelGatewayInboundTurnTest.kt
index 174cd038a..795169460 100644
--- a/app/src/test/kotlin/com/hermesandroid/relay/viewmodel/ChatViewModelGatewayInboundTurnTest.kt
+++ b/app/src/test/kotlin/com/hermesandroid/relay/viewmodel/ChatViewModelGatewayInboundTurnTest.kt
@@ -2170,6 +2170,38 @@ class ChatViewModelGatewayInboundTurnTest {
         assertTrue(viewModel.serverCommands.value.isEmpty())
     }
 
+    @Test
+    fun unsolicitedMissingTerminalReplacesPartialTextWithAuthoritativeHistory() =
+        verifyUnsolicitedMissingTerminal(boundedWindow = false)
+
+    @Test
+    fun unsolicitedMissingTerminalRecoversWhenLatestHistoryWindowDoesNotGrow() =
+        verifyUnsolicitedMissingTerminal(boundedWindow = true)
+
+    private fun verifyUnsolicitedMissingTerminal(boundedWindow: Boolean) {
+        val spoken = mutableListOf<String>()
+        viewModel.gatewayInboundSpeechReceiver = { { text -> spoken.add(text); Unit } }
+        if (boundedWindow) {
+            handler.loadMessageHistory(persistedAnswerHistory("Earlier saved answer", "earlier"))
+        }
+        serverWs.send(gatewayHarness.eventFrame("message.start", null, "live-resumed"))
+        serverWs.send(gatewayHarness.eventFrame("message.delta", buildJsonObject {
+            put("text", "Partial background answer")
+        }, "live-resumed"))
+        awaitCondition { handler.messages.value.lastOrNull()?.content == "Partial background answer" }
+        persistedHistory = persistedAnswerHistory()
+        serverWs.send(gatewayHarness.eventFrame("session.info", buildJsonObject {
+            put("running", false)
+        }, "live-resumed"))
+
+        awaitCondition {
+            !handler.isStreaming.value &&
+                handler.messages.value.singleOrNull()?.content == BACKGROUND_ANSWER
+        }
+        assertTrue(spoken.isEmpty())
+        assertFalse(gatewayHarness.rpcLog.any { it.first in setOf("prompt.submit", "session.interrupt") })
+    }
+
     @Test
     fun unsolicitedGatewayCompletionAppearsAsOneAssistantTurnAndSettles() {
         val spoken = mutableListOf<String>()
diff --git a/docs/decisions.md b/docs/decisions.md
index ce3b20da3..7f8044a19 100644
--- a/docs/decisions.md
+++ b/docs/decisions.md
@@ -3741,11 +3741,16 @@ prewarm; it did not cover this already-active turn state.
 
 **Decision.** A Gateway turn may settle from `session.activate`, exact-session
 `session.info`, or an exact live/durable `session.active_list` row only when the
-turn is Android-owned, has already received turn-scoped activity, and upstream
-reports `running=false` or Idle. The active-list request captures the exact turn
+turn has already received exact-session turn-scoped activity or its owned
+reconnect activation confirmed `running=true` without a queued successor, and
+upstream subsequently reports `running=false` or Idle. The active-list request captures the exact turn
 and its progress generation; a session/profile switch, cancellation, newer turn,
 or intervening live event rejects the delayed snapshot. Pre-start idle snapshots
-and passively observed Desktop/TUI turns are never eligible. This backstop is a
+are never eligible. A lost start frame can therefore recover through positive
+running proof without treating an Idle-only startup snapshot as completion.
+Exact unsolicited turns already admitted to the local stream
+may also settle from read-only state; this grants no authority to attach or stop
+the producing client's runtime. This backstop is a
 successful server-owned settle, not cancellation or transport failure: Android
 completes the local stream, keeps the durable session identity, performs bounded
 identity-fenced history reconciliation, consumes one late terminal without
@@ -3766,6 +3771,23 @@ ownership, Relay remains optional, and unmodified upstream compatibility is
 preserved. Physical certification across the reported device/network matrix
 remains tracked in `docs/project/TODO.md`.
 
+**Quiet-turn policy (2026-10-07).** Client-visible silence is not evidence that a
+turn failed. The idle watchdog performs a read-only `session.active_list` check
+using the same exact runtime/durable identity and progress-generation fences.
+Working, Waiting, Starting, absent, unsupported, and failed reads do not imply
+completion. Exact Idle settles the local stream and reconciles history.
+Confirmed server-state completion permits an authoritative bounded history read
+when the final text is unknown. Partial live text must not require an exact saved
+text match, nor must the count grow when an older row leaves the latest window.
+Ordinary completion after reconnect retains the final-text persistence check.
+Quiet tools, provider waits, and compaction retain the stream; there are no per-kind
+kill leases or short post-retarget failure deadline. Pending native requests
+retain their server-owned deadline. Explicit cancellation remains the sole
+turn-handle path to `session.interrupt`; unsolicited reconnect warms only the
+socket and reads the existing open-request snapshot to restore or retire pending
+asks without activating the producer's runtime.
+Client shutdown detaches an unsolicited stream without cancelling its runtime.
+
 ---
 
 ## ADR 65 — Gateway client contracts use reusable on-demand scenario fixtures
diff --git a/docs/gateway-contract-testing.md b/docs/gateway-contract-testing.md
index ce374e73c..06656a00e 100644
--- a/docs/gateway-contract-testing.md
+++ b/docs/gateway-contract-testing.md
@@ -68,6 +68,7 @@ the upstream contract identifiers it depends on.
 |---|---|
 | `initial_history_bind` | Durable, profile-scoped history is already available when the client resumes and first binds its rendered transcript |
 | `ordinary_turn` | Normal message start, deltas, completion, and persisted history |
+| `quiet_turn_recovery` | Quiet tool and provider waits retain the local stream; a subsequent unsolicited turn missing its final frame settles from exact server state and HTTP history |
 | `notification_terminal_outcomes` | Successful, failed, and interrupted terminal envelopes remain distinguishable for detached-turn alerts |
 | `unsolicited_voice_completions` | One submitted turn followed by live same-session process, watch, and delegation answers, including duplicate start/terminal frames; Standard Voice receives each admitted answer once |
 | `clarify_legacy` | Top-level single question and unkeyed `clarify.respond` |
@@ -90,6 +91,14 @@ the upstream contract identifiers it depends on.
 | `active_status_unsupported` | An older Gateway returns JSON-RPC method-not-found; the client retains Unknown rather than inventing Idle or Working |
 | `cross_client_observation` | A second client observes a Desktop-owned working session through active status and history without resume, activate, submit, or interrupt; the producing client receives the terminal event |
 
+The activation and active-list contracts also cover a start frame lost during
+socket replacement. `GatewayChatClientTest` verifies that exact running activation
+establishes activity proof before a later scoped Idle state, while Idle alone
+cannot complete a pre-start turn. The embedded real-socket fixture in
+`GatewayForegroundRecoveryInstrumentedTest#lostStartRunningActivation_recoversEndedTurnFromHistory`
+verifies that the rendered stream settles and loads the saved answer without a
+duplicate submit or an interrupt, even when no original turn frame was received.
+
 Activity receipts join canonical completion metadata by exact delegation identity
 within the connection/profile/session owner. Process notices use the canonical
 process ID; ambiguous reused process generations do not attach cached output.
@@ -182,6 +191,16 @@ redacted.
 
 ## Current-upstream conformance
 
+For quiet-turn regression coverage, run the `quiet_turn_recovery` fixture and
+select `GatewayExternalFixtureInstrumentedTest#quietTurns_surviveToolAndProviderSilenceAndRecoverUnsolicitedHistory`
+on `standardPhoneApi36`, with `gatewayFixtureBaseUrl` pointing to the host fixture
+(for example `http://10.0.2.2:8765` on the emulator). The adapter shortens only
+the diagnostic/probe interval, observes streaming across multiple quiet windows,
+then verifies final HTTP history, one submit, and zero interrupts or activations.
+The focused `GatewayChatClientTest` wire suite additionally covers read-only Idle
+settlement of exact unsolicited streams, delayed/stale snapshots, pending asks,
+explicit Stop, and route replacement. These lanes make no physical-device claim.
+
 The `secure_link_gateway_auth` scenario exercises query and subprotocol ticket
 admission through the real Secure Link proxy, single-use rejection, a completed
 turn, and live-session activation after reconnect. Run its wire regression with
diff --git a/docs/spec.md b/docs/spec.md
index 1b5a1cd33..f3300c2a5 100644
--- a/docs/spec.md
+++ b/docs/spec.md
@@ -723,7 +723,7 @@ Bottom navigation bar with 4 tabs:
 - **Bot group projection** — Android merges the bounded `ui_meta["hermes-bots-groups"]` v3 projection across gateways by durable room identity and newest revision. Rooms and recent messages are visibly read-only; Android does not create, rename, disband, join, send, coordinate member turns, or become a second room-log authority. Binary room images are ignored at this metadata boundary.
 - **Session drawer** (swipe from left or hamburger icon) — session list with title, timestamp, message count. On wide Chat layouts (at least 840 dp available), a saved opt-in presentation choice can keep the same browser visible in a 320 dp sidebar; unpinning returns it to the on-demand drawer. Compact layouts keep the drawer, and supervised history restrictions suppress the sidebar. Create, switch, rename, delete, pin/unpin, and archive/restore. A profile switch marks the replacement list loading before clearing the previous profile's rows and keeps that state until the exact-profile fetch settles, so an empty-state claim never flashes before server truth arrives. The process-owned conversation binding is the single connection/profile/session identity for Chat; selecting an All Profiles row atomically makes its owner the selected agent and persists that profile/session, while merely browsing All Profiles changes no agent state. Lifecycle or locale-driven Activity recreation cannot replace an explicit binding with stale persisted state, and asynchronous list/history/mutation work is accepted only for the binding's exact namespace. A profile lock hides All Profiles and rejects stale/deep-linked cross-profile opens. The All Profiles browser mode otherwise survives Activity state restoration and refetches its rows after recreation. Pin and archive are durable upstream session fields loaded and patched through the owning connection/profile's Dashboard session API; Android does not keep a second local flag registry. Archived rows are requested explicitly so they remain restorable after recreation. Failed mutations roll back the optimistic row, while refresh and deletion reconcile from server truth. When a persisted title is absent, use upstream's first-user-message `preview`, matching the Hermes Desktop session picker; show "Untitled" only when neither value exists.
 - **Cold profile hydration** — a persisted named profile scopes its Dashboard session directory and last-session restore immediately, before `/api/profiles` metadata is available. Server-default selection waits for the lightweight active-profile scope. Roster, avatars, pets, skills, and model metadata never precede the first directory result. The startup sphere releases after route selection; Chat keeps identity and cached rows mounted while its existing animated status surfaces show Gateway wake, session restore, and directory loading.
-- **Authoritative session activity** — one composite registry keyed by connection, normalized profile, and durable session id drives the drawer, filters, grouping, animation, accessibility, and the visible composer. Exact pending approval/clarify/sudo/secret/MCP requests produce **Needs input**; the Gateway's process-wide `session.active_list` supplies **Starting**, **Working**, and **Idle**; an exact terminal, `session.info {running:false}`, or an exact live/durable active-list row reporting Idle can settle only the matching Android-owned turn and progress generation. Because active-list rows normally have no profile metadata, Android assigns a row through exact foreground/detached ownership already held by that client, explicit profile metadata if a future upstream sends it, or the currently selected passive session when its durable id has exactly one owner in the current connection directory. Duplicate same-id owners across profiles remain unresolved and create no status. Resolved rows from a partial snapshot may update their exact owners, but disappearance settles a scope only when the successful process-wide snapshot was completely and unambiguously resolved for it. Restart/checkpoint recovery is **Checking**; a failed or unsupported live refresh is **Unavailable**, never inferred Idle. REST `is_active` remains recency metadata only. `process.list` may add a separate **Background work** indicator and never keeps the parent conversation Working. Old socket generations, ambiguous bare session ids, delayed snapshots, and snapshots crossed by newer turn events cannot settle or revive a newer generation.
+- **Authoritative session activity** — one composite registry keyed by connection, normalized profile, and durable session id drives the drawer, filters, grouping, animation, accessibility, and the visible composer. Exact pending approval/clarify/sudo/secret/MCP requests produce **Needs input**; the Gateway's process-wide `session.active_list` supplies **Starting**, **Working**, and **Idle**; an exact terminal, `session.info {running:false}`, or an exact live/durable active-list row reporting Idle can settle only the matching locally admitted turn and progress generation. This includes unsolicited turns without granting runtime control. Silence during tools, provider waits, or compaction never interrupts or fails a turn; the quiet-period check reads authoritative active state and preserves pending native requests. Because active-list rows normally have no profile metadata, Android assigns a row through exact foreground/detached ownership already held by that client, explicit profile metadata if a future upstream sends it, or the currently selected passive session when its durable id has exactly one owner in the current connection directory. Duplicate same-id owners across profiles remain unresolved and create no status. Resolved rows from a partial snapshot may update their exact owners, but disappearance settles a scope only when the successful process-wide snapshot was completely and unambiguously resolved for it. Restart/checkpoint recovery is **Checking**; a failed or unsupported live refresh is **Unavailable**, never inferred Idle. REST `is_active` remains recency metadata only. `process.list` may add a separate **Background work** indicator and never keeps the parent conversation Working. Old socket generations, ambiguous bare session ids, delayed snapshots, and snapshots crossed by newer turn events cannot settle or revive a newer generation.
 - **Concurrent Gateway chats** — switching sessions, profiles, drafts, or Threads detaches the visible Android-owned turn without sending `session.interrupt`; each Android-owned running chat keeps a connection/profile/session-scoped checkpoint and reattaches to its live Gateway session when reopened. Opening, foregrounding, or selecting a saved session without that exact checkpoint is read-only observation: Android warms only the socket, reads profile-scoped history, and polls `session.active_list` without `session.resume`, `session.activate`, `prompt.submit`, or `session.interrupt`. A Desktop/TUI-owned turn therefore remains owned by its producing client; Android refreshes persisted progress and performs one final history read when the runtime settles. Explicit send/config actions may resume the destination session, explicit Stop still interrupts, and Direct API compatibility chat stays single-stream and cancels on navigation.
 - **Queued Gateway follow-ups** — every local queued item is scoped to its originating connection, profile, stored session, transport, and predecessor run; only that predecessor's completion can make it eligible, and switching sessions shows only that session's queue. Removing or pulling a queued item into an empty composer reconnects its successor to the removed item's predecessor. Restored text queues retain the complete chain; unavailable/deleted destinations and non-restorable attachment queues fail visibly instead of following the current composer. Stop pauses pending work; Resume explicitly continues the exact destination, and Clear is the separate discard action. Queue-only checkpoints restore paused messages without reattaching the stopped turn. Drained messages add `queued: true` to `prompt.submit`; ordinary sends omit the field. Authoritative submit rejections (`4004`, `4018`, `4028`, `4029`, `4030`, `4090`, `5008`, `5070`, and `5071`) preserve the server message and never fall through to API-server SSE.
 - **Busy Chat input** — Chat settings owns the device-local Correct now / Queue next default, initially correction. Chat and Bot Chat show the effective action in a slim translucent rear tray whose lower edge sits beneath the foreground composer. Plain-text choices slide upward without covering the input; selecting one overrides the current message and returns to the saved default after send. Attachments, pending input, context compaction, slash commands, and unavailable or policy-disabled correction use the queue without replacing the saved preference. Stop remains separate from sending typed text. Pulling a queued message out for editing keeps that send queued unless the user explicitly selects correction.
diff --git a/scripts/check-gateway-scenario-conformance.py b/scripts/check-gateway-scenario-conformance.py
index 8a9259af8..fa8251254 100644
--- a/scripts/check-gateway-scenario-conformance.py
+++ b/scripts/check-gateway-scenario-conformance.py
@@ -206,14 +206,14 @@ def _check_session_initialization(server: SourceFile) -> CheckResult:
         return CheckResult(SESSION_INITIALIZATION, False, (), str(exc))
 
 
-def _check_settled_info(server: SourceFile) -> CheckResult:
+def _check_settled_info(server: SourceFile, helper_source: SourceFile, turn_source: SourceFile) -> CheckResult:
     contract = GATEWAY_SETTLED_INFO
     try:
         info = server.function("_session_info")
-        helper = server.function("_emit_settled_session_info")
-        turn = server.function("_run_prompt_submit")
+        helper = helper_source.function("_emit_settled_session_info")
+        turn = turn_source.function("_run_prompt_submit")
         info_strings = _string_constants(info)
-        helper_text = server.segment(helper)
+        helper_text = helper_source.segment(helper)
         if "running" not in info_strings or ".get(\"running\")" not in server.segment(info):
             raise ValueError("session.info no longer derives running from live session state")
         if '"session.info"' not in helper_text or "_session_info(" not in helper_text:
@@ -229,8 +229,8 @@ def _check_settled_info(server: SourceFile) -> CheckResult:
             True,
             (
                 server.evidence(info, "session.info includes running"),
-                server.evidence(helper, "settled session.info emission"),
-                f"{SERVER}:{running_line}:running=false before line {settle_line}",
+                helper_source.evidence(helper, "settled session.info emission"),
+                f"{turn_source.relative}:{running_line}:running=false before line {settle_line}",
             ),
         )
     except ValueError as exc:
@@ -586,7 +586,8 @@ def _check_server_requests(root: Path) -> CheckResult:
                 if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
                 and node.func.id == "server_request" and node.args and isinstance(node.args[0], ast.Constant)}
     passed = (
-        expected == declared
+        # Additive request methods can be declined with JSON-RPC method-not-found.
+        expected <= declared
         and {"jsonrpc", "id", "method", "params", "session_id"} <= _string_constants(frame)
         and "answers" in _string_constants(snapshot)
         and {"error", "result", "answers"} <= _string_constants(resolve)
@@ -640,7 +641,11 @@ def audit_sources(root: Path, requirements: Iterable[str]) -> list[CheckResult]:
     prompt_methods = SourceFile(root, PROMPT_METHODS)
     active_sessions = SourceFile(root, ACTIVE_SESSIONS)
     api = SourceFile(root, API_SERVER)
-    source_files = (server, methods, prompt_methods, active_sessions, api)
+    registry = (
+        SourceFile(root, "tui_gateway/session_registry.py")
+        if (root / "tui_gateway/session_registry.py").is_file() else server
+    )
+    source_files = (server, registry, methods, prompt_methods, active_sessions, api)
     fork_hits = [
         f"{source.relative}:{marker}"
         for source in source_files
@@ -661,10 +666,16 @@ def audit_sources(root: Path, requirements: Iterable[str]) -> list[CheckResult]:
             SourceFile(root, "tui_gateway/prompt_turn.py")
             if (root / "tui_gateway/prompt_turn.py").is_file() else server
         ),
-        GATEWAY_SETTLED_INFO: lambda: _check_settled_info(server),
-        SESSION_ACTIVATE: lambda: _check_activate(server, methods),
+        GATEWAY_SETTLED_INFO: lambda: _check_settled_info(
+            server,
+            SourceFile(root, "tui_gateway/session_workdir.py")
+            if (root / "tui_gateway/session_workdir.py").is_file() else server,
+            SourceFile(root, "tui_gateway/prompt_turn.py")
+            if (root / "tui_gateway/prompt_turn.py").is_file() else server,
+        ),
+        SESSION_ACTIVATE: lambda: _check_activate(registry, methods),
         SESSION_RESUME: lambda: _check_resume(methods),
-        SESSION_ACTIVE_LIST: lambda: _check_active_list(server, methods),
+        SESSION_ACTIVE_LIST: lambda: _check_active_list(registry, methods),
         SESSION_EXCLUSIVE_SUBMIT: lambda: _check_session_exclusive_submit(
             server, methods, prompt_methods, active_sessions,
         ),
diff --git a/scripts/tests/check_gateway_scenario_conformance_test.py b/scripts/tests/check_gateway_scenario_conformance_test.py
index 58333dcc8..4e684bb96 100644
--- a/scripts/tests/check_gateway_scenario_conformance_test.py
+++ b/scripts/tests/check_gateway_scenario_conformance_test.py
@@ -218,6 +218,11 @@ def test_native_contract_rejects_missing_capability_gate_and_methods(self):
         contract.write_text(contract.read_text().replace("'clarify'", "'clarify.future'"), encoding="utf-8")
         self.assertFalse(module.audit_sources(self.root, [module.SERVER_REQUESTS])[0].passed)
 
+    def test_native_contract_allows_additive_request_methods(self):
+        contract = self.root / "tui_gateway/contracts/server_requests.py"
+        contract.write_text(contract.read_text() + '\nserver_request("setup_choose", params=Params, result=Result)\n', encoding="utf-8")
+        self.assertTrue(module.audit_sources(self.root, [module.SERVER_REQUESTS])[0].passed)
+
     def setUp(self):
         self.temp = tempfile.TemporaryDirectory()
         self.root = Path(self.temp.name)
@@ -264,6 +269,17 @@ def test_complete_contract_fixture_passes(self):
         self.assertEqual(module.ALL_CONTRACTS, tuple(result.contract for result in results))
         self.assertTrue(all(result.passed for result in results), results)
 
+    def test_extracted_registry_keeps_identity_and_status_requirements(self):
+        path = self.root / "tui_gateway/session_registry.py"
+        path.write_text(SERVER_SOURCE, encoding="utf-8")
+        requirements = (module.SESSION_ACTIVATE, module.SESSION_ACTIVE_LIST)
+        self.assertTrue(all(result.passed for result in module.audit_sources(self.root, requirements)))
+        # Intact legacy helpers must not mask a broken extracted implementation.
+        path.write_text(SERVER_SOURCE.replace('"session_key"', '"other_key"'), encoding="utf-8")
+        self.assertTrue(all(not result.passed for result in module.audit_sources(self.root, requirements)))
+        path.write_text(SERVER_SOURCE.replace('"waiting"', '"other_state"'), encoding="utf-8")
+        self.assertFalse(module.audit_sources(self.root, (module.SESSION_ACTIVE_LIST,))[0].passed)
+
     def test_ticket_protocol_requires_public_selection_and_single_use_admission(self):
         for relative, before, after in (
             ("hermes_cli/web_server_chat.py", "consume_ticket(ticket)", "accept(ticket)"),
@@ -352,6 +368,16 @@ def test_settlement_must_clear_running_before_info(self):
         self.assertFalse(result.passed)
         self.assertIn("before settled", result.problem)
 
+    def test_decomposed_settlement_retains_ordering_check(self):
+        helper = self.root / "tui_gateway/session_workdir.py"
+        helper.write_text('def _emit_settled_session_info(sid, session, agent):\n    _emit("session.info", sid, _session_info(agent, session))\n', encoding="utf-8")
+        prompt = self.root / "tui_gateway/prompt_turn.py"
+        source = 'def _run_prompt_submit():\n    try:\n        pass\n    finally:\n        session["running"] = False\n        _emit_settled_session_info(sid, session, agent)\n'
+        prompt.write_text(source, encoding="utf-8")
+        self.assertTrue(module.audit_sources(self.root, (module.GATEWAY_SETTLED_INFO,))[0].passed)
+        prompt.write_text(source.replace('session["running"] = False', 'session["running"] = True'), encoding="utf-8")
+        self.assertFalse(module.audit_sources(self.root, (module.GATEWAY_SETTLED_INFO,))[0].passed)
+
     def test_manifest_selects_known_contract_subset(self):
         manifest = self.root / "scenario.json"
         manifest.write_text(
diff --git a/test-fixtures/vanilla-gateway/tests/test_fixture.py b/test-fixtures/vanilla-gateway/tests/test_fixture.py
index d9246249c..eedb2aa24 100644
--- a/test-fixtures/vanilla-gateway/tests/test_fixture.py
+++ b/test-fixtures/vanilla-gateway/tests/test_fixture.py
@@ -18,6 +18,19 @@
 
 
 class FixtureTestCase(unittest.IsolatedAsyncioTestCase):
+    async def test_quiet_turns_finish_and_publish_authoritative_unsolicited_state(self) -> None:
+        fixture, base_url = await self.start("quiet_turn_recovery")
+        ws, _ = await self.connect(base_url)
+        await self.rpc(ws, 1, "prompt.submit", {"text": "Exercise quiet turns."})
+        frames = await self.frames_until(ws, lambda f: f.get("params", {}).get("type") == "session.info")
+        events = [f["params"] for f in frames if f.get("method") == "event"]
+        self.assertEqual(2, sum(e["type"] == "message.start" for e in events))
+        self.assertEqual(1, sum(e["type"] == "message.complete" for e in events))
+        self.assertFalse(events[-1]["payload"]["running"])
+        self.assertFalse(fixture.running)
+        async with self.session.get(f"{base_url}/api/sessions/{fixture.scenario.stored_session_id}/messages") as response:
+            self.assertIn("Background work finished quietly.", await response.text())
+
     async def test_dashboard_setup_signin_expiry_and_loopback_are_distinct(self) -> None:
         fixture, base_url = await self.start("dashboard_onboarding")
         async with self.session.get(f"{base_url}/api/status") as response:
diff --git a/test-fixtures/vanilla-gateway/vanilla_gateway/scenarios/quiet_turn_recovery.json b/test-fixtures/vanilla-gateway/vanilla_gateway/scenarios/quiet_turn_recovery.json
new file mode 100644
index 000000000..2d5500dde
--- /dev/null
+++ b/test-fixtures/vanilla-gateway/vanilla_gateway/scenarios/quiet_turn_recovery.json
@@ -0,0 +1,32 @@
+{
+  "name": "quiet_turn_recovery",
+  "live_session_id": "fixture-live-1",
+  "stored_session_id": "20260821_120000_fixture",
+  "profile": "default",
+  "contract_requirements": ["gateway.message_complete", "gateway.settled_session_info", "gateway.session_active_list"],
+  "turns": [{
+    "steps": [
+      {"op": "event", "type": "message.start"},
+      {"op": "event", "type": "tool.start", "payload": {"tool_id": "quiet-tool", "name": "terminal"}},
+      {"op": "sleep", "milliseconds": 1000},
+      {"op": "event", "type": "tool.complete", "payload": {"tool_id": "quiet-tool", "name": "terminal"}},
+      {"op": "sleep", "milliseconds": 1000},
+      {"op": "persist", "messages": [
+        {"id": 1, "role": "user", "content": "Exercise quiet turns.", "timestamp": 1.0},
+        {"id": 2, "role": "assistant", "content": "Quiet tool finished.", "timestamp": 2.0, "finish_reason": "stop"}
+      ]},
+      {"op": "set_running", "value": false},
+      {"op": "event", "type": "message.complete", "payload": {"text": "Quiet tool finished.", "status": "complete"}},
+      {"op": "sleep", "milliseconds": 1000},
+      {"op": "set_running", "value": true},
+      {"op": "event", "type": "message.start"},
+      {"op": "event", "type": "message.delta", "payload": {"text": "Background work"}},
+      {"op": "sleep", "milliseconds": 1500},
+      {"op": "persist", "messages": [
+        {"id": 3, "role": "assistant", "content": "Background work finished quietly.", "timestamp": 3.0, "finish_reason": "stop"}
+      ]},
+      {"op": "set_running", "value": false},
+      {"op": "event", "type": "session.info", "payload": {"running": false}}
+    ]
+  }]
+}