hermes-agent/tests/agent/test_plugin_llm.py

"""Unit tests for the plugin LLM facade (``agent.plugin_llm``).

These tests exercise the trust gate, JSON parsing, schema validation,
image input encoding, and the auxiliary-client invocation contract.
The auxiliary client itself is stubbed via ``make_plugin_llm_for_test``
so we don't hit real providers.
"""

from __future__ import annotations

import asyncio
import base64
import json
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock

import pytest

from agent.plugin_llm import (
    PluginLlm,
    PluginLlmCompleteResult,
    PluginLlmImageInput,
    PluginLlmStructuredResult,
    PluginLlmTextInput,
    PluginLlmTrustError,
    _build_structured_messages,
    _check_overrides,
    _coerce_allowlist,
    _parse_structured_text,
    _strip_code_fences,
    _TrustPolicy,
    make_plugin_llm_for_test,
)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _fake_response(text: str, *, prompt: int = 4, completion: int = 6) -> SimpleNamespace:
    """Build an OpenAI-shaped response with the given text + token usage."""
    return SimpleNamespace(
        choices=[
            SimpleNamespace(
                message=SimpleNamespace(content=text, role="assistant"),
                finish_reason="stop",
            )
        ],
        usage=SimpleNamespace(
            prompt_tokens=prompt,
            completion_tokens=completion,
            total_tokens=prompt + completion,
        ),
    )


def _trusted_policy(plugin_id: str = "trusted-plugin", **overrides: Any) -> _TrustPolicy:
    defaults = dict(
        allow_provider_override=True,
        allowed_providers=None,
        allow_any_provider=True,
        allow_model_override=True,
        allowed_models=None,
        allow_any_model=True,
        allow_agent_id_override=True,
        allow_profile_override=True,
    )
    defaults.update(overrides)
    return _TrustPolicy(plugin_id=plugin_id, **defaults)


# ---------------------------------------------------------------------------
# Trust gate
# ---------------------------------------------------------------------------


class TestTrustGate:
    def test_default_policy_blocks_provider_override(self):
        policy = _TrustPolicy(plugin_id="locked")
        with pytest.raises(PluginLlmTrustError, match="cannot override the provider"):
            _check_overrides(
                policy,
                requested_provider="anthropic",
                requested_model=None,
                requested_agent_id=None,
                requested_profile=None,
            )

    def test_default_policy_blocks_model_override(self):
        policy = _TrustPolicy(plugin_id="locked")
        with pytest.raises(PluginLlmTrustError, match="cannot override the model"):
            _check_overrides(
                policy,
                requested_provider=None,
                requested_model="claude-3-5-sonnet",
                requested_agent_id=None,
                requested_profile=None,
            )

    def test_default_policy_blocks_agent_override(self):
        policy = _TrustPolicy(plugin_id="locked")
        with pytest.raises(PluginLlmTrustError, match="non-default agent id"):
            _check_overrides(
                policy,
                requested_provider=None,
                requested_model=None,
                requested_agent_id="ada",
                requested_profile=None,
            )

    def test_default_policy_blocks_profile_override(self):
        policy = _TrustPolicy(plugin_id="locked")
        with pytest.raises(PluginLlmTrustError, match="cannot override the auth profile"):
            _check_overrides(
                policy,
                requested_provider=None,
                requested_model=None,
                requested_agent_id=None,
                requested_profile="work",
            )

    def test_overrides_independent(self):
        """Each override is gated independently — turning on
        ``allow_model_override`` does NOT also grant provider override."""
        policy = _TrustPolicy(
            plugin_id="model-only",
            allow_model_override=True,
            allow_any_model=True,
        )
        # model alone passes
        _, m, _, _ = _check_overrides(
            policy,
            requested_provider=None,
            requested_model="gpt-4o",
            requested_agent_id=None,
            requested_profile=None,
        )
        assert m == "gpt-4o"
        # provider alone is still denied
        with pytest.raises(PluginLlmTrustError, match="cannot override the provider"):
            _check_overrides(
                policy,
                requested_provider="anthropic",
                requested_model=None,
                requested_agent_id=None,
                requested_profile=None,
            )

    def test_provider_allowlist_rejects_non_listed(self):
        policy = _TrustPolicy(
            plugin_id="restricted",
            allow_provider_override=True,
            allowed_providers=frozenset({"openrouter", "anthropic"}),
            allow_any_provider=False,
        )
        with pytest.raises(PluginLlmTrustError, match="not in plugins.entries"):
            _check_overrides(
                policy,
                requested_provider="openai",
                requested_model=None,
                requested_agent_id=None,
                requested_profile=None,
            )

    def test_provider_allowlist_accepts_listed_case_insensitively(self):
        policy = _TrustPolicy(
            plugin_id="restricted",
            allow_provider_override=True,
            allowed_providers=frozenset({"openrouter"}),
            allow_any_provider=False,
        )
        p, _, _, _ = _check_overrides(
            policy,
            requested_provider="OpenRouter",
            requested_model=None,
            requested_agent_id=None,
            requested_profile=None,
        )
        assert p == "OpenRouter"

    def test_model_allowlist_rejects_non_listed(self):
        policy = _TrustPolicy(
            plugin_id="restricted",
            allow_model_override=True,
            allowed_models=frozenset({"openai/gpt-4o-mini"}),
            allow_any_model=False,
        )
        with pytest.raises(PluginLlmTrustError, match="not in plugins.entries"):
            _check_overrides(
                policy,
                requested_provider=None,
                requested_model="anthropic/claude-3-opus",
                requested_agent_id=None,
                requested_profile=None,
            )

    def test_model_allowlist_accepts_listed_case_insensitively(self):
        policy = _TrustPolicy(
            plugin_id="restricted",
            allow_model_override=True,
            allowed_models=frozenset({"openai/gpt-4o-mini"}),
            allow_any_model=False,
        )
        _, m, _, _ = _check_overrides(
            policy,
            requested_provider=None,
            requested_model="OpenAI/GPT-4o-mini",
            requested_agent_id=None,
            requested_profile=None,
        )
        assert m == "OpenAI/GPT-4o-mini"

    def test_no_overrides_passes_through(self):
        policy = _TrustPolicy(plugin_id="locked")
        result = _check_overrides(
            policy,
            requested_provider=None,
            requested_model=None,
            requested_agent_id=None,
            requested_profile=None,
        )
        assert result == (None, None, None, None)

    def test_all_overrides_when_fully_trusted(self):
        policy = _trusted_policy()
        result = _check_overrides(
            policy,
            requested_provider="openrouter",
            requested_model="anthropic/claude-3-5-sonnet",
            requested_agent_id="ada",
            requested_profile="work",
        )
        assert result == ("openrouter", "anthropic/claude-3-5-sonnet", "ada", "work")


class TestAllowlistCoercion:
    def test_missing_yields_none(self):
        ranges, allow_any = _coerce_allowlist(None)
        assert ranges is None
        assert allow_any is False

    def test_list_of_strings(self):
        ranges, allow_any = _coerce_allowlist(["A", "B"])
        assert ranges == frozenset({"a", "b"})
        assert allow_any is False

    def test_star_alone_means_any(self):
        ranges, allow_any = _coerce_allowlist(["*"])
        assert ranges == frozenset()
        assert allow_any is True

    def test_star_plus_specific_keeps_specifics(self):
        ranges, allow_any = _coerce_allowlist(["*", "openrouter"])
        assert ranges == frozenset({"openrouter"})
        assert allow_any is True

    def test_non_list_yields_none(self):
        ranges, allow_any = _coerce_allowlist("openrouter")
        assert ranges is None
        assert allow_any is False


# ---------------------------------------------------------------------------
# Structured message building
# ---------------------------------------------------------------------------


class TestStructuredMessageBuilding:
    def test_text_only_input(self):
        messages = _build_structured_messages(
            instructions="Extract the action items",
            inputs=[PluginLlmTextInput(text="meeting notes go here")],
            json_mode=False,
            json_schema=None,
            schema_name=None,
            system_prompt=None,
        )
        assert len(messages) == 1
        assert messages[0]["role"] == "user"
        parts = messages[0]["content"]
        assert parts[0]["type"] == "text"
        assert "Extract the action items" in parts[0]["text"]
        assert parts[1] == {"type": "text", "text": "meeting notes go here"}

    def test_json_mode_adds_system_directive(self):
        messages = _build_structured_messages(
            instructions="Summarise",
            inputs=[PluginLlmTextInput(text="content")],
            json_mode=True,
            json_schema=None,
            schema_name=None,
            system_prompt=None,
        )
        assert messages[0]["role"] == "system"
        assert "JSON object" in messages[0]["content"]

    def test_schema_name_appended_to_header(self):
        messages = _build_structured_messages(
            instructions="Extract fields",
            inputs=[PluginLlmTextInput(text="data")],
            json_mode=False,
            json_schema=None,
            schema_name="action.items",
            system_prompt=None,
        )
        header = messages[0]["content"][0]["text"]
        assert "Schema name: action.items" in header

    def test_image_bytes_encoded_as_data_url(self):
        png_bytes = b"\x89PNG\r\n\x1a\nfake"
        messages = _build_structured_messages(
            instructions="Read the image",
            inputs=[
                PluginLlmImageInput(data=png_bytes, mime_type="image/png"),
                PluginLlmTextInput(text="prefer printed text"),
            ],
            json_mode=False,
            json_schema=None,
            schema_name=None,
            system_prompt=None,
        )
        parts = messages[0]["content"]
        assert parts[1]["type"] == "image_url"
        url = parts[1]["image_url"]["url"]
        assert url.startswith("data:image/png;base64,")
        decoded = base64.b64decode(url.split(",", 1)[1])
        assert decoded == png_bytes
        assert parts[2] == {"type": "text", "text": "prefer printed text"}

    def test_image_url_passed_through(self):
        messages = _build_structured_messages(
            instructions="Caption this",
            inputs=[PluginLlmImageInput(url="https://example.com/cat.jpg")],
            json_mode=False,
            json_schema=None,
            schema_name=None,
            system_prompt=None,
        )
        img_part = messages[0]["content"][1]
        assert img_part["type"] == "image_url"
        assert img_part["image_url"]["url"] == "https://example.com/cat.jpg"

    def test_dict_inputs_normalized(self):
        messages = _build_structured_messages(
            instructions="Test",
            inputs=[
                {"type": "text", "text": "hello"},
                {"type": "image", "url": "https://x.example/y.png"},
            ],
            json_mode=False,
            json_schema=None,
            schema_name=None,
            system_prompt=None,
        )
        parts = messages[0]["content"]
        assert parts[1]["text"] == "hello"
        assert parts[2]["image_url"]["url"] == "https://x.example/y.png"

    def test_invalid_input_block_rejected(self):
        with pytest.raises(ValueError, match="Unknown input block"):
            _build_structured_messages(
                instructions="Test",
                inputs=[{"type": "audio", "data": b""}],
                json_mode=False,
                json_schema=None,
                schema_name=None,
                system_prompt=None,
            )


# ---------------------------------------------------------------------------
# JSON parsing
# ---------------------------------------------------------------------------


class TestJsonParsing:
    def test_strip_code_fences_with_json_label(self):
        assert _strip_code_fences('```json\n{"a":1}\n```') == '{"a":1}'

    def test_strip_code_fences_without_label(self):
        assert _strip_code_fences("```\nfoo\n```") == "foo"

    def test_strip_code_fences_no_fence(self):
        assert _strip_code_fences('{"a":1}') == '{"a":1}'

    def test_parse_returns_text_when_not_json_mode(self):
        parsed, ct = _parse_structured_text(
            text='{"a": 1}', json_mode=False, json_schema=None
        )
        assert parsed is None
        assert ct == "text"

    def test_parse_valid_json_with_json_mode(self):
        parsed, ct = _parse_structured_text(
            text='{"language": "French", "is_question": true}',
            json_mode=True,
            json_schema=None,
        )
        assert parsed == {"language": "French", "is_question": True}
        assert ct == "json"

    def test_parse_strips_code_fences_before_loading(self):
        parsed, ct = _parse_structured_text(
            text='Here you go:\n```json\n{"ok": true}\n```',
            json_mode=True,
            json_schema=None,
        )
        assert parsed == {"ok": True}
        assert ct == "json"

    def test_parse_returns_text_on_invalid_json(self):
        parsed, ct = _parse_structured_text(
            text="not even close to json",
            json_mode=True,
            json_schema=None,
        )
        assert parsed is None
        assert ct == "text"

    def test_schema_validation_rejects_mismatch(self):
        pytest.importorskip("jsonschema")
        schema = {
            "type": "object",
            "properties": {"language": {"type": "string"}},
            "required": ["language"],
        }
        with pytest.raises(ValueError, match="did not match schema"):
            _parse_structured_text(
                text='{"is_question": true}',
                json_mode=False,
                json_schema=schema,
            )

    def test_schema_validation_accepts_match(self):
        pytest.importorskip("jsonschema")
        schema = {
            "type": "object",
            "properties": {"language": {"type": "string"}},
            "required": ["language"],
        }
        parsed, ct = _parse_structured_text(
            text='{"language": "French"}',
            json_mode=False,
            json_schema=schema,
        )
        assert parsed == {"language": "French"}
        assert ct == "json"


# ---------------------------------------------------------------------------
# End-to-end facade
# ---------------------------------------------------------------------------


class TestPluginLlmFacade:
    def test_complete_uses_active_model_by_default(self):
        captured: dict = {}

        def fake_caller(**kwargs):
            captured.update(kwargs)
            return "auto", "default", _fake_response("Hello world.")

        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_TrustPolicy(plugin_id="my-plugin"),
            sync_caller=fake_caller,
        )
        result = llm.complete([{"role": "user", "content": "hi"}])
        assert isinstance(result, PluginLlmCompleteResult)
        assert result.text == "Hello world."
        assert captured["provider_override"] is None
        assert captured["model_override"] is None
        assert captured["profile_override"] is None
        assert result.usage.input_tokens == 4
        assert result.usage.total_tokens == 10

    def test_complete_rejects_provider_override_without_trust(self):
        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_TrustPolicy(plugin_id="my-plugin"),
            sync_caller=lambda **_: ("x", "y", _fake_response("")),
        )
        with pytest.raises(PluginLlmTrustError, match="cannot override the provider"):
            llm.complete(
                [{"role": "user", "content": "hi"}],
                provider="openrouter",
            )

    def test_complete_rejects_model_override_without_trust(self):
        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_TrustPolicy(plugin_id="my-plugin"),
            sync_caller=lambda **_: ("x", "y", _fake_response("")),
        )
        with pytest.raises(PluginLlmTrustError, match="cannot override the model"):
            llm.complete(
                [{"role": "user", "content": "hi"}],
                model="anthropic/claude-3-opus",
            )

    def test_complete_passes_through_trusted_overrides(self):
        captured: dict = {}

        def fake_caller(**kwargs):
            captured.update(kwargs)
            return "anthropic", "claude-3-opus", _fake_response("ok")

        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_trusted_policy("my-plugin"),
            sync_caller=fake_caller,
        )
        result = llm.complete(
            [{"role": "user", "content": "hi"}],
            provider="anthropic",
            model="claude-3-opus",
            profile="work",
            agent_id="ada",
            temperature=0.0,
            max_tokens=128,
            timeout=10.0,
            purpose="extract",
        )
        # The recorded provider/model in the result come from the override,
        # since the stub caller echoed those values.
        assert result.provider == "anthropic"
        assert result.model == "claude-3-opus"
        assert captured["provider_override"] == "anthropic"
        assert captured["model_override"] == "claude-3-opus"
        assert captured["profile_override"] == "work"
        assert captured["temperature"] == 0.0
        assert captured["max_tokens"] == 128
        assert captured["timeout"] == 10.0

    def test_complete_structured_returns_parsed_json(self):
        def fake_caller(**_kwargs):
            return "openai", "gpt-4o", _fake_response(
                '{"language": "French", "is_question": true, "confidence": 0.99}'
            )

        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_TrustPolicy(plugin_id="my-plugin"),
            sync_caller=fake_caller,
        )
        result = llm.complete_structured(
            instructions="Detect language",
            input=[PluginLlmTextInput(text="Comment ça va?")],
            json_mode=True,
        )
        assert isinstance(result, PluginLlmStructuredResult)
        assert result.parsed == {
            "language": "French",
            "is_question": True,
            "confidence": 0.99,
        }
        assert result.content_type == "json"

    def test_complete_structured_returns_text_on_unparseable_response(self):
        def fake_caller(**_kwargs):
            return "openai", "gpt-4o", _fake_response("Sorry, I can't help with that.")

        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_TrustPolicy(plugin_id="my-plugin"),
            sync_caller=fake_caller,
        )
        result = llm.complete_structured(
            instructions="Detect language",
            input=[PluginLlmTextInput(text="x")],
            json_mode=True,
        )
        assert result.parsed is None
        assert result.content_type == "text"
        assert result.text.startswith("Sorry")

    def test_complete_structured_validates_against_schema(self):
        pytest.importorskip("jsonschema")

        def fake_caller(**_kwargs):
            return "openai", "gpt-4o", _fake_response('{"unrelated": "field"}')

        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_TrustPolicy(plugin_id="my-plugin"),
            sync_caller=fake_caller,
        )
        schema = {
            "type": "object",
            "properties": {"language": {"type": "string"}},
            "required": ["language"],
        }
        with pytest.raises(ValueError, match="did not match schema"):
            llm.complete_structured(
                instructions="Detect language",
                input=[PluginLlmTextInput(text="x")],
                json_schema=schema,
            )

    def test_complete_structured_requires_instructions(self):
        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_TrustPolicy(plugin_id="my-plugin"),
            sync_caller=MagicMock(),
        )
        with pytest.raises(ValueError, match="non-empty instructions"):
            llm.complete_structured(
                instructions="   ",
                input=[PluginLlmTextInput(text="x")],
            )

    def test_complete_structured_requires_at_least_one_input(self):
        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_TrustPolicy(plugin_id="my-plugin"),
            sync_caller=MagicMock(),
        )
        with pytest.raises(ValueError, match="at least one input"):
            llm.complete_structured(
                instructions="Extract",
                input=[],
            )

    def test_complete_structured_emits_response_format_extra_body(self):
        captured: dict = {}

        def fake_caller(**kwargs):
            captured.update(kwargs)
            return "openai", "gpt-4o", _fake_response('{"a": 1}')

        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_TrustPolicy(plugin_id="my-plugin"),
            sync_caller=fake_caller,
        )
        schema = {"type": "object"}
        llm.complete_structured(
            instructions="Test",
            input=[PluginLlmTextInput(text="x")],
            json_schema=schema,
        )
        rf = captured["extra_body"]["response_format"]
        assert rf["type"] == "json_schema"
        assert rf["json_schema"]["schema"] == schema

    def test_complete_structured_with_image_passes_image_url_part(self):
        captured: dict = {}

        def fake_caller(**kwargs):
            captured.update(kwargs)
            return "openai", "gpt-4o", _fake_response('{"caption": "ok"}')

        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_TrustPolicy(plugin_id="my-plugin"),
            sync_caller=fake_caller,
        )
        png = b"fake-bytes"
        llm.complete_structured(
            instructions="Caption this",
            input=[PluginLlmImageInput(data=png, mime_type="image/png")],
            json_mode=True,
        )
        msgs = captured["messages"]
        user_msg = next(m for m in msgs if m["role"] == "user")
        image_parts = [p for p in user_msg["content"] if p.get("type") == "image_url"]
        assert len(image_parts) == 1
        assert image_parts[0]["image_url"]["url"].startswith("data:image/png;base64,")


# ---------------------------------------------------------------------------
# Async surface
# ---------------------------------------------------------------------------


class TestAsyncSurface:
    def test_acomplete_uses_async_caller(self):
        async def fake_async(**_kwargs):
            return "openai", "gpt-4o", _fake_response("async hello")

        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_TrustPolicy(plugin_id="my-plugin"),
            async_caller=fake_async,
        )

        async def _run() -> PluginLlmCompleteResult:
            return await llm.acomplete([{"role": "user", "content": "hi"}])

        result = asyncio.run(_run())
        assert result.text == "async hello"
        assert result.provider == "openai"

    def test_acomplete_structured_parses_json(self):
        async def fake_async(**_kwargs):
            return "openai", "gpt-4o", _fake_response('{"x": 42}')

        llm = make_plugin_llm_for_test(
            plugin_id="my-plugin",
            policy=_TrustPolicy(plugin_id="my-plugin"),
            async_caller=fake_async,
        )

        async def _run() -> PluginLlmStructuredResult:
            return await llm.acomplete_structured(
                instructions="Extract x",
                input=[PluginLlmTextInput(text="data")],
                json_mode=True,
            )

        result = asyncio.run(_run())
        assert result.parsed == {"x": 42}
        assert result.content_type == "json"


# ---------------------------------------------------------------------------
# Config-driven trust gate (round-trip via plugins.entries.<id>.llm)
# ---------------------------------------------------------------------------


class TestConfigDrivenPolicy:
    def test_policy_loaded_from_yaml(self, tmp_path, monkeypatch):
        from agent.plugin_llm import _resolve_trust_policy

        hermes_home = tmp_path / ".hermes"
        hermes_home.mkdir()
        (hermes_home / "config.yaml").write_text(
            """
plugins:
  entries:
    my-plugin:
      llm:
        allow_provider_override: true
        allowed_providers: [openrouter, anthropic]
        allow_model_override: true
        allowed_models:
          - openai/gpt-4o-mini
          - anthropic/claude-3-5-haiku
        allow_profile_override: false
""",
            encoding="utf-8",
        )
        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
        from hermes_cli import config as _config_mod
        _config_mod._config_cache = None  # type: ignore[attr-defined]

        policy = _resolve_trust_policy("my-plugin")
        assert policy.allow_provider_override is True
        assert policy.allow_model_override is True
        assert policy.allow_profile_override is False
        assert policy.allowed_providers == frozenset({"openrouter", "anthropic"})
        assert policy.allowed_models == frozenset({
            "openai/gpt-4o-mini", "anthropic/claude-3-5-haiku",
        })

    def test_missing_plugin_entry_yields_default_deny(self, tmp_path, monkeypatch):
        from agent.plugin_llm import _resolve_trust_policy

        hermes_home = tmp_path / ".hermes"
        hermes_home.mkdir()
        (hermes_home / "config.yaml").write_text("plugins: {}\n", encoding="utf-8")
        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
        from hermes_cli import config as _config_mod
        _config_mod._config_cache = None  # type: ignore[attr-defined]

        policy = _resolve_trust_policy("never-configured")
        assert policy.allow_provider_override is False
        assert policy.allow_model_override is False
        assert policy.allow_profile_override is False
        assert policy.allow_agent_id_override is False


# ---------------------------------------------------------------------------
# Plugin context wiring
# ---------------------------------------------------------------------------


class TestPluginContextIntegration:
    def test_ctx_llm_is_lazy_singleton(self):
        from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager

        manifest = PluginManifest(name="test-plugin", source="test", key="test-plugin")
        manager = PluginManager()
        ctx = PluginContext(manifest, manager)
        first = ctx.llm
        second = ctx.llm
        assert first is second
        assert isinstance(first, PluginLlm)
        assert first._plugin_id == "test-plugin"  # type: ignore[attr-defined]

    def test_ctx_llm_uses_manifest_key_for_policy(self):
        from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager

        manifest = PluginManifest(
            name="bare-name", source="test", key="image_gen/openai"
        )
        manager = PluginManager()
        ctx = PluginContext(manifest, manager)
        assert ctx.llm._plugin_id == "image_gen/openai"  # type: ignore[attr-defined]


# ---------------------------------------------------------------------------
# Attribution (result.provider / result.model / audit log)
# ---------------------------------------------------------------------------


class TestAttribution:
    """Verifies that the result object and the audit log carry the real
    provider/model that ``call_llm`` ended up using, NOT the placeholder
    fallbacks ('auto', 'default') from earlier drafts."""

    def test_explicit_overrides_recorded_when_no_response_model(self):
        from agent.plugin_llm import _resolve_attribution

        # Response with no .model attribute — overrides win.
        response = SimpleNamespace(choices=[], usage=None)
        provider, model = _resolve_attribution(
            provider_override="openrouter",
            model_override="anthropic/claude-3-5-sonnet",
            response=response,
        )
        assert provider == "openrouter"
        assert model == "anthropic/claude-3-5-sonnet"

    def test_response_model_wins_over_model_override(self):
        """Providers often canonicalise the model name (e.g. ``gpt-4o``
        → ``gpt-4o-2024-08-06``). Whatever they actually returned wins
        for the recorded model so the audit log reflects reality."""
        from agent.plugin_llm import _resolve_attribution

        response = SimpleNamespace(model="gpt-4o-2024-08-06", choices=[])
        provider, model = _resolve_attribution(
            provider_override="openrouter",
            model_override="openai/gpt-4o",
            response=response,
        )
        assert model == "gpt-4o-2024-08-06"
        # Provider override is unaffected by response.model.
        assert provider == "openrouter"

    def test_falls_back_to_main_provider_and_model_when_no_overrides(self, monkeypatch):
        """When the plugin doesn't override anything, attribution
        reflects the user's active main provider/model rather than
        misleading placeholders."""
        from agent import plugin_llm
        import agent.auxiliary_client as ac

        monkeypatch.setattr(ac, "_read_main_provider", lambda: "openrouter")
        monkeypatch.setattr(ac, "_read_main_model", lambda: "anthropic/claude-3-5-sonnet")

        response = SimpleNamespace(choices=[])  # no .model attribute
        provider, model = plugin_llm._resolve_attribution(
            provider_override=None,
            model_override=None,
            response=response,
        )
        assert provider == "openrouter"
        assert model == "anthropic/claude-3-5-sonnet"

    def test_response_model_used_even_when_no_overrides(self, monkeypatch):
        """The provider's canonical model name should still flow through
        when no overrides are set."""
        from agent import plugin_llm
        import agent.auxiliary_client as ac

        monkeypatch.setattr(ac, "_read_main_provider", lambda: "openrouter")
        monkeypatch.setattr(ac, "_read_main_model", lambda: "openai/gpt-4o")

        response = SimpleNamespace(model="openai/gpt-4o-2024-08-06", choices=[])
        provider, model = plugin_llm._resolve_attribution(
            provider_override=None,
            model_override=None,
            response=response,
        )
        assert provider == "openrouter"
        assert model == "openai/gpt-4o-2024-08-06"

    def test_placeholder_fallback_only_when_everything_is_empty(self, monkeypatch):
        """If main_provider/main_model are unset AND there's no override
        AND the response has no .model, fall through to the safety
        placeholders so the result object never has empty strings."""
        from agent import plugin_llm
        import agent.auxiliary_client as ac

        monkeypatch.setattr(ac, "_read_main_provider", lambda: "")
        monkeypatch.setattr(ac, "_read_main_model", lambda: "")

        response = SimpleNamespace(choices=[])
        provider, model = plugin_llm._resolve_attribution(
            provider_override=None,
            model_override=None,
            response=response,
        )
        assert provider == "auto"
        assert model == "default"


# ---------------------------------------------------------------------------
# Hook-mode integration (ctx.llm called from a post_tool_call callback)
# ---------------------------------------------------------------------------


class TestHookMode:
    """The docs page promises ``ctx.llm`` works from inside lifecycle
    hooks. This exercises that path: register a ``post_tool_call``
    callback that calls ``ctx.llm.complete``, fire the hook through
    the real ``invoke_hook`` machinery, and check the call landed."""

    def test_complete_works_from_post_tool_call_hook(self):
        from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager

        manifest = PluginManifest(name="hook-plugin", source="test", key="hook-plugin")
        manager = PluginManager()
        ctx = PluginContext(manifest, manager)

        # Replace ctx.llm with a stub that records what the hook called.
        captured: list = []

        def fake_caller(**kwargs):
            captured.append(kwargs)
            return "openrouter", "openai/gpt-4o", _fake_response("rewrote it")

        ctx._llm = make_plugin_llm_for_test(  # type: ignore[attr-defined]
            plugin_id="hook-plugin",
            policy=_TrustPolicy(plugin_id="hook-plugin"),
            sync_caller=fake_caller,
        )

        # Plugin registers a hook that runs ctx.llm.complete on every tool call.
        def rewrite_error_hook(*, tool_name, args, result, **_):
            if "Traceback" in (result or ""):
                rewritten = ctx.llm.complete(
                    messages=[
                        {"role": "system", "content": "Rewrite errors plainly."},
                        {"role": "user", "content": result},
                    ],
                    max_tokens=64,
                    purpose="hook-plugin.rewrite",
                )
                # Real hook would return the rewritten text via
                # transform_tool_result; here we just capture for the assert.
                captured.append({"hook_returned": rewritten.text})

        ctx.register_hook("post_tool_call", rewrite_error_hook)

        # Fire the hook the same way the agent core does it.
        manager.invoke_hook(
            "post_tool_call",
            tool_name="terminal",
            args={"command": "boom"},
            result="Traceback (most recent call last):\n  RuntimeError",
        )

        # Verify ctx.llm.complete fired through the hook.
        assert len(captured) == 2  # one llm call + one hook return record
        llm_call = captured[0]
        assert "messages" in llm_call
        assert any("rewrite" in m.get("content", "").lower()
                   for m in llm_call["messages"] if isinstance(m, dict))
        hook_record = captured[1]
        assert hook_record["hook_returned"] == "rewrote it"

    def test_complete_works_from_post_tool_call_hook_when_async_caller_set(self):
        """Hooks fired synchronously should still work with sync
        ctx.llm.complete even if other callsites use async."""
        from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager

        manifest = PluginManifest(name="hook-async", source="test", key="hook-async")
        manager = PluginManager()
        ctx = PluginContext(manifest, manager)

        def fake_caller(**_):
            return "openrouter", "model-x", _fake_response("ok")

        ctx._llm = make_plugin_llm_for_test(  # type: ignore[attr-defined]
            plugin_id="hook-async",
            policy=_TrustPolicy(plugin_id="hook-async"),
            sync_caller=fake_caller,
        )

        called: list = []

        def hook(**kwargs):
            r = ctx.llm.complete(messages=[{"role": "user", "content": "x"}])
            called.append(r.text)

        ctx.register_hook("post_tool_call", hook)
        manager.invoke_hook("post_tool_call", tool_name="x", args={}, result="y")
        assert called == ["ok"]