69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.workflows.agent_graph import AgentWorkflow
|
|
|
|
|
|
def test_output_guardrail_context_exposes_structural_handoff_evidence():
|
|
state = {
|
|
'context': {},
|
|
'user_text': 'quero falar com um atendente',
|
|
'route': 'human_handoff',
|
|
'intent': 'human_handoff',
|
|
'session_control': 'HUMAN_HANDOFF',
|
|
'human_handoff_requested': True,
|
|
'route_decision': {
|
|
'route': 'human_handoff',
|
|
'agent': 'human_handoff',
|
|
'intent': 'human_handoff',
|
|
'handoff': True,
|
|
'metadata': {'session_control': 'HUMAN_HANDOFF'},
|
|
},
|
|
'mcp_results': [{'tool_name': 'invoice_explanation', 'ok': True, 'result': {'status': 'PAUSED'}}],
|
|
}
|
|
|
|
ctx = AgentWorkflow._output_guardrail_context(state)
|
|
|
|
assert ctx['current_route'] == 'human_handoff'
|
|
assert ctx['current_intent'] == 'human_handoff'
|
|
assert ctx['session_control'] == 'HUMAN_HANDOFF'
|
|
assert ctx['human_handoff_requested'] is True
|
|
assert ctx['handoff'] is True
|
|
assert ctx['route_decision']['handoff'] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_human_handoff_clears_live_paused_workflow_and_transaction_state():
|
|
graph = object.__new__(AgentWorkflow)
|
|
|
|
class _Span:
|
|
async def __aenter__(self): return self
|
|
async def __aexit__(self, *args): return False
|
|
|
|
class _Telemetry:
|
|
def span(self, *args, **kwargs): return _Span()
|
|
async def event(self, *args, **kwargs): return None
|
|
|
|
class _Settings:
|
|
HUMAN_HANDOFF_MESSAGE = 'Vou encaminhar seu atendimento para uma pessoa.'
|
|
|
|
graph.telemetry = _Telemetry()
|
|
graph.settings = _Settings()
|
|
graph.tool_router = None
|
|
|
|
result = await graph.human_handoff({
|
|
'session_id': 's1',
|
|
'pending_domain_workflow': {'execution_id': 'wf1', 'status': 'PAUSED'},
|
|
'active_transaction': {'tool_name': 'invoice_explanation', 'status': 'WORKFLOW_PAUSED'},
|
|
'mcp_results': [{'tool_name': 'invoice_explanation', 'ok': True}],
|
|
'route_decision': {'reason': 'pedido explicito'},
|
|
})
|
|
|
|
assert result['session_control'] == 'HUMAN_HANDOFF'
|
|
assert result['pending_domain_workflow'] is None
|
|
assert result['active_transaction'] is None
|
|
assert result['mcp_results'] == []
|
|
assert result['transaction_status'] == 'CANCELLED'
|
|
assert result['confirmation_required'] is False
|