238 lines
8.5 KiB
Python
238 lines
8.5 KiB
Python
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
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revprec_allows_plain_invoice_explanation_without_llm():
|
|
text = "[FaturasAgent] Analisando a sua fatura atual, a variação foi de R$ 46,99 com Tamboro Mensal de R$ 14,99."
|
|
|
|
class MustNotRunLLM:
|
|
async def ainvoke(self, *args, **kwargs):
|
|
raise AssertionError("explicação informacional não deve ser julgada como execução prematura")
|
|
|
|
out = await TimPrematureActionRail().evaluate(
|
|
text,
|
|
{"guardrail_llm": MustNotRunLLM(), "mcp_results": []},
|
|
)
|
|
assert out.allowed is True
|
|
assert out.reason == "no_operational_completion_claim"
|
|
assert out.metadata["mechanism"] == "deterministic_claim_scope"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revprec_allows_composed_contestation_when_current_execution_structurally_proves_claims():
|
|
text = (
|
|
"Sua contestação do valor R$ 14,99 referente ao Tamboro Mensal foi registrada com sucesso. "
|
|
"O protocolo aberto é 1234567890. Um novo boleto será emitido e você recebeu um SMS."
|
|
)
|
|
|
|
class MustNotRunLLM:
|
|
async def ainvoke(self, *args, **kwargs):
|
|
raise AssertionError("evidência estrutural suficiente não deve ser re-julgada probabilisticamente")
|
|
|
|
ctx = {
|
|
"guardrail_llm": MustNotRunLLM(),
|
|
"mcp_results": [
|
|
{
|
|
"tool_name": "qualquer_tool_transacional",
|
|
"ok": True,
|
|
"result": {
|
|
"status": "COMPLETED",
|
|
"output": {
|
|
"registrar": {
|
|
"success": True,
|
|
"contestation_registered": True,
|
|
"contested_invoice_amount": "14.99",
|
|
"sr": "1234567890",
|
|
"sms_sent": True,
|
|
"barcode": "34191.79001 01043.510047",
|
|
"items": [
|
|
{"itemName": "Tamboro Mensal", "message": "Contestação criada com sucesso"}
|
|
],
|
|
}
|
|
},
|
|
},
|
|
}
|
|
],
|
|
}
|
|
out = await TimPrematureActionRail().evaluate(text, ctx)
|
|
assert out.allowed is True
|
|
assert out.reason == "current_execution_structurally_grounded"
|
|
assert out.metadata["mechanism"] == "deterministic_structured_execution_evidence"
|