"""Tests for the inter-plugin event bus (PluginContext.emit / subscribe).

Covers:
  - Two plugins communicate via emit/subscribe; emit returns listener count
  - Namespace is FORCED to the emitting plugin's own key
  - Namespace spoofing (hermes:, foreign, already-colon'd) is rejected
  - Non-blocking bounded delivery for synchronous subscribers
  - Per-callback isolation and deep-copied payload ownership
  - Async subscribers resolved through the loop-safe host path
  - Owner unload / generation reset cancel zombie callbacks
  - Recursion cap: mutually-emitting plugins terminate + warn
  - Manifest emits/listens parsed as optional advisory fields
  - `hermes plugins show` output includes emits/listens
"""

from __future__ import annotations

import asyncio
import logging
import threading

import pytest

from hermes_cli.plugins import (
    _EVENT_EMIT_DEPTH_CAP,
    PluginContext,
    PluginManager,
    PluginManifest,
)


# ── Fixtures ─────────────────────────────────────────────────────────────────


def _make_ctx(manager: PluginManager, name: str, key: str = "") -> PluginContext:
    """Build a PluginContext for *name* wired to *manager*."""
    manifest = PluginManifest(name=name, key=key)
    return PluginContext(manifest, manager)


def _fresh_manager() -> PluginManager:
    manager = PluginManager()
    manager._discovered = True  # skip auto-discovery
    return manager


def _drain(manager: PluginManager) -> None:
    assert manager._wait_for_event_dispatch(timeout=2.0)


# ── 1. Two plugins communicate ───────────────────────────────────────────────


def test_two_plugins_communicate():
    manager = _fresh_manager()
    ctx_a = _make_ctx(manager, "plugin_a", key="a")
    ctx_b = _make_ctx(manager, "plugin_b", key="b")

    received = []

    def on_ping(**payload):
        received.append(payload)

    # A subscribes to b:ping; B emits the bare name "ping".
    ctx_a.subscribe("b:ping", on_ping)
    count = ctx_b.emit("ping", {"n": 42})
    _drain(manager)

    assert count == 1  # one listener invoked
    assert received == [{"n": 42}]


def test_emit_none_payload_delivers_empty_kwargs():
    manager = _fresh_manager()
    ctx_a = _make_ctx(manager, "plugin_a", key="a")
    ctx_b = _make_ctx(manager, "plugin_b", key="b")

    seen = []
    ctx_a.subscribe("b:ping", lambda **p: seen.append(p))
    count = ctx_b.emit("ping")  # payload omitted
    _drain(manager)

    assert count == 1
    assert seen == [{}]


# ── 2. Namespace is forced to the emitter's own key ──────────────────────────


def test_namespace_forced_to_emitter_key():
    manager = _fresh_manager()
    ctx_a = _make_ctx(manager, "plugin_a", key="a")
    ctx_b = _make_ctx(manager, "plugin_b", key="b")

    delivered_events = []

    # Subscribe to what we expect the fully-qualified name to be.
    ctx_a.subscribe("b:ping", lambda **p: delivered_events.append("b:ping"))
    # A wrong-namespace subscription must NOT fire.
    ctx_a.subscribe("a:ping", lambda **p: delivered_events.append("a:ping"))

    ctx_b.emit("ping")
    _drain(manager)

    # Delivered under the emitter's own key ("b"), never "a".
    assert delivered_events == ["b:ping"]


def test_namespace_falls_back_to_name_when_key_empty():
    manager = _fresh_manager()
    # No key → namespace derives from name.
    ctx = _make_ctx(manager, "plugin_named", key="")
    got = []
    ctx.subscribe("plugin_named:evt", lambda **p: got.append(p))
    count = ctx.emit("evt", {"v": 1})
    _drain(manager)
    assert count == 1
    assert got == [{"v": 1}]


# ── 3. Namespace spoofing is rejected (fail-closed) ──────────────────────────


@pytest.mark.parametrize(
    "bad_event",
    [
        "hermes:x",   # reserved core prefix
        "a:x",        # foreign namespace
        "b:x",        # even the plugin's own colon'd name — must pass bare only
        "other:evt",
        ":x",
        "x:",
    ],
)
def test_emit_rejects_namespaced_names(bad_event):
    manager = _fresh_manager()
    ctx_b = _make_ctx(manager, "plugin_b", key="b")

    fired = []
    # Subscribe to every plausible delivery target so we can prove no delivery.
    for name in (bad_event, f"b:{bad_event}", "hermes:x", "a:x", "b:x"):
        ctx_b.subscribe(name, lambda **p: fired.append(name))

    with pytest.raises(ValueError):
        ctx_b.emit(bad_event)

    assert fired == []  # nothing delivered


def test_emit_rejects_empty_event():
    manager = _fresh_manager()
    ctx_b = _make_ctx(manager, "plugin_b", key="b")
    with pytest.raises(ValueError):
        ctx_b.emit("")


# ── 4. Per-callback isolation ────────────────────────────────────────────────


def test_per_callback_isolation(caplog):
    manager = _fresh_manager()
    ctx_a = _make_ctx(manager, "plugin_a", key="a")
    ctx_b = _make_ctx(manager, "plugin_b", key="b")

    received = []

    def boom(**payload):
        raise RuntimeError("subscriber exploded")

    def good(**payload):
        received.append(payload)

    # Registration order: raising subscriber first, healthy one second.
    ctx_a.subscribe("b:ping", boom)
    ctx_a.subscribe("b:ping", good)

    with caplog.at_level(logging.WARNING):
        count = ctx_b.emit("ping", {"ok": True})
        _drain(manager)

    # Both listeners were invoked despite the first raising.
    assert count == 2
    assert received == [{"ok": True}]
    assert any("subscriber exploded" in r.message or "raised" in r.message
               for r in caplog.records)


def test_emit_returns_before_blocking_subscriber_finishes():
    manager = _fresh_manager()
    ctx_a = _make_ctx(manager, "plugin_a", key="a")
    ctx_b = _make_ctx(manager, "plugin_b", key="b")
    entered = threading.Event()
    release = threading.Event()
    emit_returned = threading.Event()
    result = {}

    def blocking(**payload):
        entered.set()
        release.wait(timeout=2.0)

    def call_emit():
        result["count"] = ctx_b.emit("ping")
        emit_returned.set()

    ctx_a.subscribe("b:ping", blocking)
    emitter = threading.Thread(target=call_emit)
    emitter.start()
    try:
        assert emit_returned.wait(timeout=1.0)
        assert entered.wait(timeout=1.0)
    finally:
        release.set()
        emitter.join(timeout=2.0)
    _drain(manager)
    assert result["count"] == 1


def test_pending_budget_drops_new_event_without_blocking(monkeypatch, caplog):
    from hermes_cli import plugins_dispatch

    monkeypatch.setattr(plugins_dispatch, "_EVENT_PENDING_CAP", 1)
    manager = _fresh_manager()
    ctx_a = _make_ctx(manager, "plugin_a", key="a")
    ctx_b = _make_ctx(manager, "plugin_b", key="b")
    entered = threading.Event()
    release = threading.Event()

    def blocking(**payload):
        entered.set()
        release.wait(timeout=2.0)

    ctx_a.subscribe("b:ping", blocking)
    assert ctx_b.emit("ping") == 1
    assert entered.wait(timeout=1.0)
    try:
        with caplog.at_level(logging.WARNING):
            assert ctx_b.emit("ping") == 0
    finally:
        release.set()
    _drain(manager)
    assert "pending budget" in caplog.text


def test_each_subscriber_receives_deep_copied_payload():
    manager = _fresh_manager()
    ctx_a = _make_ctx(manager, "plugin_a", key="a")
    ctx_b = _make_ctx(manager, "plugin_b", key="b")
    original = {"nested": {"value": 1}}
    observed = []

    def mutate(**payload):
        payload["nested"]["value"] = 99

    def observe(**payload):
        observed.append(payload["nested"]["value"])

    ctx_a.subscribe("b:ping", mutate)
    ctx_a.subscribe("b:ping", observe)
    assert ctx_b.emit("ping", original) == 2
    _drain(manager)

    assert original == {"nested": {"value": 1}}
    assert observed == [1]


def test_async_subscriber_is_awaited():
    manager = _fresh_manager()
    ctx_a = _make_ctx(manager, "plugin_a", key="a")
    ctx_b = _make_ctx(manager, "plugin_b", key="b")
    observed = []

    async def on_ping(**payload):
        await asyncio.sleep(0)
        observed.append(payload["value"])

    ctx_a.subscribe("b:ping", on_ping)
    assert ctx_b.emit("ping", {"value": 7}) == 1
    _drain(manager)
    assert observed == [7]


def test_remove_plugin_subscriptions_cancels_owner_entries():
    manager = _fresh_manager()
    ctx_a = _make_ctx(manager, "plugin_a", key="a")
    ctx_b = _make_ctx(manager, "plugin_b", key="b")
    observed = []

    ctx_a.subscribe("b:ping", lambda **payload: observed.append(payload))
    manager._remove_plugin_subscriptions("a")

    assert ctx_b.emit("ping", {"value": 1}) == 0
    _drain(manager)
    assert observed == []
    assert "b:ping" not in manager._subscriptions


def test_owner_removal_cancels_callback_already_snapshotted_in_queue():
    manager = _fresh_manager()
    ctx_gate = _make_ctx(manager, "gate", key="gate")
    ctx_a = _make_ctx(manager, "plugin_a", key="a")
    ctx_b = _make_ctx(manager, "plugin_b", key="b")
    entered = threading.Event()
    release = threading.Event()
    observed = []

    def blocking(**payload):
        entered.set()
        release.wait(timeout=2.0)

    ctx_gate.subscribe("b:ping", blocking)
    ctx_a.subscribe("b:ping", lambda **payload: observed.append(payload))
    assert ctx_b.emit("ping", {"value": 1}) == 2
    assert entered.wait(timeout=1.0)
    manager._remove_plugin_subscriptions("a")
    release.set()
    _drain(manager)

    assert observed == []


# ── 5. Recursion cap ─────────────────────────────────────────────────────────


def test_recursion_cap_terminates(caplog):
    manager = _fresh_manager()
    ctx_a = _make_ctx(manager, "plugin_a", key="a")
    ctx_b = _make_ctx(manager, "plugin_b", key="b")

    calls = {"a": 0, "b": 0}

    # A hears b:ping and re-emits a:ping; B hears a:ping and re-emits b:ping.
    def a_on_bping(**payload):
        calls["a"] += 1
        ctx_a.emit("ping")

    def b_on_aping(**payload):
        calls["b"] += 1
        ctx_b.emit("ping")

    ctx_a.subscribe("b:ping", a_on_bping)
    ctx_b.subscribe("a:ping", b_on_aping)

    with caplog.at_level(logging.WARNING):
        # Kick off the loop — must terminate, not hang or RecursionError.
        result = ctx_b.emit("ping")
        _drain(manager)

    # Returned cleanly.
    assert result == 1
    # Bounded by the depth cap — nowhere near unbounded.
    assert calls["a"] + calls["b"] <= _EVENT_EMIT_DEPTH_CAP + 1
    # Exactly the recursion-cap warning fired.
    assert any("recursion cap" in r.message.lower() for r in caplog.records)


# ── 6. Manifest emits/listens parsed as optional ─────────────────────────────


def test_manifest_parse_reads_emits_listens(tmp_path):
    """parse_manifest_file picks up optional emits/listens from plugin.yaml."""
    import hermes_yaml as yaml

    plugin_dir = tmp_path / "myplug"
    plugin_dir.mkdir()
    manifest_file = plugin_dir / "plugin.yaml"
    manifest_file.write_text(
        yaml.safe_dump(
            {
                "name": "myplug",
                "emits": ["ping"],
                "listens": ["other:evt"],
            }
        ),
        encoding="utf-8",
    )

    from hermes_cli.plugins import parse_manifest_file

    manifest = parse_manifest_file(manifest_file, plugin_dir, "user", "")
    assert manifest is not None
    assert manifest.emits == ["ping"]
    assert manifest.listens == ["other:evt"]


# ── 7. plugins show output includes emits/listens ────────────────────────────


def test_plugins_show_includes_emits_listens(tmp_path, monkeypatch, capsys):
    import hermes_yaml as yaml
    from hermes_cli import plugins_cmd

    plugin_dir = tmp_path / "showplug"
    plugin_dir.mkdir()
    (plugin_dir / "plugin.yaml").write_text(
        yaml.safe_dump(
            {
                "name": "showplug",
                "version": "1.2.3",
                "description": "a demo plugin",
                "emits": ["ping", "pong"],
                "listens": ["other:ready"],
            }
        ),
        encoding="utf-8",
    )

    # entry = (name, version, description, source, dir_path, key)
    entry = ("showplug", "1.2.3", "a demo plugin", "user", str(plugin_dir), "showplug")
    monkeypatch.setattr(plugins_cmd, "_discover_all_plugins", lambda: [entry])
    monkeypatch.setattr(plugins_cmd, "_get_enabled_set", lambda: set())
    monkeypatch.setattr(plugins_cmd, "_get_disabled_set", lambda: set())

    plugins_cmd.cmd_show("showplug")

    out = capsys.readouterr().out
    assert "showplug" in out
    assert "ping" in out
    assert "pong" in out
    assert "other:ready" in out


def test_plugins_show_not_found_exits(monkeypatch):
    from hermes_cli import plugins_cmd

    monkeypatch.setattr(plugins_cmd, "_discover_all_plugins", lambda: [])
    with pytest.raises(SystemExit) as exc:
        plugins_cmd.cmd_show("nope")
    assert exc.value.code not in (0, None)