fix(mobile-web): keep model selection scoped to remote sessions

.github/workflows/ci.yml

@@ -700,12 +700,13 @@ jobs:
         timeout-minutes: 5
         run: pnpm --dir src/mobile-web run test:account-browser
 
-      - name: Test mobile web composer and rollback layouts
+      - name: Test mobile web chat controls and layouts
         if: needs.build-impact.outputs.frontend_required != 'false'
         timeout-minutes: 3
         run: |
           pnpm --dir src/mobile-web run test:prompt-suggestion-browser
           pnpm --dir src/mobile-web run test:rollback-browser
+          pnpm --dir src/mobile-web run test:session-model-browser
 
       - name: Upload mobile input layout screenshots
         if: always() && needs.build-impact.outputs.frontend_required != 'false'

src/apps/cli/src/account.rs

@@ -332,22 +332,25 @@ impl CliAccountRoutingHost {
             relay_url,
             cancellation: Some(cancelled.clone()),
         };
-        tokio::select! {
-            biased;
-            _ = cancelled.changed() => { tracing::debug!("Retired pending RPC after account routing transition"); }
-            _ = async {
-        ACTIVE_PEER_FANOUT_OWNER
-            .scope(fanout_owner, async {
-                self.handle_current_relay_event(
-                    event,
-                    relay_client,
-                    account_generation,
-                    expected_token,
-                )
-                .await;
-            })
-            .await;
-            } => {}
+        let device_rpc = matches!(&event, RelayEvent::DeviceMessageReceived { .. });
+        let mut event_cancellation = cancelled.clone();
+        let handling = ACTIVE_PEER_FANOUT_OWNER.scope(
+            fanout_owner,
+            self.handle_current_relay_event(
+                event,
+                relay_client,
+                account_generation,
+                expected_token,
+                &mut event_cancellation,
+            ),
+        );
+        if device_rpc {
+            // RPC execution has its own cancellation boundary. Do not drop an
+            // accepted model write or lose its task-local owner and routing lease.
+            handling.await;
+        } else {
+            let _ =
+                remote_connect::rpc_execution::run_rpc_stage(&mut cancelled, false, handling).await;
         }
     }
 
@@ -357,6 +360,7 @@ impl CliAccountRoutingHost {
         relay_client: &Arc<RelayClient>,
         account_generation: u64,
         expected_token: &str,
+        cancellation: &mut tokio::sync::watch::Receiver<u64>,
     ) {
         let runtime = match self.runtime() {
             Ok(runtime) => runtime,
@@ -438,54 +442,63 @@ impl CliAccountRoutingHost {
                 encrypted_data,
                 nonce,
             } => {
-                let Ok((session, relay_url)) = runtime
-                    .read_account_context_for_generation(account_generation)
-                    .await
-                else {
-                    return;
-                };
-                if session.token != expected_token
-                    || !self
-                        .routing_loop_is_current(account_generation, relay_client)
-                        .await
-                {
-                    return;
-                }
-                let plaintext = match session
-                    .decrypt_from_peer(&relay_url, &source_device_id, &encrypted_data, &nonce)
-                    .await
-                {
-                    Ok(plaintext) => plaintext,
-                    Err(error) => {
-                        tracing::warn!("Failed to decrypt device message: {error}");
-                        return;
-                    }
-                };
                 use remote_connect::remote_server::{RemoteCommand, RemoteResponse};
-                let command: RemoteCommand = match serde_json::from_str(&plaintext) {
-                    Ok(command) => command,
-                    Err(error) => {
-                        tracing::warn!("Could not parse device command: {error}");
-                        return;
+                use remote_connect::rpc_execution::run_rpc_stage;
+
+                let prepare = async {
+                    let Ok((session, relay_url)) = runtime
+                        .read_account_context_for_generation(account_generation)
+                        .await
+                    else {
+                        return None;
+                    };
+                    if session.token != expected_token
+                        || !self
+                            .routing_loop_is_current(account_generation, relay_client)
+                            .await
+                    {
+                        return None;
                     }
+                    let plaintext = match session
+                        .decrypt_from_peer(&relay_url, &source_device_id, &encrypted_data, &nonce)
+                        .await
+                    {
+                        Ok(plaintext) => plaintext,
+                        Err(error) => {
+                            tracing::warn!("Failed to decrypt device message: {error}");
+                            return None;
+                        }
+                    };
+                    let command: RemoteCommand = match serde_json::from_str(&plaintext) {
+                        Ok(command) => command,
+                        Err(error) => {
+                            tracing::warn!("Could not parse device command: {error}");
+                            return None;
+                        }
+                    };
+                    tracing::info!(
+                        "Device command received: source={source_device_id} corr={correlation_id}"
+                    );
+                    let peer_key = match session
+                        .peer_message_key(&relay_url, &source_device_id)
+                        .await
+                    {
+                        Ok(key) => key,
+                        Err(error) => {
+                            tracing::warn!("Failed to resolve peer message key: {error}");
+                            return None;
+                        }
+                    };
+                    Some((session, command, peer_key))
                 };
-                tracing::info!(
-                    "Device command received: source={source_device_id} corr={correlation_id}"
-                );
-                let peer_key = match session
-                    .peer_message_key(&relay_url, &source_device_id)
-                    .await
-                {
-                    Ok(key) => key,
-                    Err(error) => {
-                        tracing::warn!("Failed to resolve peer message key: {error}");
-                        return;
-                    }
+                let Some((session, command, peer_key)) =
+                    run_rpc_stage(cancellation, false, prepare).await.flatten()
+                else {
+                    return;
                 };
-                // Host streams are answered from this host's memory for the
-                // requesting device; every other command goes to the dispatcher.
-                let hub = self.host_stream_hub().await;
-                let response =
+                let model_write = matches!(&command, RemoteCommand::SetSessionModel { .. });
+                let execute = async {
+                    let hub = self.host_stream_hub().await;
                     match openbitfun_core::service::remote_connect::handle_host_stream_command(
                         hub.as_ref(),
                         &source_device_id,
@@ -499,50 +512,55 @@ impl CliAccountRoutingHost {
                                 crate::peer_host::handle_host_invoke(command, args.clone()).await
                             }
                             RemoteCommand::DeviceEvent { event, payload } => {
-                                // A host is also a controller of its peers: stream
-                                // hints addressed to this device feed local subscribers.
                                 session.deliver_device_event(&source_device_id, event, payload);
                                 crate::peer_host::handle_device_event_command()
                             }
                             other => RemoteServer::new(peer_key).dispatch(other).await,
                         },
-                    };
-                if !self
-                    .routing_loop_is_current(account_generation, relay_client)
-                    .await
-                {
-                    return;
-                }
-                let response_json = serde_json::to_string(&response).unwrap_or_else(|error| {
-                    serde_json::to_string(&RemoteResponse::Error {
-                        message: format!("failed to serialize RPC response: {error}"),
-                    })
-                    .unwrap_or_else(|_| {
-                        r#"{"resp":"error","message":"serialize failed"}"#.to_string()
-                    })
-                });
-                let Ok((encrypted_response, response_nonce)) =
-                    encryption::encrypt_to_base64(&peer_key, &response_json)
-                else {
-                    tracing::warn!("Failed to encrypt RPC response");
+                    }
+                };
+                let Some(response) = run_rpc_stage(cancellation, model_write, execute).await else {
                     return;
                 };
-                let reply_target = if source_device_id == "rpc" {
-                    "rpc"
-                } else {
-                    source_device_id.as_str()
+                let reply = async {
+                    if !self
+                        .routing_loop_is_current(account_generation, relay_client)
+                        .await
+                    {
+                        return;
+                    }
+                    let response_json = serde_json::to_string(&response).unwrap_or_else(|error| {
+                        serde_json::to_string(&RemoteResponse::Error {
+                            message: format!("failed to serialize RPC response: {error}"),
+                        })
+                        .unwrap_or_else(|_| {
+                            r#"{"resp":"error","message":"serialize failed"}"#.to_string()
+                        })
+                    });
+                    let Ok((encrypted_response, response_nonce)) =
+                        encryption::encrypt_to_base64(&peer_key, &response_json)
+                    else {
+                        tracing::warn!("Failed to encrypt RPC response");
+                        return;
+                    };
+                    let reply_target = if source_device_id == "rpc" {
+                        "rpc"
+                    } else {
+                        source_device_id.as_str()
+                    };
+                    if let Err(error) = relay_client
+                        .send_device_message(
+                            reply_target,
+                            &correlation_id,
+                            &encrypted_response,
+                            &response_nonce,
+                        )
+                        .await
+                    {
+                        tracing::warn!("Failed to send RPC response: {error}");
+                    }
                 };
-                if let Err(error) = relay_client
-                    .send_device_message(
-                        reply_target,
-                        &correlation_id,
-                        &encrypted_response,
-                        &response_nonce,
-                    )
-                    .await
-                {
-                    tracing::warn!("Failed to send RPC response: {error}");
-                }
+                let _ = run_rpc_stage(cancellation, false, reply).await;
             }
             RelayEvent::Disconnected => {
                 tracing::info!("Device routing disconnected");

src/apps/cli/src/dispatch/mod.rs

@@ -724,6 +724,8 @@ async fn inspect_model_readiness() -> Result<ModelReadiness> {
                 default_models: Default::default(),
                 models_dev_reasoning_catalog: None,
                 reasoning_preset_selection_supported: true,
+                session_model_write_fence_supported: false,
+                session_model_write_token: None,
                 session_model_id: None,
                 session_reasoning_preset: None,
             });