agent_contas/tests/migration/test_revprec_epistemic_uncertainty.py

from __future__ import annotations

import pytest

from app.extensions.tim_guardrails import TimPrematureActionRail


class FakeLLM:
    def __init__(self, content: str):
        self.content = content

    async def ainvoke(self, messages, **kwargs):
        return self.content


@pytest.mark.asyncio
async def test_revprec_allows_structured_insufficient_evidence_without_calling_llm():
    text = "Identifiquei dados de desconto, mas os dados disponíveis não informam o motivo da retirada ou do término do desconto."

    class MustNotRunLLM:
        async def ainvoke(self, *args, **kwargs):
            raise AssertionError("LLM não deveria ser chamada no bypass estrutural")

    ctx = {
        "guardrail_llm": MustNotRunLLM(),
        "mcp_results": [
            {
                "tool_name": "termino_desconto",
                "ok": True,
                "result": {
                    "output": {
                        "formatar": {
                            "mensagem": text,
                            "epistemic_status": "insufficient_evidence",
                            "discount_reason_grounded": False,
                        }
                    }
                },
            }
        ],
    }
    out = await TimPrematureActionRail().evaluate(text, ctx)
    assert out.allowed is True
    assert out.reason == "current_execution_authoritative_output"
    assert out.metadata["mechanism"] == "deterministic_current_execution_evidence"


@pytest.mark.asyncio
async def test_revprec_does_not_bypass_if_message_does_not_match_structured_result():
    structured = "Os dados disponíveis não informam o motivo."
    text = "Cancelei o serviço com sucesso."
    ctx = {
        "guardrail_llm": FakeLLM("1"),
        "mcp_results": [{"result": {"output": {"mensagem": structured, "epistemic_status": "insufficient_evidence"}}}],
    }
    out = await TimPrematureActionRail().evaluate(text, ctx)
    assert out.allowed is False


@pytest.mark.asyncio
async def test_revprec_still_blocks_real_premature_action_without_structured_marker():
    out = await TimPrematureActionRail().evaluate(
        "Cancelei o serviço com sucesso.",
        {"guardrail_llm": FakeLLM("1"), "mcp_results": []},
    )
    assert out.allowed is False


def test_termino_desconto_declares_epistemic_status():
    from app.domain.contas.service import ContasDomainService
    from app.domain.contas.workflow_actions import build_contas_workflow_actions

    action = build_contas_workflow_actions(ContasDomainService()).get("formatar_capability_resposta")
    no_reason = action({"tipo": "termino_desconto"}, {"input": {}})
    with_reason = action(
        {"tipo": "termino_desconto", "discount_evidence": {"discount_reason": "fim da campanha"}},
        {"input": {}},
    )
    assert no_reason["epistemic_status"] == "insufficient_evidence"
    assert with_reason["epistemic_status"] == "grounded_fact"

@pytest.mark.asyncio
async def test_revprec_allows_exact_authoritative_message_from_any_successful_tool_without_llm():
    text = "Operação concluída com sucesso. Protocolo 1234567890."

    class MustNotRunLLM:
        async def ainvoke(self, *args, **kwargs):
            raise AssertionError("LLM não deveria re-julgar mensagem autoritativa exata")

    ctx = {
        "guardrail_llm": MustNotRunLLM(),
        "mcp_results": [
            {
                "tool_name": "qualquer_tool_futura",
                "ok": True,
                "result": {
                    "status": "COMPLETED",
                    "output": {"finalizar": {"mensagem": text, "success": True}},
                },
            }
        ],
    }
    out = await TimPrematureActionRail().evaluate(text, ctx)
    assert out.allowed is True
    assert out.metadata["mechanism"] == "deterministic_current_execution_evidence"


@pytest.mark.asyncio
async def test_revprec_composed_success_uses_current_execution_evidence_in_prompt():
    class InspectLLM:
        async def ainvoke(self, messages, **kwargs):
            prompt = messages[-1]["content"]
            assert 'qualquer_tool' in prompt
            assert '"success": true' in prompt.lower()
            assert "Serviço X" in prompt
            return "0"

    ctx = {
        "guardrail_llm": InspectLLM(),
        "mcp_results": [
            {
                "tool_name": "qualquer_tool",
                "ok": True,
                "result": {
                    "status": "COMPLETED",
                    "output": {"result": {"success": True, "subject": "Serviço X", "protocol": "ABC123"}},
                },
            }
        ],
    }
    out = await TimPrematureActionRail().evaluate("O Serviço X foi processado com sucesso.", ctx)
    assert out.allowed is True
    assert out.metadata["mechanism"] == "llm_current_execution_evidence"


@pytest.mark.asyncio
async def test_revprec_blocks_success_claim_when_current_execution_failed():
    class InspectLLM:
        async def ainvoke(self, messages, **kwargs):
            prompt = messages[-1]["content"]
            assert '"ok": false' in prompt.lower()
            assert "backend indisponível" in prompt
            return "1"

    ctx = {
        "guardrail_llm": InspectLLM(),
        "mcp_results": [
            {
                "tool_name": "qualquer_tool",
                "ok": False,
                "error": "backend indisponível",
                "result": {"status": "FAILED", "success": False},
            }
        ],
    }
    out = await TimPrematureActionRail().evaluate("A operação foi concluída com sucesso.", ctx)
    assert out.allowed is False
    assert "sem suporte" in out.reason


@pytest.mark.asyncio
async def test_revprec_does_not_use_state_or_history_as_authoritative_exact_message():
    text = "Cancelei o serviço com sucesso."
    ctx = {
        "guardrail_llm": FakeLLM("1"),
        "mcp_results": [
            {
                "tool_name": "qualquer_tool",
                "ok": True,
                "result": {
                    "status": "COMPLETED",
                    "state": {"history": [{"mensagem": text}]},
                    "output": {"result": {"success": False}},
                },
            }
        ],
    }
    out = await TimPrematureActionRail().evaluate(text, ctx)
    assert out.allowed is False