hermes_overlay/tests/test_external_ai.py

"""Small mocked checks for advisory auxiliary routing and deterministic safety."""
import asyncio
import json
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

import external_ai.advisory as advisory
import external_ai.advisory_schema as advisory_schema
import trading.trade_cycle as trade_cycle


def _ok():
    return {
        "market_bias": "neutral",
        "confidence": 0.4,
        "summary": "mock advisory",
        "supporting_points": ["context"],
        "risk_warnings": ["not a signal"],
    }, "hermes_auxiliary", "mock-model", None, 12.5


async def _fake_invoke(prompt):
    return _ok()


def test_auxiliary_task_is_primary_provider(monkeypatch):
    calls = []

    async def fake(prompt):
        calls.append(prompt)
        return _ok()

    monkeypatch.setattr(advisory, "_invoke_auxiliary_advisory", fake)
    monkeypatch.setenv("EXTERNAL_AI_ENABLED", "true")
    result = asyncio.run(advisory.get_advisory({"noTradeGuard": True}))

    assert calls
    assert result["provider"] == "hermes_auxiliary"
    assert result["provider_order"] == [advisory.AUXILIARY_TASK]
    assert "mock-model" in json.dumps(result)


def test_auxiliary_success_is_structured(monkeypatch):
    monkeypatch.setattr(advisory, "_invoke_auxiliary_advisory", _fake_invoke)
    monkeypatch.setenv("EXTERNAL_AI_ENABLED", "true")
    result = asyncio.run(advisory.get_advisory({}))
    assert result["status"] == "available"
    assert result["market_bias"] == "neutral"
    assert result["confidence"] == 0.4
    assert result["advisory_id"]


def test_auxiliary_invalid_payload_returns_unavailable(monkeypatch):
    async def fake(prompt):
        return None, "hermes_auxiliary", "mock-model", "invalid_advisory", 1.0

    monkeypatch.setattr(advisory, "_invoke_auxiliary_advisory", fake)
    monkeypatch.setenv("EXTERNAL_AI_ENABLED", "true")
    result = asyncio.run(advisory.get_advisory({}))
    assert result["provider"] == "external_ai_unavailable"
    assert result["status"] == "unavailable"
    assert result["market_bias"] == "unknown"


def test_auxiliary_timeout_returns_unavailable(monkeypatch):
    async def fake(prompt):
        await asyncio.sleep(2)
        return _ok()

    monkeypatch.setattr(advisory, "_invoke_auxiliary_advisory", fake)
    monkeypatch.setenv("EXTERNAL_AI_ENABLED", "true")
    monkeypatch.setenv("EXTERNAL_AI_TOTAL_TIMEOUT_SECONDS", "1")
    result = asyncio.run(advisory.get_advisory({}))
    assert result["provider"] == "external_ai_unavailable"
    assert result["status"] == "unavailable"
    assert result["attempts"][0]["error"] == "timeout"


def test_auxiliary_unavailable_when_runtime_missing(monkeypatch):
    async def fake(prompt):
        return None, "external_ai_unavailable", "", "hermes_runtime_unavailable", 0.0

    monkeypatch.setattr(advisory, "_invoke_auxiliary_advisory", fake)
    monkeypatch.setenv("EXTERNAL_AI_ENABLED", "true")
    result = asyncio.run(advisory.get_advisory({"noTradeGuard": True}))
    assert result["provider"] == "external_ai_unavailable"
    assert result["market_bias"] == "unknown"
    assert result["confidence"] == 0.0


def test_disabled_advisory_returns_safe_result(monkeypatch):
    monkeypatch.setenv("EXTERNAL_AI_ENABLED", "false")
    result = asyncio.run(advisory.get_advisory({}))
    assert result["error"] == "disabled"
    assert result["market_bias"] == "unknown"


def test_advisory_cannot_clear_ds4_no_trade_guard(monkeypatch):
    async def context(symbol):
        return {
            "symbol": symbol,
            "primary": {"dataState": "PARTIAL"},
            "merged": {},
            "sources": {},
            "warnings": [],
            "noTradeGuard": True,
            "noTradeReasons": ["DS4 guard"],
        }

    async def fake_advisory(snapshot):
        return {
            "provider": "hermes_auxiliary",
            "model": "mock",
            "market_bias": "bullish",
            "confidence": 1.0,
            "summary": "bullish",
            "supporting_points": [],
            "risk_warnings": [],
            "latency_ms": 1,
            "fallback_used": False,
            "error": None,
        }

    monkeypatch.setattr(trade_cycle, "get_market_context", context)
    monkeypatch.setattr(trade_cycle, "get_advisory", fake_advisory)
    result = asyncio.run(
        trade_cycle.run_futures_cycle("BTCUSDT", execute=False, include_external_context=True)
    )

    assert result["decision"] == "NO_TRADE"
    assert result["external_advisory"]["market_bias"] == "bullish"
    assert result["risk_approved"] is False


def test_advisory_parser_accepts_json_wrapped_in_provider_preamble():
    parsed = advisory_schema.parse_advisory_payload(
        "Here is the requested JSON:\n"
        '{"market_bias":"neutral","confidence":0.5,"summary":"Mixed",'
        '"supporting_points":["Momentum is mixed"],"risk_warnings":["No clear direction"]}'
        "\nEnd of response."
    )
    assert parsed is not None
    assert parsed["market_bias"] == "neutral"
    assert parsed["confidence"] == 0.5


def test_skill_parse_script_roundtrip():
    import subprocess

    script = os.path.join(
        os.path.dirname(__file__),
        "..",
        "skills",
        "futures",
        "market-context-advisory",
        "scripts",
        "parse_advisory_response.py",
    )
    payload = (
        '{"market_bias":"bearish","confidence":0.7,"summary":"Weak",'
        '"supporting_points":["Funding elevated"],"risk_warnings":["Volatility"]}'
    )
    completed = subprocess.run(
        [sys.executable, script, payload],
        capture_output=True,
        text=True,
        check=False,
    )
    assert completed.returncode == 0
    parsed = json.loads(completed.stdout)
    assert parsed["market_bias"] == "bearish"