"""Tests for the Hermes plugin system (hermes_cli.plugins)."""

import logging
import sys
import types
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest
import yaml

from hermes_cli.plugins import (
    ENTRY_POINTS_GROUP,
    VALID_HOOKS,
    PluginContext,
    PluginManager,
    PluginManifest,
    get_plugin_command_handler,
    get_plugin_commands,
    get_pre_tool_call_block_message,
    has_middleware,
    resolve_plugin_command_result,
)
from hermes_cli.middleware import (
    VALID_MIDDLEWARE,
    apply_llm_request_middleware,
    apply_tool_request_middleware,
    run_tool_execution_middleware,
)


# ── Helpers ────────────────────────────────────────────────────────────────


def _make_plugin_dir(base: Path, name: str, *, register_body: str = "pass",
                     manifest_extra: dict | None = None,
                     auto_enable: bool = True) -> Path:
    """Create a minimal plugin directory with plugin.yaml + __init__.py.

    If *auto_enable* is True (default), also write the plugin's name into
    ``<hermes_home>/config.yaml`` under ``plugins.enabled``. Plugins are
    opt-in by default, so tests that expect the plugin to actually load
    need this. Pass ``auto_enable=False`` for tests that exercise the
    unenabled path.

    *base* is expected to be ``<hermes_home>/plugins/``; we derive
    ``<hermes_home>`` from it by walking one level up.
    """
    plugin_dir = base / name
    plugin_dir.mkdir(parents=True, exist_ok=True)

    manifest = {"name": name, "version": "0.1.0", "description": f"Test plugin {name}"}
    if manifest_extra:
        manifest.update(manifest_extra)

    (plugin_dir / "plugin.yaml").write_text(yaml.dump(manifest))
    (plugin_dir / "__init__.py").write_text(
        f"def register(ctx):\n    {register_body}\n"
    )

    if auto_enable:
        # Write/merge plugins.enabled in <HERMES_HOME>/config.yaml.
        # Config is always read from HERMES_HOME (not from the project
        # dir for project plugins), so that's where we opt in.
        import os
        hermes_home_str = os.environ.get("HERMES_HOME")
        if hermes_home_str:
            hermes_home = Path(hermes_home_str)
        else:
            hermes_home = base.parent
        hermes_home.mkdir(parents=True, exist_ok=True)
        cfg_path = hermes_home / "config.yaml"
        cfg: dict = {}
        if cfg_path.exists():
            try:
                cfg = yaml.safe_load(cfg_path.read_text()) or {}
            except Exception:
                cfg = {}
        plugins_cfg = cfg.setdefault("plugins", {})
        enabled = plugins_cfg.setdefault("enabled", [])
        if isinstance(enabled, list) and name not in enabled:
            enabled.append(name)
        cfg_path.write_text(yaml.safe_dump(cfg))

    return plugin_dir


# ── TestPluginDiscovery ────────────────────────────────────────────────────


class TestPluginDiscovery:
    """Tests for plugin discovery from directories and entry points."""

    def test_discover_user_plugins(self, tmp_path, monkeypatch):
        """Plugins in ~/.hermes/plugins/ are discovered."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(plugins_dir, "hello_plugin")
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        assert "hello_plugin" in mgr._plugins
        assert mgr._plugins["hello_plugin"].enabled

    def test_plugin_can_register_and_invoke_middleware(self, tmp_path, monkeypatch):
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir,
            "mw_plugin",
            register_body=(
                "ctx.register_middleware('llm_request', "
                "lambda **kw: {'request': {**kw['request'], 'mw': True}})\n"
                "    ctx.register_middleware('tool_request', "
                "lambda **kw: {'args': {**kw['args'], 'mw': True}})"
            ),
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        assert "llm_request" in VALID_MIDDLEWARE
        assert "tool_request" in VALID_MIDDLEWARE
        assert set(mgr._plugins["mw_plugin"].middleware_registered) == {"llm_request", "tool_request"}
        assert mgr.invoke_middleware("llm_request", request={"messages": []}) == [
            {"request": {"messages": [], "mw": True}}
        ]
        assert mgr.invoke_middleware("tool_request", args={"path": "README.md"}) == [
            {"args": {"path": "README.md", "mw": True}}
        ]
        assert mgr.has_middleware("llm_request") is True

    def test_execution_middleware_does_not_retry_downstream_failure(self, monkeypatch):
        calls = []

        def middleware(**kwargs):
            return kwargs["next_call"](kwargs["args"])

        manager = types.SimpleNamespace(_middleware={"tool_execution": [middleware]})
        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)

        def terminal(args):
            calls.append(args)
            raise RuntimeError("tool failed")

        with pytest.raises(RuntimeError, match="tool failed"):
            run_tool_execution_middleware("terminal", {"command": "false"}, terminal)

        assert calls == [{"command": "false"}]

    def test_middleware_helpers_skip_no_listener_work(self, monkeypatch):
        manager = types.SimpleNamespace(_middleware={})
        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)

        request = {"messages": []}
        args = {"path": "README.md"}

        llm_result = apply_llm_request_middleware(request)
        tool_result = apply_tool_request_middleware("read_file", args)

        assert llm_result.payload is request
        assert llm_result.original_payload is request
        assert llm_result.changed is False
        assert llm_result.trace == []
        assert tool_result.payload is args
        assert tool_result.original_payload is args
        assert tool_result.changed is False
        assert tool_result.trace == []
        assert run_tool_execution_middleware("terminal", args, lambda payload: payload) is args
        assert has_middleware("tool_request") is False

    def test_request_middleware_changed_tracks_trace_not_deep_equality(self, monkeypatch):
        def same_payload_middleware(**kwargs):
            return {"args": kwargs["args"], "source": "same-payload"}

        manager = types.SimpleNamespace(
            _middleware={"tool_request": [same_payload_middleware]},
            invoke_middleware=lambda kind, **kwargs: [same_payload_middleware(**kwargs)],
        )
        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)

        args = {"path": "README.md"}
        result = apply_tool_request_middleware("read_file", args)

        assert result.payload == args
        assert result.original_payload == args
        assert result.changed is True
        assert result.trace == [{"source": "same-payload"}]

    def test_execution_middleware_post_next_call_error_does_not_retry(self, monkeypatch):
        calls = []

        def middleware(**kwargs):
            result = kwargs["next_call"](kwargs["args"])
            raise RuntimeError(f"post-processing failed after {result}")

        manager = types.SimpleNamespace(_middleware={"tool_execution": [middleware]})
        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)

        def terminal(args):
            calls.append(args)
            return "terminal-result"

        result = run_tool_execution_middleware("terminal", {"command": "printf ok"}, terminal)

        assert result == "terminal-result"
        assert calls == [{"command": "printf ok"}]

    def test_execution_middleware_pre_next_call_error_fails_open_to_remaining_chain(self, monkeypatch):
        calls = []

        def failing_middleware(**kwargs):
            calls.append("failing")
            raise RuntimeError("middleware setup failed")

        def downstream_middleware(**kwargs):
            calls.append("downstream")
            return kwargs["next_call"]({**kwargs["args"], "rewritten": True})

        manager = types.SimpleNamespace(_middleware={"tool_execution": [failing_middleware, downstream_middleware]})
        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)

        def terminal(args):
            calls.append(("terminal", args))
            return args

        result = run_tool_execution_middleware("terminal", {"command": "printf ok"}, terminal)

        assert result == {"command": "printf ok", "rewritten": True}
        assert calls == ["failing", "downstream", ("terminal", {"command": "printf ok", "rewritten": True})]

    def test_execution_middleware_translated_downstream_failure_is_not_masked(self, monkeypatch):
        calls = []

        def middleware(**kwargs):
            try:
                return kwargs["next_call"](kwargs["args"])
            except Exception as exc:
                raise RuntimeError(f"translated downstream failure: {exc}") from exc

        manager = types.SimpleNamespace(_middleware={"tool_execution": [middleware]})
        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)

        def terminal(args):
            calls.append(args)
            raise RuntimeError("terminal failed")

        with pytest.raises(RuntimeError, match="translated downstream failure: terminal failed"):
            run_tool_execution_middleware("terminal", {"command": "false"}, terminal)

        assert calls == [{"command": "false"}]

    def test_execution_middleware_downstream_base_exception_is_not_wrapped(self, monkeypatch):
        calls = []

        def middleware(**kwargs):
            try:
                return kwargs["next_call"](kwargs["args"])
            except Exception as exc:
                raise RuntimeError(f"middleware should not catch base exception: {exc}") from exc

        manager = types.SimpleNamespace(_middleware={"tool_execution": [middleware]})
        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)

        def terminal(args):
            calls.append(args)
            raise KeyboardInterrupt()

        with pytest.raises(KeyboardInterrupt):
            run_tool_execution_middleware("terminal", {"command": "interrupt"}, terminal)

        assert calls == [{"command": "interrupt"}]

    def test_execution_middleware_double_next_call_does_not_run_terminal_twice(self, monkeypatch):
        calls = []

        def middleware(**kwargs):
            first = kwargs["next_call"](kwargs["args"])
            # Deliberate misuse: a second next_call() must not re-run the
            # downstream tool. The chain surfaces it as an error and preserves
            # the first (successful) downstream result.
            kwargs["next_call"](kwargs["args"])
            return first

        manager = types.SimpleNamespace(_middleware={"tool_execution": [middleware]})
        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)

        def terminal(args):
            calls.append(args)
            return "terminal-result"

        result = run_tool_execution_middleware("terminal", {"command": "printf ok"}, terminal)

        assert result == "terminal-result"
        assert calls == [{"command": "printf ok"}]

    def test_request_middleware_tolerates_non_deepcopyable_payload(self, monkeypatch):
        import threading

        recorded = {}

        def middleware(**kwargs):
            recorded["args"] = kwargs["args"]
            return None

        manager = types.SimpleNamespace(
            _middleware={"tool_request": [middleware]},
            invoke_middleware=lambda kind, **kwargs: [middleware(**kwargs)],
        )
        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)

        # threading.Lock is not deepcopyable; a hard deepcopy would raise.
        args = {"command": "noop", "lock": threading.Lock()}
        result = apply_tool_request_middleware("terminal", args)

        # Middleware ran (payload was copied via the shallow fallback) and the
        # non-deepcopyable member is shared by reference rather than aborting.
        assert recorded["args"]["command"] == "noop"
        assert result.payload["command"] == "noop"
        assert result.payload["lock"] is args["lock"]

    def test_discover_project_plugins(self, tmp_path, monkeypatch):
        """Plugins in ./.hermes/plugins/ are discovered."""
        project_dir = tmp_path / "project"
        project_dir.mkdir()
        monkeypatch.chdir(project_dir)
        monkeypatch.setenv("HERMES_ENABLE_PROJECT_PLUGINS", "true")
        plugins_dir = project_dir / ".hermes" / "plugins"
        _make_plugin_dir(plugins_dir, "proj_plugin")

        mgr = PluginManager()
        mgr.discover_and_load()

        assert "proj_plugin" in mgr._plugins
        assert mgr._plugins["proj_plugin"].enabled

    def test_discover_project_plugins_skipped_by_default(self, tmp_path, monkeypatch):
        """Project plugins are not discovered unless explicitly enabled."""
        project_dir = tmp_path / "project"
        project_dir.mkdir()
        monkeypatch.chdir(project_dir)
        plugins_dir = project_dir / ".hermes" / "plugins"
        _make_plugin_dir(plugins_dir, "proj_plugin")

        mgr = PluginManager()
        mgr.discover_and_load()

        assert "proj_plugin" not in mgr._plugins

    def test_discover_is_idempotent(self, tmp_path, monkeypatch):
        """Calling discover_and_load() twice does not duplicate plugins."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(plugins_dir, "once_plugin")
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()
        mgr.discover_and_load()  # second call should no-op

        # Filter out bundled plugins — they're always discovered.
        non_bundled = {
            n: p for n, p in mgr._plugins.items()
            if p.manifest.source != "bundled"
        }
        assert len(non_bundled) == 1

    def test_failed_discovery_is_not_cached(self, tmp_path, monkeypatch):
        """A sweep that raises must not cache 'discovered' with no plugins.

        Regression for the stranded-empty-registry class of failures: callers
        (e.g. tools.web_tools._ensure_web_plugins_loaded) swallow discovery
        exceptions as warnings, so if a failed sweep flipped ``_discovered``
        permanently, every later call would early-return against an empty
        registry ("No web provider configured") for the process lifetime.
        """
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(plugins_dir, "retry_plugin")
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()

        def _boom(self_inner):
            raise RuntimeError("sweep failed")

        monkeypatch.setattr(PluginManager, "_discover_and_load_inner", _boom)
        with pytest.raises(RuntimeError, match="sweep failed"):
            mgr.discover_and_load()
        assert mgr._discovered is False, "failed sweep was cached as discovered"

        # A later call (with discovery healthy again) must do the real scan.
        monkeypatch.undo()
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
        mgr.discover_and_load()
        assert mgr._discovered is True
        non_bundled = {
            n: p for n, p in mgr._plugins.items()
            if p.manifest.source != "bundled"
        }
        assert len(non_bundled) == 1

    def test_discover_skips_dir_without_manifest(self, tmp_path, monkeypatch):
        """Directories without plugin.yaml are silently skipped."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        (plugins_dir / "no_manifest").mkdir(parents=True)
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        # Filter out bundled plugins — they're always discovered.
        non_bundled = {
            n: p for n, p in mgr._plugins.items()
            if p.manifest.source != "bundled"
        }
        assert len(non_bundled) == 0

    def test_entry_points_scanned(self, tmp_path, monkeypatch):
        """Entry-point based plugins are discovered (mocked)."""
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        fake_module = types.ModuleType("fake_ep_plugin")
        fake_module.register = lambda ctx: None  # type: ignore[attr-defined]

        fake_ep = MagicMock()
        fake_ep.name = "ep_plugin"
        fake_ep.value = "fake_ep_plugin:register"
        fake_ep.group = ENTRY_POINTS_GROUP
        fake_ep.load.return_value = fake_module

        def fake_entry_points():
            result = MagicMock()
            result.select = MagicMock(return_value=[fake_ep])
            return result

        with patch("importlib.metadata.entry_points", fake_entry_points):
            mgr = PluginManager()
            mgr.discover_and_load()

        assert "ep_plugin" in mgr._plugins

    def test_force_rediscover_clears_all_plugin_registries(self, monkeypatch):
        """force=True must clear every plugin-populated registry.

        Regression: ``_plugin_platform_names`` was populated by
        ``register_platform`` but omitted from the ``discover_and_load(force=True)``
        clear block, so a platform plugin disabled between force-rediscovers
        left a stale entry behind forever (the set diverged from the real
        platform_registry / _plugins truth). This asserts the clear block
        empties the full set of per-plugin registries so no future addition
        silently leaks across a force pass either.
        """
        mgr = PluginManager()

        # Seed every registry that a plugin's register() can populate, then
        # mark discovery done so force=True takes the clear path (we stub the
        # inner sweep so the test doesn't depend on any on-disk plugins).
        mgr._plugins["p"] = MagicMock()
        mgr._hooks["pre_tool_call"] = [lambda **_: None]
        mgr._middleware["llm_request"] = [lambda **_: None]
        mgr._plugin_tool_names.add("some_tool")
        mgr._plugin_platform_names.add("irc")
        mgr._cli_commands["c"] = {"plugin": "p"}
        mgr._plugin_commands["cmd"] = {"plugin": "p"}
        mgr._plugin_skills["p:skill"] = {}
        mgr._aux_tasks["task"] = {"plugin": "p"}
        mgr._slack_action_handlers.append(("aid", lambda **_: None, "p"))
        mgr._discovered = True

        monkeypatch.setattr(PluginManager, "_discover_and_load_inner", lambda self_inner: None)
        mgr.discover_and_load(force=True)

        assert mgr._plugins == {}
        assert mgr._hooks == {}
        assert mgr._middleware == {}
        assert mgr._plugin_tool_names == set()
        assert mgr._plugin_platform_names == set(), (
            "_plugin_platform_names was not cleared on force-rediscover"
        )
        assert mgr._cli_commands == {}
        assert mgr._plugin_commands == {}
        assert mgr._plugin_skills == {}
        assert mgr._aux_tasks == {}
        assert mgr._slack_action_handlers == []


# ── TestPluginLoading ──────────────────────────────────────────────────────


class TestPluginLoading:
    """Tests for plugin module loading."""

    def test_load_missing_init(self, tmp_path, monkeypatch):
        """Plugin dir without __init__.py records an error."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        plugin_dir = plugins_dir / "bad_plugin"
        plugin_dir.mkdir(parents=True)
        (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "bad_plugin"}))
        # Explicitly enable so the loader tries to import it and hits the
        # missing-init error.
        hermes_home = tmp_path / "hermes_test"
        (hermes_home / "config.yaml").write_text(
            yaml.safe_dump({"plugins": {"enabled": ["bad_plugin"]}})
        )
        monkeypatch.setenv("HERMES_HOME", str(hermes_home))

        mgr = PluginManager()
        mgr.discover_and_load()

        assert "bad_plugin" in mgr._plugins
        assert not mgr._plugins["bad_plugin"].enabled
        assert mgr._plugins["bad_plugin"].error is not None
        # Should be the missing-init error, not "not enabled".
        assert "not enabled" not in mgr._plugins["bad_plugin"].error

    def test_load_missing_register_fn(self, tmp_path, monkeypatch):
        """Plugin without register() function records an error."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        plugin_dir = plugins_dir / "no_reg"
        plugin_dir.mkdir(parents=True)
        (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "no_reg"}))
        (plugin_dir / "__init__.py").write_text("# no register function\n")
        # Explicitly enable it so the loader actually tries to import.
        hermes_home = tmp_path / "hermes_test"
        (hermes_home / "config.yaml").write_text(
            yaml.safe_dump({"plugins": {"enabled": ["no_reg"]}})
        )
        monkeypatch.setenv("HERMES_HOME", str(hermes_home))

        mgr = PluginManager()
        mgr.discover_and_load()

        assert "no_reg" in mgr._plugins
        assert not mgr._plugins["no_reg"].enabled
        assert "no register()" in mgr._plugins["no_reg"].error

    def test_load_registers_namespace_module(self, tmp_path, monkeypatch):
        """Directory plugins are importable under hermes_plugins.<name>."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(plugins_dir, "ns_plugin")
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        # Clean up any prior namespace module
        sys.modules.pop("hermes_plugins.ns_plugin", None)

        mgr = PluginManager()
        mgr.discover_and_load()

        assert "hermes_plugins.ns_plugin" in sys.modules

    def test_user_memory_plugin_auto_coerced_to_exclusive(self, tmp_path, monkeypatch):
        """User-installed memory plugins must NOT be loaded by the general
        PluginManager — they belong to plugins/memory discovery.

        Regression test for the mempalace crash:
            'PluginContext' object has no attribute 'register_memory_provider'

        A plugin that calls ``ctx.register_memory_provider`` in its
        ``__init__.py`` should be auto-detected and treated as
        ``kind: exclusive`` so the general loader records the manifest but
        does not import/register() it. The real activation happens through
        ``plugins/memory/__init__.py`` via ``memory.provider`` config.
        """
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        plugin_dir = plugins_dir / "mempalace"
        plugin_dir.mkdir(parents=True)
        # No explicit `kind:` — the heuristic should kick in.
        (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "mempalace"}))
        (plugin_dir / "__init__.py").write_text(
            "class MemPalaceProvider:\n"
            "    pass\n"
            "def register(ctx):\n"
            "    ctx.register_memory_provider('mempalace', MemPalaceProvider)\n"
        )
        # Even if the user explicitly enables it in config, the loader
        # should still treat it as exclusive and skip general loading.
        hermes_home = tmp_path / "hermes_test"
        (hermes_home / "config.yaml").write_text(
            yaml.safe_dump({"plugins": {"enabled": ["mempalace"]}})
        )
        monkeypatch.setenv("HERMES_HOME", str(hermes_home))

        mgr = PluginManager()
        mgr.discover_and_load()

        assert "mempalace" in mgr._plugins
        entry = mgr._plugins["mempalace"]
        assert entry.manifest.kind == "exclusive", (
            f"Expected auto-coerced kind='exclusive', got {entry.manifest.kind}"
        )
        # Not loaded by general manager (no register() call, no AttributeError).
        assert not entry.enabled
        assert entry.module is None
        assert "exclusive" in (entry.error or "").lower()

    def test_explicit_standalone_kind_not_coerced(self, tmp_path, monkeypatch):
        """If a plugin explicitly declares ``kind: standalone`` in its
        manifest, the memory-provider heuristic must NOT override it —
        even if the source happens to mention ``MemoryProvider``.
        """
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        plugin_dir = plugins_dir / "not_memory"
        plugin_dir.mkdir(parents=True)
        (plugin_dir / "plugin.yaml").write_text(
            yaml.dump({"name": "not_memory", "kind": "standalone"})
        )
        (plugin_dir / "__init__.py").write_text(
            "# This plugin inspects MemoryProvider docs but isn't one.\n"
            "def register(ctx):\n    pass\n"
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        assert mgr._plugins["not_memory"].manifest.kind == "standalone"


# ── TestPluginHooks ────────────────────────────────────────────────────────


class TestPluginHooks:
    """Tests for lifecycle hook registration and invocation."""

    def test_valid_hooks_include_request_scoped_api_hooks(self):
        assert "pre_api_request" in VALID_HOOKS
        assert "post_api_request" in VALID_HOOKS
        assert "api_request_error" in VALID_HOOKS
        assert "subagent_start" in VALID_HOOKS
        assert "transform_terminal_output" in VALID_HOOKS
        assert "transform_tool_result" in VALID_HOOKS
        assert "transform_llm_output" in VALID_HOOKS

    def test_valid_hooks_include_pre_gateway_dispatch(self):
        assert "pre_gateway_dispatch" in VALID_HOOKS

    def test_pre_gateway_dispatch_collects_action_dicts(self, tmp_path, monkeypatch):
        """pre_gateway_dispatch callbacks return action dicts (skip/rewrite/allow)."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir, "predispatch_plugin",
            register_body=(
                'ctx.register_hook("pre_gateway_dispatch", '
                'lambda **kw: {"action": "skip", "reason": "test"})'
            ),
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        results = mgr.invoke_hook(
            "pre_gateway_dispatch",
            event=object(),
            gateway=object(),
            session_store=object(),
        )
        assert len(results) == 1
        assert results[0] == {"action": "skip", "reason": "test"}

    def test_register_and_invoke_hook(self, tmp_path, monkeypatch):
        """Registered hooks are called on invoke_hook()."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir, "hook_plugin",
            register_body='ctx.register_hook("pre_tool_call", lambda **kw: None)',
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        # Should not raise
        mgr.invoke_hook("pre_tool_call", tool_name="test", args={}, task_id="t1")

    def test_invoke_hook_adds_observer_schema_version(self, tmp_path, monkeypatch):
        """invoke_hook() supplies the observer schema version for all hooks."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir,
            "schema_plugin",
            register_body=(
                'ctx.register_hook("pre_tool_call", '
                'lambda **kw: kw.get("telemetry_schema_version"))'
            ),
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        assert mgr.invoke_hook("pre_tool_call", tool_name="test", args={}) == [
            "hermes.observer.v1"
        ]

    def test_hook_exception_does_not_propagate(self, tmp_path, monkeypatch):
        """A hook callback that raises does NOT crash the caller."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir, "bad_hook",
            register_body='ctx.register_hook("post_tool_call", lambda **kw: 1/0)',
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        # Should not raise despite 1/0
        mgr.invoke_hook("post_tool_call", tool_name="x", args={}, result="r", task_id="")

    def test_hook_return_values_collected(self, tmp_path, monkeypatch):
        """invoke_hook() collects non-None return values from callbacks."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir, "ctx_plugin",
            register_body=(
                'ctx.register_hook("pre_llm_call", '
                'lambda **kw: {"context": "memory from plugin"})'
            ),
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        results = mgr.invoke_hook("pre_llm_call", session_id="s1", user_message="hi",
                                  conversation_history=[], is_first_turn=True, model="test")
        assert len(results) == 1
        assert results[0] == {"context": "memory from plugin"}

    def test_hook_none_returns_excluded(self, tmp_path, monkeypatch):
        """invoke_hook() excludes None returns from the result list."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir, "none_hook",
            register_body='ctx.register_hook("post_llm_call", lambda **kw: None)',
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        results = mgr.invoke_hook("post_llm_call", session_id="s1",
                                  user_message="hi", assistant_response="bye", model="test")
        assert results == []

    def test_request_hooks_are_invokeable(self, tmp_path, monkeypatch):
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir, "request_hook",
            register_body=(
                'ctx.register_hook("pre_api_request", '
                'lambda **kw: {"seen": kw.get("api_call_count"), '
                '"mc": kw.get("message_count"), "tc": kw.get("tool_count")})'
            ),
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        assert mgr.has_hook("pre_api_request") is True
        assert mgr.has_hook("post_api_request") is False
        results = mgr.invoke_hook(
            "pre_api_request",
            session_id="s1",
            task_id="t1",
            model="test",
            api_call_count=2,
            message_count=5,
            tool_count=3,
            approx_input_tokens=100,
            request_char_count=400,
            max_tokens=8192,
        )
        assert results == [{"seen": 2, "mc": 5, "tc": 3}]

    def test_transform_terminal_output_hook_can_be_registered_and_invoked(self, tmp_path, monkeypatch):
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir, "transform_hook",
            register_body=(
                'ctx.register_hook("transform_terminal_output", '
                'lambda **kw: f"{kw[\'command\']}|{kw[\'returncode\']}|{kw[\'env_type\']}|{kw[\'task_id\']}|{len(kw[\'output\'])}")'
            ),
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        results = mgr.invoke_hook(
            "transform_terminal_output",
            command="echo hello",
            output="abcdef",
            returncode=7,
            task_id="task-1",
            env_type="local",
        )
        assert results == ["echo hello|7|local|task-1|6"]

    def test_invalid_hook_name_warns(self, tmp_path, monkeypatch, caplog):
        """Registering an unknown hook name logs a warning."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir, "warn_plugin",
            register_body='ctx.register_hook("on_banana", lambda **kw: None)',
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"):
            mgr = PluginManager()
            mgr.discover_and_load()

        assert any("on_banana" in record.message for record in caplog.records)

class TestPreToolCallBlocking:
    """Tests for the pre_tool_call block directive helper."""

    def test_block_message_returned_for_valid_directive(self, monkeypatch):
        monkeypatch.setattr(
            "hermes_cli.plugins.invoke_hook",
            lambda hook_name, **kwargs: [{"action": "block", "message": "blocked by plugin"}],
        )
        assert get_pre_tool_call_block_message("todo", {}, task_id="t1") == "blocked by plugin"

    def test_invalid_returns_are_ignored(self, monkeypatch):
        """Various malformed hook returns should not trigger a block."""
        monkeypatch.setattr(
            "hermes_cli.plugins.invoke_hook",
            lambda hook_name, **kwargs: [
                "block",                                 # not a dict
                123,                                     # not a dict
                {"action": "block"},                     # missing message
                {"action": "deny", "message": "nope"},   # wrong action
                {"message": "missing action"},            # no action key
                {"action": "block", "message": 123},     # message not str
            ],
        )
        assert get_pre_tool_call_block_message("todo", {}, task_id="t1") is None

    def test_none_when_no_hooks(self, monkeypatch):
        monkeypatch.setattr(
            "hermes_cli.plugins.invoke_hook",
            lambda hook_name, **kwargs: [],
        )
        assert get_pre_tool_call_block_message("web_search", {"q": "test"}) is None

    def test_first_valid_block_wins(self, monkeypatch):
        monkeypatch.setattr(
            "hermes_cli.plugins.invoke_hook",
            lambda hook_name, **kwargs: [
                {"action": "allow"},
                {"action": "block", "message": "first blocker"},
                {"action": "block", "message": "second blocker"},
            ],
        )
        assert get_pre_tool_call_block_message("terminal", {}) == "first blocker"


class TestThreadToolWhitelist:
    """Tests for the thread-local tool whitelist used by background review forks."""

    def test_allowed_tool_passes_through_to_hooks(self, monkeypatch):
        from hermes_cli.plugins import (
            set_thread_tool_whitelist,
            clear_thread_tool_whitelist,
        )

        monkeypatch.setattr(
            "hermes_cli.plugins.invoke_hook",
            lambda hook_name, **kwargs: [],
        )
        set_thread_tool_whitelist({"memory", "skill_manage"})
        try:
            assert get_pre_tool_call_block_message("memory", {}) is None
        finally:
            clear_thread_tool_whitelist()

    def test_disallowed_tool_blocked_with_message(self, monkeypatch):
        from hermes_cli.plugins import (
            set_thread_tool_whitelist,
            clear_thread_tool_whitelist,
        )

        monkeypatch.setattr(
            "hermes_cli.plugins.invoke_hook",
            lambda hook_name, **kwargs: [],
        )
        set_thread_tool_whitelist(
            {"memory"}, deny_msg_fmt="denied: {tool_name}"
        )
        try:
            msg = get_pre_tool_call_block_message("terminal", {})
            assert msg == "denied: terminal"
        finally:
            clear_thread_tool_whitelist()

    def test_clear_restores_unrestricted_behavior(self, monkeypatch):
        from hermes_cli.plugins import (
            set_thread_tool_whitelist,
            clear_thread_tool_whitelist,
        )

        monkeypatch.setattr(
            "hermes_cli.plugins.invoke_hook",
            lambda hook_name, **kwargs: [],
        )
        set_thread_tool_whitelist({"memory"})
        clear_thread_tool_whitelist()
        # After clearing, any tool should pass through to plugin hooks (which
        # return [] here, so result is None).
        assert get_pre_tool_call_block_message("terminal", {}) is None

    def test_whitelist_is_thread_local(self, monkeypatch):
        """Setting a whitelist in one thread must NOT leak into another."""
        import threading

        from hermes_cli.plugins import (
            set_thread_tool_whitelist,
            clear_thread_tool_whitelist,
        )

        monkeypatch.setattr(
            "hermes_cli.plugins.invoke_hook",
            lambda hook_name, **kwargs: [],
        )

        # Main thread: install a restrictive whitelist.
        set_thread_tool_whitelist({"memory"})
        try:
            assert get_pre_tool_call_block_message("terminal", {}) is not None

            # Worker thread: should NOT inherit main thread's whitelist.
            result = {}

            def worker():
                result["msg"] = get_pre_tool_call_block_message("terminal", {})

            t = threading.Thread(target=worker)
            t.start()
            t.join()
            assert result["msg"] is None, (
                "thread-local whitelist leaked across threads"
            )
        finally:
            clear_thread_tool_whitelist()


# ── TestPluginContext ──────────────────────────────────────────────────────


class TestPluginContext:
    """Tests for the PluginContext facade."""

    def test_register_tool_adds_to_registry(self, tmp_path, monkeypatch):
        """PluginContext.register_tool() puts the tool in the global registry."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        plugin_dir = plugins_dir / "tool_plugin"
        plugin_dir.mkdir(parents=True)
        (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "tool_plugin"}))
        (plugin_dir / "__init__.py").write_text(
            'def register(ctx):\n'
            '    ctx.register_tool(\n'
            '        name="plugin_echo",\n'
            '        toolset="plugin_tool_plugin",\n'
            '        schema={"name": "plugin_echo", "description": "Echo", "parameters": {"type": "object", "properties": {}}},\n'
            '        handler=lambda args, **kw: "echo",\n'
            '    )\n'
        )
        hermes_home = tmp_path / "hermes_test"
        (hermes_home / "config.yaml").write_text(
            yaml.safe_dump({"plugins": {"enabled": ["tool_plugin"]}})
        )
        monkeypatch.setenv("HERMES_HOME", str(hermes_home))

        mgr = PluginManager()
        mgr.discover_and_load()

        assert "plugin_echo" in mgr._plugin_tool_names

        from tools.registry import registry
        assert "plugin_echo" in registry._tools

    def test_register_tool_rejects_shadow_without_override(self, tmp_path, monkeypatch, caplog):
        """Without override=True, registering a tool name claimed by a different toolset is rejected."""
        from tools.registry import registry

        # Seed an existing entry from a non-plugin toolset.
        registry.register(
            name="shadow_target",
            toolset="terminal",
            schema={"name": "shadow_target", "description": "Built-in", "parameters": {"type": "object", "properties": {}}},
            handler=lambda args, **kw: "built-in",
        )
        original_handler = registry._tools["shadow_target"].handler
        try:
            plugins_dir = tmp_path / "hermes_test" / "plugins"
            plugin_dir = plugins_dir / "shadow_plugin"
            plugin_dir.mkdir(parents=True)
            (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "shadow_plugin"}))
            (plugin_dir / "__init__.py").write_text(
                'def register(ctx):\n'
                '    ctx.register_tool(\n'
                '        name="shadow_target",\n'
                '        toolset="plugin_shadow_plugin",\n'
                '        schema={"name": "shadow_target", "description": "Plugin", "parameters": {"type": "object", "properties": {}}},\n'
                '        handler=lambda args, **kw: "plugin",\n'
                '    )\n'
            )
            hermes_home = tmp_path / "hermes_test"
            (hermes_home / "config.yaml").write_text(
                yaml.safe_dump({"plugins": {"enabled": ["shadow_plugin"]}})
            )
            monkeypatch.setenv("HERMES_HOME", str(hermes_home))

            with caplog.at_level(logging.ERROR, logger="tools.registry"):
                mgr = PluginManager()
                mgr.discover_and_load()

            # Original handler must still be in place — registration was rejected.
            assert registry._tools["shadow_target"].handler is original_handler
            assert registry._tools["shadow_target"].toolset == "terminal"
            # And an ERROR was logged explaining why and how to opt in.
            assert any("override=True" in r.message for r in caplog.records)
        finally:
            registry.deregister("shadow_target")

    def test_register_tool_override_replaces_existing(self, tmp_path, monkeypatch, caplog):
        """override=True lets a plugin replace an existing built-in tool."""
        from tools.registry import registry

        registry.register(
            name="override_target",
            toolset="terminal",
            schema={"name": "override_target", "description": "Built-in", "parameters": {"type": "object", "properties": {}}},
            handler=lambda args, **kw: "built-in",
        )
        try:
            plugins_dir = tmp_path / "hermes_test" / "plugins"
            plugin_dir = plugins_dir / "override_plugin"
            plugin_dir.mkdir(parents=True)
            (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "override_plugin"}))
            (plugin_dir / "__init__.py").write_text(
                'def register(ctx):\n'
                '    ctx.register_tool(\n'
                '        name="override_target",\n'
                '        toolset="plugin_override_plugin",\n'
                '        schema={"name": "override_target", "description": "Plugin", "parameters": {"type": "object", "properties": {}}},\n'
                '        handler=lambda args, **kw: "plugin",\n'
                '        override=True,\n'
                '    )\n'
            )
            hermes_home = tmp_path / "hermes_test"
            (hermes_home / "config.yaml").write_text(
                yaml.safe_dump({"plugins": {"enabled": ["override_plugin"]}})
            )
            monkeypatch.setenv("HERMES_HOME", str(hermes_home))

            with caplog.at_level(logging.INFO, logger="tools.registry"):
                mgr = PluginManager()
                mgr.discover_and_load()

            # Plugin handler replaced the built-in one.
            assert registry._tools["override_target"].toolset == "plugin_override_plugin"
            assert registry._tools["override_target"].handler({}, ) == "plugin"
            # Override is audit-logged at INFO.
            assert any(
                "overriding existing" in r.message and "override_target" in r.message
                for r in caplog.records
            )
            # Plugin tracks it.
            assert "override_target" in mgr._plugin_tool_names
        finally:
            registry.deregister("override_target")

    def test_register_tool_override_on_new_name_is_noop_path(self, tmp_path, monkeypatch):
        """override=True on a brand-new name still registers cleanly (no existing entry to replace)."""
        from tools.registry import registry

        plugins_dir = tmp_path / "hermes_test" / "plugins"
        plugin_dir = plugins_dir / "new_override_plugin"
        plugin_dir.mkdir(parents=True)
        (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "new_override_plugin"}))
        (plugin_dir / "__init__.py").write_text(
            'def register(ctx):\n'
            '    ctx.register_tool(\n'
            '        name="brand_new_override_tool",\n'
            '        toolset="plugin_new_override_plugin",\n'
            '        schema={"name": "brand_new_override_tool", "description": "New", "parameters": {"type": "object", "properties": {}}},\n'
            '        handler=lambda args, **kw: "ok",\n'
            '        override=True,\n'
            '    )\n'
        )
        hermes_home = tmp_path / "hermes_test"
        (hermes_home / "config.yaml").write_text(
            yaml.safe_dump({"plugins": {"enabled": ["new_override_plugin"]}})
        )
        monkeypatch.setenv("HERMES_HOME", str(hermes_home))

        try:
            mgr = PluginManager()
            mgr.discover_and_load()
            assert "brand_new_override_tool" in registry._tools
        finally:
            registry.deregister("brand_new_override_tool")


# ── TestPluginToolVisibility ───────────────────────────────────────────────


class TestPluginToolVisibility:
    """Plugin-registered tools appear in get_tool_definitions()."""

    def test_plugin_tools_in_definitions(self, tmp_path, monkeypatch):
        """Plugin tools are included when their toolset is in enabled_toolsets."""
        import hermes_cli.plugins as plugins_mod

        plugins_dir = tmp_path / "hermes_test" / "plugins"
        plugin_dir = plugins_dir / "vis_plugin"
        plugin_dir.mkdir(parents=True)
        (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "vis_plugin"}))
        (plugin_dir / "__init__.py").write_text(
            'def register(ctx):\n'
            '    ctx.register_tool(\n'
            '        name="vis_tool",\n'
            '        toolset="plugin_vis_plugin",\n'
            '        schema={"name": "vis_tool", "description": "Visible", "parameters": {"type": "object", "properties": {}}},\n'
            '        handler=lambda args, **kw: "ok",\n'
            '    )\n'
        )
        hermes_home = tmp_path / "hermes_test"
        (hermes_home / "config.yaml").write_text(
            yaml.safe_dump({"plugins": {"enabled": ["vis_plugin"]}})
        )
        monkeypatch.setenv("HERMES_HOME", str(hermes_home))

        mgr = PluginManager()
        mgr.discover_and_load()
        monkeypatch.setattr(plugins_mod, "_plugin_manager", mgr)

        from model_tools import get_tool_definitions

        # Plugin tools are included when their toolset is explicitly enabled
        tools = get_tool_definitions(enabled_toolsets=["terminal", "plugin_vis_plugin"], quiet_mode=True)
        tool_names = [t["function"]["name"] for t in tools]
        assert "vis_tool" in tool_names

        # Plugin tools are excluded when only other toolsets are enabled
        tools2 = get_tool_definitions(enabled_toolsets=["terminal"], quiet_mode=True)
        tool_names2 = [t["function"]["name"] for t in tools2]
        assert "vis_tool" not in tool_names2

        # Plugin tools are included when no toolset filter is active (all enabled)
        tools3 = get_tool_definitions(quiet_mode=True)
        tool_names3 = [t["function"]["name"] for t in tools3]
        assert "vis_tool" in tool_names3


# ── TestPluginManagerList ──────────────────────────────────────────────────


class TestPluginManagerList:
    """Tests for PluginManager.list_plugins()."""

    def test_list_empty(self):
        """Empty manager returns empty list."""
        mgr = PluginManager()
        assert mgr.list_plugins() == []

    def test_list_returns_sorted(self, tmp_path, monkeypatch):
        """list_plugins() returns results sorted by key."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(plugins_dir, "zulu")
        _make_plugin_dir(plugins_dir, "alpha")
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        listing = mgr.list_plugins()
        # list_plugins sorts by key (path-derived, e.g. ``image_gen/openai``),
        # not by display name, so that category plugins group together.
        keys = [p["key"] for p in listing]
        assert keys == sorted(keys)

    def test_list_with_plugins(self, tmp_path, monkeypatch):
        """list_plugins() returns info dicts for each discovered plugin."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(plugins_dir, "alpha")
        _make_plugin_dir(plugins_dir, "beta")
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        listing = mgr.list_plugins()
        names = [p["name"] for p in listing]
        assert "alpha" in names
        assert "beta" in names
        for p in listing:
            assert "enabled" in p
            assert "tools" in p
            assert "hooks" in p

    def test_shared_hook_name_credited_to_every_plugin(self, tmp_path, monkeypatch):
        """Two plugins registering the SAME hook name are each credited.

        Regression: hook/middleware/tool attribution diffed names against all
        already-loaded plugins, so when a later plugin registered a hook name
        an earlier plugin had already used, the shared name was attributed to
        the first plugin only and the later plugin reported 0 hooks in
        `hermes plugins list`. Attribution now counts what each plugin's own
        register() added (per-registration delta), so both get credit.
        """
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir, "first_hooker",
            register_body='ctx.register_hook("post_tool_call", lambda **kw: None)',
        )
        _make_plugin_dir(
            plugins_dir, "second_hooker",
            register_body='ctx.register_hook("post_tool_call", lambda **kw: None)',
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        by_name = {p["name"]: p for p in mgr.list_plugins()}
        assert by_name["first_hooker"]["hooks"] == 1
        assert by_name["second_hooker"]["hooks"] == 1, (
            "second plugin sharing a hook name was not credited with its hook"
        )



class TestPreLlmCallTargetRouting:
    """Tests for pre_llm_call hook return format with target-aware routing.

    The routing logic lives in run_agent.py, but the return format is collected
    by invoke_hook(). These tests verify the return format works correctly and
    that downstream code can route based on the 'target' key.
    """

    def _make_pre_llm_plugin(self, plugins_dir, name, return_expr):
        """Create a plugin that returns a specific value from pre_llm_call."""
        _make_plugin_dir(
            plugins_dir, name,
            register_body=(
                f'ctx.register_hook("pre_llm_call", lambda **kw: {return_expr})'
            ),
        )

    def test_context_dict_returned(self, tmp_path, monkeypatch):
        """Plugin returning a context dict is collected by invoke_hook."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        self._make_pre_llm_plugin(
            plugins_dir, "basic_plugin",
            '{"context": "basic context"}',
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        results = mgr.invoke_hook(
            "pre_llm_call", session_id="s1", user_message="hi",
            conversation_history=[], is_first_turn=True, model="test",
        )
        assert len(results) == 1
        assert results[0]["context"] == "basic context"
        assert "target" not in results[0]

    def test_plain_string_return(self, tmp_path, monkeypatch):
        """Plain string returns are collected as-is (routing treats them as user_message)."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        self._make_pre_llm_plugin(
            plugins_dir, "str_plugin",
            '"plain string context"',
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        results = mgr.invoke_hook(
            "pre_llm_call", session_id="s1", user_message="hi",
            conversation_history=[], is_first_turn=True, model="test",
        )
        assert len(results) == 1
        assert results[0] == "plain string context"

    def test_multiple_plugins_context_collected(self, tmp_path, monkeypatch):
        """Multiple plugins returning context are all collected."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        self._make_pre_llm_plugin(
            plugins_dir, "aaa_memory",
            '{"context": "memory context"}',
        )
        self._make_pre_llm_plugin(
            plugins_dir, "bbb_guardrail",
            '{"context": "guardrail text"}',
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        results = mgr.invoke_hook(
            "pre_llm_call", session_id="s1", user_message="hi",
            conversation_history=[], is_first_turn=True, model="test",
        )
        assert len(results) == 2
        contexts = [r["context"] for r in results]
        assert "memory context" in contexts
        assert "guardrail text" in contexts

    def test_routing_logic_all_to_user_message(self, tmp_path, monkeypatch):
        """Simulate the routing logic from run_agent.py.

        All plugin context — dicts and plain strings — ends up in a single
        user message context string. There is no system_prompt target.
        """
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        self._make_pre_llm_plugin(
            plugins_dir, "aaa_mem",
            '{"context": "memory A"}',
        )
        self._make_pre_llm_plugin(
            plugins_dir, "bbb_guard",
            '{"context": "rule B"}',
        )
        self._make_pre_llm_plugin(
            plugins_dir, "ccc_plain",
            '"plain text C"',
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        results = mgr.invoke_hook(
            "pre_llm_call", session_id="s1", user_message="hi",
            conversation_history=[], is_first_turn=True, model="test",
        )

        # Replicate run_agent.py routing logic — everything goes to user msg
        _ctx_parts = []
        for r in results:
            if isinstance(r, dict) and r.get("context"):
                _ctx_parts.append(str(r["context"]))
            elif isinstance(r, str) and r.strip():
                _ctx_parts.append(r)

        assert _ctx_parts == ["memory A", "rule B", "plain text C"]
        _plugin_user_context = "\n\n".join(_ctx_parts)
        assert "memory A" in _plugin_user_context
        assert "rule B" in _plugin_user_context
        assert "plain text C" in _plugin_user_context


# ── TestPluginCommands ────────────────────────────────────────────────────


class TestPluginCommands:
    """Tests for plugin slash command registration via register_command()."""

    def test_register_command_basic(self):
        """register_command() stores handler, description, and plugin name."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        handler = lambda args: f"echo {args}"
        ctx.register_command("mycmd", handler, description="My custom command")

        assert "mycmd" in mgr._plugin_commands
        entry = mgr._plugin_commands["mycmd"]
        assert entry["handler"] is handler
        assert entry["description"] == "My custom command"
        assert entry["plugin"] == "test-plugin"
        # args_hint defaults to empty string when not passed.
        assert entry["args_hint"] == ""

    def test_register_command_with_args_hint(self):
        """args_hint is stored and surfaced for gateway-native UI registration."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        ctx.register_command(
            "metricas",
            lambda a: a,
            description="Metrics dashboard",
            args_hint="dias:7 formato:json",
        )

        entry = mgr._plugin_commands["metricas"]
        assert entry["args_hint"] == "dias:7 formato:json"

    def test_register_command_args_hint_whitespace_trimmed(self):
        """args_hint leading/trailing whitespace is stripped."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        ctx.register_command("foo", lambda a: a, args_hint="  <file>  ")
        assert mgr._plugin_commands["foo"]["args_hint"] == "<file>"

    def test_register_command_normalizes_name(self):
        """Names are lowercased, stripped, and leading slashes removed."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        ctx.register_command("/MyCmd ", lambda a: a, description="test")
        assert "mycmd" in mgr._plugin_commands
        assert "/MyCmd " not in mgr._plugin_commands

    def test_register_command_empty_name_rejected(self, caplog):
        """Empty name after normalization is rejected with a warning."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"):
            ctx.register_command("", lambda a: a)
        assert len(mgr._plugin_commands) == 0
        assert "empty name" in caplog.text

    def test_register_command_builtin_conflict_rejected(self, caplog):
        """Commands that conflict with built-in names are rejected."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"):
            ctx.register_command("help", lambda a: a)
        assert "help" not in mgr._plugin_commands
        assert "conflicts" in caplog.text.lower()

    def test_register_command_default_description(self):
        """Missing description defaults to 'Plugin command'."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        ctx.register_command("status-cmd", lambda a: a)
        assert mgr._plugin_commands["status-cmd"]["description"] == "Plugin command"

    def test_get_plugin_command_handler_found(self):
        """get_plugin_command_handler() returns the handler for a registered command."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        handler = lambda args: f"result: {args}"
        ctx.register_command("mycmd", handler, description="test")

        with patch("hermes_cli.plugins._plugin_manager", mgr):
            result = get_plugin_command_handler("mycmd")
            assert result is handler

    def test_get_plugin_command_handler_not_found(self):
        """get_plugin_command_handler() returns None for unregistered commands."""
        mgr = PluginManager()
        with patch("hermes_cli.plugins._plugin_manager", mgr):
            assert get_plugin_command_handler("nonexistent") is None

    def test_get_plugin_commands_returns_dict(self):
        """get_plugin_commands() returns the full commands dict."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)
        ctx.register_command("cmd-a", lambda a: a, description="A")
        ctx.register_command("cmd-b", lambda a: a, description="B")

        with patch("hermes_cli.plugins._plugin_manager", mgr):
            cmds = get_plugin_commands()
            assert "cmd-a" in cmds
            assert "cmd-b" in cmds
            assert cmds["cmd-a"]["description"] == "A"

    def test_get_plugin_command_handler_discovers_plugins_lazily(self, tmp_path, monkeypatch):
        """Handler lookup should work before any explicit discover_plugins() call."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir,
            "cmd-plugin",
            register_body='ctx.register_command("lazycmd", lambda a: f"ok:{a}", description="Lazy")',
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        import hermes_cli.plugins as plugins_mod

        with patch.object(plugins_mod, "_plugin_manager", None):
            handler = get_plugin_command_handler("lazycmd")
            assert handler is not None
            assert handler("x") == "ok:x"

    def test_get_plugin_commands_discovers_plugins_lazily(self, tmp_path, monkeypatch):
        """Command listing should trigger plugin discovery on first access."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir,
            "cmd-plugin",
            register_body='ctx.register_command("lazycmd", lambda a: a, description="Lazy")',
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        import hermes_cli.plugins as plugins_mod

        with patch.object(plugins_mod, "_plugin_manager", None):
            cmds = get_plugin_commands()
            assert "lazycmd" in cmds
            assert cmds["lazycmd"]["description"] == "Lazy"

    def test_get_plugin_context_engine_discovers_plugins_lazily(self, tmp_path, monkeypatch):
        """Context engine lookup should work before any explicit discover_plugins() call."""
        hermes_home = tmp_path / "hermes_test"
        plugins_dir = hermes_home / "plugins"
        plugin_dir = plugins_dir / "engine-plugin"
        plugin_dir.mkdir(parents=True, exist_ok=True)
        (plugin_dir / "plugin.yaml").write_text(
            yaml.dump({
                "name": "engine-plugin",
                "version": "0.1.0",
                "description": "Test engine plugin",
            })
        )
        (plugin_dir / "__init__.py").write_text(
            "from agent.context_engine import ContextEngine\n\n"
            "class StubEngine(ContextEngine):\n"
            "    @property\n"
            "    def name(self):\n"
            "        return 'stub-engine'\n\n"
            "    def update_from_response(self, usage):\n"
            "        return None\n\n"
            "    def should_compress(self, prompt_tokens):\n"
            "        return False\n\n"
            "    def compress(self, messages, current_tokens):\n"
            "        return messages\n\n"
            "def register(ctx):\n"
            "    ctx.register_context_engine(StubEngine())\n"
        )
        # Opt-in: plugins are opt-in by default, so enable in config.yaml
        (hermes_home / "config.yaml").write_text(
            yaml.safe_dump({"plugins": {"enabled": ["engine-plugin"]}})
        )
        monkeypatch.setenv("HERMES_HOME", str(hermes_home))

        import hermes_cli.plugins as plugins_mod

        with patch.object(plugins_mod, "_plugin_manager", None):
            engine = plugins_mod.get_plugin_context_engine()
            assert engine is not None
            assert engine.name == "stub-engine"

    def test_commands_tracked_on_loaded_plugin(self, tmp_path, monkeypatch):
        """Commands registered during discover_and_load() are tracked on LoadedPlugin."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        _make_plugin_dir(
            plugins_dir, "cmd-plugin",
            register_body=(
                'ctx.register_command("mycmd", lambda a: "ok", description="Test")'
            ),
        )
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))

        mgr = PluginManager()
        mgr.discover_and_load()

        loaded = mgr._plugins["cmd-plugin"]
        assert loaded.enabled
        assert "mycmd" in loaded.commands_registered

    def test_commands_in_list_plugins_output(self, tmp_path, monkeypatch):
        """list_plugins() includes command count."""
        plugins_dir = tmp_path / "hermes_test" / "plugins"
        # Set HERMES_HOME BEFORE _make_plugin_dir so auto-enable targets
        # the right config.yaml.
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
        _make_plugin_dir(
            plugins_dir, "cmd-plugin",
            register_body=(
                'ctx.register_command("mycmd", lambda a: "ok", description="Test")'
            ),
        )

        mgr = PluginManager()
        mgr.discover_and_load()

        info = mgr.list_plugins()
        # Filter out bundled plugins — they're always discovered.
        cmd_info = [p for p in info if p["name"] == "cmd-plugin"]
        assert len(cmd_info) == 1
        assert cmd_info[0]["commands"] == 1

    def test_handler_receives_raw_args(self):
        """The handler is called with the raw argument string."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        received = []
        ctx.register_command("echo", lambda args: received.append(args) or "ok")

        handler = mgr._plugin_commands["echo"]["handler"]
        handler("hello world")
        assert received == ["hello world"]

    def test_multiple_plugins_register_different_commands(self):
        """Multiple plugins can each register their own commands."""
        mgr = PluginManager()

        for plugin_name, cmd_name in [("plugin-a", "cmd-a"), ("plugin-b", "cmd-b")]:
            manifest = PluginManifest(name=plugin_name, source="user")
            ctx = PluginContext(manifest, mgr)
            ctx.register_command(cmd_name, lambda a: a, description=f"From {plugin_name}")

        assert "cmd-a" in mgr._plugin_commands
        assert "cmd-b" in mgr._plugin_commands
        assert mgr._plugin_commands["cmd-a"]["plugin"] == "plugin-a"
        assert mgr._plugin_commands["cmd-b"]["plugin"] == "plugin-b"


class TestPluginCommandResultResolution:
    def test_returns_sync_values_unchanged(self):
        assert resolve_plugin_command_result("ok") == "ok"

    def test_awaits_async_result_without_running_loop(self):
        async def _handler():
            return "async-ok"

        assert resolve_plugin_command_result(_handler()) == "async-ok"

    def test_awaits_async_result_with_running_loop(self, monkeypatch):
        class _Loop:
            pass

        async def _handler():
            return "threaded-ok"

        monkeypatch.setattr("hermes_cli.plugins.asyncio.get_running_loop", lambda: _Loop())
        assert resolve_plugin_command_result(_handler()) == "threaded-ok"

    def test_running_loop_timeout_does_not_hang_forever(self, monkeypatch):
        """Threaded path must abort a hung async handler instead of blocking the caller."""
        import asyncio as _asyncio

        class _Loop:
            pass

        async def _slow_handler():
            await _asyncio.sleep(10)
            return "should-not-reach"

        monkeypatch.setattr("hermes_cli.plugins.asyncio.get_running_loop", lambda: _Loop())
        monkeypatch.setattr("hermes_cli.plugins._PLUGIN_COMMAND_AWAIT_TIMEOUT_SECS", 0.1)

        with pytest.raises(TimeoutError):
            resolve_plugin_command_result(_slow_handler())


# ── TestPluginDispatchTool ────────────────────────────────────────────────


class TestPluginDispatchTool:
    """Tests for PluginContext.dispatch_tool() — tool dispatch with agent context."""

    def test_dispatch_tool_calls_registry(self):
        """dispatch_tool() delegates to registry.dispatch()."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        mock_registry = MagicMock()
        mock_registry.dispatch.return_value = '{"result": "ok"}'

        with patch("hermes_cli.plugins.PluginContext.dispatch_tool.__module__", "hermes_cli.plugins"):
            with patch.dict("sys.modules", {}):
                with patch("tools.registry.registry", mock_registry):
                    result = ctx.dispatch_tool("web_search", {"query": "test"})

        assert result == '{"result": "ok"}'

    def test_dispatch_tool_injects_parent_agent_from_cli_ref(self):
        """When _cli_ref has an agent, it's passed as parent_agent."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        mock_agent = MagicMock()
        mock_cli = MagicMock()
        mock_cli.agent = mock_agent
        mgr._cli_ref = mock_cli

        mock_registry = MagicMock()
        mock_registry.dispatch.return_value = '{"ok": true}'

        with patch("tools.registry.registry", mock_registry):
            ctx.dispatch_tool("delegate_task", {"goal": "test"})

        mock_registry.dispatch.assert_called_once()
        call_kwargs = mock_registry.dispatch.call_args
        assert call_kwargs[1].get("parent_agent") is mock_agent

    def test_dispatch_tool_no_parent_agent_when_no_cli_ref(self):
        """When _cli_ref is None (gateway mode), no parent_agent is injected."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)
        mgr._cli_ref = None

        mock_registry = MagicMock()
        mock_registry.dispatch.return_value = '{"ok": true}'

        with patch("tools.registry.registry", mock_registry):
            ctx.dispatch_tool("delegate_task", {"goal": "test"})

        call_kwargs = mock_registry.dispatch.call_args
        assert "parent_agent" not in call_kwargs[1]

    def test_dispatch_tool_no_parent_agent_when_agent_is_none(self):
        """When cli_ref exists but agent is None (not yet initialized), skip parent_agent."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        mock_cli = MagicMock()
        mock_cli.agent = None
        mgr._cli_ref = mock_cli

        mock_registry = MagicMock()
        mock_registry.dispatch.return_value = '{"ok": true}'

        with patch("tools.registry.registry", mock_registry):
            ctx.dispatch_tool("delegate_task", {"goal": "test"})

        call_kwargs = mock_registry.dispatch.call_args
        assert "parent_agent" not in call_kwargs[1]

    def test_dispatch_tool_respects_explicit_parent_agent(self):
        """Explicit parent_agent kwarg is not overwritten by _cli_ref.agent."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)

        cli_agent = MagicMock(name="cli_agent")
        mock_cli = MagicMock()
        mock_cli.agent = cli_agent
        mgr._cli_ref = mock_cli

        explicit_agent = MagicMock(name="explicit_agent")

        mock_registry = MagicMock()
        mock_registry.dispatch.return_value = '{"ok": true}'

        with patch("tools.registry.registry", mock_registry):
            ctx.dispatch_tool("delegate_task", {"goal": "test"}, parent_agent=explicit_agent)

        call_kwargs = mock_registry.dispatch.call_args
        assert call_kwargs[1]["parent_agent"] is explicit_agent

    def test_dispatch_tool_forwards_extra_kwargs(self):
        """Extra kwargs are forwarded to registry.dispatch()."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)
        mgr._cli_ref = None

        mock_registry = MagicMock()
        mock_registry.dispatch.return_value = '{"ok": true}'

        with patch("tools.registry.registry", mock_registry):
            ctx.dispatch_tool("some_tool", {"x": 1}, task_id="test-123")

        call_kwargs = mock_registry.dispatch.call_args
        assert call_kwargs[1]["task_id"] == "test-123"

    def test_dispatch_tool_returns_json_string(self):
        """dispatch_tool() returns the raw JSON string from the registry."""
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        ctx = PluginContext(manifest, mgr)
        mgr._cli_ref = None

        mock_registry = MagicMock()
        mock_registry.dispatch.return_value = '{"error": "Unknown tool: fake"}'

        with patch("tools.registry.registry", mock_registry):
            result = ctx.dispatch_tool("fake", {})

        assert '"error"' in result


class TestPluginDebugLogging:
    """HERMES_PLUGINS_DEBUG opt-in stderr handler for plugin developers."""

    def test_debug_handler_not_installed_when_env_var_absent(self, monkeypatch):
        """Without the env var, no stderr handler is attached."""
        monkeypatch.delenv("HERMES_PLUGINS_DEBUG", raising=False)
        from hermes_cli import plugins as plugins_mod

        # Snapshot, then force a re-evaluation.
        original_installed = plugins_mod._DEBUG_HANDLER_INSTALLED
        original_debug = plugins_mod._PLUGINS_DEBUG
        original_handlers = list(plugins_mod.logger.handlers)
        try:
            plugins_mod._DEBUG_HANDLER_INSTALLED = False
            plugins_mod._install_plugin_debug_handler(force=True)
            assert plugins_mod._PLUGINS_DEBUG is False
            assert plugins_mod._DEBUG_HANDLER_INSTALLED is False
            # No new stderr handler was attached.
            assert plugins_mod.logger.handlers == original_handlers
        finally:
            plugins_mod._DEBUG_HANDLER_INSTALLED = original_installed
            plugins_mod._PLUGINS_DEBUG = original_debug
            plugins_mod.logger.handlers = original_handlers

    def test_debug_handler_installed_when_env_var_set(self, monkeypatch):
        """With HERMES_PLUGINS_DEBUG=1, a DEBUG-level stderr handler is attached."""
        monkeypatch.setenv("HERMES_PLUGINS_DEBUG", "1")
        from hermes_cli import plugins as plugins_mod

        original_installed = plugins_mod._DEBUG_HANDLER_INSTALLED
        original_debug = plugins_mod._PLUGINS_DEBUG
        original_level = plugins_mod.logger.level
        original_handlers = list(plugins_mod.logger.handlers)
        try:
            plugins_mod._DEBUG_HANDLER_INSTALLED = False
            plugins_mod._install_plugin_debug_handler(force=True)
            assert plugins_mod._PLUGINS_DEBUG is True
            assert plugins_mod._DEBUG_HANDLER_INSTALLED is True
            assert plugins_mod.logger.level == logging.DEBUG
            new_handlers = [
                h for h in plugins_mod.logger.handlers if h not in original_handlers
            ]
            assert len(new_handlers) == 1
            assert isinstance(new_handlers[0], logging.StreamHandler)
            assert new_handlers[0].level == logging.DEBUG
        finally:
            plugins_mod._DEBUG_HANDLER_INSTALLED = original_installed
            plugins_mod._PLUGINS_DEBUG = original_debug
            plugins_mod.logger.setLevel(original_level)
            plugins_mod.logger.handlers = original_handlers

    def test_debug_handler_idempotent(self, monkeypatch):
        """Calling install twice (without force) does not double-attach."""
        monkeypatch.setenv("HERMES_PLUGINS_DEBUG", "1")
        from hermes_cli import plugins as plugins_mod

        original_installed = plugins_mod._DEBUG_HANDLER_INSTALLED
        original_debug = plugins_mod._PLUGINS_DEBUG
        original_level = plugins_mod.logger.level
        original_handlers = list(plugins_mod.logger.handlers)
        try:
            plugins_mod._DEBUG_HANDLER_INSTALLED = False
            plugins_mod._install_plugin_debug_handler(force=True)
            count_after_first = len(plugins_mod.logger.handlers)
            plugins_mod._install_plugin_debug_handler()  # no force
            count_after_second = len(plugins_mod.logger.handlers)
            assert count_after_first == count_after_second
        finally:
            plugins_mod._DEBUG_HANDLER_INSTALLED = original_installed
            plugins_mod._PLUGINS_DEBUG = original_debug
            plugins_mod.logger.setLevel(original_level)
            plugins_mod.logger.handlers = original_handlers


class TestPluginContextProfileName:
    """ctx.profile_name resolves from HERMES_HOME in every context."""

    def _ctx(self):
        mgr = PluginManager()
        manifest = PluginManifest(name="test-plugin", source="user")
        return PluginContext(manifest, mgr)

    def test_default_profile(self, tmp_path, monkeypatch):
        """HERMES_HOME at the root resolves to 'default'."""
        home = tmp_path / ".hermes"
        home.mkdir()
        monkeypatch.setattr(Path, "home", lambda: tmp_path)
        monkeypatch.setenv("HERMES_HOME", str(home))
        assert self._ctx().profile_name == "default"

    def test_named_profile(self, tmp_path, monkeypatch):
        """HERMES_HOME under profiles/<name> resolves to that name."""
        prof = tmp_path / ".hermes" / "profiles" / "coder"
        prof.mkdir(parents=True)
        monkeypatch.setattr(Path, "home", lambda: tmp_path)
        monkeypatch.setenv("HERMES_HOME", str(prof))
        assert self._ctx().profile_name == "coder"

    def test_works_without_cli_ref(self, tmp_path, monkeypatch):
        """profile_name does not depend on _cli_ref (None in worker sessions)."""
        prof = tmp_path / ".hermes" / "profiles" / "worker1"
        prof.mkdir(parents=True)
        monkeypatch.setattr(Path, "home", lambda: tmp_path)
        monkeypatch.setenv("HERMES_HOME", str(prof))
        ctx = self._ctx()
        assert ctx._manager._cli_ref is None
        assert ctx.profile_name == "worker1"


class TestDispatchToolWithoutCliRef:
    """ctx.dispatch_tool works in worker/hook contexts (no _cli_ref).

    This pins the contract the plugin docs rely on: a plugin can drive
    tools from a hook callback even when running in the gateway or a
    kanban-spawned worker session, where _cli_ref is None.
    """

    def test_dispatch_tool_invokes_handler_without_cli_ref(self):
        from tools.registry import registry

        mgr = PluginManager()
        assert mgr._cli_ref is None  # worker/hook context
        ctx = PluginContext(PluginManifest(name="test-plugin", source="user"), mgr)

        calls = []
        registry.register(
            name="_test_dispatch_probe",
            toolset="debugging",
            schema={"name": "_test_dispatch_probe", "description": "probe",
                    "parameters": {"type": "object", "properties": {}}},
            handler=lambda args, **kw: calls.append((args, kw)) or '{"ok": true}',
        )
        try:
            result = ctx.dispatch_tool("_test_dispatch_probe", {"x": 1})
            assert result == '{"ok": true}'
            assert calls and calls[0][0] == {"x": 1}
            # parent_agent is not forced when there's no CLI agent to resolve.
            assert calls[0][1].get("parent_agent") is None
        finally:
            registry.deregister("_test_dispatch_probe")
