backend/tests/test_agent_loop.py

"""Stage 7 integration tests — the AgentLoop end-to-end (live LLM)."""
import os

import pytest

from backend.agent.kb import KnowledgeBase
from backend.agent.loop import AgentLoop

LIVE = os.environ.get("GEMINI_API_KEY") is not None
pytestmark = pytest.mark.skipif(not LIVE, reason="GEMINI_API_KEY not set")


@pytest.fixture(scope="module")
def kb():
    return KnowledgeBase("PO")


def _new_loop(kb):
    return AgentLoop(kb)


def test_turn_1_create_po_intent_detected(kb):
    loop = _new_loop(kb)
    r = loop.handle_turn("I want to create a new purchase order")
    assert loop.state.current_journey == "Create Direct Purchase Order"
    assert r.reply
    # At least one trace event of kind 'intent_classified'
    kinds = [e["kind"] for e in r.trace_events]
    assert "intent_classified" in kinds


def test_turn_offtopic_redirect(kb):
    loop = _new_loop(kb)
    loop.state.current_journey = "Create Direct Purchase Order"
    r = loop.handle_turn("tell me a joke")
    kinds = [e["kind"] for e in r.trace_events]
    assert "route_decided" in kinds


def test_capital_adhoc_violation_surfaced(kb):
    """Capital + adhoc should be caught by the resolver and the agent should explain it."""
    loop = _new_loop(kb)
    loop.state.current_journey = "Create Direct Purchase Order"
    # Force the forbidden combination directly
    loop.state.set_slot("potypeenum", "4")
    loop.state.set_slot("adhocitemclassml", "ML01")
    r = loop.handle_turn("here are some details")
    kinds = [e["kind"] for e in r.trace_events]
    assert "resolved" in kinds
    # Look at the most recent resolved event
    resolved_events = [e for e in r.trace_events if e["kind"] == "resolved"]
    assert resolved_events
    last = resolved_events[-1]
    assert last["data"]["locked_status"] == "blocked_by_violation"
    assert len(last["data"]["violations"]) >= 1


def test_multi_slot_extraction_advances(kb):
    """Several slots in one user turn should all land in state."""
    loop = _new_loop(kb)
    loop.state.current_journey = "Create Direct Purchase Order"
    r = loop.handle_turn("buyer is Maria, currency USD, supplier ABC")
    # Buyer or currency should be in state now
    assert any(k in loop.state.slots for k in ("buyerhdr", "currencycode", "supplier_code"))


def test_question_mid_flow_keeps_state(kb):
    loop = _new_loop(kb)
    loop.state.current_journey = "Create Direct Purchase Order"
    loop.state.set_slot("potypeenum", "2")
    r = loop.handle_turn("what does PO type capital mean exactly?")
    # State should be preserved
    assert loop.state.current_journey == "Create Direct Purchase Order"
    assert loop.state.slots.get("potypeenum") == "2"
    kinds = [e["kind"] for e in r.trace_events]
    # Q&A should have fired
    assert "qa_answered" in kinds or "route_decided" in kinds


def test_termination_when_locked(kb):
    """After enough slots are filled and no violations, an explicit 'go ahead' fires the APIs."""
    loop = _new_loop(kb)
    loop.state.current_journey = "Create Direct Purchase Order"
    # Fill enough slots to pass resolver (avoid violations).
    pilot = kb.journey("Create Direct Purchase Order")
    # Required slots come from resolver — fill them with placeholder values.
    from backend.agent.resolver import resolve as _resolve
    res = _resolve(pilot, {})
    for req in res.active_requirements:
        loop.state.set_slot(req.slot, "X")
    # Also fill gating dropdowns with their first vocab value
    for gs in pilot.gating_slots[:6]:
        vocab = pilot.slot_value_vocab.get(gs, [])
        if vocab:
            loop.state.set_slot(gs, vocab[0])
    r = loop.handle_turn("go ahead and create it")
    # Either we terminated (success) OR we got blocked by a violation surfaced by the
    # additional constraints — both are legitimate outcomes for this synthetic input.
    kinds = [e["kind"] for e in r.trace_events]
    assert any(k in kinds for k in ("fire_apis_done", "fire_apis_failed", "fire_apis_begin", "resolved"))