"""
Plugin system: discovery, loading, context API, and tool wrapping.

General plugins live in ~/.mugi/plugins/<name>/; dashboard-only plugins live in
~/.mugi/dashboard_plugins/dash_<name>/. Each requires plugin.toml and plugin.py register(ctx).

PluginManager is held by ChatAppDelegate and drives the full lifecycle:
discover -> approve -> load -> provide tools/renderers/routes/hooks to the app.
"""
from __future__ import annotations

import asyncio
import importlib.util
import inspect
import json
import os
import re
import sys
import threading
import time
import uuid
from collections import deque
from dataclasses import dataclass, field
from typing import Any

from starlette.requests import Request

_dash_plugin_get_hits: dict[str, deque[float]] = {}
_dash_plugin_get_lock = threading.Lock()
_DASH_PLUGIN_GET_LIMIT = 180
_DASH_PLUGIN_GET_WINDOW_S = 60.0


def _dash_plugin_get_rate_limited(slug: str, subject: str) -> bool:
    """Return True when per-token GET budget is exceeded (dashboard tile fetches)."""
    key = f"{subject}:{slug}"
    now = time.monotonic()
    with _dash_plugin_get_lock:
        dq = _dash_plugin_get_hits.setdefault(key, deque())
        while dq and now - dq[0] > _DASH_PLUGIN_GET_WINDOW_S:
            dq.popleft()
        if len(dq) >= _DASH_PLUGIN_GET_LIMIT:
            return True
        dq.append(now)
        return False

from plugin_security import (
    PLUGIN_KIND_DASHBOARD,
    PLUGIN_KIND_GENERAL,
    PluginManifest,
    hash_plugin_directory,
    get_plugin_approval_state,
    set_plugin_approval,
    disable_plugin_approval,
    remove_plugin_approval,
    get_all_plugin_states,
)
from tool_policy import (
    register_plugin_tools_for_slug,
    unregister_plugin_tool_capabilities,
)
from tools._logging import _debug_log, emit_live_view_event as _emit_lv
from utils.path_safety import safe_resolve_path, is_path_allowed
from plugin_runtime_host import PluginRuntimeHost, RuntimeState
from plugin_fs_broker import PluginFsBroker

# Global lock for the builtins.__import__ patch during plugin loading.
# builtins.__import__ is process-global; concurrent plugin loads would race without this.
_plugin_load_lock = threading.Lock()

# Default plugins directories — import from persistence for single source of truth
try:
    from persistence import MUGI_PLUGINS_DIR as PLUGINS_DIR, MUGI_DASHBOARD_PLUGINS_DIR
except ImportError:
    PLUGINS_DIR = os.path.expanduser("~/.mugi/plugins")
    MUGI_DASHBOARD_PLUGINS_DIR = os.path.expanduser("~/.mugi/dashboard_plugins")

# Dashboard plugin folder names must be dash_* and match slug from [plugin].name.
_DASH_SLUG_RE = re.compile(r"^dash_[a-z0-9_]+$")

# ``PluginRuntimeHost.invoke_tool`` defaults to 60s. Neko CDP tools can exceed
# that when chaining direct page WS, in-container runner, browser attach, and a
# stale-target retry (each leg may use multi-second or ~45s CDP deadlines).
_PLUGIN_TOOL_INVOKE_TIMEOUT_OVERRIDES: dict[str, float] = {
    "neko_page_visible_text": 150.0,
    "neko_wait_for": 120.0,
}

# Identifier used to derive safe slugs for plugin names
_SLUG_RE = re.compile(r"[^a-z0-9_]")


def _plugin_slug(name: str) -> str:
    """Lowercase alphanumeric + underscore slug from a plugin name."""
    return _SLUG_RE.sub("_", name.lower().strip()).strip("_") or "unnamed"


def _coerce_number(value) -> float | None:
    from tools.job_stream_schema import coerce_number
    return coerce_number(value)


def _normalize_stream_controls(controls: list[dict] | None) -> list[dict]:
    """Validate/normalize control definitions for dashboard reverse channel."""
    if not isinstance(controls, list):
        return []
    normalized: list[dict] = []
    seen_ids: set[str] = set()
    allowed_types = {"slider", "toggle", "select", "button"}
    for raw in controls:
        if not isinstance(raw, dict):
            continue
        cid = str(raw.get("id") or "").strip()
        ctype = str(raw.get("type") or "").strip().lower()
        if not cid or cid in seen_ids or ctype not in allowed_types:
            continue
        label = str(raw.get("label") or cid).strip() or cid
        item: dict = {
            "id": cid,
            "type": ctype,
            "label": label,
        }
        if "description" in raw:
            item["description"] = str(raw.get("description") or "")
        if ctype == "slider":
            min_v = _coerce_number(raw.get("min"))
            max_v = _coerce_number(raw.get("max"))
            if min_v is None or max_v is None or max_v <= min_v:
                continue
            step_v = _coerce_number(raw.get("step"))
            if step_v is None or step_v <= 0:
                step_v = 1.0
            default_v = _coerce_number(raw.get("default"))
            if default_v is None:
                default_v = min_v
            default_v = min(max(default_v, min_v), max_v)
            item.update(
                {
                    "min": float(min_v),
                    "max": float(max_v),
                    "step": float(step_v),
                    "default": float(default_v),
                }
            )
        elif ctype == "toggle":
            item["default"] = bool(raw.get("default", False))
        elif ctype == "select":
            options_raw = raw.get("options")
            options: list[str] = []
            if isinstance(options_raw, list):
                for opt in options_raw:
                    text = str(opt or "").strip()
                    if text:
                        options.append(text)
            if not options:
                continue
            default_opt = str(raw.get("default") or options[0]).strip()
            if default_opt not in options:
                default_opt = options[0]
            item["options"] = options
            item["default"] = default_opt
        elif ctype == "button":
            action_id = str(raw.get("action_id") or cid).strip() or cid
            item["action_id"] = action_id
            if "style" in raw:
                style = str(raw.get("style") or "").strip().lower()
                if style in {"default", "destructive", "primary"}:
                    item["style"] = style
            if "confirm" in raw:
                item["confirm"] = str(raw.get("confirm") or "")
        seen_ids.add(cid)
        normalized.append(item)
    return normalized


# =============================================================================
# Tool wrapping -- error isolation, attribution, future permission hooks
# =============================================================================

def _wrap_plugin_tool(
    plugin_name: str,
    original_tool,
    *,
    realtime: bool = False,
    permissions: dict | None = None,
    visible_in_tier0: bool = False,
    runtime: PluginRuntimeHost | None = None,
):
    """Wrap a plugin tool with error isolation and attributed logging.

    Returns a new callable that quacks like a LangChain StructuredTool:
    it has .name, .description, .args_schema, .metadata, and supports .invoke().
    If realtime is True (from manifest or original tool metadata), metadata
    includes {"realtime": True} so the agent loop auto-dispatches to background.
    """
    tool_name = getattr(original_tool, "name", None) or "unknown"
    description = getattr(original_tool, "description", None) or ""
    args_schema = getattr(original_tool, "args_schema", None)
    orig_meta = getattr(original_tool, "metadata", None) or {}
    merged_meta = dict(orig_meta) if isinstance(orig_meta, dict) else {}
    if realtime or merged_meta.get("realtime"):
        merged_meta["realtime"] = True
    if visible_in_tier0:
        merged_meta["visible_in_tier0"] = True

    perms = permissions or {}
    _network_tool_names = {
        "search_web",
        "deep_web_search",
        "read_webpage",
        "read_and_analyze_page",
        "download_file",
        "screenshot_webpage",
    }
    _shell_tool_names = {"shell_execute"}

    class _WrappedPluginTool:
        """Thin wrapper providing error isolation around a plugin tool."""

        def __init__(self):
            self.name = tool_name
            self.description = description
            self.args_schema = args_schema
            self.metadata = merged_meta

        def invoke(self, input_data, config=None, **kwargs):
            _debug_log(f"[PLUGIN:{plugin_name}] tool {tool_name} called", "TOOL")
            try:
                if tool_name in _network_tool_names and not bool(perms.get("network", False)):
                    return (
                        f"Plugin tool error ({plugin_name}/{tool_name}): "
                        "network permission is required by this plugin."
                    )
                if tool_name in _shell_tool_names and not bool(perms.get("shell", False)):
                    return (
                        f"Plugin tool error ({plugin_name}/{tool_name}): "
                        "shell permission is required by this plugin."
                    )
                if runtime is not None:
                    _invoke_timeout = _PLUGIN_TOOL_INVOKE_TIMEOUT_OVERRIDES.get(
                        tool_name, 60.0
                    )
                    return runtime.invoke_tool(
                        tool_name=tool_name,
                        args=input_data if isinstance(input_data, dict) else {},
                        capability_context={},
                        timeout_s=_invoke_timeout,
                    )
                if hasattr(original_tool, "invoke"):
                    return original_tool.invoke(input_data, config=config, **kwargs)
                return original_tool(**input_data)
            except Exception as e:
                _debug_log(f"[PLUGIN:{plugin_name}] tool {tool_name} failed: {e}", "ERROR")
                return f"Plugin tool error ({plugin_name}/{tool_name}): {e}"

        def __call__(self, *args, **kwargs):
            return self.invoke(kwargs)

    return _WrappedPluginTool()


# =============================================================================
# PluginContext -- the restricted API object given to each plugin's register()
# =============================================================================

class PluginContext:
    """Interface a plugin uses to interact with the app.

    Grows across phases:
      Phase 1: register_tools, log, emit_live_view_event, path helpers
      Phase 2: register_live_view_renderer
      Phase 3: register_routes, register_settings, get_setting, set_setting
      Phase 4: register_hook
    """

    def __init__(self, plugin_name: str, manifest: PluginManifest):
        self._name = plugin_name
        self._slug = _plugin_slug(plugin_name)
        self._manifest = manifest
        self._tools: list = []
        self._live_view_renderers: dict[str, dict] = {}
        self._route_setups: list = []
        self._settings_schema: dict | None = None
        self._hooks: dict[str, list] = {}
        self._stream_emitter = None
        self._stream_controls_by_id: dict[str, list[dict]] = {}
        self._control_handlers: list = []
        self._tool_registration_warnings: list[str] = []

    # -- Phase 1: Tools -------------------------------------------------------

    def register_tools(self, tools: list) -> None:
        """Register LangChain @tool decorated functions."""
        for t in tools:
            if not hasattr(t, "args_schema"):
                _msg = f"Skipping non-tool object (missing args_schema): {t!r}"
                self._tool_registration_warnings.append(_msg)
                _debug_log(f"[PLUGIN:{self._name}] {_msg}", "WARN")
                continue
            self._tools.append(t)

    # -- Phase 1: Logging & events --------------------------------------------

    def log(self, message: str, level: str = "INFO") -> None:
        """Write to the terminal log, attributed to this plugin."""
        _debug_log(f"[PLUGIN:{self._name}] {message}", level)

    def emit_live_view_event(self, event: dict) -> None:
        """Emit a Live View event. Source attribution is forced."""
        event = dict(event)
        event["source"] = f"plugin:{self._name}"
        _emit_lv(event)

    def request_user_decision(
        self,
        kind: str,
        *,
        context: dict | None = None,
        timeout_s: float = 300.0,
        thread_id: str | None = None,
    ) -> str | None:
        """Request a user decision via core IPC and block until response/timeout."""
        kind_norm = (kind or "").strip()
        allowed = set(self._manifest.user_decision_allowed_kinds or set())
        if not kind_norm or kind_norm not in allowed:
            _debug_log(
                f"[PLUGIN:{self._name}] request_user_decision denied for kind={kind_norm!r}",
                "WARN",
            )
            return None
        try:
            from user_decision_ipc import request_user_decision as _request_user_decision
            return _request_user_decision(
                kind=kind_norm,
                context=context if isinstance(context, dict) else {},
                source=f"plugin:{self._slug}",
                thread_id=thread_id,
                timeout_s=timeout_s,
            )
        except Exception as e:
            _debug_log(f"[PLUGIN:{self._name}] request_user_decision failed: {e}", "ERROR")
            return None

    def emit_display_window(self, filename: str, title: str = "Display") -> None:
        """Open the display window with a preview file. File must be in ~/Documents/Mugi.

        Supports images, video, HTML (.html, .htm), and plain text (.txt).
        Use this when your plugin generates results that should be shown prominently.
        """
        import os
        from tools._logging import emit_chat_event
        from utils.path_safety import get_user_output_dir

        output_dir = get_user_output_dir()
        base = os.path.basename((filename or "").strip())
        if not base:
            return
        path = os.path.join(output_dir, base)
        if not os.path.isfile(path):
            _debug_log(f"[PLUGIN:{self._name}] emit_display_window: file not found {base}", "WARN")
            return
        ext = os.path.splitext(base)[1].lower()
        image_exts = (".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp")
        video_exts = (".mp4", ".webm", ".mov", ".avi", ".m4v")
        if ext in video_exts:
            content_type = "video"
        elif ext in image_exts:
            content_type = "image"
        elif ext in (".html", ".htm"):
            content_type = "html"
        elif ext == ".txt":
            content_type = "text"
        else:
            _debug_log(f"[PLUGIN:{self._name}] emit_display_window: unsupported format {ext}", "WARN")
            return
        emit_chat_event({"type": "display_window", "files": [base], "title": title or "Display", "content_type": content_type})

    # -- Phase 1.5: Real-time stream events -----------------------------------

    def set_stream_emitter(self, emitter) -> None:
        """Set stream event emitter callback. Internal use by PluginManager."""
        self._stream_emitter = emitter

    def _emit_stream_event(self, event: dict) -> None:
        """Emit a typed real-time stream event if a stream emitter is wired."""
        if not callable(self._stream_emitter):
            return
        evt = dict(event or {})
        evt.setdefault("schema", "mugi.plugin_stream.v1")
        evt.setdefault("schema_version", 1)
        evt.setdefault("plugin_id", self._slug)
        evt.setdefault("plugin_name", self._name)
        evt.setdefault("timestamp_ms", int(time.time() * 1000))
        try:
            self._stream_emitter(evt)
        except Exception as e:
            _debug_log(f"[PLUGIN:{self._name}] stream emit failed: {e}", "WARN")

    def open_stream(
        self,
        stream_id: str | None = None,
        *,
        stream_type: str = "timeseries",
        title: str | None = None,
        metadata: dict | None = None,
        thread_id: str | None = None,
    ) -> str:
        """Open a real-time stream and return the stream_id."""
        sid = (stream_id or "").strip() or f"{self._slug}_{uuid.uuid4().hex[:12]}"
        self._emit_stream_event(
            {
                "event": "open",
                "stream_id": sid,
                "stream_type": stream_type,
                "title": title or sid,
                "metadata": metadata or {},
                "thread_id": thread_id,
            }
        )
        return sid

    def push_stream_update(
        self,
        stream_id: str,
        payload: dict,
        *,
        thread_id: str | None = None,
    ) -> None:
        """Push a real-time update payload for an open stream."""
        sid = (stream_id or "").strip()
        if not sid:
            return
        self._emit_stream_event(
            {
                "event": "update",
                "stream_id": sid,
                "data": payload or {},
                "thread_id": thread_id,
            }
        )

    def close_stream(
        self,
        stream_id: str,
        *,
        status: str = "completed",
        payload: dict | None = None,
        thread_id: str | None = None,
    ) -> None:
        """Close a real-time stream."""
        sid = (stream_id or "").strip()
        if not sid:
            return
        self._emit_stream_event(
            {
                "event": "close",
                "stream_id": sid,
                "status": status,
                "data": payload or {},
                "thread_id": thread_id,
            }
        )
        self._stream_controls_by_id.pop(sid, None)

    def emit_stream_error(
        self,
        stream_id: str,
        message: str,
        *,
        payload: dict | None = None,
        thread_id: str | None = None,
    ) -> None:
        """Emit an error event for a stream."""
        sid = (stream_id or "").strip()
        if not sid:
            return
        self._emit_stream_event(
            {
                "event": "error",
                "stream_id": sid,
                "message": message,
                "data": payload or {},
                "thread_id": thread_id,
            }
        )

    def register_stream_controls(
        self,
        stream_id: str,
        controls: list[dict],
        *,
        thread_id: str | None = None,
    ) -> list[dict]:
        """Register controls for a stream and emit a control-schema event."""
        sid = (stream_id or "").strip()
        if not sid:
            return []
        normalized = _normalize_stream_controls(controls)
        self._stream_controls_by_id[sid] = normalized
        self._emit_stream_event(
            {
                "event": "controls",
                "stream_id": sid,
                "data": {"controls": normalized},
                "thread_id": thread_id,
            }
        )
        return normalized

    def register_control_handler(self, callback) -> None:
        """Register a callback to handle dashboard control events."""
        if not callable(callback):
            return
        self._control_handlers.append(callback)

    def get_stream_controls(self, stream_id: str) -> list[dict]:
        """Return normalized controls for a stream."""
        sid = (stream_id or "").strip()
        return list(self._stream_controls_by_id.get(sid, []))

    # -- Phase 1: Path safety helpers -----------------------------------------

    def resolve_safe_path(self, path: str, allow_write: bool = False) -> tuple[str | None, str]:
        """Resolve a path with the same safety checks as built-in tools."""
        from core import get_workspace_path
        workspace = get_workspace_path()
        return safe_resolve_path(workspace, path, allow_write=allow_write)

    def is_path_allowed(self, path: str) -> bool:
        return is_path_allowed(path)

    # -- Phase 2: Live View renderers -----------------------------------------

    def register_live_view_renderer(self, event_type: str, renderer: dict) -> None:
        """Register a JS renderer for a custom Live View event type.

        event_type must be prefixed with 'plugin_<slug>_'.
        renderer should contain a 'js' key with a JS function body:
            function(event) { return '<div>...</div>'; }
        """
        prefix = f"plugin_{self._slug}_"
        if not event_type.startswith(prefix):
            raise ValueError(
                f"Event type must start with '{prefix}', got '{event_type}'"
            )
        self._live_view_renderers[event_type] = renderer

    # -- Phase 3: HTTP routes -------------------------------------------------

    def register_routes(self, setup_fn) -> None:
        """Register a route setup function: setup_fn(fastapi_sub_app)."""
        self._route_setups.append(setup_fn)

    # -- Phase 3: Settings ----------------------------------------------------

    def register_settings(self, schema: dict) -> None:
        """Declare plugin settings (shown in Preferences > Plugins).

        schema example:
            {"name": "My Plugin", "fields": [
                {"key": "timeout", "type": "number", "default": 5, "label": "Timeout (s)"},
                {"key": "verbose", "type": "bool", "default": False, "label": "Verbose logging"},
                {"key": "api_token", "type": "secret", "default": "", "label": "API token"},
            ]}

        Field types: ``string``, ``secret`` (masked in UI; stored like string), ``bool`` or ``boolean``, ``number``.
        """
        self._settings_schema = schema

    def get_setting(self, key: str, default=None):
        """Read a plugin setting from app_prefs.json."""
        from plugin_security import get_plugin_settings_values

        settings = get_plugin_settings_values(self._slug, schema=self._settings_schema)
        return settings.get(key, default)

    def set_setting(self, key: str, value) -> None:
        """Write a plugin setting to app_prefs.json."""
        from plugin_security import save_plugin_settings_values

        save_plugin_settings_values(
            self._slug,
            {key: value},
            schema=self._settings_schema,
        )

    # -- Phase 4: Lifecycle hooks ---------------------------------------------

    def register_hook(self, event_name: str, callback) -> None:
        """Register a callback for a lifecycle event.

        Supported event_name values:
            on_agent_start, on_tool_call, on_tool_result, on_tool_call_blocked, on_agent_done
        """
        valid = (
            "on_agent_start",
            "on_tool_call",
            "on_tool_result",
            "on_tool_call_blocked",
            "on_agent_done",
        )
        if event_name not in valid:
            _debug_log(
                f"[PLUGIN:{self._name}] unknown hook '{event_name}'; valid: {valid}",
                "WARN",
            )
            return
        self._hooks.setdefault(event_name, []).append(callback)


# =============================================================================
# LoadedPlugin -- holds a loaded plugin's context and metadata
# =============================================================================

@dataclass
class LoadedPlugin:
    name: str
    slug: str
    manifest: PluginManifest
    context: Any
    directory: str
    current_hash: str
    error: str | None = None
    warnings: list[str] = field(default_factory=list)
    runtime: PluginRuntimeHost | None = None


@dataclass
class RuntimeToolSnapshot:
    name: str
    description: str = ""
    args_schema: dict | None = None
    metadata: dict = field(default_factory=dict)


class RuntimeContextSnapshot:
    """Core-side readonly snapshot returned by runtime handshake."""

    def __init__(self, *, settings_schema: dict | None = None):
        self._tools: list[RuntimeToolSnapshot] = []
        self._live_view_renderers: dict[str, dict] = {}
        self._route_setups: list = []
        self._settings_schema: dict | None = settings_schema
        self._hooks: dict[str, list] = {}
        self._control_handlers: list = []
        self._stream_controls_by_id: dict[str, list[dict]] = {}
        self._dashboard_tiles: list[dict] = []
        #: True when sandbox bootstrap reported ``route_mode == "handler"`` (``register_route_handler`` used).
        self._route_handler_registered: bool = False


# =============================================================================
# PluginManager -- discovery, approval, loading, and aggregation
# =============================================================================

class PluginManager:
    """Singleton-ish manager held by ChatAppDelegate."""

    def __init__(self, plugins_dir: str | None = None):
        self._plugins_dir = plugins_dir or PLUGINS_DIR
        self._dashboard_plugins_dir = MUGI_DASHBOARD_PLUGINS_DIR
        self._lock = threading.Lock()
        # Lock-order rule: always acquire _op_lock before _lock for lifecycle ops.
        self._op_lock = threading.RLock()
        self._loaded: dict[str, LoadedPlugin] = {}
        self._pending_approvals: list[tuple[PluginManifest, str]] = []
        self._discovered_manifests: dict[str, tuple[PluginManifest, str]] = {}
        self._stream_emitter = None
        self._sandbox_enabled = True
        self._sandbox_kill_switch = False
        self._quarantined: dict[str, str] = {}
        self._fs_broker = PluginFsBroker()
        self._load_sandbox_state()
        # Set when discover() finishes (including empty/no-op runs) so the agent/UI never
        # observe a partially discovered manifest set while another thread holds _op_lock.
        self._initial_discover_done = threading.Event()

    def _sandbox_state_path(self) -> str:
        try:
            from utils.app_prefs_store import app_prefs_path
            return app_prefs_path()
        except Exception:
            return os.path.expanduser("~/.mugi/app_prefs.json")

    def _load_sandbox_state(self) -> None:
        try:
            from utils.app_prefs_store import load_app_prefs_cached
            data = load_app_prefs_cached()
            sandbox = data.get("plugin_sandbox") if isinstance(data, dict) else {}
            if isinstance(sandbox, dict):
                self._sandbox_kill_switch = bool(sandbox.get("kill_switch", False))
                q = sandbox.get("quarantined")
                if isinstance(q, dict):
                    self._quarantined = {str(k): str(v) for k, v in q.items() if str(k).strip()}
        except Exception:
            pass

    def _save_sandbox_state(self) -> None:
        try:
            from utils.app_prefs_store import load_app_prefs_cached, save_app_prefs
            data = load_app_prefs_cached()
            data["plugin_sandbox"] = {
                "kill_switch": bool(self._sandbox_kill_switch),
                "quarantined": dict(self._quarantined),
            }
            save_app_prefs(data)
        except Exception:
            pass

    def set_stream_emitter(self, emitter) -> None:
        """Set stream emitter callback for all loaded and future plugin contexts."""
        self._stream_emitter = emitter
        with self._lock:
            for lp in self._loaded.values():
                if hasattr(lp.context, "set_stream_emitter"):
                    lp.context.set_stream_emitter(emitter)

    def _forward_runtime_stream_event(self, event: dict) -> None:
        """Forward stream events from sandbox runtime processes to the manager's emitter."""
        if not callable(self._stream_emitter):
            return
        evt = dict(event or {})
        # Runtime subprocess historically emitted plugin_slug only; clients expect plugin_id.
        evt.setdefault("plugin_id", str(evt.get("plugin_slug") or "").strip())
        evt.setdefault("schema", "mugi.plugin_stream.v1")
        evt.setdefault("schema_version", 1)
        try:
            self._stream_emitter(evt)
        except Exception as e:
            _debug_log(f"[PLUGIN] runtime stream forward failed: {e}", "WARN")

    @property
    def plugins_dir(self) -> str:
        return self._plugins_dir

    # -- Discovery ------------------------------------------------------------

    def _try_discover_plugin(
        self,
        plugin_dir: str,
        entry: str,
        *,
        dashboard_root: bool,
    ) -> tuple[PluginManifest, str, str, str] | None:
        """Parse and register one plugin directory. Returns (manifest, dir, hash, slug) or None."""
        toml_path = os.path.join(plugin_dir, "plugin.toml")
        if not os.path.isdir(plugin_dir) or not os.path.isfile(toml_path):
            return None
        try:
            manifest = PluginManifest.from_toml(toml_path)
        except Exception as e:
            _debug_log(f"[PLUGIN] Failed to parse {toml_path}: {e}", "WARN")
            return None
        slug = _plugin_slug(manifest.name)
        if dashboard_root:
            if not _DASH_SLUG_RE.match(entry):
                _debug_log(
                    f"[PLUGIN] Dashboard plugin folder {entry!r} must match ^dash_[a-z0-9_]+$; "
                    f"skipping {plugin_dir}",
                    "WARN",
                )
                return None
            if manifest.kind != PLUGIN_KIND_DASHBOARD:
                _debug_log(
                    f"[PLUGIN] {plugin_dir}: [plugin].kind must be 'dashboard' under dashboard_plugins; skipping",
                    "WARN",
                )
                return None
            if slug != entry:
                _debug_log(
                    f"[PLUGIN] Dashboard folder {entry!r} must match slug from [plugin].name ({slug!r}); "
                    f"skipping {plugin_dir}",
                    "WARN",
                )
                return None
        else:
            if manifest.kind == PLUGIN_KIND_DASHBOARD:
                _debug_log(
                    f"[PLUGIN] {plugin_dir}: kind=dashboard belongs in ~/.mugi/dashboard_plugins/; skipping",
                    "WARN",
                )
                return None
            if entry.startswith("dash_"):
                _debug_log(
                    f"[PLUGIN] Folder {entry!r} uses dash_ prefix under general plugins — "
                    f"use ~/.mugi/dashboard_plugins/ or rename; skipping",
                    "WARN",
                )
                return None
        if slug in self._discovered_manifests:
            _debug_log(
                f"[PLUGIN] Duplicate plugin slug '{slug}' from {plugin_dir}; keeping first discovered plugin.",
                "WARN",
            )
            return None
        try:
            current_hash = hash_plugin_directory(plugin_dir)
        except Exception as e:
            _debug_log(f"[PLUGIN] Failed to hash {plugin_dir}: {e}", "WARN")
            return None
        self._discovered_manifests[slug] = (manifest, current_hash)
        return (manifest, plugin_dir, current_hash, slug)

    def discover(self) -> list[tuple[PluginManifest, str, str]]:
        """Scan general and dashboard plugin directories.

        Returns list of (manifest, directory, hash).
        Populates _pending_approvals for plugins that are new or whose hash changed.
        Auto-loads plugins that are already approved and enabled.
        Plugin runtime starts are parallelized so handshake timeouts don't stack.
        """
        with self._op_lock:
            results: list[tuple[PluginManifest, str, str]] = []
            try:
                try:
                    from persistence import ensure_bundled_dashboard_plugins, ensure_bundled_plugins
                    ensure_bundled_plugins()
                    ensure_bundled_dashboard_plugins()
                except Exception:
                    pass
                os.makedirs(self._plugins_dir, exist_ok=True)
                os.makedirs(self._dashboard_plugins_dir, exist_ok=True)
                self._pending_approvals.clear()
                self._discovered_manifests.clear()

                to_load: list[tuple[str, PluginManifest, str]] = []

                for root, dashboard_root in (
                    (self._plugins_dir, False),
                    (self._dashboard_plugins_dir, True),
                ):
                    try:
                        entries = sorted(os.listdir(root))
                    except OSError:
                        entries = []
                    for entry in entries:
                        plugin_dir = os.path.join(root, entry)
                        got = self._try_discover_plugin(
                            plugin_dir, entry, dashboard_root=dashboard_root
                        )
                        if not got:
                            continue
                        manifest, pdir, current_hash, slug = got
                        results.append((manifest, pdir, current_hash))

                        state = get_plugin_approval_state(slug)
                        if not state:
                            self._pending_approvals.append((manifest, current_hash))
                        elif state.get("approved_hash") != current_hash:
                            self._pending_approvals.append((manifest, current_hash))
                        elif state.get("enabled", False):
                            to_load.append((pdir, manifest, current_hash))

                if to_load:
                    from concurrent.futures import ThreadPoolExecutor, as_completed
                    with ThreadPoolExecutor(
                        max_workers=min(len(to_load), 8),
                        thread_name_prefix="plugin-load",
                    ) as pool:
                        futures = {
                            pool.submit(self._load_plugin, pdir, mfest, chash): mfest.name
                            for pdir, mfest, chash in to_load
                        }
                        for fut in as_completed(futures):
                            try:
                                fut.result()
                            except Exception as e:
                                _debug_log(
                                    f"[PLUGIN] Parallel load of '{futures[fut]}' raised: {e}",
                                    "ERROR",
                                )

                return results
            finally:
                self._initial_discover_done.set()

    def wait_initial_discover(self, timeout_s: float = 120.0) -> bool:
        """Return True once discover() has completed at least once (may be empty)."""
        return self._initial_discover_done.wait(timeout_s)

    def mark_discover_skipped(self) -> None:
        """Unblock wait_initial_discover when discover() could not finish (startup error path)."""
        self._initial_discover_done.set()

    # -- Approval management --------------------------------------------------

    def get_pending_approvals(self) -> list[tuple[PluginManifest, str]]:
        """Return plugins waiting for user approval: [(manifest, hash), ...]."""
        return list(self._pending_approvals)

    def approve_plugin(self, plugin_name: str) -> bool:
        """Approve and load a pending plugin. Returns True on success."""
        with self._op_lock:
            return self._approve_plugin_locked(plugin_name)

    def _resync_plugin_hash_unload(self, slug: str) -> tuple[PluginManifest, str] | None:
        """Re-hash plugin dir from disk, update discovery cache, stop runtime and drop stale load.

        Needed after files change on disk while the app is running: otherwise ``_load_plugin``
        compares a stale *current_hash* to *live_hash* and refuses to start (user cannot Enable).
        """
        entry = self._discovered_manifests.get(slug)
        if not entry:
            return None
        manifest, _ = entry
        plugin_dir = manifest.path
        if not plugin_dir or not os.path.isdir(plugin_dir):
            return None
        try:
            fresh_hash = hash_plugin_directory(plugin_dir)
        except OSError:
            return None
        self._discovered_manifests[slug] = (manifest, fresh_hash)
        with self._lock:
            lp = self._loaded.pop(slug, None)
        if lp and lp.runtime:
            lp.runtime.stop()
        unregister_plugin_tool_capabilities(slug)
        return manifest, fresh_hash

    def _approve_plugin_locked(self, plugin_name: str) -> bool:
        """Call only while holding _op_lock."""
        slug = _plugin_slug(plugin_name)
        synced = self._resync_plugin_hash_unload(slug)
        if not synced:
            return False
        manifest, fresh_hash = synced
        set_plugin_approval(slug, enabled=True, approved_hash=fresh_hash)
        self._load_plugin(manifest.path, manifest, fresh_hash)
        self._pending_approvals = [
            (m, h) for m, h in self._pending_approvals
            if _plugin_slug(m.name) != slug
        ]
        return True

    def disable_plugin(self, plugin_name: str) -> None:
        """Disable a plugin (keeps approval record)."""
        with self._op_lock:
            self._disable_plugin_locked(plugin_name)

    def _disable_plugin_locked(self, plugin_name: str) -> None:
        """Call only while holding _op_lock."""
        slug = _plugin_slug(plugin_name)
        disable_plugin_approval(slug)
        unregister_plugin_tool_capabilities(slug)
        with self._lock:
            lp = self._loaded.pop(slug, None)
        if lp and lp.runtime:
            lp.runtime.stop()
        self._pending_approvals = [
            (m, h) for m, h in self._pending_approvals
            if _plugin_slug(m.name) != slug
        ]

    def enable_plugin(self, plugin_name: str) -> bool:
        """Re-enable a previously disabled plugin. Returns True on success."""
        with self._op_lock:
            return self._enable_plugin_locked(plugin_name)

    def _enable_plugin_locked(self, plugin_name: str) -> bool:
        """Call only while holding _op_lock."""
        slug = _plugin_slug(plugin_name)
        synced = self._resync_plugin_hash_unload(slug)
        if not synced:
            return False
        manifest, fresh_hash = synced
        set_plugin_approval(slug, enabled=True, approved_hash=fresh_hash)
        self._load_plugin(manifest.path, manifest, fresh_hash)
        return True

    def remove_plugin(self, plugin_name: str) -> bool:
        """Disable and remove approval state. Does NOT delete files."""
        with self._op_lock:
            return self._remove_plugin_locked(plugin_name)

    def _remove_plugin_locked(self, plugin_name: str) -> bool:
        """Call only while holding _op_lock."""
        slug = _plugin_slug(plugin_name)
        unregister_plugin_tool_capabilities(slug)
        with self._lock:
            lp = self._loaded.pop(slug, None)
        if lp and lp.runtime:
            lp.runtime.stop()
        remove_plugin_approval(slug)
        self._discovered_manifests.pop(slug, None)
        return True

    # -- Loading --------------------------------------------------------------

    def _load_plugin(self, plugin_dir: str, manifest: PluginManifest, current_hash: str) -> None:
        """Start isolated plugin runtime, then bootstrap metadata via IPC.

        Thread-safe: may be called from parallel workers during discover().
        """
        slug = _plugin_slug(manifest.name)
        with self._lock:
            unregister_plugin_tool_capabilities(slug)
        if self._sandbox_kill_switch:
            lp = LoadedPlugin(
                name=manifest.name,
                slug=slug,
                manifest=manifest,
                context=RuntimeContextSnapshot(),
                directory=plugin_dir,
                current_hash=current_hash,
            )
            lp.error = "Sandbox kill-switch is enabled; plugin runtime start blocked."
            with self._lock:
                self._loaded[slug] = lp
            return
        if slug in self._quarantined:
            lp = LoadedPlugin(
                name=manifest.name,
                slug=slug,
                manifest=manifest,
                context=RuntimeContextSnapshot(),
                directory=plugin_dir,
                current_hash=current_hash,
            )
            lp.error = f"Plugin is quarantined: {self._quarantined.get(slug)}"
            with self._lock:
                self._loaded[slug] = lp
            return

        # Re-verify hash at load time (not just at discovery/approval time) to detect
        # modifications to plugin files after they were approved.
        live_hash = hash_plugin_directory(plugin_dir)
        if live_hash != current_hash:
            _debug_log(
                f"[PLUGIN] '{manifest.name}' hash mismatch — files changed since approval. "
                f"Re-approval required. Expected {current_hash!r}, got {live_hash!r}.",
                "ERROR",
            )
            lp = LoadedPlugin(
                name=manifest.name,
                slug=slug,
                manifest=manifest,
                context=RuntimeContextSnapshot(),
                directory=plugin_dir,
                current_hash=live_hash,
            )
            lp.error = "Hash mismatch: plugin files changed since last approval. Re-approve in Settings > Plugins."
            with self._lock:
                self._loaded[slug] = lp
            return
        ctx = RuntimeContextSnapshot()
        lp = LoadedPlugin(
            name=manifest.name,
            slug=slug,
            manifest=manifest,
            context=ctx,
            directory=plugin_dir,
            current_hash=current_hash,
        )
        try:
            def _settings_get(key: str, default=None):
                from plugin_security import get_plugin_settings_values
                values = get_plugin_settings_values(slug, schema=ctx._settings_schema)
                return values.get(key, default)

            def _settings_set(key: str, value) -> None:
                from plugin_security import save_plugin_settings_values
                save_plugin_settings_values(slug, {key: value}, schema=ctx._settings_schema)

            runtime = PluginRuntimeHost(
                plugin_slug=slug,
                plugin_name=manifest.name,
                plugin_dir=plugin_dir,
                plugin_hash=live_hash,
                permissions=manifest.permissions,
                settings_get=_settings_get,
                settings_set=_settings_set,
                on_stream_event=self._forward_runtime_stream_event,
            )
            bootstrap = runtime.start(timeout_s=12.0)
            tools_raw = bootstrap.get("tools") if isinstance(bootstrap.get("tools"), list) else []
            for tr in tools_raw:
                if not isinstance(tr, dict):
                    continue
                name = str(tr.get("name") or "").strip()
                if not name:
                    continue
                ctx._tools.append(
                    RuntimeToolSnapshot(
                        name=name,
                        description=str(tr.get("description") or ""),
                        args_schema=tr.get("args_schema") if isinstance(tr.get("args_schema"), dict) else {},
                        metadata=tr.get("metadata") if isinstance(tr.get("metadata"), dict) else {},
                    )
                )
            hooks_raw = bootstrap.get("hooks") if isinstance(bootstrap.get("hooks"), list) else []
            for hn in hooks_raw:
                event_name = str(hn or "").strip()
                if event_name:
                    ctx._hooks[event_name] = ["runtime"]
            if isinstance(bootstrap.get("live_view_renderers"), dict):
                ctx._live_view_renderers = dict(bootstrap.get("live_view_renderers"))
            if isinstance(bootstrap.get("settings_schema"), dict):
                ctx._settings_schema = dict(bootstrap.get("settings_schema"))
                try:
                    from plugin_security import get_plugin_settings_values
                    _settings_values = get_plugin_settings_values(slug, schema=ctx._settings_schema)
                    runtime.request("sync_settings", {"settings": _settings_values}, timeout_s=5.0)
                except Exception:
                    pass
            dt_raw = bootstrap.get("dashboard_tiles")
            if isinstance(dt_raw, list):
                ctx._dashboard_tiles = [x for x in dt_raw if isinstance(x, dict)]
            ctx._route_handler_registered = bootstrap.get("route_mode") == "handler"
            lp.warnings = [str(w) for w in (bootstrap.get("warnings") or []) if str(w).strip()]
            lp.runtime = runtime
            boot_names = {t.name for t in ctx._tools}
            for perm_tool in sorted((manifest.tool_permissions or {}).keys()):
                if perm_tool and perm_tool not in boot_names:
                    lp.warnings.append(
                        f"plugin.toml declares [tools.{perm_tool}] but the running plugin did not register "
                        f"that tool—update plugin.py to match plugin.toml, then Reload Plugins."
                    )
            with self._lock:
                register_plugin_tools_for_slug(
                    slug,
                    [t.name for t in ctx._tools],
                    manifest.permissions,
                    tool_permissions=manifest.tool_permissions,
                )
            _debug_log(
                f"[PLUGIN] Runtime-loaded '{manifest.name}' v{manifest.version} "
                f"({len(ctx._tools)} tools, {len(ctx._live_view_renderers)} renderers, "
                f"{1 if bootstrap.get('route_mode') == 'handler' else 0} route handlers, "
                f"{len(ctx._hooks)} hook groups)",
                "INFO",
            )
        except Exception as e:
            lp.error = str(e)
            _debug_log(f"[PLUGIN] {manifest.name} failed runtime load: {e}", "ERROR")
            if lp.runtime:
                if lp.runtime.state == RuntimeState.QUARANTINED:
                    with self._lock:
                        self._quarantined[slug] = lp.runtime.quarantined_reason or "runtime quarantined"
                    self._save_sandbox_state()
                lp.runtime.stop()
                lp.runtime = None
        with self._lock:
            self._loaded[slug] = lp

    def reload_all(self) -> list[tuple[PluginManifest, str]]:
        """Re-discover and reload all plugins. Returns new pending approvals."""
        with self._op_lock:
            with self._lock:
                old_loaded = list(self._loaded.values())
                for _slug in list(self._loaded.keys()):
                    unregister_plugin_tool_capabilities(_slug)
                self._loaded.clear()
            for lp in old_loaded:
                if lp.runtime:
                    lp.runtime.stop()
            self.discover()
            return self.get_pending_approvals()

    # -- Aggregation (used by the rest of the app) ----------------------------

    @property
    def loaded_plugins(self) -> list[LoadedPlugin]:
        with self._lock:
            return list(self._loaded.values())

    def get_all_discovered(self) -> list[tuple[str, PluginManifest, str, bool, str | None]]:
        """Return info for all discovered plugins for the Preferences UI.

        Returns: [(slug, manifest, hash, is_loaded, error_or_None), ...]

        Holds ``_op_lock`` so callers never see an empty ``_discovered_manifests`` or a
        manifest list whose enabled plugins are not yet present in ``_loaded`` while
        ``discover()`` is running on another thread.
        """
        with self._op_lock:
            with self._lock:
                manifests = dict(self._discovered_manifests)
                loaded_snap = {s: self._loaded.get(s) for s in manifests}
        result = []
        for slug, (manifest, current_hash) in manifests.items():
            lp = loaded_snap.get(slug)
            is_loaded = lp is not None and lp.error is None
            error = lp.error if lp else None
            result.append((slug, manifest, current_hash, is_loaded, error))
        return result

    def get_load_warnings(self, slug: str) -> list[str]:
        """Return plugin load warnings for *slug* (non-fatal), or [] if none."""
        with self._lock:
            lp = self._loaded.get(slug)
            if lp is None:
                return []
            return list(lp.warnings or [])

    def set_sandbox_kill_switch(self, enabled: bool) -> None:
        with self._op_lock:
            self._sandbox_kill_switch = bool(enabled)
            self._save_sandbox_state()
            if self._sandbox_kill_switch:
                with self._lock:
                    loaded = list(self._loaded.values())
                for lp in loaded:
                    if lp.runtime:
                        lp.runtime.stop()

    def quarantine_plugin(self, slug: str, reason: str) -> None:
        with self._op_lock:
            s = _plugin_slug(slug)
            self._quarantined[s] = str(reason or "quarantined")
            self._save_sandbox_state()
            with self._lock:
                lp = self._loaded.get(s)
            if lp and lp.runtime:
                lp.runtime.quarantine(self._quarantined[s])

    def clear_quarantine(self, slug: str) -> None:
        with self._op_lock:
            s = _plugin_slug(slug)
            self._quarantined.pop(s, None)
            self._save_sandbox_state()

    def rollback_sandbox_state(self) -> dict:
        """Deterministic rollback path for stuck sandbox lifecycle state."""
        with self._op_lock:
            self._sandbox_kill_switch = False
            self._quarantined = {}
            self._save_sandbox_state()
            return {"kill_switch": False, "quarantined": {}}

    def get_runtime_diagnostics(self, slug: str) -> dict:
        s = _plugin_slug(slug)
        with self._lock:
            lp = self._loaded.get(s)
        if lp is None:
            return {
                "state": RuntimeState.STOPPED,
                "runtime_error": "",
                "last_exit_code": None,
                "last_heartbeat": None,
                "quarantined_reason": self._quarantined.get(s, ""),
            }
        rt = lp.runtime
        state = rt.state if rt else RuntimeState.STOPPED
        if rt and state in {RuntimeState.HEALTHY, RuntimeState.DEGRADED}:
            try:
                rt.health_check(timeout_s=1.5)
            except Exception:
                state = RuntimeState.DEGRADED
        return {
            "state": state,
            "runtime_error": (rt.last_error if rt else "") or "",
            "failure_code": (rt.last_failure_code if rt else "") or "",
            "last_exit_code": rt.last_exit_code if rt else None,
            "last_heartbeat": rt.last_heartbeat_ms if rt else None,
            "quarantined_reason": self._quarantined.get(s, ""),
            "metrics": {
                "invoke_count": (rt.metrics.invoke_count if rt else 0),
                "timeout_count": (rt.metrics.timeout_count if rt else 0),
                "crash_count": (rt.metrics.crash_count if rt else 0),
                "consecutive_failures": (rt.consecutive_failures if rt else 0),
                "restart_backoff_s": (rt.current_backoff_s if rt else 0.0),
                "queue_depth": (rt.queue_depth if rt else 0),
                "invoke_latency_p50_ms": (rt.metrics.p50_ms if rt else 0.0),
                "invoke_latency_p95_ms": (rt.metrics.p95_ms if rt else 0.0),
            },
        }

    def restart_plugin_runtime(self, slug: str) -> dict:
        s = _plugin_slug(slug)
        with self._op_lock:
            with self._lock:
                lp = self._loaded.get(s)
            if lp is None:
                return {"ok": False, "error": "plugin not loaded"}
            if lp.runtime is None:
                return {"ok": False, "error": "runtime not initialized"}
            try:
                lp.runtime.start(timeout_s=10.0)
                return {"ok": True}
            except Exception as e:
                return {"ok": False, "error": str(e)}

    def capture_runtime_crash_dump(self, slug: str) -> dict:
        s = _plugin_slug(slug)
        with self._lock:
            lp = self._loaded.get(s)
        if lp is None or lp.runtime is None:
            return {"ok": False, "error": "runtime not initialized"}
        rt = lp.runtime
        return {
            "ok": True,
            "slug": s,
            "state": rt.state,
            "last_exit_code": rt.last_exit_code,
            "failure_code": rt.last_failure_code,
            "runtime_error": rt.last_error,
            "quarantined_reason": rt.quarantined_reason,
            "captured_at_ms": int(time.time() * 1000),
        }

    def sync_runtime_settings(self, slug: str, settings: dict) -> None:
        s = _plugin_slug(slug)
        with self._lock:
            lp = self._loaded.get(s)
        if lp is None or lp.runtime is None:
            return
        lp.runtime.request("sync_settings", {"settings": dict(settings or {})}, timeout_s=5.0)

    def issue_fs_handle(
        self,
        *,
        slug: str,
        workspace_path: str,
        requested_path: str,
        allow_write: bool = False,
        ttl_s: float = 120.0,
    ) -> str:
        return self._fs_broker.issue_handle(
            plugin_slug=_plugin_slug(slug),
            workspace_path=workspace_path,
            requested_path=requested_path,
            allow_write=allow_write,
            ttl_s=ttl_s,
        )

    def resolve_fs_handle(self, *, slug: str, handle: str, want_write: bool = False) -> str:
        return self._fs_broker.resolve_handle(
            plugin_slug=_plugin_slug(slug),
            handle=handle,
            want_write=want_write,
        )

    def get_settings_schema(self, slug: str) -> dict | None:
        """Return ``register_settings`` schema dict for a loaded plugin, or None."""
        with self._lock:
            lp = self._loaded.get(slug)
            if lp is None or lp.error:
                return None
            s = lp.context._settings_schema
            if not s or not isinstance(s, dict):
                return None
            return s

    def get_all_tools(self) -> list:
        """Return wrapped tools from all loaded plugins."""
        tools = []
        with self._lock:
            for lp in self._loaded.values():
                if lp.error:
                    continue
                if lp.manifest.kind == PLUGIN_KIND_DASHBOARD:
                    continue
                rt_names = lp.manifest.realtime_tools
                for t in lp.context._tools:
                    t_name = getattr(t, "name", "")
                    tools.append(
                        _wrap_plugin_tool(
                            lp.name,
                            t,
                            realtime=(t_name in rt_names),
                            permissions=lp.manifest.permissions,
                            visible_in_tier0=lp.manifest.visible_in_tier0,
                            runtime=lp.runtime,
                        )
                    )
        return tools

    def get_dashboard_manifest(self, base_url: str) -> dict[str, Any]:
        """Aggregate dashboard tile metadata from enabled, loaded dashboard plugins.

        ``base_url`` is kept for call-site compatibility; tile ``url`` values are **path-only**
        (e.g. ``/plugin/dash_foo/tile``) so the dashboard WKWebView always fetches same-origin
        (``localhost`` vs ``127.0.0.1`` in the server ``request.base_url`` would otherwise break
        ``fetch`` and leave tiles on skeletons).
        """
        tiles_out: list[dict[str, Any]] = []
        plugins_out: list[dict[str, str]] = []
        with self._lock:
            for lp in self._loaded.values():
                if lp.error or lp.manifest.kind != PLUGIN_KIND_DASHBOARD:
                    continue
                st = get_plugin_approval_state(lp.slug)
                if not st.get("enabled"):
                    continue
                plugins_out.append(
                    {
                        "slug": lp.slug,
                        "name": lp.manifest.name,
                        "version": lp.manifest.version,
                    }
                )
                raw_tiles = getattr(lp.context, "_dashboard_tiles", None) or []
                route_ok = bool(getattr(lp.context, "_route_handler_registered", False))
                for t in raw_tiles:
                    if not isinstance(t, dict):
                        continue
                    path = str(t.get("path") or "/").strip()
                    if not path.startswith("/"):
                        path = "/" + path
                    title = str(t.get("title") or lp.manifest.name).strip()
                    tid = str(t.get("id") or title).strip()
                    rs = t.get("refresh_seconds")
                    entry: dict[str, Any] = {
                        "slug": lp.slug,
                        "title": title,
                        "id": tid,
                        "path": path,
                    }
                    if route_ok:
                        entry["url"] = f"/plugin/{lp.slug}{path}"
                    else:
                        entry["url"] = ""
                        entry["error"] = (
                            "No HTTP route handler registered; use ctx.register_route_handler(...) in "
                            "plugin.py so /dashboard can fetch tile content."
                        )
                    if isinstance(rs, (int, float)) and rs > 0:
                        entry["refresh_seconds"] = int(rs)
                    pr = t.get("priority")
                    if isinstance(pr, (int, float)):
                        entry["priority"] = int(pr)
                    layout = str(t.get("layout") or "").strip().lower()
                    if layout in {"hero", "wide", "feature"}:
                        entry["layout"] = layout
                    tiles_out.append(entry)
        tiles_out.sort(key=lambda x: int(x.get("priority", 999)))
        return {"tiles": tiles_out, "plugins": plugins_out}

    def get_live_view_js(self) -> str:
        """Return concatenated JS for all plugin Live View renderers."""
        parts: list[str] = []
        with self._lock:
            for lp in self._loaded.values():
                if lp.error:
                    continue
                for event_type, renderer in lp.context._live_view_renderers.items():
                    js = renderer.get("js", "")
                    if js:
                        parts.append(
                            f"window._pluginRenderers[{event_type!r}] = {js};\n"
                        )
        return "\n".join(parts)

    def get_route_setups(self) -> list[tuple[str, list]]:
        """Return proxy route setup fns for loaded runtime plugins."""
        def _setup_proxy(slug: str, runtime: PluginRuntimeHost):
            def _setup(sub_app):
                try:
                    from fastapi.responses import Response, JSONResponse, StreamingResponse
                except Exception:
                    return

                @sub_app.websocket("/{path:path}")
                async def _plugin_proxy_ws(_ws, path: str):
                    # Phase-2 websocket proxy is intentionally not enabled yet.
                    await _ws.close(code=1008, reason=f"websocket proxy disabled for plugin route '{slug}/{path}'")

                # ``Request`` must be module-resolved (see ``from __future__ import annotations``); a local import
                # inside this closure breaks FastAPI's type resolution and ``request`` becomes a query param.
                @sub_app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"])
                async def _plugin_proxy(path: str, req: Request):
                    body = await req.body()
                    if len(body) > (1024 * 1024):
                        return JSONResponse(status_code=413, content={"ok": False, "error": "payload too large"})
                    try:
                        _auth_subj = (
                            "authenticated_local"
                            if (req.cookies.get("mugi_auth") or req.headers.get("x-auth-token"))
                            else "localhost"
                        )
                    except Exception:
                        _auth_subj = "localhost"
                    if str(slug).startswith("dash_") and req.method.upper() == "GET":
                        if _dash_plugin_get_rate_limited(str(slug), _auth_subj):
                            return JSONResponse(
                                status_code=429,
                                content={"ok": False, "error": "dashboard_tile_rate_limited"},
                            )
                    headers = {}
                    allowed_headers = {
                        "accept",
                        "content-type",
                        "cache-control",
                        "if-none-match",
                        "if-modified-since",
                        "authorization",
                        "x-requested-with",
                    }
                    for k, v in req.headers.items():
                        lk = k.lower().strip()
                        if lk in {"host", "content-length", "connection"}:
                            continue
                        if lk not in allowed_headers and not lk.startswith("x-"):
                            continue
                        headers[str(k)] = str(v)
                    try:
                        auth_subject = "authenticated_local" if (req.cookies.get("mugi_auth") or req.headers.get("x-auth-token")) else "localhost"
                        req_id = req.headers.get("x-request-id") or str(uuid.uuid4())
                        thread_id = req.query_params.get("thread_id") or req.headers.get("x-thread-id") or ""
                        method_s = str(req.method)
                        path_s = "/" + str(path or "")
                        query_s = str(req.url.query or "")
                        ctx = {
                            "request_id": str(req_id),
                            "thread_id": str(thread_id),
                            "auth_subject": auth_subject,
                            "user_meta_redacted": True,
                        }

                        def _call_proxy() -> dict[str, Any]:
                            return runtime.proxy_route(
                                method=method_s,
                                path=path_s,
                                query=query_s,
                                headers=headers,
                                body=body,
                                context=ctx,
                            )

                        loop = asyncio.get_running_loop()
                        out = await loop.run_in_executor(None, _call_proxy)
                        resp = out.get("response") if isinstance(out.get("response"), dict) else {}
                        status = int(resp.get("status") or 200)
                        rheaders = resp.get("headers") if isinstance(resp.get("headers"), dict) else {}
                        if len(rheaders) > 64:
                            return JSONResponse(status_code=502, content={"ok": False, "error": "too many response headers"})
                        body_limit = 2 * 1024 * 1024
                        if isinstance(resp.get("stream"), dict):
                            stream = resp["stream"]
                            stype = str(stream.get("type") or "").lower()
                            chunks = stream.get("chunks") if isinstance(stream.get("chunks"), list) else []
                            if stype == "sse":
                                def _iter_sse():
                                    total = 0
                                    for ch in chunks:
                                        text = str(ch)
                                        total += len(text.encode("utf-8"))
                                        if total > body_limit:
                                            break
                                        yield text
                                media_type = str(rheaders.get("content-type") or "text/event-stream")
                                return StreamingResponse(_iter_sse(), status_code=status, headers=rheaders, media_type=media_type)
                            return JSONResponse(status_code=501, content={"ok": False, "error": f"unsupported stream type '{stype}'"})
                        text_body = str(resp.get("body") or "")
                        if len(text_body.encode("utf-8")) > body_limit:
                            return JSONResponse(status_code=502, content={"ok": False, "error": "response body too large"})
                        return Response(content=text_body, status_code=status, headers=rheaders)
                    except Exception as e:
                        return JSONResponse(status_code=502, content={"ok": False, "error": str(e)})
            return _setup

        result = []
        with self._lock:
            for lp in self._loaded.values():
                if lp.error or lp.runtime is None:
                    continue
                result.append((lp.slug, [_setup_proxy(lp.slug, lp.runtime)]))
        return result

    def get_all_hooks(self, event_name: str) -> list[tuple[str, list]]:
        """Return loaded hook registrations (for diagnostics/tests)."""
        result = []
        with self._lock:
            for lp in self._loaded.values():
                if lp.error:
                    continue
                hooks = lp.context._hooks.get(event_name, [])
                if hooks:
                    result.append((lp.name, hooks))
        return result

    def call_hooks(self, event_name: str, *args, **kwargs) -> None:
        """Dispatch hooks through runtime IPC. Failures are logged, never raised."""
        with self._lock:
            loaded = list(self._loaded.values())
        for lp in loaded:
            if lp.error or lp.runtime is None:
                continue
            if (
                lp.manifest.kind == PLUGIN_KIND_DASHBOARD
                and not lp.manifest.agent_hooks
            ):
                continue
            if event_name not in lp.context._hooks:
                continue
            try:
                lp.runtime.call_hook(
                    event_name=event_name,
                    args=list(args),
                    kwargs=kwargs if isinstance(kwargs, dict) else {},
                    timeout_s=5.0,
                )
            except Exception as e:
                _debug_log(
                    f"[PLUGIN:{lp.name}] hook {event_name} runtime dispatch failed: {e}",
                    "ERROR",
                )

    def dispatch_control_event(
        self,
        *,
        plugin_id: str,
        stream_id: str,
        control_id: str,
        value=None,
        action_id: str | None = None,
        thread_id: str | None = None,
        event_id: str | None = None,
        timestamp_ms: int | None = None,
    ) -> dict:
        """Validate and dispatch a dashboard control event to plugin handlers."""
        slug = _plugin_slug(plugin_id or "")
        sid = (stream_id or "").strip()
        cid = (control_id or "").strip()
        if not slug or not sid or not cid:
            return {"ok": False, "status": "rejected", "message": "Missing plugin_id, stream_id, or control_id."}
        with self._lock:
            lp = self._loaded.get(slug)
        if lp is None or lp.error:
            return {"ok": False, "status": "rejected", "message": f"Unknown or unloaded plugin '{plugin_id}'."}
        if lp.runtime is not None:
            event = {
                "plugin_id": lp.slug,
                "plugin_name": lp.name,
                "stream_id": sid,
                "control_id": cid,
                "value": value,
                "action_id": action_id,
                "thread_id": (thread_id or "").strip() or None,
                "event_id": (event_id or "").strip() or None,
                "timestamp_ms": int(timestamp_ms) if timestamp_ms is not None else int(time.time() * 1000),
            }
            try:
                result = lp.runtime.dispatch_control(event, timeout_s=30.0)
                status = str(result.get("status") or "applied")
                msg = str(result.get("message") or "Control applied.")
                out = {
                    "ok": status not in {"rejected", "error"},
                    "status": status,
                    "message": msg,
                    "plugin_id": lp.slug,
                    "stream_id": sid,
                    "control_id": cid,
                    "value": value,
                    "action_id": action_id,
                }
                for k in ("open_url", "copy_text"):
                    if result.get(k) is not None:
                        out[k] = str(result[k])
                return out
            except Exception as e:
                _debug_log(f"[PLUGIN:{lp.name}] runtime control dispatch failed: {e}", "ERROR")
                return {"ok": False, "status": "error", "message": f"Runtime control dispatch failed: {e}"}
        controls = lp.context.get_stream_controls(sid)
        by_id = {str(c.get("id")): c for c in controls if isinstance(c, dict)}
        cdef = by_id.get(cid)
        if cdef is None:
            return {"ok": False, "status": "rejected", "message": f"Control '{cid}' is not registered for stream '{sid}'."}
        ok, coerced, err = self._validate_control_value(cdef, value, action_id=action_id)
        if not ok:
            return {"ok": False, "status": "rejected", "message": err or "Invalid control value."}
        if not lp.context._control_handlers:
            return {"ok": False, "status": "rejected", "message": "Plugin has no control handler registered."}

        event = {
            "plugin_id": lp.slug,
            "plugin_name": lp.name,
            "stream_id": sid,
            "control_id": cid,
            "control_type": cdef.get("type"),
            "value": coerced,
            "action_id": action_id or cdef.get("action_id"),
            "thread_id": (thread_id or "").strip() or None,
            "event_id": (event_id or "").strip() or None,
            "timestamp_ms": int(timestamp_ms) if timestamp_ms is not None else int(time.time() * 1000),
            "definition": cdef,
        }
        message = "Control applied."
        _open_url = None
        _copy_text = None
        for cb in list(lp.context._control_handlers):
            try:
                sig = inspect.signature(cb)
                if len(sig.parameters) >= 2:
                    result = cb(event, lp.context)
                else:
                    result = cb(event)
                if isinstance(result, dict):
                    if str(result.get("status") or "").strip().lower() in {"rejected", "error"}:
                        return {
                            "ok": False,
                            "status": str(result.get("status") or "rejected"),
                            "message": str(result.get("message") or "Control handler rejected update."),
                        }
                    if result.get("message"):
                        message = str(result.get("message"))
                    # Forward client-action hints so backend can include in control_ack
                    _open_url = result.get("open_url")
                    _copy_text = result.get("copy_text")
            except Exception as e:
                _debug_log(f"[PLUGIN:{lp.name}] control handler failed: {e}", "ERROR")
                return {"ok": False, "status": "error", "message": f"Control handler error: {e}"}
        out = {
            "ok": True,
            "status": "applied",
            "message": message,
            "plugin_id": lp.slug,
            "stream_id": sid,
            "control_id": cid,
            "value": coerced,
            "action_id": action_id or cdef.get("action_id"),
        }
        if _open_url is not None:
            out["open_url"] = str(_open_url)
        if _copy_text is not None:
            out["copy_text"] = str(_copy_text)
        return out

    @staticmethod
    def _validate_control_value(control_def: dict, value, *, action_id: str | None = None) -> tuple[bool, object, str | None]:
        from tools.job_stream_schema import validate_control_value
        ok, coerced, _cdef, err, _ctype = validate_control_value(control_def, value, action_id=action_id)
        return ok, coerced, err
