# For connect to plugin runtime.
from __future__ import annotations

import asyncio
import io
import time
import zipfile
from typing import Any
import typing
import os
import sys
import httpx
import sqlalchemy
import yaml
from async_lru import alru_cache
from langbot_plugin.api.entities.builtin.pipeline.query import provider_session

from ..core import app
from . import handler
from ..utils import platform
from ..utils.managed_runtime import ManagedRuntimeConnector
from langbot_plugin.runtime.io.controllers.stdio import (
    client as stdio_client_controller,
)
from langbot_plugin.runtime.io.controllers.ws import client as ws_client_controller
from langbot_plugin.api.entities import events
from langbot_plugin.api.entities import context
import langbot_plugin.runtime.io.connection as base_connection
from langbot_plugin.api.definition.components.manifest import ComponentManifest
from langbot_plugin.api.entities.builtin.command import (
    context as command_context,
    errors as command_errors,
)
from langbot_plugin.runtime.plugin.mgr import PluginInstallSource
from ..core import taskmgr
from ..entity.persistence import plugin as persistence_plugin


class PluginRuntimeNotConnectedError(RuntimeError):
    """Raised when plugin runtime operations are requested before connection."""


class PluginRuntimeConnector(ManagedRuntimeConnector):
    """Plugin runtime connector"""

    handler: handler.RuntimeConnectionHandler

    handler_task: asyncio.Task

    heartbeat_task: asyncio.Task | None = None

    stdio_client_controller: stdio_client_controller.StdioClientController

    ctrl: stdio_client_controller.StdioClientController | ws_client_controller.WebSocketClientController

    runtime_disconnect_callback: typing.Callable[
        [PluginRuntimeConnector], typing.Coroutine[typing.Any, typing.Any, None]
    ]

    is_enable_plugin: bool = True
    """Mark if the plugin system is enabled"""

    def __init__(
        self,
        ap: app.Application,
        runtime_disconnect_callback: typing.Callable[
            [PluginRuntimeConnector], typing.Coroutine[typing.Any, typing.Any, None]
        ],
    ):
        super().__init__(ap)
        self.runtime_disconnect_callback = runtime_disconnect_callback
        self.is_enable_plugin = self.ap.instance_config.data.get('plugin', {}).get('enable', True)

    async def heartbeat_loop(self):
        while True:
            await asyncio.sleep(20)
            try:
                await self.ping_plugin_runtime()
                self.ap.logger.debug('Heartbeat to plugin runtime success.')
            except Exception as e:
                self.ap.logger.debug(f'Failed to heartbeat to plugin runtime: {e}')

    async def initialize(self):
        if not self.is_enable_plugin:
            self.ap.logger.info('Plugin system is disabled.')
            return

        async def new_connection_callback(connection: base_connection.Connection):
            async def disconnect_callback(
                rchandler: handler.RuntimeConnectionHandler,
            ) -> bool:
                if platform.get_platform() == 'docker' or platform.use_websocket_to_connect_plugin_runtime():
                    self.ap.logger.error('Disconnected from plugin runtime, trying to reconnect...')
                    await self.runtime_disconnect_callback(self)
                    return False
                else:
                    self.ap.logger.error(
                        'Disconnected from plugin runtime, cannot automatically reconnect while LangBot connects to plugin runtime via stdio, please restart LangBot.'
                    )
                    return False

            self.handler = handler.RuntimeConnectionHandler(connection, disconnect_callback, self.ap)

            self.handler_task = asyncio.create_task(self.handler.run())
            _ = await self.handler.ping()
            # Push the configured marketplace (Space) URL to the runtime so it
            # downloads plugins from the same Space LangBot is bound to, rather
            # than relying on the runtime's own env/default.
            space_url = self.ap.instance_config.data.get('space', {}).get('url', '').rstrip('/')
            if space_url:
                try:
                    await self.handler.set_runtime_config(cloud_service_url=space_url)
                    self.ap.logger.info(f'Pushed marketplace URL to plugin runtime: {space_url}')
                except Exception as e:
                    self.ap.logger.warning(f'Failed to push runtime config: {e}')
            self.ap.logger.info('Connected to plugin runtime.')
            await self.handler_task

        task: asyncio.Task | None = None

        if platform.get_platform() == 'docker' or platform.use_websocket_to_connect_plugin_runtime():  # use websocket
            self.ap.logger.info('use websocket to connect to plugin runtime')
            ws_url = self.ap.instance_config.data.get('plugin', {}).get(
                'runtime_ws_url', 'ws://langbot_plugin_runtime:5400/control/ws'
            )

            async def make_connection_failed_callback(
                ctrl: ws_client_controller.WebSocketClientController,
                exc: Exception = None,
            ) -> None:
                if exc is not None:
                    self.ap.logger.error(f'Failed to connect to plugin runtime({ws_url}): {exc}')
                else:
                    self.ap.logger.error(f'Failed to connect to plugin runtime({ws_url}), trying to reconnect...')
                await self.runtime_disconnect_callback(self)

            self.ctrl = ws_client_controller.WebSocketClientController(
                ws_url=ws_url,
                make_connection_failed_callback=make_connection_failed_callback,
            )
            task = self.ctrl.run(new_connection_callback)
        elif platform.get_platform() == 'win32':
            # Due to Windows's lack of supports for both stdio and subprocess:
            # See also: https://docs.python.org/zh-cn/3.13/library/asyncio-platforms.html
            # We have to launch runtime via cmd but communicate via ws.
            self.ap.logger.info('(windows) use cmd to launch plugin runtime and communicate via ws')

            await self._start_runtime_subprocess('-m', 'langbot_plugin.cli.__init__', 'rt')

            ws_url = 'ws://localhost:5400/control/ws'

            async def make_connection_failed_callback(
                ctrl: ws_client_controller.WebSocketClientController,
                exc: Exception = None,
            ) -> None:
                if exc is not None:
                    self.ap.logger.error(f'(windows) Failed to connect to plugin runtime({ws_url}): {exc}')
                else:
                    self.ap.logger.error(
                        f'(windows) Failed to connect to plugin runtime({ws_url}), trying to reconnect...'
                    )
                await self.runtime_disconnect_callback(self)

            self.ctrl = ws_client_controller.WebSocketClientController(
                ws_url=ws_url,
                make_connection_failed_callback=make_connection_failed_callback,
            )
            task = self.ctrl.run(new_connection_callback)

        else:  # stdio
            self.ap.logger.info('use stdio to connect to plugin runtime')
            # cmd: lbp rt -s
            python_path = sys.executable
            env = os.environ.copy()
            self.ctrl = stdio_client_controller.StdioClientController(
                command=python_path,
                args=['-m', 'langbot_plugin.cli.__init__', 'rt', '-s'],
                env=env,
            )
            task = self.ctrl.run(new_connection_callback)

        if self.heartbeat_task is None:
            self.heartbeat_task = asyncio.create_task(self.heartbeat_loop())

        asyncio.create_task(task)

    async def initialize_plugins(self):
        pass

    async def ping_plugin_runtime(self):
        if not hasattr(self, 'handler'):
            raise PluginRuntimeNotConnectedError('Plugin runtime is not connected')

        return await self.handler.ping()

    def _inspect_plugin_package(
        self,
        file_bytes: bytes,
        task_context: taskmgr.TaskContext | None,
    ) -> tuple[str | None, str | None]:
        """Extract plugin identity and dependency metadata from a plugin package."""
        plugin_author = None
        plugin_name = None

        try:
            with zipfile.ZipFile(io.BytesIO(file_bytes)) as zf:
                try:
                    manifest = yaml.safe_load(zf.read('manifest.yaml').decode('utf-8', errors='ignore')) or {}
                    metadata = manifest.get('metadata', {})
                    plugin_author = metadata.get('author')
                    plugin_name = metadata.get('name')
                except Exception:
                    pass

                if task_context is not None:
                    for name in zf.namelist():
                        if name.endswith('requirements.txt'):
                            content = zf.read(name).decode('utf-8', errors='ignore')
                            deps = [
                                line.strip()
                                for line in content.splitlines()
                                if line.strip() and not line.strip().startswith('#')
                            ]
                            task_context.metadata['deps_total'] = len(deps)
                            task_context.metadata['deps_list'] = deps
                            break
        except Exception:
            pass

        return plugin_author, plugin_name

    async def _install_mcp_from_marketplace(
        self,
        mcp_data: dict[str, Any],
        task_context: taskmgr.TaskContext | None = None,
    ):
        """Install an MCP server from marketplace data.

        Marketplace MCP records carry the runtime-ready ``mode`` and
        ``extra_args`` directly (the same shape LangBot stores in
        ``mcp_servers``), so they are used as-is rather than reconstructed.
        For ``stdio`` this preserves ``command``/``args``/``env``/``box``;
        for ``http``/``sse`` it preserves ``url``/``headers``/``timeout``/
        ``ssereadtimeout``.
        """
        from ..entity.persistence import mcp as persistence_mcp
        import uuid

        mode = mcp_data.get('mode') or 'stdio'
        extra_args = mcp_data.get('extra_args') or {}
        # The MCP transport selection was simplified to two modes: 'stdio'
        # (local, Box-sandboxed) and 'remote' (the runtime auto-detects
        # Streamable HTTP vs. legacy SSE from the URL). Marketplace records may
        # still carry the older 'http'/'sse' modes — normalize them to 'remote'
        # so the installed server shows up correctly in the two-option UI. The
        # connection args (url/headers/timeout/ssereadtimeout) are preserved and
        # consumed by the auto-detecting remote transport regardless.
        if mode in ('http', 'sse'):
            mode = 'remote'
        # Marketplace records carry the rendered README markdown; persist it so
        # the detail page Docs tab works offline and without a marketplace round-trip.
        readme = mcp_data.get('readme') or ''
        # Use __ instead of / to avoid URL routing issues with slashes
        name = f'{mcp_data.get("author", "")}__{mcp_data.get("name", "")}'

        # Check if MCP server already exists
        existing = await self.ap.persistence_mgr.execute_async(
            sqlalchemy.select(persistence_mcp.MCPServer).where(persistence_mcp.MCPServer.name == name)
        )
        if existing.scalar_one_or_none():
            self.ap.logger.info(f'MCP server {name} already exists, skipping installation')
            return

        # Create MCP server record
        server_uuid = str(uuid.uuid4())
        server_data = {
            'uuid': server_uuid,
            'name': name,
            'enable': True,
            'mode': mode,
            'extra_args': extra_args,
            'readme': readme,
        }

        await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_mcp.MCPServer).values(server_data))

        # Start the MCP server
        result = await self.ap.persistence_mgr.execute_async(
            sqlalchemy.select(persistence_mcp.MCPServer).where(persistence_mcp.MCPServer.uuid == server_uuid)
        )
        server_entity = result.first()
        if server_entity:
            server_config = self.ap.persistence_mgr.serialize_model(persistence_mcp.MCPServer, server_entity)
            if self.ap.tool_mgr.mcp_tool_loader:
                mcp_task = asyncio.create_task(self.ap.tool_mgr.mcp_tool_loader.host_mcp_server(server_config))
                self.ap.tool_mgr.mcp_tool_loader._hosted_mcp_tasks.append(mcp_task)

        self.ap.logger.info(f'Installed MCP server {name} from marketplace')

    async def _install_skill_from_zip(
        self,
        file_bytes: bytes,
        filename: str,
        task_context: taskmgr.TaskContext | None = None,
    ):
        """Install a skill from marketplace ZIP data."""
        from ..api.http.service.skill import SkillService

        skill_service = SkillService(self.ap)

        self.ap.logger.info(f'Installing skill from marketplace ZIP ({len(file_bytes)} bytes)')

        # Install from ZIP using skill service
        result = await skill_service.install_from_zip_upload(
            file_bytes=file_bytes,
            filename=filename + '.zip',
        )
        self.ap.logger.info(f'Skill installed successfully: {result}')

    def _build_plugin_startup_failure_message(
        self,
        plugin_author: str,
        plugin_name: str,
        task_context: taskmgr.TaskContext | None,
    ) -> str:
        dep_hint = ''
        if task_context is not None:
            current_dep = task_context.metadata.get('current_dep')
            if current_dep:
                dep_hint = f' Last dependency: {current_dep}.'

        return (
            f'Plugin {plugin_author}/{plugin_name} failed to start after installation. '
            f'Dependency installation or plugin initialization may have failed.{dep_hint} '
            f'Please check the plugin requirements and runtime logs.'
        )

    async def _wait_for_installed_plugin_ready(
        self,
        plugin_author: str | None,
        plugin_name: str | None,
        task_context: taskmgr.TaskContext | None,
        timeout: float = 30,
    ):
        """Wait until the installed plugin is registered by the runtime.

        The plugin runtime launches plugins asynchronously. If dependency installation
        fails, the plugin process exits before registration; without this check the
        install task can incorrectly finish successfully.
        """
        if not plugin_author or not plugin_name:
            return

        deadline = time.time() + timeout
        last_error: Exception | None = None
        while time.time() < deadline:
            try:
                plugin = await self.get_plugin_info(plugin_author, plugin_name)
                if plugin is not None:
                    status = plugin.get('status')
                    if status == 'initialized':
                        return
            except Exception as e:
                last_error = e

            await asyncio.sleep(0.5)

        message = self._build_plugin_startup_failure_message(plugin_author, plugin_name, task_context)
        if last_error is not None:
            message = f'{message} Last runtime error: {last_error}'
        raise RuntimeError(message)

    async def install_plugin(
        self,
        install_source: PluginInstallSource,
        install_info: dict[str, Any],
        task_context: taskmgr.TaskContext | None = None,
    ):
        plugin_author = install_info.get('plugin_author')
        plugin_name = install_info.get('plugin_name')

        if install_source == PluginInstallSource.MARKETPLACE:
            # Handle marketplace plugin/mcp/skill installation
            plugin_author = install_info.get('plugin_author', '')
            plugin_name = install_info.get('plugin_name', '')
            space_url = (
                self.ap.instance_config.data.get('space', {}).get('url', 'https://space.langbot.app').rstrip('/')
            )

            # Try MCP endpoint first
            async with httpx.AsyncClient(trust_env=True, timeout=15) as client:
                mcp_resp = await client.get(f'{space_url}/api/v1/marketplace/mcps/{plugin_author}/{plugin_name}')
                if mcp_resp.status_code == 200:
                    mcp_data = mcp_resp.json().get('data', {}).get('mcp', {})
                    if mcp_data.get('mode'):
                        # It's an MCP - create server locally
                        self.ap.logger.info(f'Installing MCP from marketplace: {plugin_author}/{plugin_name}')
                        if task_context:
                            task_context.set_current_action('installing mcp server')
                        await self._install_mcp_from_marketplace(mcp_data, task_context)
                        # Best-effort install report (bumps marketplace install_count).
                        try:
                            await client.post(
                                f'{space_url}/api/v1/marketplace/mcps/{plugin_author}/{plugin_name}/install'
                            )
                        except Exception as report_err:
                            self.ap.logger.debug(f'Failed to report MCP install: {report_err}')
                        return
                    else:
                        raise Exception(f'MCP {plugin_author}/{plugin_name} has no mode')
                elif mcp_resp.status_code == 404:
                    # Try skill endpoint - download ZIP and install
                    self.ap.logger.info(f'Trying skill endpoint for: {plugin_author}/{plugin_name}')
                    if task_context:
                        task_context.set_current_action('checking skill marketplace')

                    # Get skill detail to find version
                    skill_resp = await client.get(
                        f'{space_url}/api/v1/marketplace/skills/{plugin_author}/{plugin_name}'
                    )
                    if skill_resp.status_code == 200:
                        self.ap.logger.info(f'Installing skill from marketplace: {plugin_author}/{plugin_name}')
                        if task_context:
                            task_context.set_current_action('installing skill from marketplace')

                        # Download the skill ZIP (no version needed - uses latest)
                        if task_context:
                            task_context.set_current_action('downloading skill package')

                        download_resp = await client.get(
                            f'{space_url}/api/v1/marketplace/skills/download/{plugin_author}/{plugin_name}'
                        )
                        if download_resp.status_code != 200:
                            raise Exception(
                                f'Failed to download skill {plugin_author}/{plugin_name}: {download_resp.status_code}'
                            )

                        file_bytes = download_resp.content
                        file_size = len(file_bytes)
                        self.ap.logger.info(f'Downloaded skill ZIP ({file_size} bytes)')

                        # Install skill from ZIP using skill service
                        await self._install_skill_from_zip(file_bytes, f'{plugin_author}-{plugin_name}', task_context)
                        return
                    elif skill_resp.status_code == 404:
                        # Try plugin endpoint - get versions and download
                        self.ap.logger.info(f'Trying plugin endpoint for: {plugin_author}/{plugin_name}')
                        if task_context:
                            task_context.set_current_action('checking plugin marketplace')

                        # Get plugin versions to find latest
                        versions_resp = await client.get(
                            f'{space_url}/api/v1/marketplace/plugins/{plugin_author}/{plugin_name}/versions'
                        )
                        if versions_resp.status_code == 200:
                            versions_data = versions_resp.json().get('data', {}).get('versions', [])
                            if versions_data:
                                latest_version = versions_data[0].get('version', '')
                                if latest_version:
                                    self.ap.logger.info(
                                        f'Installing plugin from marketplace: {plugin_author}/{plugin_name} v{latest_version}'
                                    )
                                    if task_context:
                                        task_context.set_current_action('downloading plugin package')

                                    download_resp = await client.get(
                                        f'{space_url}/api/v1/marketplace/plugins/download/{plugin_author}/{plugin_name}/{latest_version}'
                                    )
                                    if download_resp.status_code != 200:
                                        raise Exception(
                                            f'Failed to download plugin {plugin_author}/{plugin_name}: {download_resp.status_code}'
                                        )

                                    file_bytes = download_resp.content
                                    self._inspect_plugin_package(file_bytes, task_context)
                                    file_key = await self.handler.send_file(file_bytes, 'lbpkg')
                                    install_info['plugin_file_key'] = file_key
                                    self.ap.logger.info(f'Transfered file {file_key} to plugin runtime')
                                    # Continue to install via runtime
                                else:
                                    raise Exception(f'No version found for plugin {plugin_author}/{plugin_name}')
                            else:
                                raise Exception(f'Plugin {plugin_author}/{plugin_name} has no versions')
                        else:
                            raise Exception(f'Plugin {plugin_author}/{plugin_name} not found in marketplace')
                    else:
                        skill_resp.raise_for_status()
                        raise Exception(f'Failed to get skill {plugin_author}/{plugin_name}')
                else:
                    mcp_resp.raise_for_status()
                    raise Exception(f'Failed to get MCP {plugin_author}/{plugin_name}')

        if install_source == PluginInstallSource.LOCAL:
            # transfer file before install
            file_bytes = install_info['plugin_file']
            plugin_author, plugin_name = self._inspect_plugin_package(file_bytes, task_context)
            if task_context is not None and plugin_author and plugin_name:
                task_context.metadata['plugin_name'] = f'{plugin_author}/{plugin_name}'
            file_key = await self.handler.send_file(file_bytes, 'lbpkg')
            install_info['plugin_file_key'] = file_key
            del install_info['plugin_file']
            self.ap.logger.info(f'Transfered file {file_key} to plugin runtime')
        elif install_source == PluginInstallSource.GITHUB:
            # download and transfer file with streaming progress
            try:
                async with httpx.AsyncClient(
                    trust_env=True,
                    follow_redirects=True,
                    timeout=60,
                ) as client:
                    async with client.stream('GET', install_info['asset_url']) as response:
                        response.raise_for_status()
                        total = int(response.headers.get('content-length', 0))
                        downloaded = 0
                        chunks: list[bytes] = []
                        start_time = time.time()

                        if task_context is not None:
                            task_context.set_current_action('downloading plugin package')
                            task_context.metadata['download_total'] = total
                            task_context.metadata['download_current'] = 0
                            task_context.metadata['download_speed'] = 0

                        async for chunk in response.aiter_bytes(chunk_size=8192):
                            chunks.append(chunk)
                            downloaded += len(chunk)

                            if task_context is not None:
                                elapsed = time.time() - start_time
                                task_context.metadata['download_current'] = downloaded
                                task_context.metadata['download_total'] = total
                                task_context.metadata['download_speed'] = downloaded / elapsed if elapsed > 0 else 0

                    file_bytes = b''.join(chunks)
                    plugin_author, plugin_name = self._inspect_plugin_package(file_bytes, task_context)
                    if task_context is not None and plugin_author and plugin_name:
                        task_context.metadata['plugin_name'] = f'{plugin_author}/{plugin_name}'
                    file_key = await self.handler.send_file(file_bytes, 'lbpkg')
                    install_info['plugin_file_key'] = file_key
                    self.ap.logger.info(f'Transfered file {file_key} to plugin runtime')
            except Exception as e:
                self.ap.logger.error(f'Failed to download file from GitHub: {e}')
                raise Exception(f'Failed to download file from GitHub: {e}')

        async for ret in self.handler.install_plugin(install_source.value, install_info):
            current_action = ret.get('current_action', None)
            if current_action is not None:
                if task_context is not None:
                    task_context.set_current_action(current_action)

            trace = ret.get('trace', None)
            if trace is not None:
                if task_context is not None:
                    task_context.trace(trace)

            # Forward structured metadata from runtime
            metadata = ret.get('metadata', None)
            if metadata is not None and task_context is not None:
                task_context.metadata.update(metadata)

        await self._wait_for_installed_plugin_ready(plugin_author, plugin_name, task_context)

    async def upgrade_plugin(
        self,
        plugin_author: str,
        plugin_name: str,
        task_context: taskmgr.TaskContext | None = None,
    ) -> dict[str, Any]:
        async for ret in self.handler.upgrade_plugin(plugin_author, plugin_name):
            current_action = ret.get('current_action', None)
            if current_action is not None:
                if task_context is not None:
                    task_context.set_current_action(current_action)

            trace = ret.get('trace', None)
            if trace is not None:
                if task_context is not None:
                    task_context.trace(trace)

    async def delete_plugin(
        self,
        plugin_author: str,
        plugin_name: str,
        delete_data: bool = False,
        task_context: taskmgr.TaskContext | None = None,
    ) -> dict[str, Any]:
        async for ret in self.handler.delete_plugin(plugin_author, plugin_name):
            current_action = ret.get('current_action', None)
            if current_action is not None:
                if task_context is not None:
                    task_context.set_current_action(current_action)

            trace = ret.get('trace', None)
            if trace is not None:
                if task_context is not None:
                    task_context.trace(trace)

        # Clean up plugin settings and binary storage if requested
        if delete_data:
            if task_context is not None:
                task_context.trace('Cleaning up plugin configuration and storage...')
            await self.handler.cleanup_plugin_data(plugin_author, plugin_name)

    async def list_plugins(self, component_kinds: list[str] | None = None) -> list[dict[str, Any]]:
        """List plugins, optionally filtered by component kinds.

        Args:
            component_kinds: Optional list of component kinds to filter by.
                           If provided, only plugins that contain at least one
                           component of the specified kinds will be returned.
                           E.g., ['Command', 'EventListener', 'Tool'] for pipeline-related plugins.
        """
        if not self.is_enable_plugin:
            return []

        plugins = await self.handler.list_plugins()

        # Filter plugins by component kinds if specified
        if component_kinds is not None:
            filtered_plugins = []
            for plugin in plugins:
                components = plugin.get('components', [])
                has_matching_component = False
                for component in components:
                    component_kind = component.get('manifest', {}).get('manifest', {}).get('kind', '')
                    if component_kind in component_kinds:
                        has_matching_component = True
                        break
                if has_matching_component:
                    filtered_plugins.append(plugin)
            plugins = filtered_plugins

        # Sort plugins: debug plugins first, then by installation time (newest first)
        # Get installation timestamps from database in a single query
        plugin_timestamps = {}

        if plugins:
            # Build list of (author, name) tuples for all plugins
            plugin_ids = []
            for plugin in plugins:
                author = plugin.get('manifest', {}).get('manifest', {}).get('metadata', {}).get('author', '')
                name = plugin.get('manifest', {}).get('manifest', {}).get('metadata', {}).get('name', '')
                if author and name:
                    plugin_ids.append((author, name))

            # Fetch all timestamps in a single query using OR conditions
            if plugin_ids:
                conditions = [
                    sqlalchemy.and_(
                        persistence_plugin.PluginSetting.plugin_author == author,
                        persistence_plugin.PluginSetting.plugin_name == name,
                    )
                    for author, name in plugin_ids
                ]

                result = await self.ap.persistence_mgr.execute_async(
                    sqlalchemy.select(
                        persistence_plugin.PluginSetting.plugin_author,
                        persistence_plugin.PluginSetting.plugin_name,
                        persistence_plugin.PluginSetting.created_at,
                    ).where(sqlalchemy.or_(*conditions))
                )

                for row in result:
                    plugin_id = f'{row.plugin_author}/{row.plugin_name}'
                    plugin_timestamps[plugin_id] = row.created_at

        # Sort: debug plugins first (descending), then by created_at (descending)
        def sort_key(plugin):
            author = plugin.get('manifest', {}).get('manifest', {}).get('metadata', {}).get('author', '')
            name = plugin.get('manifest', {}).get('manifest', {}).get('metadata', {}).get('name', '')
            plugin_id = f'{author}/{name}'

            is_debug = plugin.get('debug', False)
            created_at = plugin_timestamps.get(plugin_id)

            # Return tuple: (not is_debug, -timestamp)
            # not is_debug: False (0) for debug plugins, True (1) for non-debug
            # -timestamp: to sort newest first (will be None for plugins without timestamp)
            timestamp_value = -created_at.timestamp() if created_at else 0
            return (not is_debug, timestamp_value)

        plugins.sort(key=sort_key)

        return plugins

    async def get_plugin_info(self, author: str, plugin_name: str) -> dict[str, Any]:
        return await self.handler.get_plugin_info(author, plugin_name)

    async def set_plugin_config(self, plugin_author: str, plugin_name: str, config: dict[str, Any]) -> dict[str, Any]:
        return await self.handler.set_plugin_config(plugin_author, plugin_name, config)

    @alru_cache(ttl=5 * 60)  # 5 minutes
    async def get_plugin_icon(self, plugin_author: str, plugin_name: str) -> dict[str, Any]:
        return await self.handler.get_plugin_icon(plugin_author, plugin_name)

    @alru_cache(ttl=5 * 60)  # 5 minutes
    async def get_plugin_readme(self, plugin_author: str, plugin_name: str, language: str = 'en') -> str:
        return await self.handler.get_plugin_readme(plugin_author, plugin_name, language)

    async def get_plugin_logs(
        self,
        plugin_author: str,
        plugin_name: str,
        limit: int = 200,
        level: str | None = None,
    ) -> list[dict[str, Any]]:
        # Not cached: logs are live and change constantly.
        return await self.handler.get_plugin_logs(plugin_author, plugin_name, limit, level)

    @alru_cache(ttl=5 * 60)
    async def get_plugin_assets(self, plugin_author: str, plugin_name: str, filepath: str) -> dict[str, Any]:
        return await self.handler.get_plugin_assets(plugin_author, plugin_name, filepath)

    async def handle_page_api(
        self,
        plugin_author: str,
        plugin_name: str,
        page_id: str,
        endpoint: str,
        method: str,
        body: Any = None,
    ) -> dict[str, Any]:
        return await self.handler.handle_page_api(plugin_author, plugin_name, page_id, endpoint, method, body)

    async def get_debug_info(self) -> dict[str, Any]:
        """Get debug information including debug key and WS URL"""
        if not self.is_enable_plugin:
            return {}
        return await self.handler.get_debug_info()

    async def emit_event(
        self,
        event: events.BaseEventModel,
        bound_plugins: list[str] | None = None,
    ) -> context.EventContext:
        event_ctx = context.EventContext.from_event(event)

        if not self.is_enable_plugin:
            event_ctx._emitted_plugins = []
            event_ctx._response_sources = []
            return event_ctx

        # Pass include_plugins to runtime for filtering
        event_ctx_result = await self.handler.emit_event(
            event_ctx.model_dump(serialize_as_any=False), include_plugins=bound_plugins
        )

        event_ctx = context.EventContext.model_validate(event_ctx_result['event_context'])
        event_ctx._emitted_plugins = event_ctx_result.get('emitted_plugins', [])
        if 'response_sources' in event_ctx_result:
            event_ctx._response_sources = event_ctx_result['response_sources']

        return event_ctx

    async def notify_plugin_diagnostic(self, diagnostic: dict[str, Any]) -> None:
        """Best-effort diagnostic forwarding to the plugin runtime."""
        if not self.is_enable_plugin:
            return
        try:
            await self.handler.notify_plugin_diagnostic(diagnostic)
        except Exception as e:
            self.ap.logger.debug(f'Plugin diagnostic forwarding skipped: {e}')

    async def list_tools(self, bound_plugins: list[str] | None = None) -> list[ComponentManifest]:
        if not self.is_enable_plugin:
            return []

        # Pass include_plugins to runtime for filtering
        list_tools_data = await self.handler.list_tools(include_plugins=bound_plugins)

        tools = [ComponentManifest.model_validate(tool) for tool in list_tools_data]

        return tools

    async def call_tool(
        self,
        tool_name: str,
        parameters: dict[str, Any],
        session: provider_session.Session,
        query_id: int,
        bound_plugins: list[str] | None = None,
    ) -> dict[str, Any]:
        if not self.is_enable_plugin:
            return {'error': 'Tool not found: plugin system is disabled'}

        # Pass include_plugins to runtime for validation
        return await self.handler.call_tool(
            tool_name, parameters, session.model_dump(serialize_as_any=True), query_id, include_plugins=bound_plugins
        )

    async def list_commands(self, bound_plugins: list[str] | None = None) -> list[ComponentManifest]:
        if not self.is_enable_plugin:
            return []

        # Pass include_plugins to runtime for filtering
        list_commands_data = await self.handler.list_commands(include_plugins=bound_plugins)

        commands = [ComponentManifest.model_validate(command) for command in list_commands_data]

        return commands

    async def execute_command(
        self, command_ctx: command_context.ExecuteContext, bound_plugins: list[str] | None = None
    ) -> typing.AsyncGenerator[command_context.CommandReturn, None]:
        if not self.is_enable_plugin:
            yield command_context.CommandReturn(error=command_errors.CommandNotFoundError(command_ctx.command))
            return

        # Pass include_plugins to runtime for validation
        gen = self.handler.execute_command(command_ctx.model_dump(serialize_as_any=True), include_plugins=bound_plugins)

        async for ret in gen:
            cmd_ret = command_context.CommandReturn.model_validate(ret)

            yield cmd_ret

    async def retrieve_knowledge(
        self,
        plugin_author: str,
        plugin_name: str,
        retriever_name: str,
        retrieval_context: dict[str, Any],
    ) -> dict[str, Any]:
        """Retrieve knowledge using a KnowledgeEngine instance."""
        if not self.is_enable_plugin:
            return {'results': []}

        return await self.handler.retrieve_knowledge(plugin_author, plugin_name, retriever_name, retrieval_context)

    def dispose(self):
        # On non-Windows stdio mode, terminate via the controller's process handle.
        # On Windows, the managed subprocess is cleaned up by the base class.
        if (
            self.is_enable_plugin
            and hasattr(self, 'ctrl')
            and isinstance(self.ctrl, stdio_client_controller.StdioClientController)
        ):
            self.ap.logger.info('Terminating plugin runtime process...')
            self.ctrl.process.terminate()

        self._dispose_subprocess()

        if self.heartbeat_task is not None:
            self.heartbeat_task.cancel()
            self.heartbeat_task = None

    @staticmethod
    def _parse_plugin_id(plugin_id: str) -> tuple[str, str]:
        """Parse a plugin ID string into (author, name).

        Args:
            plugin_id: Plugin ID in 'author/name' format.

        Returns:
            Tuple of (plugin_author, plugin_name).

        Raises:
            ValueError: If plugin_id is not in the expected 'author/name' format.
        """
        segments = plugin_id.split('/')
        if len(segments) != 2 or not all(segments):
            raise ValueError(
                f"Invalid plugin_id format: '{plugin_id}'. Expected 'author/name' format (e.g. 'langbot/rag-engine')."
            )
        return segments[0], segments[1]

    async def call_rag_ingest(self, plugin_id: str, context_data: dict[str, Any]) -> dict[str, Any]:
        """Call plugin to ingest document.

        Args:
            plugin_id: Target plugin ID (author/name).
            context_data: IngestionContext data.
        """
        plugin_author, plugin_name = self._parse_plugin_id(plugin_id)
        return await self.handler.rag_ingest_document(plugin_author, plugin_name, context_data)

    async def call_rag_delete_document(self, plugin_id: str, document_id: str, kb_id: str) -> bool:
        plugin_author, plugin_name = self._parse_plugin_id(plugin_id)
        return await self.handler.rag_delete_document(plugin_author, plugin_name, document_id, kb_id)

    async def get_rag_creation_schema(self, plugin_id: str) -> dict[str, Any]:
        plugin_author, plugin_name = self._parse_plugin_id(plugin_id)
        return await self.handler.get_rag_creation_schema(plugin_author, plugin_name)

    async def get_rag_retrieval_schema(self, plugin_id: str) -> dict[str, Any]:
        plugin_author, plugin_name = self._parse_plugin_id(plugin_id)
        return await self.handler.get_rag_retrieval_schema(plugin_author, plugin_name)

    async def rag_on_kb_create(self, plugin_id: str, kb_id: str, config: dict[str, Any]) -> dict[str, Any]:
        """Notify plugin about KB creation."""
        plugin_author, plugin_name = self._parse_plugin_id(plugin_id)
        return await self.handler.rag_on_kb_create(plugin_author, plugin_name, kb_id, config)

    async def rag_on_kb_delete(self, plugin_id: str, kb_id: str) -> dict[str, Any]:
        """Notify plugin about KB deletion."""
        plugin_author, plugin_name = self._parse_plugin_id(plugin_id)
        return await self.handler.rag_on_kb_delete(plugin_author, plugin_name, kb_id)

    async def call_rag_retrieve(self, plugin_id: str, retrieval_context: dict[str, Any]) -> dict[str, Any]:
        """Call plugin to retrieve knowledge.

        Args:
            plugin_id: Target plugin ID (author/name).
            retrieval_context: RetrievalContext data.
        """
        plugin_author, plugin_name = self._parse_plugin_id(plugin_id)
        return await self.handler.retrieve_knowledge(plugin_author, plugin_name, '', retrieval_context)

    async def list_knowledge_engines(self) -> list[dict[str, Any]]:
        """List all available Knowledge Engines from plugins.

        Returns a list of Knowledge Engines with their capabilities and configuration schemas.
        """
        if not self.is_enable_plugin:
            return []

        return await self.handler.list_knowledge_engines()

    async def list_parsers(self) -> list[dict[str, Any]]:
        """List all available parsers from plugins."""
        if not self.is_enable_plugin:
            return []
        return await self.handler.list_parsers()

    async def call_parser(self, plugin_id: str, context_data: dict[str, Any], file_bytes: bytes) -> dict[str, Any]:
        """Call plugin to parse a document."""
        plugin_author, plugin_name = self._parse_plugin_id(plugin_id)
        return await self.handler.parse_document(plugin_author, plugin_name, context_data, file_bytes)
