async def connect(self) -> None:
        """Start the action_execution_server on the local machine or connect to an existing one."""
        self.send_status_message('STATUS$STARTING_RUNTIME')

        # Check if there's already a server running for this session ID
        if self.sid in _RUNNING_SERVERS:
            self.log('info', f'Connecting to existing server for session {self.sid}')
            server_info = _RUNNING_SERVERS[self.sid]
            self.server_process = server_info.process
            self._execution_server_port = server_info.execution_server_port
            self._log_thread = server_info.log_thread
            self._log_thread_exit_event = server_info.log_thread_exit_event
            self._vscode_port = server_info.vscode_port
            self._app_ports = server_info.app_ports
            self._temp_workspace = server_info.temp_workspace
            self.config.workspace_mount_path_in_sandbox = (
                server_info.workspace_mount_path
            )
            self.api_url = (
                f'{self.config.sandbox.local_runtime_url}:{self._execution_server_port}'
            )
        elif self.attach_to_existing:
            # If we're supposed to attach to an existing server but none exists, raise an error
            self.log('error', f'No existing server found for session {self.sid}')
            raise AgentRuntimeDisconnectedError(
                f'No existing server found for session {self.sid}'
            )
        else:
            # Set up workspace directory
            if self.config.workspace_base is not None:
                logger.warning(
                    f'Workspace base path is set to {self.config.workspace_base}. '
                    'It will be used as the path for the agent to run in. '
                    'Be careful, the agent can EDIT files in this directory!'
                )
                self.config.workspace_mount_path_in_sandbox = self.config.workspace_base
                self._temp_workspace = None
            else:
                # A temporary directory is created for the agent to run in
                logger.warning(
                    'Workspace base path is NOT set. Agent will run in a temporary directory.'
                )
                self._temp_workspace = tempfile.mkdtemp(
                    prefix=f'openhands_workspace_{self.sid}',
                )
                self.config.workspace_mount_path_in_sandbox = self._temp_workspace

            logger.info(
                f'Using workspace directory: {self.config.workspace_mount_path_in_sandbox}'
            )

            # Start a new server
            self._execution_server_port = self._find_available_port(
                EXECUTION_SERVER_PORT_RANGE
            )
            self._vscode_port = int(
                os.getenv('VSCODE_PORT')
                or str(self._find_available_port(VSCODE_PORT_RANGE))
            )
            self._app_ports = [
                int(
                    os.getenv('APP_PORT_1')
                    or str(self._find_available_port(APP_PORT_RANGE_1))
                ),
                int(
                    os.getenv('APP_PORT_2')
                    or str(self._find_available_port(APP_PORT_RANGE_2))
                ),
            ]
            self.api_url = (
                f'{self.config.sandbox.local_runtime_url}:{self._execution_server_port}'
            )

            # Start the server process
            cmd = get_action_execution_server_startup_command(
                server_port=self._execution_server_port,
                plugins=self.plugins,
                app_config=self.config,
                python_prefix=['poetry', 'run'],
                override_user_id=self._user_id,
                override_username=self._username,
            )

            self.log('debug', f'Starting server with command: {cmd}')
            env = os.environ.copy()
            # Get the code repo path
            code_repo_path = os.path.dirname(os.path.dirname(openhands.__file__))
            env['PYTHONPATH'] = os.pathsep.join(
                [code_repo_path, env.get('PYTHONPATH', '')]
            )
            env['OPENHANDS_REPO_PATH'] = code_repo_path
            env['LOCAL_RUNTIME_MODE'] = '1'
            env['VSCODE_PORT'] = str(self._vscode_port)

            # Derive environment paths using sys.executable
            interpreter_path = sys.executable
            python_bin_path = os.path.dirname(interpreter_path)
            env_root_path = os.path.dirname(python_bin_path)

            # Prepend the interpreter's bin directory to PATH for subprocesses
            env['PATH'] = f'{python_bin_path}{os.pathsep}{env.get("PATH", "")}'
            logger.debug(f'Updated PATH for subprocesses: {env["PATH"]}')

            # Check dependencies using the derived env_root_path if not skipped
            if os.getenv('SKIP_DEPENDENCY_CHECK', '') != '1':
                check_dependencies(code_repo_path, env_root_path)

            self.server_process = subprocess.Popen(  # noqa: S603
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                universal_newlines=True,
                bufsize=1,
                env=env,
                cwd=code_repo_path,  # Explicitly set the working directory
            )

            # Start a thread to read and log server output
            def log_output() -> None:
                if not self.server_process or not self.server_process.stdout:
                    self.log(
                        'error', 'Server process or stdout not available for logging.'
                    )
                    return

                try:
                    # Read lines while the process is running and stdout is available
                    while self.server_process.poll() is None:
                        if self._log_thread_exit_event.is_set():  # Check exit event
                            self.log('info', 'Log thread received exit signal.')
                            break  # Exit loop if signaled
                        line = self.server_process.stdout.readline()
                        if not line:
                            # Process might have exited between poll() and readline()
                            break
                        self.log('info', f'Server: {line.strip()}')

                    # Capture any remaining output after the process exits OR if signaled
                    if (
                        not self._log_thread_exit_event.is_set()
                    ):  # Check again before reading remaining
                        self.log(
                            'info', 'Server process exited, reading remaining output.'
                        )
                        for line in self.server_process.stdout:
                            if (
                                self._log_thread_exit_event.is_set()
                            ):  # Check inside loop too
                                self.log(
                                    'info',
                                    'Log thread received exit signal while reading remaining output.',
                                )
                                break
                            self.log('info', f'Server (remaining): {line.strip()}')

                except Exception as e:
                    # Log the error, but don't prevent the thread from potentially exiting
                    self.log('error', f'Error reading server output: {e}')
                finally:
                    self.log(
                        'info', 'Log output thread finished.'
                    )  # Add log for thread exit

            self._log_thread = threading.Thread(target=log_output, daemon=True)
            self._log_thread.start()

            # Store the server process in the global dictionary
            _RUNNING_SERVERS[self.sid] = ActionExecutionServerInfo(
                process=self.server_process,
                execution_server_port=self._execution_server_port,
                vscode_port=self._vscode_port,
                app_ports=self._app_ports,
                log_thread=self._log_thread,
                log_thread_exit_event=self._log_thread_exit_event,
                temp_workspace=self._temp_workspace,
                workspace_mount_path=self.config.workspace_mount_path_in_sandbox,
            )

        self.log('info', f'Waiting for server to become ready at {self.api_url}...')
        self.send_status_message('STATUS$WAITING_FOR_CLIENT')

        await call_sync_from_async(self._wait_until_alive)

        if not self.attach_to_existing:
            await call_sync_from_async(self.setup_initial_env)

        self.log(
            'debug',
            f'Server initialized with plugins: {[plugin.name for plugin in self.plugins]}',
        )
        if not self.attach_to_existing:
            self.send_status_message(' ')
        self._runtime_initialized = True
Start the action_execution_server on the local machine or connect to an existing one.
openhands/runtime/impl/local/local_runtime.py