hermes-mcp/tests/test_action_dispatch.py

from types import SimpleNamespace

import pytest

from hermes_mcp.decision_engine import DecisionSnapshot
from hermes_mcp.server import _build_trader_control_payload, _maybe_dispatch_trader_action


def test_build_trader_control_payload_maps_replace_to_switch():
    concern = {"id": "c1", "account_id": "acct-1", "market_symbol": "xrpusd"}
    decision = DecisionSnapshot(
        mode="act",
        action="replace_with_exposure_protector",
        target_strategy="protect-1",
        reason_summary="inventory repair should start",
        confidence=0.81,
        requires_action=True,
        payload={"current_primary_strategy": "grid-1"},
    )

    payload = _build_trader_control_payload(decision_id="d1", concern=concern, decision=decision)

    assert payload == {
        "decision_id": "d1",
        "concern_id": "c1",
        "account_id": "acct-1",
        "market_symbol": "xrpusd",
        "action": "switch",
        "target_strategy_id": "protect-1",
        "expected_active_strategy_id": "grid-1",
        "risk_mode": None,
        "reason": "inventory repair should start",
        "confidence": 0.81,
        "dry_run": False,
        "override": False,
        "source": "hermes-mcp",
        "source_action": "replace_with_exposure_protector",
    }


@pytest.mark.anyio
async def test_dispatch_is_blocked_when_hermes_allow_actions_is_false():
    concern = {"id": "c1", "account_id": "acct-1", "market_symbol": "xrpusd"}
    decision = DecisionSnapshot(
        mode="act",
        action="replace_with_grid",
        target_strategy="grid-1",
        reason_summary="range conditions support grid again",
        confidence=0.77,
        requires_action=True,
        payload={"current_primary_strategy": "protect-1"},
    )
    cfg = SimpleNamespace(hermes_allow_actions=False, trader_url="http://trader.test/mcp/sse")

    result = await _maybe_dispatch_trader_action(cfg=cfg, decision_id="d2", concern=concern, decision=decision)

    assert result["dispatch"] == "blocked"
    assert result["reason"] == "HERMES_ALLOW_ACTIONS is false"
    assert result["payload"]["action"] == "switch"
    assert result["payload"]["target_strategy_id"] == "grid-1"


@pytest.mark.anyio
async def test_dispatch_calls_trader_when_gate_is_open(monkeypatch):
    concern = {"id": "c1", "account_id": "acct-1", "market_symbol": "xrpusd"}
    decision = DecisionSnapshot(
        mode="act",
        action="replace_with_trend_follower",
        target_strategy="trend-1",
        reason_summary="persistent breakout pressure favors trend capture",
        confidence=0.84,
        requires_action=True,
        payload={"current_primary_strategy": "grid-1"},
    )
    cfg = SimpleNamespace(hermes_allow_actions=True, trader_url="http://trader.test/mcp/sse")
    seen = {}

    async def fake_apply(base_url: str, payload: dict):
        seen["base_url"] = base_url
        seen["payload"] = payload
        return {"ok": True, "status": "applied", "decision_id": payload["decision_id"]}

    monkeypatch.setattr("hermes_mcp.server.trader_apply_control_decision", fake_apply)

    result = await _maybe_dispatch_trader_action(cfg=cfg, decision_id="d3", concern=concern, decision=decision)

    assert result["dispatch"] == "sent"
    assert result["result"]["ok"] is True
    assert seen["base_url"] == "http://trader.test/mcp/sse"
    assert seen["payload"]["action"] == "switch"
    assert seen["payload"]["expected_active_strategy_id"] == "grid-1"