fix(android): keep passive gateway observation read-only

CHANGELOG.md

@@ -17,8 +17,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/), and this
 
 ### Fixed
 
+- **Opening Android no longer claims or interrupts a turn already running in Hermes Desktop/TUI.** Passive foreground and session browsing now use read-only Gateway status plus profile-scoped history; live-session resume remains reserved for explicit Android actions and exact Android-owned recovery.
 - **The visible Android Sphere keeps its smooth procedural motion across startup and chat.** Backgrounded and motion-disabled surfaces remain still without reducing foreground animation to a stepped ambient pulse.
-- **Android New Chat keeps the current profile and stays fresh across profile switches.** Starting from All Profiles no longer forces the literal default profile, choosing another profile from an empty draft no longer reopens that profile's previous session after route settlement or restart, and leaving a provisional phone Thread cannot route the next turn to its old chat under the new profile.
 
 ### Removed
 

TODO.md

@@ -26,6 +26,9 @@ model device-certified:
   renders as Working.
 - Run a background process that outlives its parent turn and verify Background
   work remains separate from the conversation's Idle state.
+- On a physical phone, open and repeatedly foreground Android while the same
+  session is working in official Desktop/TUI; verify Android sends no live
+  attach/interrupt RPC, the producer completes, and final history appears.
 - Pursue an upstream `session.active_list` profile field/filter or an aggregate
   activity route with explicit profile ownership so multi-profile clients do
   not need to resolve process-wide rows from durable keys.

app/src/androidTest/kotlin/com/hermesandroid/relay/viewmodel/GatewayForegroundRecoveryInstrumentedTest.kt

@@ -239,6 +239,60 @@ class GatewayForegroundRecoveryInstrumentedTest {
         assertEquals(0, fixture.requestsTo("/v1/chat/completions"))
     }
 
+    @Test
+    fun desktopOwnedTurn_remainsReadOnlyAcrossAndroidForegroundLifecycle() {
+        viewModel.setChatVisible(false)
+        viewModel.updateGatewayClient(null)
+        gatewayClient.shutdown()
+        gatewayScope.cancel()
+
+        val controlMethods = setOf(
+            "session.resume",
+            "session.activate",
+            "session.interrupt",
+            "prompt.submit",
+        )
+        val baseline = controlMethods.associateWith(fixture::rpcCount)
+        val baselineActiveList = fixture.rpcCount("session.active_list")
+        fixture.activeSessionStatus = "working"
+        gatewayScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+        val okHttp = OkHttpClient()
+        gatewayClient = GatewayChatClient(
+            initialDashboardClient = DashboardApiClient(
+                baseUrl = fixture.server.url("/").toString().trimEnd('/'),
+                okHttpClient = okHttp,
+            ),
+            okHttpClient = okHttp,
+            callbackDispatcher = { block -> Handler(Looper.getMainLooper()).post(block) },
+            scope = gatewayScope,
+            reconnectJitterUnit = { 0.0 },
+        )
+        viewModel.setChatTurnCheckpointStore(null)
+        viewModel.updateGatewayClient(gatewayClient)
+
+        viewModel.setChatVisible(true)
+        compose.activityRule.scenario.moveToState(Lifecycle.State.STARTED)
+        compose.activityRule.scenario.moveToState(Lifecycle.State.RESUMED)
+        viewModel.setChatVisible(false)
+        viewModel.setChatVisible(true)
+        fixture.awaitRpcCount("session.active_list", baselineActiveList + 1)
+
+        controlMethods.forEach { method ->
+            assertEquals(
+                "passive lifecycle sent $method",
+                baseline.getValue(method),
+                fixture.rpcCount(method),
+            )
+        }
+        viewModel.updateGatewayClient(null)
+        gatewayClient.shutdown()
+        assertEquals(
+            "observer teardown interrupted the Desktop turn",
+            baseline.getValue("session.interrupt"),
+            fixture.rpcCount("session.interrupt"),
+        )
+    }
+
     private companion object {
         const val STORED_SESSION_ID = "20260821_120000_fixture"
         const val LIVE_SESSION_ID = "fixture-live-1"
@@ -263,6 +317,9 @@ internal class AndroidGatewayContractFixture {
     @Volatile
     var recoveryRunning = false
 
+    @Volatile
+    var activeSessionStatus: String? = null
+
     private val listener = object : WebSocketListener() {
         override fun onOpen(webSocket: WebSocket, response: Response) {
             sockets.add(webSocket)
@@ -282,6 +339,18 @@ internal class AndroidGatewayContractFixture {
                 "session.activate" -> sessionSnapshot(
                     (params["session_id"] as? JsonPrimitive)?.contentOrNull ?: "fixture-live-1",
                 )
+                "session.active_list" -> buildJsonObject {
+                    put("sessions", kotlinx.serialization.json.buildJsonArray {
+                        activeSessionStatus?.let { status ->
+                            add(buildJsonObject {
+                                put("id", LIVE_SESSION_ID)
+                                put("session_key", STORED_SESSION_ID)
+                                put("status", status)
+                                put("last_active", 1.0)
+                            })
+                        }
+                    })
+                }
                 "prompt.submit", "session.interrupt" -> buildJsonObject { put("ok", true) }
                 else -> JsonObject(emptyMap())
             }
@@ -345,6 +414,15 @@ internal class AndroidGatewayContractFixture {
         error("Gateway RPC $method not observed; saw ${rpcLog.map { it.first }}")
     }
 
+    fun awaitRpcCount(method: String, count: Int) {
+        val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5)
+        while (System.nanoTime() < deadline) {
+            if (rpcCount(method) >= count) return
+            Thread.sleep(20)
+        }
+        error("Gateway RPC $method count $count not observed; saw ${rpcLog.map { it.first }}")
+    }
+
     fun requestsTo(path: String): Int = requestPaths.count { it.startsWith(path) }
 
     fun rpcCount(method: String): Int = rpcLog.count { it.first == method }
@@ -353,4 +431,9 @@ internal class AndroidGatewayContractFixture {
         allSockets.forEach { socket -> runCatching { socket.close(1001, "teardown") } }
         runCatching { server.shutdown() }
     }
+
+    private companion object {
+        const val STORED_SESSION_ID = "20260821_120000_fixture"
+        const val LIVE_SESSION_ID = "fixture-live-1"
+    }
 }

app/src/main/kotlin/com/hermesandroid/relay/network/upstream/GatewayChatClient.kt

@@ -906,6 +906,30 @@ class GatewayChatClient(
         scope.launch { prewarmAwait(storedSessionId) }
     }
 
+    /**
+     * Establish only the shared Gateway socket for read-only observation.
+     *
+     * `session.resume` and `session.activate` attach a live runtime to this
+     * transport. Opening Chat, foreground restoration, and selecting a saved
+     * transcript must not claim a turn that another Desktop/TUI client owns,
+     * so those paths use this socket-only warmup and observe through REST
+     * history plus `session.active_list` instead.
+     */
+    fun observe(onReady: (() -> Unit)? = null) {
+        scope.launch {
+            if (observeAwait() && onReady != null) callbackDispatcher(onReady)
+        }
+    }
+
+    /** Suspending [observe]; returns true once the read-only socket is ready. */
+    suspend fun observeAwait(): Boolean = try {
+        connectMutex.withLock { ensureConnected() }
+        true
+    } catch (e: Exception) {
+        Log.d(TAG, "Gateway observation warmup skipped: ${e.message}")
+        false
+    }
+
     /**
      * Suspending [prewarm]: establishes the socket and (when [storedSessionId]
      * is non-null) resumes the existing session, returning only once that work

app/src/main/kotlin/com/hermesandroid/relay/runtime/HermesRuntimeBinder.kt

@@ -196,7 +196,6 @@ internal class HermesRuntimeBinder(
         chat.profileSessionPinner = connection::setSessionPinned
         chat.profileSessionArchiver = connection::setSessionArchived
         chat.onSessionChanged = connection::saveLastSessionId
-        chat.onFreshDraftSelected = connection::saveFreshDraft
         chat.setDemoModeWiring(
             isDemo = { connection.isDemoMode.value },
             handler = { connection.chatHandler },

app/src/main/kotlin/com/hermesandroid/relay/ui/components/SessionDrawer.kt

@@ -215,6 +215,7 @@ fun SessionDrawerContent(
     /** Opens the separate Bot Mode messenger workspace; never changes drawer filters. */
     onOpenBotMode: (() -> Unit)? = null,
     onNewChat: () -> Unit,
+    onNewDefaultChat: (() -> Unit)? = null,
     onSelectSession: (String) -> Unit,
     onDeleteSession: (String) -> Unit,
     onRenameSession: (String, String) -> Unit,
@@ -534,7 +535,13 @@ fun SessionDrawerContent(
 
             // New Chat button
             Button(
-                onClick = onNewChat,
+                onClick = {
+                    if (showAllProfiles) {
+                        onNewDefaultChat?.invoke() ?: onNewChat()
+                    } else {
+                        onNewChat()
+                    }
+                },
                 modifier = Modifier.fillMaxWidth(),
                 enabled = newChatEnabled,
             ) {

app/src/main/kotlin/com/hermesandroid/relay/ui/screens/ChatScreen.kt

@@ -1119,12 +1119,16 @@ fun ChatScreen(
     }
 
     // Recover any durable in-flight chat checkpoint whenever Chat returns to
-    // the foreground. On Gateway this also pre-warms/re-attaches the socket;
-    // sessions-SSE falls back to bounded persisted-history reconciliation.
+    // the foreground. setChatVisible owns that edge; an ordinary Gateway open
+    // warms only the observation socket and never attaches a saved session.
     val appForeground by com.hermesandroid.relay.util.AppForegroundTracker.isForeground.collectAsState()
     LaunchedEffect(isGatewayTransport, appForeground, chatReady) {
-        chatViewModel.setChatVisible(appForeground && chatReady)
-        if (appForeground && chatReady) {
+        val chatVisible = appForeground && chatReady
+        val visibilityChanged = chatViewModel.setChatVisible(chatVisible)
+        if (isGatewayTransport && chatVisible && !visibilityChanged) {
+            // Gateway availability can settle after Chat was already visible.
+            // Repeat the socket-only warmup for that edge; ordinary observation
+            // still cannot resume or activate a session.
             chatViewModel.prewarmGateway()
         }
         if (isGatewayTransport && appForeground && chatReady) {
@@ -2385,15 +2389,8 @@ fun ChatScreen(
     }
     val selectProfileFromShelf: (com.hermesandroid.relay.data.Profile?) -> Unit = { profile ->
         if (AgentDisplay.profileSessionKey(profile?.name) != selectedProfileKey) {
-            val profileName = profile?.name
-            chatViewModel.selectProfileFromHeader(
-                profileName = profileName,
-                profile = profile,
-                contextKey = AgentDisplay.profileContextKey(
-                    connectionId = activeConnection?.id,
-                    profileName = profileName,
-                ),
-            )
+            connectionViewModel.selectProfile(profile)
+            chatViewModel.activateGatewayProfile(profile)
         }
     }
     val hasLiveConversationSurface = messages.isNotEmpty() || isStreaming
@@ -2476,6 +2473,25 @@ fun ChatScreen(
                         scope.launch { drawerState.close() }
                     }
                 },
+                onNewDefaultChat = {
+                    if (isProfileLocked) return@SessionDrawerContent
+                    val defaultProfile = agentProfiles.firstOrNull {
+                        it.name.equals("default", ignoreCase = true)
+                    } ?: com.hermesandroid.relay.data.Profile(
+                        name = "default",
+                        model = "",
+                        description = "Default",
+                    )
+                    val opened = chatViewModel.createProfileChat(
+                        profileName = "default",
+                        profile = defaultProfile,
+                        contextKey = AgentDisplay.profileContextKey(
+                            connectionId = activeConnection?.id,
+                            profileName = "default",
+                        ),
+                    )
+                    if (opened) scope.launch { drawerState.close() }
+                },
                 onSelectSession = { sessionId ->
                     chatViewModel.switchSession(sessionId)
                     scope.launch { drawerState.close() }

app/src/main/kotlin/com/hermesandroid/relay/viewmodel/ChatViewModel.kt

@@ -37,7 +37,6 @@ import com.hermesandroid.relay.data.applyMessageReaction
 import com.hermesandroid.relay.data.parseChatQuotedPrompt
 import com.hermesandroid.relay.data.prepareTextTransportAttachments
 import com.hermesandroid.relay.data.Profile
-import com.hermesandroid.relay.data.SessionTransport
 import com.hermesandroid.relay.data.ProactiveInboxEntry
 import com.hermesandroid.relay.data.RealtimeConversationContextMessage
 import com.hermesandroid.relay.data.RealtimeTurnTrace
@@ -409,6 +408,9 @@ class ChatViewModel : ViewModel() {
     private val sessionActivityGeneration = AtomicLong(0L)
     private val sessionActivityPollMutex = Mutex()
     private var sessionActivityPollJob: Job? = null
+    private var passiveGatewayHistoryRefreshJob: Job? = null
+    private var passivelyObservedGatewaySessionId: String? = null
+    private var passiveObservationCatchupPendingSessionId: String? = null
     private var sessionActivityDirectory: Set<SessionActivityOwner> = emptySet()
     private var lastProjectedProcessIds: Set<String> = emptySet()
     private var lastProjectedProcessOwner: SessionActivityOwner? = null
@@ -661,7 +663,6 @@ class ChatViewModel : ViewModel() {
 
     /** Callback to persist session ID — set by RelayApp */
     var onSessionChanged: ((String?) -> Unit)? = null
-    var onFreshDraftSelected: ((String?, SessionTransport) -> Unit)? = null
 
     /**
      * Send a user message into an agent **Thread** (a `source=phone` session)
@@ -680,7 +681,6 @@ class ChatViewModel : ViewModel() {
      */
     private data class PendingThread(val chatId: String, val name: String)
     private var pendingThread: PendingThread? = null
-    private val threadNavigationGeneration = AtomicLong(0L)
 
     /**
      * A "+ New Thread" whose first message has been sent — we're now polling for
@@ -696,17 +696,6 @@ class ChatViewModel : ViewModel() {
     )
     private var creatingThread: CreatingThread? = null
 
-    /**
-     * Provisional phone Threads are route-owned drafts, not transferable chat
-     * drafts. Leaving that surface retires only the pending local route; durable
-     * inbox/session rows and learned session-to-chat-id mappings stay intact.
-     */
-    private fun exitProvisionalThread() {
-        threadNavigationGeneration.incrementAndGet()
-        pendingThread = null
-        creatingThread = null
-    }
-
     /**
      * `sessionId` → phone-platform `chat_id`, learned for threads this app
      * created ([switchToCreatedThread]) or received a message in
@@ -2266,6 +2255,8 @@ class ChatViewModel : ViewModel() {
     }
 
     private suspend fun pollSessionActivity(client: GatewayChatClient) {
+        var hasPassiveCurrentLiveWork = false
+        var hasPassiveCatchupPending = false
         sessionActivityPollMutex.withLock {
             if (gatewayClient !== client || !chatVisible || streamingEndpoint != "gateway") return
             val generation = sessionActivityGeneration.get()
@@ -2286,6 +2277,43 @@ class ChatViewModel : ViewModel() {
             when (val result = client.listActiveSessions()) {
                 is GatewayActiveSessionsResult.Success -> {
                     if (gatewayClient !== client || generation != sessionActivityGeneration.get()) return
+                    val currentStoredId = currentOwner?.storedSessionId
+                    val passiveCurrentRows = if (currentStoredId == null) {
+                        emptyList()
+                    } else {
+                        result.sessions.filter { row ->
+                            row.storedSessionId == currentStoredId &&
+                                client.knownSessionOwner(row.runtimeSessionId) == null
+                        }
+                    }
+                    hasPassiveCurrentLiveWork = passiveCurrentRows.any { row ->
+                        row.status != GatewayActiveSessionStatus.Idle
+                    }
+                    val initialCatchupPending =
+                        passiveObservationCatchupPendingSessionId == currentStoredId
+                    val needsFinalPassiveRefresh =
+                        passivelyObservedGatewaySessionId == currentStoredId &&
+                            !hasPassiveCurrentLiveWork
+                    if (hasPassiveCurrentLiveWork) {
+                        currentStoredId?.let(::refreshPassivelyObservedGatewayHistory)
+                        if (initialCatchupPending) {
+                            passiveObservationCatchupPendingSessionId = null
+                        }
+                    } else if (needsFinalPassiveRefresh || initialCatchupPending) {
+                        val scheduled = currentStoredId?.let { storedId ->
+                            refreshPassivelyObservedGatewayHistory(
+                                storedSessionId = storedId,
+                                retryUntilChanged = true,
+                            )
+                        } == true
+                        if (scheduled && initialCatchupPending) {
+                            passiveObservationCatchupPendingSessionId = null
+                        }
+                    }
+                    passivelyObservedGatewaySessionId =
+                        currentStoredId?.takeIf { hasPassiveCurrentLiveWork }
+                    hasPassiveCatchupPending =
+                        passiveObservationCatchupPendingSessionId == currentStoredId
                     val resolved = resolveGatewayActiveSessions(
                         sessions = result.sessions,
                         directory = directory,
@@ -2347,6 +2375,18 @@ class ChatViewModel : ViewModel() {
                 GatewayActiveSessionsResult.Unsupported,
                 is GatewayActiveSessionsResult.TransientFailure -> {
                     if (gatewayClient !== client || generation != sessionActivityGeneration.get()) return
+                    val currentStoredId = currentOwner?.storedSessionId
+                    if (passiveObservationCatchupPendingSessionId == currentStoredId) {
+                        val scheduled = currentStoredId?.let { storedId ->
+                            refreshPassivelyObservedGatewayHistory(
+                                storedSessionId = storedId,
+                                retryUntilChanged = true,
+                            )
+                        } == true
+                        if (scheduled) passiveObservationCatchupPendingSessionId = null
+                    }
+                    hasPassiveCatchupPending =
+                        passiveObservationCatchupPendingSessionId == currentStoredId
                     val scopes = directory.mapTo(mutableSetOf()) {
                         SessionActivityScope.of(it.connectionId, it.profile)
                     }.apply { add(currentScope) }
@@ -2365,7 +2405,9 @@ class ChatViewModel : ViewModel() {
             record.freshness == SessionActivityFreshness.Confirmed &&
                 record.phase(System.currentTimeMillis()) != SessionActivityPhase.Idle
         }
-        val delayMs = if (hasConfirmedLiveWork) 1_500L else 30_000L
+        val delayMs = if (
+            hasConfirmedLiveWork || hasPassiveCurrentLiveWork || hasPassiveCatchupPending
+        ) 1_500L else 30_000L
         sessionActivityPollJob = viewModelScope.launch {
             delay(delayMs)
             if (gatewayClient === client && chatVisible) pollSessionActivity(client)
@@ -2438,6 +2480,10 @@ class ChatViewModel : ViewModel() {
             clearProjectedBackgroundProcesses()
             sessionActivityPollJob?.cancel()
             sessionActivityPollJob = null
+            passiveGatewayHistoryRefreshJob?.cancel()
+            passiveGatewayHistoryRefreshJob = null
+            passivelyObservedGatewaySessionId = null
+            passiveObservationCatchupPendingSessionId = null
             sessionActivityGeneration.incrementAndGet()
             sessionActivityDirectory = emptySet()
             lastLocalActivityOwner = null
@@ -2557,8 +2603,9 @@ class ChatViewModel : ViewModel() {
                             // Foreground can race OkHttp's delayed close callback:
                             // the first prewarm sees the old socket as Ready, then
                             // the callback moves it to Idle. Re-run from this exact
-                            // client transition so the visible durable session is
-                            // resumed and its authoritative history reconciled.
+                            // client transition so the observation socket is
+                            // restored; only an exact Android checkpoint may
+                            // resume/activate a live runtime.
                             prewarmGateway()
                         }
                     }
@@ -3013,6 +3060,103 @@ class ChatViewModel : ViewModel() {
         }
     }
 
+    /**
+     * Refresh a Desktop/TUI-owned turn through the profile-scoped history
+     * surface without attaching its live runtime. `session.active_list` drives
+     * the bounded cadence; one final read follows Working/Waiting -> Idle.
+     */
+    private fun refreshPassivelyObservedGatewayHistory(
+        storedSessionId: String,
+        retryUntilChanged: Boolean = false,
+    ): Boolean {
+        if (passiveGatewayHistoryRefreshJob?.isActive == true) return false
+        if (_isLoadingHistory.value) return false
+        val handler = chatHandler ?: return false
+        val contextKey = activeProfileContextKey
+        val profileName = currentSessionProfileName()
+        val refreshJob = viewModelScope.launch(start = CoroutineStart.LAZY) {
+            try {
+                repeat(if (retryUntilChanged) 8 else 1) { attempt ->
+                    val serverMessages = runCatching {
+                        loadGatewaySessionHistory(
+                            sessionId = storedSessionId,
+                            requireProfileScope = true,
+                            profileName = profileName,
+                        )
+                    }.getOrNull() ?: return@launch
+                    if (
+                        chatHandler !== handler ||
+                        activeProfileContextKey != contextKey ||
+                        currentSessionProfileName() != profileName ||
+                        handler.currentSessionId.value != storedSessionId ||
+                        _isLoadingHistory.value ||
+                        activeStream != null ||
+                        handler.isStreaming.value
+                    ) return@launch
+
+                    val visibleSignature = handler.messages.value
+                        .filterNot { it.clientOnly }
+                        .map { message ->
+                            Triple(
+                                message.role.name.lowercase(),
+                                message.content,
+                                message.thinkingContent,
+                            )
+                        }
+                    val serverSignature = serverMessages.map { message ->
+                        Triple(
+                            message.role.lowercase(),
+                            message.contentText.orEmpty(),
+                            message.resolvedReasoning.orEmpty(),
+                        )
+                    }
+                    if (visibleSignature != serverSignature) {
+                        handler.loadMessageHistory(serverMessages)
+                        refreshSessions()
+                        scheduleTitleReconcile(storedSessionId)
+                        return@launch
+                    }
+                    if (attempt < 7 && retryUntilChanged) delay(250L)
+                }
+            } finally {
+                if (passiveGatewayHistoryRefreshJob === coroutineContext[Job]) {
+                    passiveGatewayHistoryRefreshJob = null
+                }
+            }
+        }
+        passiveGatewayHistoryRefreshJob = refreshJob
+        refreshJob.start()
+        return true
+    }
+
+    /** Open the read-only socket off Main, then publish observation ownership on Main. */
+    private fun observeGatewaySession(
+        client: GatewayChatClient?,
+        handler: ChatHandler,
+        storedSessionId: String,
+    ) {
+        val observer = client ?: return
+        val contextKey = activeProfileContextKey
+        val profileName = currentSessionProfileName()
+        observer.observe {
+            if (
+                chatVisible &&
+                gatewayClient === observer &&
+                chatHandler === handler &&
+                activeProfileContextKey == contextKey &&
+                currentSessionProfileName() == profileName &&
+                handler.currentSessionId.value == storedSessionId
+            ) {
+                passiveObservationCatchupPendingSessionId = storedSessionId
+                refreshPassivelyObservedGatewayHistory(
+                    storedSessionId = storedSessionId,
+                    retryUntilChanged = true,
+                )
+                requestSessionActivityRefresh()
+            }
+        }
+    }
+
     /** One-shot `config.get personality` over a ready socket → drives the collector. */
     private fun seedServerPersonality(client: GatewayChatClient) {
         viewModelScope.launch {
@@ -3023,11 +3167,11 @@ class ChatViewModel : ViewModel() {
     }
 
     /**
-     * Warm the gateway socket (and resume the current session) when the chat
-     * surface is visible and the gateway is the resolved transport, so the
-     * first send is warm instead of paying the cold connect + `session.resume`
-     * on the send path. No-op without a gateway client; idempotent when warm.
-     * Driven by a foreground/visibility effect in ChatScreen.
+     * Warm the Gateway socket when Chat is visible without claiming a runtime
+     * that may belong to Desktop/TUI. Exact Android-owned checkpoints recover
+     * through `session.activate`/`session.resume`; an ordinary open observes
+     * through REST history and `session.active_list` until the user performs
+     * an explicit action that needs session ownership.
      */
     fun prewarmGateway() {
         val client = gatewayClient
@@ -3035,17 +3179,16 @@ class ChatViewModel : ViewModel() {
         val sessionId = handler.currentSessionId.value
         selectBackgroundProcessSession(sessionId)
         if (sessionId == null) {
-            client?.prewarm(null)
+            client?.observe()
         } else {
             // Preserve the original warm-up path before persistence wiring is
             // available (early composition and JVM tests). Production installs
             // the store from initializeMedia before Chat becomes ready.
             if (chatTurnCheckpointStore == null) {
                 val gateway = client ?: return
-                // GatewayChatClient owns an IO scope, so this can progress even
-                // while a paused/blocked UI dispatcher is being recreated.
-                // Its cold-ready listener performs history/process refresh.
-                gateway.prewarm(sessionId)
+                // GatewayChatClient owns the socket IO scope, so the dial can
+                // progress while a paused UI dispatcher is being recreated.
+                observeGatewaySession(gateway, handler, sessionId)
                 return
             }
             if (activeStream == null && (streamRecovery == null || client != null)) {
@@ -3057,26 +3200,29 @@ class ChatViewModel : ViewModel() {
                         chatHandler === handler &&
                         handler.currentSessionId.value == sessionId
                     ) {
-                        if (client?.prewarmAwait(sessionId) == true) {
-                            gatewayProcessController.sessionReady(sessionId)
-                        }
+                        observeGatewaySession(client, handler, sessionId)
                     }
                     checkpointRecoveryJob = null
                 }
                 return
             }
-            // prewarm() only emits the existing "cold ready" callback when it
-            // had to resume. An already-live session still needs its initial
-            // process snapshot when Chat opens, so confirm it explicitly.
-            viewModelScope.launch {
-                if (
-                    client?.prewarmAwait(sessionId) == true &&
-                    gatewayClient === client &&
-                    chatHandler === handler &&
-                    handler.currentSessionId.value == sessionId
-                ) {
-                    gatewayProcessController.sessionReady(sessionId)
+            // A locally-owned live mapper may revalidate its existing binding.
+            // A passive transcript must remain socket-only: resuming it here
+            // can replace another client's transport and turn Android teardown
+            // into a later session.interrupt.
+            if (client?.hasActiveTurnForSession(sessionId) == true) {
+                viewModelScope.launch {
+                    if (client.prewarmAwait(sessionId) &&
+                        gatewayClient === client &&
+                        chatHandler === handler &&
+                        handler.currentSessionId.value == sessionId
+                    ) {
+                        gatewayProcessController.sessionReady(sessionId)
+                        requestSessionActivityRefresh()
+                    }
                 }
+            } else {
+                observeGatewaySession(client, handler, sessionId)
             }
         }
     }
@@ -3086,7 +3232,7 @@ class ChatViewModel : ViewModel() {
      * Gateway chat owns automatic idle-socket reattachment; other tabs and a
      * backgrounded app retain the normal no-reconnect behavior.
      */
-    fun setChatVisible(visible: Boolean) {
+    fun setChatVisible(visible: Boolean): Boolean {
         val changed = chatVisible != visible
         chatVisible = visible
         if (visible && changed) {
@@ -3095,7 +3241,12 @@ class ChatViewModel : ViewModel() {
         } else if (!visible) {
             sessionActivityPollJob?.cancel()
             sessionActivityPollJob = null
+            passiveGatewayHistoryRefreshJob?.cancel()
+            passiveGatewayHistoryRefreshJob = null
+            passivelyObservedGatewaySessionId = null
+            passiveObservationCatchupPendingSessionId = null
         }
+        return changed
     }
 
     // === Gateway desktop-parity state ===
@@ -4222,7 +4373,6 @@ class ChatViewModel : ViewModel() {
                 sessionRefreshJob?.cancel()
                 _isLoadingSessions.value = false
                 conversationBindingController.reset()
-                exitProvisionalThread()
                 relayCapabilityGeneration.incrementAndGet()
                 relayReasoningCapabilities.value = emptyMap()
                 _reasoningCapabilityRevision.value += 1L
@@ -4270,7 +4420,6 @@ class ChatViewModel : ViewModel() {
         sessionId: String,
     ): Boolean {
         if (!selectConversationProfile(profileName, profile)) return false
-        exitProvisionalThread()
         // Detach the old live gateway session without reading launch/global
         // model options: session.info for the resumed owner is authoritative.
         activateGatewayProfile(profile, refreshModelOptions = false)
@@ -4280,7 +4429,6 @@ class ChatViewModel : ViewModel() {
             sessionId = sessionId,
             explicitProfileName = profileName,
             explicitDisplayProfile = profile,
-            explicitBinding = true,
         )
         return true
     }
@@ -4291,15 +4439,11 @@ class ChatViewModel : ViewModel() {
      * `default` profile wins over the server's sticky active profile everywhere.
      */
     fun createProfileChat(
-        profileName: String?,
+        profileName: String,
         profile: Profile?,
         contextKey: String,
     ): Boolean {
         if (!selectConversationProfile(profileName, profile)) return false
-        // A provisional phone Thread belongs to its original connection/chat_id
-        // and cannot transfer to another profile. Exit it before binding or
-        // persisting the destination draft so the next send uses session.create.
-        exitProvisionalThread()
         activateGatewayProfile(profile, refreshModelOptions = false)
         refreshActiveAgentName(profile, relabelGenericMessages = true)
         switchProfileContextInternal(
@@ -4307,49 +4451,11 @@ class ChatViewModel : ViewModel() {
             sessionId = null,
             explicitProfileName = profileName,
             explicitDisplayProfile = profile,
-            explicitBinding = true,
         )
-        // Selection has already moved persistence to the target profile, so
-        // clear that profile/transport's stored last-session slot as part of
-        // the same draft transfer. A restart must reopen the draft, not the
-        // target profile's previous conversation.
-        persistFreshDraft(profileName)
         AppAnalytics.onSessionCreated()
         return true
     }
 
-    /**
-     * Atomic owner switch for the Chat header.
-     *
-     * Empty ordinary drafts and provisional phone Threads both become a fresh
-     * destination-profile draft, but only after provisional routing is retired.
-     * Durable sessions keep the established profile-selection lifecycle, whose
-     * binder may restore the destination profile's compatible last session.
-     */
-    fun selectProfileFromHeader(
-        profileName: String?,
-        profile: Profile?,
-        contextKey: String,
-    ): Boolean {
-        val handler = chatHandler ?: return false
-        val currentSessionId = handler.currentSessionId.value
-        val activeSession = handler.sessions.value.firstOrNull {
-            it.sessionId == currentSessionId
-        }
-        if (currentSessionId == null || activeSession?.source == "phone") {
-            return createProfileChat(profileName, profile, contextKey)
-        }
-        if (!selectConversationProfile(profileName, profile)) return false
-        exitProvisionalThread()
-        activateGatewayProfile(profile)
-        return true
-    }
-
-    private fun persistFreshDraft(profileName: String?) {
-        val transport = SessionTransport.forEndpoint(streamingEndpoint)
-        onFreshDraftSelected?.invoke(profileName, transport) ?: onSessionChanged?.invoke(null)
-    }
-
     fun switchProfileContext(contextKey: String, sessionId: String?) {
         clearOpenedSessionOwner()
         switchProfileContextInternal(contextKey, sessionId)
@@ -4372,19 +4478,14 @@ class ChatViewModel : ViewModel() {
         sessionId: String?,
         explicitProfileName: String? = null,
         explicitDisplayProfile: Profile? = null,
-        explicitBinding: Boolean = false,
         reconciliation: Boolean = false,
     ) {
         val handler = chatHandler ?: return
         dismissChatFailure()
         val previousBinding = conversationBinding.value
         val isInitialContextBinding = !previousBinding.isBound
-        val targetProfileName = if (explicitBinding) {
-            explicitProfileName
-        } else {
-            sessionProfileNameProvider()
-        }
-        if (explicitBinding) {
+        val targetProfileName = explicitProfileName ?: sessionProfileNameProvider()
+        if (explicitProfileName != null) {
             val accepted = conversationBindingController.openExplicit(
                 contextKey = contextKey,
                 profileName = explicitProfileName,
@@ -4529,7 +4630,9 @@ class ChatViewModel : ViewModel() {
                     )
                     if (stillCurrent()) {
                         handler.loadMessageHistory(messages)
-                        if (streamingEndpoint == "gateway") gatewayClient?.prewarm(sessionId)
+                        if (streamingEndpoint == "gateway") {
+                            observeGatewaySession(gatewayClient, handler, sessionId)
+                        }
                     }
                 }
             } catch (e: kotlinx.coroutines.CancellationException) {
@@ -4742,12 +4845,9 @@ class ChatViewModel : ViewModel() {
         if (supervisedModePolicy.enabled && !supervisedModePolicy.capabilities.newChat) return
         val handler = chatHandler ?: return
         recordPreResetEvidence(handler, "new_chat")
-        // A new chat clears only the durable session identity. Keep the bound
-        // profile/context so an All Profiles conversation becomes a fresh
-        // draft for that same owner instead of falling back to the globally
-        // restored default profile.
-        conversationBindingController.startFreshDraft()
-        exitProvisionalThread()
+        clearOpenedSessionOwner()
+        pendingThread = null
+        creatingThread = null
 
         // Gateway turns continue as detached siblings; SSE remains exclusive.
         releaseTurnForNavigation(handler)
@@ -4777,7 +4877,7 @@ class ChatViewModel : ViewModel() {
             _fastEnabled.value = null
             approvalModeRevision.incrementAndGet()
             pendingYolo = null
-            persistFreshDraft(currentSessionProfileName())
+            onSessionChanged?.invoke(null)
             AppAnalytics.onSessionCreated()
             onReady?.invoke(null)
             return
@@ -4847,7 +4947,6 @@ class ChatViewModel : ViewModel() {
      */
     fun startNewThread(name: String) {
         val handler = chatHandler ?: return
-        exitProvisionalThread()
         recordPreResetEvidence(handler, "new_thread")
         releaseTurnForNavigation(handler)
         cancelAnswerRecovery(settleUi = false)
@@ -4884,7 +4983,6 @@ class ChatViewModel : ViewModel() {
             .sortedBy { it.receivedAt }
         if (ordered.isEmpty()) return
 
-        exitProvisionalThread()
         recordPreResetEvidence(handler, "open_proactive_thread")
 
         releaseTurnForNavigation(handler)
@@ -4894,6 +4992,7 @@ class ChatViewModel : ViewModel() {
             chatId = normalizedChatId,
             name = ordered.last().title.ifBlank { "Hermes" },
         )
+        creatingThread = null
         gatewayClient?.clearSession()
         handler.setSessionId(null)
         selectBackgroundProcessSession(null)
@@ -4946,20 +5045,11 @@ class ChatViewModel : ViewModel() {
      */
     private fun switchToCreatedThread() {
         val creating = creatingThread ?: return
-        val generation = threadNavigationGeneration.get()
         viewModelScope.launch {
             for (delayMs in longArrayOf(900L, 1300L, 1800L, 2500L, 3500L, 4500L)) {
                 delay(delayMs)
-                if (
-                    threadNavigationGeneration.get() != generation ||
-                    creatingThread != creating
-                ) return@launch
                 refreshSessions()
                 delay(400L) // let the refresh job land in the sessions flow
-                if (
-                    threadNavigationGeneration.get() != generation ||
-                    creatingThread != creating
-                ) return@launch
                 val match = chatHandler?.sessions?.value?.firstOrNull {
                     it.source == "phone" && it.sessionId !in creating.knownIds
                 }
@@ -4989,7 +5079,8 @@ class ChatViewModel : ViewModel() {
         val handler = chatHandler ?: return
         dismissChatFailure()
         if (streamingEndpoint != "gateway" && apiClient == null) return
-        exitProvisionalThread()
+        pendingThread = null
+        creatingThread = null
 
         // Keep a Gateway sibling alive and detach its callbacks. SSE remains a
         // single exclusive stream and is interrupted on navigation.
@@ -5052,7 +5143,9 @@ class ChatViewModel : ViewModel() {
                         handler.currentSessionId.value == sessionId
                     ) {
                         handler.loadMessageHistory(messages)
-                        if (streamingEndpoint == "gateway") gatewayClient?.prewarm(sessionId)
+                        if (streamingEndpoint == "gateway") {
+                            observeGatewaySession(gatewayClient, handler, sessionId)
+                        }
                     }
                 } catch (e: kotlinx.coroutines.CancellationException) {
                     throw e
@@ -6642,6 +6735,10 @@ class ChatViewModel : ViewModel() {
      * SSE cannot, so it retains the existing interrupt/cancel behavior.
      */
     private fun releaseTurnForNavigation(handler: ChatHandler) {
+        passiveGatewayHistoryRefreshJob?.cancel()
+        passiveGatewayHistoryRefreshJob = null
+        passivelyObservedGatewaySessionId = null
+        passiveObservationCatchupPendingSessionId = null
         val gateway = gatewayClient
         val canBackground = streamingEndpoint == "gateway" &&
             activeStreamIsGateway && activeStream != null && gateway != null

app/src/main/kotlin/com/hermesandroid/relay/viewmodel/ConnectionViewModel.kt

@@ -6884,7 +6884,6 @@ class ConnectionViewModel(application: Application) : AndroidViewModel(applicati
                     // ground truth about which transport can resume it, robust to a
                     // turn that fell back from gateway to SSE.
                     val transport = SessionTransport.forSessionId(sessionId)
-                    profileController.markSessionPersisted(connectionId, profileName, transport)
                     profileController.profileSessionStore.setSessionId(
                         connectionId,
                         profileName,
@@ -6921,20 +6920,6 @@ class ConnectionViewModel(application: Application) : AndroidViewModel(applicati
         }
     }
 
-    /** Persist an intentional empty draft without conflating it with transient null state. */
-    fun saveFreshDraft(profileName: String?, transport: SessionTransport) {
-        _lastSessionId.value = null
-        val connectionId = activeConnectionId.value ?: return
-        profileController.markFreshDraft(connectionId, profileName, transport)
-        if (profileName == null) {
-            viewModelScope.launch {
-                getApplication<Application>().relayDataStore.edit { preferences ->
-                    preferences.remove(KEY_LAST_SESSION_ID)
-                }
-            }
-        }
-    }
-
     // --- Shared methods ---
 
     fun setTheme(theme: String) {

app/src/main/kotlin/com/hermesandroid/relay/viewmodel/ConversationBindingController.kt

@@ -102,18 +102,6 @@ internal class ConversationBindingController {
         )
     }
 
-    /** A user-requested draft keeps its owner and fences persisted-session reconciliation. */
-    fun startFreshDraft() {
-        val current = _state.value
-        if (!current.isBound) return
-        if (current.sessionId == null && current.hasExplicitOwner) return
-        _state.value = current.copy(
-            sessionId = null,
-            origin = ConversationBindingOrigin.ExplicitSession,
-            revision = current.revision + 1,
-        )
-    }
-
     fun releaseExplicitOwner() {
         if (!_state.value.hasExplicitOwner) return
         reset()

app/src/main/kotlin/com/hermesandroid/relay/viewmodel/connection/ProfileController.kt

@@ -155,14 +155,6 @@ class ProfileController(
     private val avatarRefreshGeneration = AtomicLong(0L)
     private val petRefreshGeneration = AtomicLong(0L)
     private val petGalleryGeneration = AtomicLong(0L)
-    private val sessionRestoreGeneration = AtomicLong(0L)
-    private val freshDraftScopes = ConcurrentHashMap.newKeySet<SessionScopeKey>()
-
-    private data class SessionScopeKey(
-        val connectionId: String,
-        val profileName: String?,
-        val transport: SessionTransport,
-    )
     private val petThumbnailRequests = ConcurrentHashMap.newKeySet<String>()
 
     val agentProfiles: StateFlow<List<Profile>> = combine(
@@ -1471,48 +1463,10 @@ class ProfileController(
         }
     }
 
-    /**
-     * Persist a user-requested empty draft for one exact conversation scope.
-     *
-     * The in-memory marker fences any stored-session read that was already in
-     * flight, while clearing the exact transport slot makes the draft survive a
-     * process restart. Other profiles, connections, transports, and the server's
-     * actual session/history rows are untouched.
-     */
-    fun markFreshDraft(
-        connectionId: String,
-        profileName: String?,
-        transport: SessionTransport,
-    ) {
-        val scopeKey = SessionScopeKey(connectionId, profileName, transport)
-        freshDraftScopes += scopeKey
-        sessionRestoreGeneration.incrementAndGet()
-        if (
-            activeConnectionId.value == connectionId &&
-            _selectedProfile.value?.name == profileName
-        ) {
-            setLastSessionId(null)
-        }
-        scope.launch {
-            profileSessionStore.setSessionId(connectionId, profileName, transport, null)
-        }
-    }
-
-    /** A real session supersedes the fresh-draft marker for its exact scope. */
-    fun markSessionPersisted(
-        connectionId: String,
-        profileName: String?,
-        transport: SessionTransport,
-    ) {
-        freshDraftScopes -= SessionScopeKey(connectionId, profileName, transport)
-        sessionRestoreGeneration.incrementAndGet()
-    }
-
     fun refreshLastSessionForProfile(
         connectionId: String?,
         profileName: String?,
     ) {
-        val generation = sessionRestoreGeneration.incrementAndGet()
         setLastSessionId(null)
         if (connectionId == null) return
         // Defer until the active transport is known — restoring an id the
@@ -1524,8 +1478,6 @@ class ProfileController(
         // `default` (or any other name), but its last-session slot must remain
         // distinct from explicitly selecting that named profile.
         val sessionProfileName = profileName
-        val scopeKey = SessionScopeKey(connectionId, sessionProfileName, transport)
-        if (scopeKey in freshDraftScopes) return
         scope.launch {
             val profileScoped = profileSessionStore
                 .sessionIdFlow(connectionId, sessionProfileName, transport)
@@ -1541,8 +1493,6 @@ class ProfileController(
                 null
             }
             if (
-                sessionRestoreGeneration.get() == generation &&
-                scopeKey !in freshDraftScopes &&
                 activeConnectionId.value == connectionId &&
                 _selectedProfile.value?.name == profileName &&
                 activeSessionTransport() == transport

app/src/test/kotlin/com/hermesandroid/relay/data/ProfileSessionStoreTest.kt

@@ -17,8 +17,7 @@ import kotlinx.coroutines.sync.withLock
 
 class ProfileSessionStoreTest {
 
-    private val dataStore = InMemoryPreferencesDataStore()
-    private val store = ProfileSessionStore(dataStore)
+    private val store = ProfileSessionStore(InMemoryPreferencesDataStore())
 
     @Test
     fun setAndGet_defaultProfileSession() = runBlocking {
@@ -86,26 +85,6 @@ class ProfileSessionStoreTest {
         assertEquals("session-sse", store.sessionIdFlow("conn-1", "mizu", SSE).first())
     }
 
-    @Test
-    fun clearedDraftSurvivesStoreRecreationAndPreservesOtherScopes() = runBlocking {
-        store.setSessionId("conn-1", "mizu", GATEWAY, "session-gw")
-        store.setSessionId("conn-1", "mizu", SSE, "session-sse")
-        store.setSessionId("conn-2", "mizu", GATEWAY, "session-other")
-
-        store.setSessionId("conn-1", "mizu", GATEWAY, null)
-        val restartedStore = ProfileSessionStore(dataStore)
-
-        assertNull(restartedStore.sessionIdFlow("conn-1", "mizu", GATEWAY).first())
-        assertEquals(
-            "session-sse",
-            restartedStore.sessionIdFlow("conn-1", "mizu", SSE).first(),
-        )
-        assertEquals(
-            "session-other",
-            restartedStore.sessionIdFlow("conn-2", "mizu", GATEWAY).first(),
-        )
-    }
-
     @Test
     fun clearConnectionRemovesAllProfilesAndTransportsForThatConnectionOnly() = runBlocking {
         store.setSessionId("conn-1", null, GATEWAY, "session-default")

app/src/test/kotlin/com/hermesandroid/relay/network/upstream/GatewayChatClientTest.kt

@@ -231,11 +231,14 @@ class GatewayClientHarness(
     val suppressAckMethods: MutableSet<String> = ConcurrentHashMap.newKeySet()
     val pendingAcks = LinkedBlockingQueue<PendingAck>()
 
+    @Volatile
+    var suppressGatewayReady: Boolean = false
+
     private val wsListener = object : WebSocketListener() {
         override fun onOpen(webSocket: WebSocket, response: okhttp3.Response) {
             serverSockets.add(webSocket)
             allServerSockets.add(webSocket)
-            webSocket.send(eventFrame("gateway.ready", null, null))
+            if (!suppressGatewayReady) sendGatewayReady(webSocket)
         }
 
         override fun onMessage(webSocket: WebSocket, text: String) {
@@ -627,6 +630,10 @@ class GatewayClientHarness(
     fun awaitServerSocket(): WebSocket =
         serverSockets.poll(5, TimeUnit.SECONDS) ?: error("server socket never opened")
 
+    fun sendGatewayReady(webSocket: WebSocket) {
+        webSocket.send(eventFrame("gateway.ready", null, null))
+    }
+
     fun awaitRpc(method: String): JsonObject {
         val deadline = System.currentTimeMillis() + 5_000
         while (System.currentTimeMillis() < deadline) {
@@ -1407,6 +1414,27 @@ class GatewayChatClientTest {
         assertEquals(listOf("stored-session"), resumedSessions.toList())
     }
 
+    @Test
+    fun `observation warmup never claims or interrupts a foreign runtime`() = runBlocking {
+        val registrations = AtomicInteger(0)
+        client.setUnsolicitedTurnProvider {
+            registrations.incrementAndGet()
+            GatewayInboundTurnRegistration(Recorder().callbacks) { true }
+        }
+
+        assertTrue(client.observeAwait())
+        val serverWs = harness.awaitServerSocket()
+        serverWs.send(harness.eventFrame("message.start", null, "foreign-runtime"))
+        delay(100)
+        client.shutdown()
+
+        assertEquals(0, registrations.get())
+        assertFalse(harness.rpcLog.any { it.first == "session.resume" })
+        assertFalse(harness.rpcLog.any { it.first == "session.activate" })
+        assertFalse(harness.rpcLog.any { it.first == "session.interrupt" })
+        assertFalse(harness.rpcLog.any { it.first == "prompt.submit" })
+    }
+
     @Test
     fun `newer prewarm selection wins when an older resume completes late`() = runBlocking {
         harness.suppressAckMethods += "session.resume"

app/src/test/kotlin/com/hermesandroid/relay/ui/components/SessionDrawerTest.kt

@@ -304,8 +304,9 @@ class SessionDrawerTest {
     }
 
     @Test
-    fun `new chat from all profiles keeps the current conversation owner`() {
+    fun `new chat from all profiles requests an explicit default draft`() {
         var scopedNewChats = 0
+        var defaultNewChats = 0
         compose.setContent {
             MaterialTheme {
                 SessionDrawerContent(
@@ -318,6 +319,7 @@ class SessionDrawerTest {
                     onRefreshAllProfiles = {},
                     onSelectProfileSession = { _, _ -> },
                     onNewChat = { scopedNewChats++ },
+                    onNewDefaultChat = { defaultNewChats++ },
                     onSelectSession = {},
                     onDeleteSession = {},
                     onRenameSession = { _, _ -> },
@@ -330,7 +332,8 @@ class SessionDrawerTest {
         compose.onNodeWithText("New Chat").performClick()
 
         compose.runOnIdle {
-            assertEquals(1, scopedNewChats)
+            assertEquals(0, scopedNewChats)
+            assertEquals(1, defaultNewChats)
         }
     }
 

app/src/test/kotlin/com/hermesandroid/relay/viewmodel/ChatViewModelGatewayInboundTurnTest.kt

@@ -14,12 +14,9 @@ import com.hermesandroid.relay.data.ChatTurnUserCheckpoint
 import com.hermesandroid.relay.data.HermesCardDispatch
 import com.hermesandroid.relay.data.MessageRole
 import com.hermesandroid.relay.data.Profile
-import com.hermesandroid.relay.data.ProactiveInboxEntry
-import com.hermesandroid.relay.data.SessionTransport
 import com.hermesandroid.relay.data.SessionActivityState
 import com.hermesandroid.relay.diagnostics.DiagnosticCategory
 import com.hermesandroid.relay.diagnostics.DiagnosticsLog
-import com.hermesandroid.relay.network.relay.ProactiveMessage
 import com.hermesandroid.relay.network.upstream.ChatHandler
 import com.hermesandroid.relay.network.upstream.DashboardApiClient
 import com.hermesandroid.relay.network.upstream.GatewayChatClient
@@ -434,17 +431,8 @@ class ChatViewModelGatewayInboundTurnTest {
         assertEquals(owner.name, viewModel.conversationBinding.value.profileName)
 
         viewModel.createNewChat()
-        assertTrue(viewModel.conversationBinding.value.hasExplicitOwner)
-        assertEquals(owner.name, viewModel.conversationBinding.value.profileName)
-        assertNull(viewModel.conversationBinding.value.sessionId)
-        assertEquals(owner.name, gatewayClient.sessionProfileProvider())
-
-        viewModel.reconcileProfileContext(
-            AgentDisplay.profileContextKey("connection-a", owner.name),
-            sessionId = "x-bot-session",
-        )
-        assertNull(viewModel.conversationBinding.value.sessionId)
-        assertNull(handler.currentSessionId.value)
+        assertFalse(viewModel.conversationBinding.value.hasExplicitOwner)
+        assertEquals(global.name, gatewayClient.sessionProfileProvider())
     }
 
     @Test
@@ -659,258 +647,7 @@ class ChatViewModelGatewayInboundTurnTest {
         assertEquals("default", gatewayClient.sessionProfileProvider())
         assertEquals(null, handler.currentSessionId.value)
         assertEquals("Hermes", handler.activeAgentName)
-        assertEquals("cleared", persistedSession)
-    }
-
-    @Test
-    fun freshDraftTransferKeepsNullableServerDefaultAndRejectsOldSessionRestore() {
-        val named = Profile(name = "x-bot", model = "grok-4.3", description = "X Bot")
-        var selected: Profile? = named
-        var persistedDraft: Pair<String?, SessionTransport>? = null
-        viewModel.setSelectedProfileProvider { selected }
-        viewModel.setSessionProfileNameProvider { selected?.name }
-        viewModel.setProfileSelectionHandler { profile ->
-            selected = profile
-            true
-        }
-        viewModel.onFreshDraftSelected = { profileName, transport ->
-            persistedDraft = profileName to transport
-        }
-
-        assertTrue(
-            viewModel.createProfileChat(
-                profileName = null,
-                profile = null,
-                contextKey = AgentDisplay.profileContextKey("connection-a", null),
-            ),
-        )
-
-        assertNull(selected)
-        assertTrue(viewModel.conversationBinding.value.hasExplicitOwner)
-        assertNull(viewModel.conversationBinding.value.profileName)
-        assertNull(viewModel.conversationBinding.value.sessionId)
-        assertEquals(null to SessionTransport.GATEWAY, persistedDraft)
-        assertNull(gatewayClient.sessionProfileProvider())
-
-        viewModel.reconcileProfileContext(
-            AgentDisplay.profileContextKey("connection-a", null),
-            sessionId = "old-default-session",
-        )
-        assertNull(viewModel.conversationBinding.value.sessionId)
-        assertNull(handler.currentSessionId.value)
-    }
-
-    @Test
-    fun freshDraftTransferToNamedProfileCreatesInsteadOfResumingItsOldSession() {
-        val alpha = Profile(name = "alpha", model = "model-a", description = "Alpha")
-        val beta = Profile(name = "beta", model = "model-b", description = "Beta")
-        var selected: Profile? = alpha
-        var persistedDraft: Pair<String?, SessionTransport>? = null
-        viewModel.setSelectedProfileProvider { selected }
-        viewModel.setSessionProfileNameProvider { selected?.name }
-        viewModel.setProfileSelectionHandler { profile ->
-            selected = profile
-            true
-        }
-        viewModel.onFreshDraftSelected = { profileName, transport ->
-            persistedDraft = profileName to transport
-        }
-
-        viewModel.openProfileSession(
-            profileName = alpha.name,
-            profile = alpha,
-            contextKey = AgentDisplay.profileContextKey("connection-a", alpha.name),
-            sessionId = "alpha-session",
-        )
-        viewModel.createNewChat()
-        assertTrue(
-            viewModel.selectProfileFromHeader(
-                profileName = beta.name,
-                profile = beta,
-                contextKey = AgentDisplay.profileContextKey("connection-a", beta.name),
-            ),
-        )
-
-        assertEquals(beta, selected)
-        assertEquals(beta.name to SessionTransport.GATEWAY, persistedDraft)
-        viewModel.reconcileProfileContext(
-            AgentDisplay.profileContextKey("connection-a", beta.name),
-            sessionId = "beta-old-session",
-        )
-        assertNull(handler.currentSessionId.value)
-
-        gatewayHarness.createdSessionProfileName = beta.name
-        val resumeCountBeforeFreshSend = gatewayHarness.rpcLog.count {
-            it.first == "session.resume"
-        }
-        viewModel.sendMessage("Fresh beta turn")
-        val create = gatewayHarness.awaitRpc("session.create")
-        assertEquals(beta.name, (create["profile"] as JsonPrimitive).content)
-        assertEquals(
-            resumeCountBeforeFreshSend,
-            gatewayHarness.rpcLog.count { it.first == "session.resume" },
-        )
-    }
-
-    @Test
-    fun headerProfileSwitchExitsProvisionalThreadBeforeFreshProfileSend() {
-        val alpha = Profile(name = "alpha", model = "model-a", description = "Alpha")
-        val beta = Profile(name = "beta", model = "model-b", description = "Beta")
-        var selected: Profile? = alpha
-        val proactiveChatIds = mutableListOf<String?>()
-        viewModel.setSelectedProfileProvider { selected }
-        viewModel.setSessionProfileNameProvider { selected?.name }
-        viewModel.setProfileSelectionHandler { profile ->
-            selected = profile
-            true
-        }
-        viewModel.onProactiveReply = { _, chatId, _, _ -> proactiveChatIds += chatId }
-
-        viewModel.openProactiveThread(
-            chatId = "old-phone-chat",
-            entries = listOf(
-                ProactiveInboxEntry(
-                    id = "inbox-1",
-                    title = "Old phone thread",
-                    text = "Continue here",
-                    receivedAt = 1L,
-                    chatId = "old-phone-chat",
-                    connectionId = "connection-a",
-                ),
-            ),
-        )
-        assertNull(handler.currentSessionId.value)
-
-        assertTrue(
-            viewModel.selectProfileFromHeader(
-                profileName = beta.name,
-                profile = beta,
-                contextKey = AgentDisplay.profileContextKey("connection-a", beta.name),
-            ),
-        )
-        viewModel.sendMessage("Fresh beta turn")
-
-        val create = gatewayHarness.awaitRpc("session.create")
-        assertEquals(beta.name, (create["profile"] as JsonPrimitive).content)
-        assertTrue(proactiveChatIds.isEmpty())
-        assertEquals(beta.name, viewModel.conversationBinding.value.profileName)
-    }
-
-    @Test
-    fun headerProfileSwitchExitsPromotedPhoneSessionWithoutReusingItsChatId() {
-        val alpha = Profile(name = "alpha", model = "model-a", description = "Alpha")
-        val beta = Profile(name = "beta", model = "model-b", description = "Beta")
-        var selected: Profile? = alpha
-        val proactiveChatIds = mutableListOf<String?>()
-        viewModel.setSelectedProfileProvider { selected }
-        viewModel.setSessionProfileNameProvider { selected?.name }
-        viewModel.setProfileSelectionHandler { profile ->
-            selected = profile
-            true
-        }
-        viewModel.onProactiveReply = { _, chatId, _, _ -> proactiveChatIds += chatId }
-        handler.addSession(
-            com.hermesandroid.relay.data.ChatSession(
-                sessionId = "promoted-phone-session",
-                title = "Promoted thread",
-                model = null,
-                source = "phone",
-            ),
-        )
-        handler.setSessionId("promoted-phone-session")
-
-        assertTrue(
-            viewModel.selectProfileFromHeader(
-                profileName = beta.name,
-                profile = beta,
-                contextKey = AgentDisplay.profileContextKey("connection-a", beta.name),
-            ),
-        )
-        assertNull(handler.currentSessionId.value)
-        viewModel.sendMessage("Fresh beta after Thread")
-
-        val create = gatewayHarness.awaitRpc("session.create")
-        assertEquals(beta.name, (create["profile"] as JsonPrimitive).content)
-        assertTrue(proactiveChatIds.isEmpty())
-    }
-
-    @Test
-    fun newChatAndConnectionSwitchRetireProvisionalThreadRouting() {
-        val entry = ProactiveInboxEntry(
-            id = "inbox-1",
-            title = "Old phone thread",
-            text = "Continue here",
-            receivedAt = 1L,
-            chatId = "old-phone-chat",
-            connectionId = "connection-a",
-        )
-        val inbound = ProactiveMessage(
-            messageId = "late-1",
-            chatId = "old-phone-chat",
-            text = "Late old-thread message",
-            title = "Old phone thread",
-            surfacing = "thread",
-            sentAt = 2L,
-        )
-
-        viewModel.openProactiveThread("old-phone-chat", listOf(entry))
-        viewModel.createNewChat()
-        assertFalse(viewModel.injectThreadMessage(inbound))
-
-        val switches = MutableSharedFlow<String>(extraBufferCapacity = 1)
-        viewModel.observeConnectionSwitches(switches)
-        viewModel.openProactiveThread("old-phone-chat", listOf(entry))
-        switches.tryEmit("connection-b")
-        awaitCondition { handler.messages.value.isEmpty() }
-        assertFalse(viewModel.injectThreadMessage(inbound))
-    }
-
-    @Test
-    fun staleThreadPromotionCannotReplaceTransferredProfileDraft() {
-        val beta = Profile(name = "beta", model = "model-b", description = "Beta")
-        var selected: Profile? = Profile(name = "alpha", model = "model-a")
-        viewModel.setSelectedProfileProvider { selected }
-        viewModel.setSessionProfileNameProvider { selected?.name }
-        viewModel.setProfileSelectionHandler { profile ->
-            selected = profile
-            true
-        }
-        viewModel.onProactiveReply = { _, _, _, _ -> }
-        viewModel.openProactiveThread(
-            "old-phone-chat",
-            listOf(
-                ProactiveInboxEntry(
-                    id = "inbox-1",
-                    title = "Old phone thread",
-                    text = "Continue here",
-                    receivedAt = 1L,
-                    chatId = "old-phone-chat",
-                    connectionId = "connection-a",
-                ),
-            ),
-        )
-        viewModel.sendMessage("Promote the old Thread")
-
-        assertTrue(
-            viewModel.selectProfileFromHeader(
-                profileName = beta.name,
-                profile = beta,
-                contextKey = AgentDisplay.profileContextKey("connection-a", beta.name),
-            ),
-        )
-        handler.addSession(
-            com.hermesandroid.relay.data.ChatSession(
-                sessionId = "late-promoted-thread",
-                title = "Late promoted thread",
-                model = null,
-                source = "phone",
-            ),
-        )
-        shadowOf(Looper.getMainLooper()).idleFor(2, TimeUnit.SECONDS)
-        Thread.sleep(100)
-
-        assertNull(handler.currentSessionId.value)
-        assertEquals(beta.name, viewModel.conversationBinding.value.profileName)
+        assertEquals("unchanged", persistedSession)
     }
 
     @Test
@@ -1255,11 +992,12 @@ class ChatViewModelGatewayInboundTurnTest {
         awaitCondition { handler.messages.value.any { it.content == "Partial A" } }
 
         viewModel.switchSession(secondSession)
-        gatewayHarness.awaitRpcCount("session.resume", 2)
         awaitCondition { handler.currentSessionId.value == secondSession && !handler.isStreaming.value }
+        assertEquals(1, gatewayHarness.rpcLog.count { it.first == "session.resume" })
         assertTrue(gatewayHarness.rpcLog.none { it.first == "session.interrupt" })
 
         viewModel.sendMessage("Run task B")
+        gatewayHarness.awaitRpcCount("session.resume", 2)
         gatewayHarness.awaitRpcCount("prompt.submit", 2)
         serverWs.send(
             gatewayHarness.eventFrame(
@@ -2087,11 +1825,12 @@ class ChatViewModelGatewayInboundTurnTest {
         awaitCondition { viewModel.queuedMessages.value == listOf("Follow up A") }
 
         viewModel.switchSession(secondSession)
-        gatewayHarness.awaitRpcCount("session.resume", 2)
         awaitCondition { handler.currentSessionId.value == secondSession && !handler.isStreaming.value }
+        assertEquals(1, gatewayHarness.rpcLog.count { it.first == "session.resume" })
         assertTrue("session B must not show A's queue", viewModel.queuedMessages.value.isEmpty())
 
         viewModel.sendMessage("Run task B")
+        gatewayHarness.awaitRpcCount("session.resume", 2)
         gatewayHarness.awaitRpcCount("prompt.submit", 2)
         serverWs.send(
             gatewayHarness.eventFrame(
@@ -2294,13 +2033,16 @@ class ChatViewModelGatewayInboundTurnTest {
         serverWs.close(1012, "test disconnect")
         awaitCondition { gatewayClient.connectionState.value == GatewayConnectionState.Idle }
 
+        viewModel.setChatVisible(true)
         viewModel.prewarmGateway()
         gatewayHarness.awaitServerSocket()
-        gatewayHarness.awaitRpcCount("session.resume", 2)
 
         awaitCondition {
             handler.messages.value.singleOrNull()?.content == BACKGROUND_ANSWER
         }
+        assertEquals(1, gatewayHarness.rpcLog.count { it.first == "session.resume" })
+        assertEquals(0, gatewayHarness.rpcLog.count { it.first == "session.activate" })
+        assertEquals(0, gatewayHarness.rpcLog.count { it.first == "session.interrupt" })
         assertFalse(handler.isStreaming.value)
     }
 
@@ -2311,16 +2053,19 @@ class ChatViewModelGatewayInboundTurnTest {
 
         // Foreground arrives while OkHttp still reports the old socket ready,
         // so the one-shot prewarm is an intentional no-op. The delayed close
-        // callback must itself trigger an exact-session reattach.
+        // callback must itself restore the observation socket and catch up
+        // history without attaching the live session.
         viewModel.prewarmGateway()
         serverWs.close(1012, "late background close")
 
         awaitCondition { gatewayHarness.ticketMints.get() >= 2 }
         serverWs = gatewayHarness.awaitServerSocket()
-        gatewayHarness.awaitRpcCount("session.resume", 2)
         awaitCondition {
             handler.messages.value.singleOrNull()?.content == BACKGROUND_ANSWER
         }
+        assertEquals(1, gatewayHarness.rpcLog.count { it.first == "session.resume" })
+        assertEquals(0, gatewayHarness.rpcLog.count { it.first == "session.activate" })
+        assertEquals(0, gatewayHarness.rpcLog.count { it.first == "session.interrupt" })
         assertFalse(handler.isStreaming.value)
     }
 
@@ -2629,36 +2374,189 @@ class ChatViewModelGatewayInboundTurnTest {
     }
 
     @Test
-    fun reconnectAfterMissedStartRecoversOnExactSessionCompletion() {
+    fun reconnectCatchupClosesCompletionBetweenFirstReadAndIdleSnapshot() {
+        viewModel.switchProfileContext(PROFILE_CONTEXT, STORED_SESSION_ID)
+        awaitCondition { !viewModel.isLoadingHistory.value }
+        val firstReadStarted = CompletableDeferred<Unit>()
+        val releaseFirstRead = CompletableDeferred<Unit>()
+        val readCount = AtomicInteger(0)
+        viewModel.setProfileMessageLoader {
+            if (readCount.incrementAndGet() == 1) {
+                firstReadStarted.complete(Unit)
+                releaseFirstRead.await()
+                Result.success(emptyList())
+            } else {
+                Result.success(persistedHistory)
+            }
+        }
         serverWs.close(1012, "missed start")
         awaitCondition { gatewayClient.connectionState.value == GatewayConnectionState.Idle }
+        viewModel.setChatVisible(true)
         viewModel.prewarmGateway()
         serverWs = gatewayHarness.awaitServerSocket()
-        gatewayHarness.awaitRpcCount("session.resume", 2)
-
-        // Reconnected midway through the synthetic turn: no message.start is
-        // replayed, so the delta is intentionally ignored and completion drives
-        // authoritative history recovery.
-        serverWs.send(
-            gatewayHarness.eventFrame(
-                "message.delta",
-                buildJsonObject { put("text", BACKGROUND_ANSWER) },
-                "live-resumed",
-            ),
-        )
+        awaitCondition { firstReadStarted.isCompleted }
+        // Completion persists after the reconnect's first catch-up read began,
+        // while the first active-list snapshot is already empty/idle. The
+        // pending final-read marker must close this exact ordering window.
         persistedHistory = persistedAnswerHistory()
-        serverWs.send(
-            gatewayHarness.eventFrame(
-                "message.complete",
-                buildJsonObject { put("text", BACKGROUND_ANSWER) },
-                "live-resumed",
-            ),
-        )
+        releaseFirstRead.complete(Unit)
 
         awaitCondition { handler.messages.value.any { it.content == BACKGROUND_ANSWER } }
+        assertEquals(1, gatewayHarness.rpcLog.count { it.first == "session.resume" })
+        assertEquals(0, gatewayHarness.rpcLog.count { it.first == "session.activate" })
+        assertEquals(0, gatewayHarness.rpcLog.count { it.first == "session.interrupt" })
         assertFalse(handler.isStreaming.value)
     }
 
+    @Test
+    fun passiveForegroundObservationNeverClaimsOrInterruptsDesktopTurn() {
+        val observerProfile = Profile(
+            name = "observer",
+            model = "model-a",
+            description = "Observer",
+        )
+        viewModel.setSelectedProfileProvider { observerProfile }
+        viewModel.setSessionProfileNameProvider { observerProfile.name }
+        viewModel.setProfileMessageLoaderWithMode { profileName, sessionId, _ ->
+            assertEquals(STORED_SESSION_ID, sessionId)
+            Result.success(
+                if (profileName == observerProfile.name) {
+                    persistedHistory
+                } else {
+                    listOf(
+                        MessageItem(
+                            id = "wrong-profile",
+                            sessionId = STORED_SESSION_ID,
+                            role = "assistant",
+                            content = JsonPrimitive("Wrong profile history"),
+                        ),
+                    )
+                },
+            )
+        }
+        viewModel.switchProfileContext(
+            AgentDisplay.profileContextKey("connection-a", observerProfile.name),
+            STORED_SESSION_ID,
+        )
+        awaitCondition { !viewModel.isLoadingHistory.value }
+        viewModel.setChatVisible(false)
+        viewModel.updateGatewayClient(null)
+        gatewayClient.shutdown()
+        gatewayScope.cancel()
+
+        val ownershipMethods = setOf(
+            "session.resume",
+            "session.activate",
+            "session.interrupt",
+            "prompt.submit",
+        )
+        val baseline = ownershipMethods.associateWith { method ->
+            gatewayHarness.rpcLog.count { it.first == method }
+        }
+        val baselineActiveList = gatewayHarness.rpcLog.count { it.first == "session.active_list" }
+        gatewayHarness.activeSessionListPayload = activeSessionPayload("working")
+        gatewayScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+        gatewayClient = GatewayChatClient(
+            initialDashboardClient = DashboardApiClient(
+                baseUrl = gatewayHarness.server.url("/").toString().trimEnd('/'),
+                okHttpClient = OkHttpClient(),
+            ),
+            okHttpClient = OkHttpClient(),
+            callbackDispatcher = { block -> Handler(Looper.getMainLooper()).post(block) },
+            scope = gatewayScope,
+        )
+        viewModel.setChatTurnCheckpointStore(MemoryCheckpointStore())
+        viewModel.updateGatewayClient(gatewayClient)
+
+        viewModel.setChatVisible(true)
+        viewModel.prewarmGateway()
+        awaitCondition {
+            gatewayHarness.rpcLog.count { it.first == "session.active_list" } > baselineActiveList
+        }
+        persistedHistory = listOf(
+            MessageItem(
+                id = "desktop-answer",
+                sessionId = STORED_SESSION_ID,
+                role = "assistant",
+                content = JsonPrimitive("Desktop completed without Android attachment."),
+            ),
+        )
+        awaitCondition {
+            handler.messages.value.singleOrNull()?.content ==
+                "Desktop completed without Android attachment."
+        }
+        ownershipMethods.forEach { method ->
+            assertEquals(
+                "passive foreground sent $method",
+                baseline.getValue(method),
+                gatewayHarness.rpcLog.count { it.first == method },
+            )
+        }
+
+        viewModel.updateGatewayClient(null)
+        gatewayClient.shutdown()
+        assertEquals(
+            "observer teardown interrupted the Desktop turn",
+            baseline.getValue("session.interrupt"),
+            gatewayHarness.rpcLog.count { it.first == "session.interrupt" },
+        )
+    }
+
+    @Test
+    fun observerReadyAfterChatHidesCannotRestartPassiveWork() {
+        viewModel.switchProfileContext(PROFILE_CONTEXT, STORED_SESSION_ID)
+        awaitCondition { !viewModel.isLoadingHistory.value }
+        viewModel.setChatVisible(false)
+        viewModel.updateGatewayClient(null)
+        gatewayClient.shutdown()
+        gatewayScope.cancel()
+
+        val historyReads = AtomicInteger(0)
+        viewModel.setProfileMessageLoader {
+            historyReads.incrementAndGet()
+            Result.success(persistedHistory)
+        }
+        val controlMethods = setOf(
+            "session.resume",
+            "session.activate",
+            "session.interrupt",
+            "prompt.submit",
+        )
+        val baseline = controlMethods.associateWith { method ->
+            gatewayHarness.rpcLog.count { it.first == method }
+        }
+        val baselineActiveList = gatewayHarness.rpcLog.count { it.first == "session.active_list" }
+        gatewayHarness.suppressGatewayReady = true
+        gatewayScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+        gatewayClient = GatewayChatClient(
+            initialDashboardClient = DashboardApiClient(
+                baseUrl = gatewayHarness.server.url("/").toString().trimEnd('/'),
+                okHttpClient = OkHttpClient(),
+            ),
+            okHttpClient = OkHttpClient(),
+            callbackDispatcher = { block -> Handler(Looper.getMainLooper()).post(block) },
+            scope = gatewayScope,
+        )
+        viewModel.setChatTurnCheckpointStore(MemoryCheckpointStore())
+        viewModel.updateGatewayClient(gatewayClient)
+
+        viewModel.setChatVisible(true)
+        val delayedSocket = gatewayHarness.awaitServerSocket()
+        viewModel.setChatVisible(false)
+        gatewayHarness.sendGatewayReady(delayedSocket)
+        shadowOf(Looper.getMainLooper()).idleFor(500, TimeUnit.MILLISECONDS)
+        Thread.sleep(100)
+
+        assertEquals(0, historyReads.get())
+        assertEquals(
+            baselineActiveList,
+            gatewayHarness.rpcLog.count { it.first == "session.active_list" },
+        )
+        controlMethods.forEach { method ->
+            assertEquals(baseline.getValue(method), gatewayHarness.rpcLog.count { it.first == method })
+        }
+    }
+
     @Test
     fun staleHistoryReadCannotEraseATurnCompletedDuringTheFetch() {
         val loadCount = AtomicInteger(0)

app/src/test/kotlin/com/hermesandroid/relay/viewmodel/ConversationBindingControllerTest.kt

@@ -58,33 +58,6 @@ class ConversationBindingControllerTest {
         assertEquals("a2", controller.state.value.sessionId)
     }
 
-    @Test
-    fun newDraftKeepsExplicitAllProfilesOwnerAndRejectsStaleRestore() {
-        val alpha = Profile("alpha", "model-a", "Alpha")
-        controller.openExplicit("c::alpha", alpha.name, "a1", alpha, null)
-
-        controller.startFreshDraft()
-
-        assertEquals("c::alpha", controller.state.value.contextKey)
-        assertEquals("alpha", controller.state.value.profileName)
-        assertNull(controller.state.value.sessionId)
-        assertEquals(alpha, controller.state.value.displayProfile)
-        assertTrue(controller.state.value.hasExplicitOwner)
-        assertFalse(controller.reconcileGlobal("c::alpha", "alpha", "a1"))
-        assertNull(controller.state.value.sessionId)
-    }
-
-    @Test
-    fun newDraftPromotesGlobalOwnerAndRejectsItsStoredSession() {
-        controller.forceGlobal("c::alpha", "alpha", "a1")
-
-        controller.startFreshDraft()
-
-        assertTrue(controller.state.value.hasExplicitOwner)
-        assertNull(controller.state.value.sessionId)
-        assertFalse(controller.reconcileGlobal("c::alpha", "alpha", "a1"))
-    }
-
     @Test
     fun profileLockRejectsOtherOwnersAndAllowsTheLockedOwner() {
         val locked = AgentDisplay.profileSessionKey("beta")

app/src/test/kotlin/com/hermesandroid/relay/viewmodel/connection/ProfileControllerLockTest.kt

@@ -4,7 +4,6 @@ import android.content.Context
 import com.hermesandroid.relay.auth.AuthManager
 import com.hermesandroid.relay.data.AgentDisplay
 import com.hermesandroid.relay.data.Profile
-import com.hermesandroid.relay.data.SessionTransport
 import com.hermesandroid.relay.network.upstream.DashboardApiClient
 import com.hermesandroid.relay.network.upstream.DashboardProfileScope
 import com.hermesandroid.relay.network.upstream.GatewayAvailability
@@ -367,42 +366,4 @@ class ProfileControllerLockTest {
         controller.selectProfile(coder)
         assertEquals(coder, controller.selectedProfile.value)
     }
-
-    @Test
-    fun freshDraftFencesRestoreAndClearsOnlyItsConnectionProfileTransport() = runBlocking {
-        val sessions = controller.profileSessionStore
-        sessions.setSessionId(connectionId, mizu.name, SessionTransport.SSE, "old-sse")
-        sessions.setSessionId(connectionId, mizu.name, SessionTransport.GATEWAY, "old-gateway")
-        sessions.setSessionId("other-connection", mizu.name, SessionTransport.SSE, "other-sse")
-        controller.selectProfile(mizu)
-        awaitSelected(mizu.name)
-
-        controller.markFreshDraft(connectionId, mizu.name, SessionTransport.SSE)
-        withTimeout(5_000) {
-            sessions.sessionIdFlow(connectionId, mizu.name, SessionTransport.SSE)
-                .first { it == null }
-        }
-
-        // Simulate an older read observing the pre-clear value: the live intent
-        // fence still wins until a real session supersedes the draft.
-        sessions.setSessionId(connectionId, mizu.name, SessionTransport.SSE, "stale-sse")
-        controller.refreshLastSessionForProfile(connectionId, mizu.name)
-        assertNull(lastSessionIds.last())
-
-        assertEquals(
-            "old-gateway",
-            sessions.sessionIdFlow(connectionId, mizu.name, SessionTransport.GATEWAY).first(),
-        )
-        assertEquals(
-            "other-sse",
-            sessions.sessionIdFlow("other-connection", mizu.name, SessionTransport.SSE).first(),
-        )
-
-        controller.markSessionPersisted(connectionId, mizu.name, SessionTransport.SSE)
-        sessions.setSessionId(connectionId, mizu.name, SessionTransport.SSE, "new-sse")
-        controller.refreshLastSessionForProfile(connectionId, mizu.name)
-        withTimeout(5_000) {
-            while (lastSessionIds.lastOrNull() != "new-sse") Thread.sleep(10)
-        }
-    }
 }

docs/decisions.md

@@ -2836,20 +2836,6 @@ session or starts a draft, never hot-swaps a live session. Model/provider,
 personality, reasoning, approval, Fast, and YOLO state reset at the ViewModel
 context boundary before destination session truth can repopulate them.
 
-New Chat retains the current concrete conversation owner even when the drawer is
-browsing All Profiles. A profile choice made from that empty draft transfers an
-explicit fresh-draft intent rather than restoring the destination's previous
-session. Android persists and generation-fences that intent by exact
-connection/profile/transport; it clears only the resumable pointer, leaving the
-stored conversation, transcript, and per-owner composer drafts intact.
-
-Phone Threads keep their connection/chat-id ownership when leaving that surface.
-They are never transferred into a different profile binding: the atomic header
-switch retires provisional or in-progress promotion state before creating the
-destination profile draft, while durable inbox rows, promoted sessions,
-notification ownership, and session-to-chat-id indexes remain untouched. A
-generation fence prevents a delayed promotion from replacing the new draft.
-
 **Consequences.** The hamburger remains exclusively the Session Drawer. Agent
 Passport stays focused on inspection and configuration. The drawer may widen
 its read-only browse scope to all profiles and organize that combined set by
@@ -3991,3 +3977,36 @@ disappearance, client-side profile isolation, and method-not-found; physical
 and current-host certification remains tracked in `TODO.md`. An upstream
 profile field/filter or explicitly owned aggregate activity route would remove
 the remaining ambiguity for multi-profile clients.
+
+---
+
+## ADR 69 — Passive Android observation never attaches another client's Gateway turn
+
+**Status:** Accepted (2026-08-28).
+
+**Context.** `session.resume` and `session.activate` are live-runtime attachment
+operations, not read-only subscriptions. Android previously called
+`session.resume` while opening or foregrounding Chat and after loading a saved
+session's history. When Desktop/TUI already owned a running turn, that passive
+prewarm could rebind the runtime transport to Android. A later Android socket,
+route, or client teardown could then strand the producer or promote the foreign
+turn into an Android `GatewayTurn` whose cancellation sends `session.interrupt`.
+The issue was distinct from the earlier stale-view and missing-terminal recovery
+paths, which concern exact Android-owned checkpoints.
+
+**Decision.** Ordinary visibility, foreground restoration, Idle-socket recovery,
+and saved-session selection establish only the shared Gateway socket. They use
+profile-scoped REST history plus process-wide `session.active_list`; while an
+unowned row with the selected durable id is live, Android performs bounded
+history refreshes and one final read after settlement. These observer paths send
+no `session.resume`, `session.activate`, `prompt.submit`, or `session.interrupt`.
+Exact Android-owned checkpoints retain `session.activate` with durable-resume
+fallback, and explicit send or session-config actions may resume because the user
+is intentionally taking control of that destination.
+
+**Consequences.** Opening Android cannot replace, stop, or later cancel a turn
+already running in Desktop/TUI. Live token frames remain with the producing
+client; Android observes durable progress and final history without inventing a
+multi-subscriber Gateway contract. The first explicit Android mutation may pay
+the resume latency that passive prewarm previously hid. Cross-client fixtures
+and Android lifecycle coverage enforce the no-control-RPC observation boundary.

docs/gateway-contract-testing.md

@@ -41,6 +41,7 @@ the upstream contract identifiers it depends on.
 | `active_status_lifecycle` | `session.active_list` reports starting, working, waiting, and idle, then a complete empty process-wide snapshot permits removal of unambiguously owned prior rows |
 | `active_status_profile_scope` | A row has no profile metadata and a caller profile hint has no effect; the client must use exact client-held ownership and reject invented attribution |
 | `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 |
 
 Fixture evidence is a bounded metadata-only ring. It records sequence,
 connection number, RPC method, event type, scope classification, and outcome.

docs/spec.md

@@ -475,13 +475,13 @@ Bottom navigation bar with 4 tabs:
 - **Upstream animated pets** — the agent sheet consumes the profile-scoped Gateway `pet.info`, `pet.gallery`, `pet.select`, and `pet.disable` contracts. Android caches the bounded PNG/WebP sprite sheet by connection, effective profile, and `spritesheetRevision`; it sends `knownRevision` on refresh and reuses the existing bounded pet renderer for the returned geometry, row taxonomy, and activity states. The active upstream pet becomes the phone companion unless the user explicitly selected a phone-local floating pet. Selection and disable write Hermes `display.pet.*` state and therefore follow the profile across current Hermes surfaces; a method-not-found response leaves older hosts on the established local pet flow.
 - **Profile creation** — Manage uses `profiles.create` on current Gateways and labels authentication as shared sign-in, copied credential snapshot, or isolated/no-copy. Android serializes `mirror_credentials` and `share_auth` explicitly, reports best-effort SOUL/model/credential results without claiming full success, and never receives or logs credentials. The user may explicitly enable the authenticated Dashboard create route as an older-host fallback only for the legacy shared/default choice; explicit isolation never degrades to an ambiguous older mutation.
 - **Deletion boundary** — Hermes exposes no `profiles.delete` Gateway RPC. Android continues to delete profiles only through authenticated Dashboard `DELETE /api/profiles/{name}`.
-- **Profile switch lifecycle** — selecting an inactive profile never changes Hermes' sticky server default and never hot-swaps a live session. Android switches connection/profile context, restores that profile's last session only from the compatible Gateway or SSE transport slot, or opens a fresh draft. New Chat from All Profiles keeps the current conversation owner, and selecting another profile while that draft is empty transfers fresh-draft intent to the destination instead of restoring its prior session. That intent is fenced and persisted by exact connection/profile/transport while the prior session and history remain available in the drawer. Provisional and promoted phone Threads are not transferable profile drafts: a header profile switch exits their local routing state before establishing the destination draft, preserves their durable inbox/session/index data, and generation-fences any pending promotion. Gateway turns detach and reconcile in their original durable session; live SSE switching is disabled. Model/provider, personality, reasoning, approval, Fast, and YOLO state are cleared before destination session truth re-seeds them.
+- **Profile switch lifecycle** — selecting an inactive profile never changes Hermes' sticky server default and never hot-swaps a live session. Android switches connection/profile context, restores that profile's last session only from the compatible Gateway or SSE transport slot, or opens a fresh draft. Gateway turns detach and reconcile in their original durable session; live SSE switching is disabled. Model/provider, personality, reasoning, approval, Fast, and YOLO state are cleared before destination session truth re-seeds them.
 - **Bot Mode workspace** — the session drawer exposes one entry into a separate full-screen messenger surface; it does not add Bot or group rows to the ordinary session taxonomy. Android refreshes every saved Dashboard/Gateway with bounded concurrency, preserves last-good rows as visibly offline, and collapses duplicate routes by upstream `install_id` before assigning source-qualified handles. Every Bot carries an immutable `(connectionId, profile)` owner; labels, installation metadata, and the currently resolved URL are presentation/routing data rather than identity. All gateways and one-gateway filters never mutate the foreground connection.
 - **Canonical Bot Chat** — each individual row resolves the exact hidden session titled `Bot Chat` on its owning Gateway. Lookup failure is not absence, so Android creates and materializes the lazy row with `session.title` only after an authoritative empty exact-title result. The dedicated Bot Chat destination retains that route's pooled Gateway client, loads history through the same connection/profile Dashboard, sends only through Gateway, and returns directly to Bot Mode without rebinding Standard Chat or the global connection. `/new` or `/reset` compacts the canonical conversation instead of forking it. The route pool mints a fresh WebSocket ticket per dial, includes the immutable profile in the WebSocket URL, isolates credentials by exact trusted connection origin, and tears down only the removed connection's clients.
 - **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. Create, switch, rename, delete, pin/unpin, and archive/restore. 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.
 - **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**; exact terminal or `session.info {running:false}` can settle the matching generation. Because active-list rows normally have no profile metadata, Android assigns a row only through exact foreground/detached ownership already held by that client, or explicit profile metadata if a future upstream sends it. A bounded REST directory never proves global uniqueness. Unresolved rows 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, bare session ids from another profile, and delayed snapshots cannot revive newer settled state.
-- **Concurrent Gateway chats** — switching sessions, profiles, drafts, or Threads detaches the visible turn without sending `session.interrupt`; each running chat keeps a connection/profile/session-scoped checkpoint and reattaches to its live Gateway session when reopened. Explicit Stop still interrupts. SSE fallback stays single-stream and cancels on navigation.
+- **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 SSE fallback stays single-stream and cancels on navigation.
 - **Queued Gateway follow-ups** — every local queued item is immutably scoped to its originating connection, profile, stored session, transport, and run generation; only that run's completion can make it eligible, and switching sessions shows only that session's queue. Restored text queues retain the same scope, while unavailable/deleted destinations and non-restorable attachment queues fail visibly instead of following the current composer. 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.
 - **Durable composer drafts** — each connection/profile/session owns one app-private draft containing text, quote/edit context, and pending attachment bytes. Metadata and content-addressed blobs live under Android's no-backup directory, are capped at 64 drafts and 128 MB of retained blobs outside the active draft, flush when Chat backgrounds, and are removed after a successful send. Session/profile/connection navigation saves the previous owner before restoring the destination; an opened cross-profile session uses its actual owning profile rather than the global picker.
 - **Large paste review** — a default-on Chat setting converts any single insertion of at least 5,000 characters into a visible `pasted-text.txt` attachment before the normal message-length limit rejects it. Gateway uses upstream `file.attach`; API-server SSE and proactive Thread paths materialize the same UTF-8 text into the outgoing prompt and remove only the synthetic attachment from that transport, so the behavior never requires Relay or silently drops content.

test-fixtures/vanilla-gateway/tests/test_fixture.py

@@ -100,6 +100,47 @@ class FixtureTestCase(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(["user", "assistant"], [row["role"] for row in history["messages"]])
         self.assertEqual(2, history["pagination"]["returned"])
 
+    async def test_cross_client_observer_never_claims_or_interrupts_producer(self) -> None:
+        fixture, base_url = await self.start("cross_client_observation")
+        producer, _ = await self.connect(base_url)
+        await self.rpc(producer, 1, "session.resume", {"session_id": fixture.scenario.stored_session_id})
+        await producer.receive_json()
+        await self.rpc(producer, 2, "prompt.submit", {"text": "producer-only content"})
+        producer_frames = await self.frames_until(
+            producer,
+            lambda frame: frame.get("params", {}).get("type") == "message.delta",
+        )
+
+        observer, _ = await self.connect(base_url)
+        await self.rpc(observer, 3, "session.active_list")
+        active = (await observer.receive_json())["result"]["sessions"]
+        self.assertEqual("working", active[0]["status"])
+        async with self.session.get(
+            f"{base_url}/api/sessions/{fixture.scenario.stored_session_id}/messages",
+            params={"profile": "default", "limit": 500, "offset": 0, "order": "asc"},
+        ) as response:
+            self.assertEqual(200, response.status)
+            self.assertIsInstance((await response.json())["messages"], list)
+        await observer.close()
+
+        producer_frames += await self.frames_until(
+            producer,
+            lambda frame: frame.get("params", {}).get("type") == "message.complete",
+        )
+        self.assertIn(
+            "message.complete",
+            [frame.get("params", {}).get("type") for frame in producer_frames],
+        )
+        async with self.session.get(f"{base_url}/__fixture__/evidence") as response:
+            evidence = await response.json()
+        observer_methods = [
+            entry.get("method")
+            for entry in evidence["entries"]
+            if entry.get("kind") == "rpc" and entry.get("connection") == 2
+        ]
+        self.assertEqual(["session.active_list"], observer_methods)
+        self.assertNotIn("session.interrupt", observer_methods)
+
     async def test_rapid_chunks_tools_and_interims_keep_wire_order(self) -> None:
         _, base_url = await self.start("rapid_tools_interims")
         ws, _ = await self.connect(base_url)
@@ -265,6 +306,7 @@ class ScenarioTestCase(unittest.TestCase):
             "active_status_lifecycle",
             "active_status_profile_scope",
             "active_status_unsupported",
+            "cross_client_observation",
             "ordinary_turn",
             "rapid_tools_interims",
             "terminal_gap_activate",
@@ -304,6 +346,10 @@ class ScenarioTestCase(unittest.TestCase):
             ("gateway.settled_session_info",),
             load_scenario("terminal_gap_session_info").contract_requirements,
         )
+        self.assertEqual(
+            ("gateway.message_complete", "gateway.session_active_list"),
+            load_scenario("cross_client_observation").contract_requirements,
+        )
 
     def test_tls_arguments_must_be_paired(self) -> None:
         with contextlib.redirect_stderr(io.StringIO()):

test-fixtures/vanilla-gateway/vanilla_gateway/scenarios/cross_client_observation.json

@@ -0,0 +1,40 @@
+{
+  "name": "cross_client_observation",
+  "live_session_id": "fixture-desktop-live",
+  "stored_session_id": "fixture-shared-session",
+  "profile": "default",
+  "contract_requirements": [
+    "gateway.message_complete",
+    "gateway.session_active_list"
+  ],
+  "initial_history": [],
+  "turns": [
+    {
+      "steps": [
+        {"op": "event", "type": "message.start"},
+        {"op": "event", "type": "message.delta", "payload": {"text": "Desktop still owns this turn."}},
+        {"op": "sleep", "milliseconds": 250},
+        {
+          "op": "persist",
+          "messages": [
+            {"id": 1, "role": "user", "content": "Desktop prompt.", "timestamp": 1.0},
+            {"id": 2, "role": "assistant", "content": "Desktop still owns this turn.", "timestamp": 2.0, "finish_reason": "stop"}
+          ]
+        },
+        {"op": "set_running", "value": false},
+        {"op": "event", "type": "message.complete", "payload": {"text": "Desktop still owns this turn.", "status": "complete"}}
+      ]
+    }
+  ],
+  "active_list": {
+    "supported": true,
+    "snapshots": [
+      [
+        {"id": "fixture-desktop-live", "session_key": "fixture-shared-session", "status": "working", "current": false}
+      ],
+      [
+        {"id": "fixture-desktop-live", "session_key": "fixture-shared-session", "status": "idle", "current": false}
+      ]
+    ]
+  }
+}