from __future__ import annotations import sys import re import unicodedata from pathlib import Path from typing import Any, Callable from dotenv import load_dotenv from fastapi import FastAPI from pydantic import BaseModel, Field PROJECT_ROOT = Path(__file__).resolve().parents[3] FRAMEWORK_SRC = PROJECT_ROOT / "agent_framework_oci" / "libs" / "agent_framework" / "src" for entry in (PROJECT_ROOT, FRAMEWORK_SRC): if str(entry) not in sys.path: sys.path.insert(0, str(entry)) load_dotenv(PROJECT_ROOT / ".env", override=False) from agent_framework.checkpoints.langgraph_saver import create_langgraph_checkpointer from agent_framework.idempotency import create_idempotency_store from agent_framework.cache.cache import create_cache from agent_framework.config.settings import get_settings from agent_framework.workflows import FileWorkflowRepository, WorkflowRuntime from app.domain.contas import ContasDomainService from app.domain.contas.workflow_actions import build_contas_workflow_actions from app.domain.contas.invoice_resolver import InvoiceResolver from app.domain.contas.invoice_context import InvoiceContextService from app.domain.contas.item_matcher import SimilarityItemMatcher from app.domain.contas.vas_cancellation_message import compose_vas_cancellation_message from app.domain.contas.contestation_validation import validate_contestation_items from contas_mcp.servers.contas_mcp_server.authorized_lines_service import AuthorizedLinesMockService from contas_mcp.servers.contas_mcp_server.discount_history_service import DiscountHistoryMockService from contas_mcp.servers.contas_mcp_server.line_policy import ( POLICY_NAME as LINE_POLICY_NAME, POLICY_DESCRIPTION as LINE_POLICY_DESCRIPTION, apply_line_policy, ) app = FastAPI(title="TIM Contas MCP - Framework Native") service = ContasDomainService() invoice_resolver = InvoiceResolver(matcher=SimilarityItemMatcher()) settings = get_settings() _workflow_runtime: WorkflowRuntime | None = None _invoice_context_service: InvoiceContextService | None = None authorized_lines_service = AuthorizedLinesMockService(PROJECT_ROOT / "app" / "domain" / "contas" / "fixtures" / "authorized_lines.json") discount_history_service = DiscountHistoryMockService(PROJECT_ROOT / "app" / "domain" / "contas" / "fixtures" / "discount_history.json") def get_invoice_context_service() -> InvoiceContextService: global _invoice_context_service if _invoice_context_service is None: _invoice_context_service = InvoiceContextService(service.client, create_cache(settings)) return _invoice_context_service def get_workflow_runtime() -> WorkflowRuntime: """Inicializa WorkflowRuntime/checkpointer/idempotência somente no primeiro uso. Evita abrir Oracle/Redis durante import, health ou tools/list e garante que transações usem o IdempotencyStore selecionado pelo framework, não memória local. """ global _workflow_runtime if _workflow_runtime is None: idempotency_store = create_idempotency_store( settings, namespace="contas", require_durable=False ) _workflow_runtime = WorkflowRuntime( FileWorkflowRepository(PROJECT_ROOT / "workflows"), actions=build_contas_workflow_actions( service, idempotency_store=idempotency_store ), checkpointer=create_langgraph_checkpointer(settings), ) return _workflow_runtime class ToolCall(BaseModel): tool_name: str arguments: dict[str, Any] = Field(default_factory=dict) TOOLS: dict[str, dict[str, Any]] = { "consultar_faturas": {"description": "Consulta faturas do cliente, incluindo valor total quando disponível na fatura detalhada.", "input_schema": {"msisdn": "string"}}, "consultar_plano": {"description": "Consulta exclusivamente o plano ou planos contratados presentes na fatura.", "input_schema": {"msisdn": "string"}}, "invoice_explanation": {"description": "Executa o workflow de explicação de fatura com pause/resume pelo WorkflowRuntime do framework.", "input_schema": {"msisdn": "string"}}, "buscar_informacao": {"description": "Preserva a capability de conhecimento e delega a recuperação ao RAG do framework.", "input_schema": {"queries": "array"}}, "consultar_vas": {"description": "Consulta VAS ativos.", "input_schema": {"msisdn": "string"}}, "consultar_linhas_autorizadas": {"description": "Mock side-effect-free da integração que retorna as linhas autorizadas/relacionadas à identidade autenticada do atendimento.", "input_schema": {"msisdn": "string", "customer_key": "string", "contract_key": "string"}}, "consultar_historico_vas": {"description": "Consulta histórico de VAS.", "input_schema": {"msisdn": "string"}}, "consultar_historico_descontos": {"description": "Consulta o histórico autoritativo de descontos, incluindo status, valores, datas e causa explícita de término quando disponível.", "input_schema": {"msisdn": "string", "nome_plano": "string"}}, "cancelar_vas_avulso": {"description": "Executa workflow de cancelamento VAS após confirmação transacional do framework.", "input_schema": {"msisdn": "string", "subject": "string", "items": "array"}}, "tratar_vas_estrategico": {"description": "Executa workflow conversacional de VAS estratégico/bundle.", "input_schema": {"msisdn": "string", "subject": "string", "items": "array"}}, "validar_vas_subject": {"description": "Pré-valida se subject resolve para uma entidade VAS concreta sem efeitos colaterais.", "input_schema": {"msisdn": "string", "subject": "string", "target_tool": "string"}}, "validar_contestacao": {"description": "Pre-valida contestação sem executar efeitos transacionais.", "input_schema": {"msisdn": "string", "subject": "string", "valor": "number", "motivo": "string", "target_tool": "string"}}, "contestar_cobranca": {"description": "Executa workflow completo de contestação/Conta Certa.", "input_schema": {"msisdn": "string", "subject": "string", "valor": "number"}}, "pro_rata": {"description": "Executa workflow conversacional de pró-rata.", "input_schema": {"msisdn": "string", "planos": "array", "has_plano_controle": "boolean"}}, "termino_desconto": {"description": "Analisa término/retirada de desconto com evidência de fatura/plano; não infere causa ausente.", "input_schema": {"msisdn": "string", "nome_plano": "string"}}, "valor_divergente": {"description": "Executa capability de valor divergente.", "input_schema": {"msisdn": "string"}}, "retomar_workflow": {"description": "Retoma um workflow pausado pelo mesmo execution_id.", "input_schema": {"workflow_name": "string", "execution_id": "string", "resposta_usuario": "string"}}, "consultar_status_solicitacao": {"description": "Consulta/atualiza status técnico de solicitação/protocolo TIM.", "input_schema": {"msisdn": "string", "protocol": "string"}}, "enviar_sms": {"description": "Envia SMS por integração TIM.", "input_schema": {"msisdn": "string", "message": "string"}}, "recuperar_fatura_pdf": {"description": "Recupera SecurePDF usando contrato criptografado TIM.", "input_schema": {"msisdn": "string", "invoice_id": "string", "customer_id": "string"}}, "finalizar_atendimento": {"description": "Executa workflow de finalização e seus efeitos de domínio.", "input_schema": {"status": "string", "summary": "string", "msisdn": "string"}}, } def _require(args: dict[str, Any], *names: str) -> None: missing = [n for n in names if args.get(n) in (None, "")] if missing: raise ValueError("Parâmetros obrigatórios ausentes: " + ", ".join(missing)) def _clarification_option(item: Any) -> dict[str, Any]: name = str(getattr(item, "canonical_name", "") or "").strip() msisdn = str(getattr(item, "msisdn", "") or "").strip() charge_date = str(getattr(item, "charge_date", "") or "").strip() value = getattr(item, "value", None) details = [] if msisdn: details.append(f"linha final {msisdn[-4:]}") if charge_date: details.append(f"data {charge_date}") if value is not None: details.append(f"R$ {value}") label = name + ((" — " + ", ".join(details)) if details else "") return {"label": label, "value": name, "msisdn": msisdn, "charge_date": charge_date} def _norm_invoice_name(value: Any) -> str: """Normalização conservadora para comparação exata de nomes de itens.""" text = str(value or "").strip().casefold().replace("+", " ") text = unicodedata.normalize("NFKD", text) text = "".join(ch for ch in text if not unicodedata.combining(ch)) text = re.sub(r"[^a-z0-9]+", " ", text) return re.sub(r"\s+", " ", text).strip() def _invoice_subject_catalog(args: dict[str, Any]) -> list[dict[str, Any]]: """Catálogo mínimo de itens reais da fatura, sem fuzzy matching. É usado como barreira de consistência antes do resolver especializado em VAS. """ found: dict[str, dict[str, Any]] = {} def add(name: Any, value: Any = None, category: Any = None, contestable: Any = None) -> None: canonical = str(name or "").strip() key = _norm_invoice_name(canonical) if not key: return item = found.setdefault(key, {"name": canonical, "value": value, "category": str(category or "").strip(), "contestable": contestable}) if item.get("value") in (None, "") and value not in (None, ""): item["value"] = value if not item.get("category") and category: item["category"] = str(category).strip() if item.get("contestable") is None and contestable is not None: item["contestable"] = contestable analysis = args.get("billing_analysis") if isinstance(analysis, dict): for section_name in ("currentInvoice", "invoiceVariation"): sections = analysis.get(section_name) if not isinstance(sections, list): continue for section in sections: if not isinstance(section, dict): continue category = section.get("type") or section.get("desc") items = section.get("items") if not isinstance(items, list): continue for item in items: if isinstance(item, dict): add(item.get("desc") or item.get("name"), item.get("value"), item.get("type") or category, item.get("contestable")) detail = args.get("invoice_detail") if isinstance(detail, dict): parsed = detail.get("parsed_content") if isinstance(detail.get("parsed_content"), dict) else detail if isinstance(parsed, dict): # parsed_content normalmente é agrupado por MSISDN. roots = list(parsed.values()) if parsed and all(isinstance(v, dict) for v in parsed.values()) else [parsed] for root in roots: if not isinstance(root, dict): continue plans = root.get("Planos") or root.get("Plano") if isinstance(plans, dict): for plan_name, plan_data in plans.items(): value = plan_data.get("valor_final") if isinstance(plan_data, dict) else None add(plan_name, value, "plano", False) for section_name, section_value in root.items(): if section_name in {"Planos", "Plano"} or not isinstance(section_value, list): continue for item in section_value: if isinstance(item, dict): add(item.get("desc") or item.get("name"), item.get("value") or item.get("valor_final"), item.get("classe") or section_name, item.get("contestable")) return list(found.values()) def _recover_explicit_contestation_subject(args: dict[str, Any]) -> dict[str, Any] | None: """Recupera item explicitamente citado no texto original da contestação. Na confirmação a mensagem corrente pode ser apenas ``sim``. A fala que abriu a transação pode chegar por campos diferentes conforme o runtime/provedor (por exemplo ``query``, ``operator_instructions``, ``message``, ``text``, ``motivo`` ou ``descricao``). Essas fontes textuais são mais fortes que um ``subject`` residual. Só corrige quando há exatamente um nome canônico da fatura explicitamente presente; não faz fuzzy matching. """ text_source_fields = ( "query", "operator_instructions", "message", "text", "motivo", "descricao", ) original = " ".join( str(args.get(k) or "").strip() for k in text_source_fields if args.get(k) ).strip() if not original: return None norm_original = f" {_norm_invoice_name(original)} " matches = [] for item in _invoice_subject_catalog(args): name_norm = _norm_invoice_name(item.get("name")) if name_norm and f" {name_norm} " in norm_original: matches.append(item) return matches[0] if len(matches) == 1 else None def _preflight_subject(name: str, args: dict[str, Any]) -> dict[str, Any] | None: """Resolve/desambigua o item contra a evidência da fatura antes do workflow. A regra é de domínio; a persistência e a retomada da clarificação ficam no AgentRuntimeMixin. Quando o cliente já escolheu uma opção, não reabre o gate. """ if name not in {"cancelar_vas_avulso", "tratar_vas_estrategico", "contestar_cobranca"}: return None if bool(args.get("clarification_resolved")): return None # A pre-validation transaction boundary may already have resolved a multi-item # cancellation into authoritative ``items[]``. At execution time that list is # the source of truth; re-parsing the presentation ``subject`` (for example # ``"Tamboro Mensal, Paramount+"``) would incorrectly turn two canonical # entities back into one ambiguous free-text reference. if name == "cancelar_vas_avulso": resolved_items = [ item for item in (args.get("items") or []) if isinstance(item, dict) and str(item.get("name") or item.get("subject") or "").strip() ] if len(resolved_items) >= 2: return None msisdn = str(args.get("msisdn") or "").strip() subject = str(args.get("subject") or "").strip() if not msisdn or not subject: return None # ``validar_vas_subject`` is the authoritative, side-effect-free resolver # used at the transaction boundary. When it has already canonicalized the # subject and redirected the action to the strategic/bundle workflow, do not # resolve the same subject a second time against invoice evidence. Repeating # resolution with a different catalog can incorrectly turn a previously # ELIGIBLE subject into NEEDS_PARAMETER (the Aya Audiobooks case). if name == "tratar_vas_estrategico" and bool(args.get("_vas_subject_prevalidated")): return None # Contestação pode referenciar qualquer categoria da fatura. Antes do # resolver especializado em serviços/VAS, preserve um item explicitamente # citado no texto que abriu a transação. Isso impede que um plano explícito # seja reinterpretado como outro VAS durante a confirmação. if name == "contestar_cobranca": explicit = _recover_explicit_contestation_subject(args) if explicit is not None: previous_subject = subject canonical = str(explicit.get("name") or subject) args["subject"] = canonical subject = canonical if explicit.get("value") not in (None, ""): args["resolved_value"] = str(explicit.get("value")) category = str(explicit.get("category") or "").strip() if category: args["resolved_category"] = category if _norm_invoice_name(previous_subject) != _norm_invoice_name(canonical): args["_subject_corrected_from"] = previous_subject # O workflow/CVAL atual de contestar_cobranca trata VAS. Um plano # existente deve ser rejeitado com o próprio nome/valor, nunca # convertido em outro serviço. if _norm_invoice_name(category) in {"plano", "planos"}: return { "status": "OUT_OF_SCOPE", "success": False, "subject": canonical, "resolved_value": args.get("resolved_value"), "category": category or "plano", "error": f"O item '{canonical}' existe na fatura e é um plano, mas não pertence a uma categoria tratável por esta operação.", "metadata": { "subject_corrected_from": args.get("_subject_corrected_from"), "subject_source": "explicit_contestation_text", }, } try: invoice_detail = args.get("billing_analysis") if not isinstance(invoice_detail, dict): invoice_detail = service.client.billing_analysis(msisdn) if not isinstance(invoice_detail, dict): return None outcome = invoice_resolver.resolve([subject], invoice_detail) except Exception: # Fail-safe de disponibilidade: a própria operação ainda valida o serviço # no backend TIM. O gate de similaridade nunca transforma falha de leitura # em autorização para um item diferente. return None if outcome.ambiguous: matches = outcome.ambiguous[0].matches options = [_clarification_option(item) for item in matches] return { "status": "NEEDS_CLARIFICATION", "parameter": "subject", "question": f"Encontrei mais de uma cobrança parecida com '{subject}'. Qual delas você quis dizer?", "options": options, } if len(outcome.resolved) == 1: resolved = outcome.resolved[0] args["subject"] = resolved.canonical_name if resolved.msisdn: args.setdefault("item_msisdn", resolved.msisdn) if resolved.charge_date: args.setdefault("charge_date", resolved.charge_date) if getattr(resolved, "value", None) is not None: args.setdefault("resolved_value", str(resolved.value)) # Paridade do backend original: se o cliente pediu cancelamento avulso, # mas a própria fatura classifica o item como estratégico/bundle, não # executamos a operação errada. A resolução determinística do domínio # redireciona a mesma solicitação para o workflow VAS estratégico. if name == "cancelar_vas_avulso" and resolved.tool_category == "vas_estrategico": args["_domain_redirect"] = "tratar_vas_estrategico" args["type"] = resolved.item_type args["items"] = [{ "type": resolved.item_type, "msisdn": resolved.msisdn or msisdn, "name": resolved.canonical_name, }] if outcome.out_of_scope and not outcome.resolved: return { "status": "OUT_OF_SCOPE", "success": False, "error": f"O item '{subject}' existe na fatura, mas não pertence a uma categoria tratável por esta operação.", } if name in {"cancelar_vas_avulso", "tratar_vas_estrategico"} and not outcome.resolved: return { "status": "NEEDS_PARAMETER", "success": False, "parameter": "subject", "reason": "subject_not_resolved", "subject": subject, "metadata": {"side_effect_free": True, "entity_resolution": "invoice_evidence"}, } return None async def _enrich_invoice_context(name: str, args: dict[str, Any]) -> None: if name not in {"consultar_faturas", "consultar_plano", "invoice_explanation", "cancelar_vas_avulso", "tratar_vas_estrategico", "validar_vas_subject", "contestar_cobranca"}: return msisdn = str(args.get("msisdn") or "").strip() if not msisdn: return # If all required evidence is already present, preserve it exactly. For # consultar_faturas/invoice_explanation the semantic amount also matters, so # a cached pair without detail must still be enriched. has_base = isinstance(args.get("complete_invoices_payload"), dict) and isinstance(args.get("billing_analysis"), dict) needs_detail = name in {"consultar_faturas", "invoice_explanation", "cancelar_vas_avulso", "validar_vas_subject", "contestar_cobranca"} has_detail = isinstance(args.get("invoice_detail"), dict) or args.get("invoice_amount") not in (None, "") if has_base and (not needs_detail or has_detail): return session_id = str(args.get("session_id") or args.get("original_session_id") or args.get("conversation_key") or "").strip() invoice_id = str(args.get("invoice_id") or args.get("current_invoice_number") or "").strip() include_detail = name in {"consultar_faturas", "invoice_explanation", "cancelar_vas_avulso", "validar_vas_subject", "contestar_cobranca"} and not isinstance(args.get("invoice_detail"), dict) try: ctx = await get_invoice_context_service().get( session_id=session_id, msisdn=msisdn, invoice_id=invoice_id, use_cache=bool(args.get("use_invoice_context_cache", True)), include_detail=include_detail, message_id=str(args.get("message_id") or args.get("interaction_key") or ""), ) except Exception: return values = ctx.as_dict() if ctx.business_events: args["_prefetch_business_events"] = list(ctx.business_events) for key, value in values.items(): if value not in (None, ""): args.setdefault(key, value) if ctx.invoice_id: args.setdefault("invoice_id", ctx.invoice_id) args.setdefault("current_invoice_number", ctx.invoice_id) if ctx.customer_id: args.setdefault("customer_id", ctx.customer_id) def _with_prefetch_events(result: Any, args: dict[str, Any]) -> Any: events = args.get("_prefetch_business_events") if isinstance(args.get("_prefetch_business_events"), list) else [] if not events or not isinstance(result, dict): return result merged = dict(result) existing = merged.get("business_events") if isinstance(merged.get("business_events"), list) else [] merged["business_events"] = [*events, *existing] return merged def _workflow_payload(name: str, args: dict[str, Any]) -> dict[str, Any]: payload = dict(args) msisdn = str(args.get("msisdn") or "") subject = str(args.get("subject") or "") if name == "cancelamento_vas_avulso" and not isinstance(payload.get("items"), list): payload["items"] = [{"msisdn": msisdn, "name": subject, "type": "avulso"}] elif name == "vas_estrategico" and not isinstance(payload.get("items"), list): payload["items"] = [{"msisdn": msisdn, "name": subject, "type": str(args.get("type") or "estrategico")}] elif name == "contestacao_tool": # Paridade do backend original: o workflow usa tipo_atendimento para # selecionar branches/contratos de contestação. O wrapper antigo sempre # fixava esse valor quando a tool contestar_cobranca era chamada. payload.setdefault("tipo_atendimento", "contestacao") payload.setdefault("servico", subject) payload.setdefault("descricao", args.get("motivo") or "") payload.setdefault("items", [{"itemName": subject, "claimedAmount": args.get("valor"), "validatedAmount": args.get("valor")}]) return payload def _result_payload(result: Any, *, workflow_name: str) -> dict[str, Any]: data = result.model_dump() if hasattr(result, "model_dump") else dict(result) trace = data.get("trace") if isinstance(data.get("trace"), list) else [] last_node = "" for row in reversed(trace): if isinstance(row, dict) and row.get("node"): last_node = str(row.get("node")) break if not last_node and isinstance(data.get("state"), dict): last_node = str(data["state"].get("current_node") or "") metadata = { "workflow_name": workflow_name, "workflow_execution_id": data.get("execution_id"), "workflow_status": data.get("status"), "workflow_last_node": last_node or None, "resume_tool": "retomar_workflow" if data.get("status") == "PAUSED" else None, } payload = {**data, "metadata": metadata} # Generic workflow terminal contract. The MCP adapter promotes structural # terminal signals from the actual last node without knowing the workflow # business meaning. This lets the framework short-circuit composition for # handoff/end-session workflows in any domain. if data.get("status") == "COMPLETED" and isinstance(data.get("output"), dict): terminal_output = data["output"].get(last_node) if last_node else None if isinstance(terminal_output, dict): session_control = str(terminal_output.get("session_control") or "").strip().upper() terminal_status = str(terminal_output.get("terminal_status") or "").strip() is_terminal = ( terminal_output.get("terminal") is True or terminal_output.get("session_ended") is True or terminal_output.get("handoff") is True or bool(terminal_status) or session_control in {"HUMAN_HANDOFF", "END_SESSION"} ) if is_terminal: for key in ( "mensagem", "user_message", "message", "session_control", "human_handoff_requested", "handoff", "session_ended", "terminal_status", "handoff_reason", ): if key in terminal_output: payload[key] = terminal_output[key] payload["terminal"] = True payload["terminal_action"] = ( "handoff" if terminal_output.get("handoff") is True or session_control == "HUMAN_HANDOFF" else "end_session" ) # Compatibilidade funcional do antigo backend, agora derivada apenas do # branch determinístico do WorkflowRuntime do framework. if data.get("status") == "COMPLETED": if workflow_name == "invoice_explanation": if last_node == "registrar_protocolo_aceite": payload["recomenda_finalizacao"] = True payload["status_finalizacao_sugerido"] = "resolvido" elif last_node == "handoff_pos_explicacao_nao": terminal_output = (data.get("output") or {}).get(last_node) if isinstance(data.get("output"), dict) else {} terminal_output = terminal_output if isinstance(terminal_output, dict) else {} payload["mensagem"] = str(terminal_output.get("mensagem") or "Para continuar com a sua solicitação, aguarde um instante.") payload["session_control"] = "HUMAN_HANDOFF" payload["human_handoff_requested"] = True payload["handoff"] = True payload["session_ended"] = True payload["terminal_status"] = "human_handoff" payload["handoff_reason"] = str(terminal_output.get("handoff_reason") or "invoice_explanation_not_resolved") elif last_node == "finalizar_nao_resolvido": payload["recomenda_finalizacao"] = True payload["status_finalizacao_sugerido"] = "nao_resolvido" elif last_node == "resposta_falha_servico": payload["success"] = False payload["service_failed"] = True payload["auto_finalize_on_failure"] = False payload["mensagem"] = "Para continuar com a sua solicitação, aguarde um instante." elif workflow_name == "pro_rata" and last_node in {"registrar_aceitou", "registrar_nao_controle"}: payload["recomenda_finalizacao"] = True payload["status_finalizacao_sugerido"] = "resolvido" return payload async def _run_workflow(workflow_name: str, args: dict[str, Any]) -> dict[str, Any]: # A workflow execution belongs to one transaction, never to the whole # conversation session. Starting a workflow is therefore always a NEW # execution. The only path allowed to reuse an execution id is # ``retomar_workflow`` below, which calls ``aresume`` explicitly. # # Do not trust/forward a residual ``workflow_execution_id`` from a previous # transaction: WorkflowRuntime.arun() will allocate a fresh UUID. payload_args = dict(args) payload_args.pop("workflow_execution_id", None) result = await get_workflow_runtime().arun( workflow_name, _workflow_payload(workflow_name, payload_args), execution_id=None, ) payload = _result_payload(result, workflow_name=workflow_name) if workflow_name == "contestacao_tool" and payload.get("status") == "FAILED": details = payload.get("error_details") if isinstance(payload.get("error_details"), dict) else {} body = details.get("body") provider_message = "" if isinstance(body, dict): provider = body.get("provider") if isinstance(body.get("provider"), dict) else {} provider_message = str(provider.get("errorMessage") or body.get("description") or "").strip() protocol_node = (payload.get("output") or {}).get("registrar_protocolo") if isinstance(payload.get("output"), dict) else {} protocol = str((protocol_node or {}).get("protocolo_id") or (protocol_node or {}).get("protocol_number") or "") if isinstance(protocol_node, dict) else "" message = provider_message or str(payload.get("error") or "Falha ao executar contestação") payload["success"] = False payload["mensagem"] = message if provider_message: payload["contestation_error_description"] = provider_message if protocol: payload["contestacao_protocol"] = protocol payload["erro_sistemico"] = not bool(provider_message) payload["erro"] = "contestacao_nao_realizada" if provider_message else "erro_falha_sistema" if workflow_name == "contestacao_tool" and payload.get("status") == "COMPLETED": output = payload.get("output") if isinstance(payload.get("output"), dict) else {} contest = output.get("abrir_contestacao_cliente") if isinstance(output.get("abrir_contestacao_cliente"), dict) else {} if contest and contest.get("success") is False: reason = str(contest.get("guardrail_reason") or contest.get("contestation_error_description") or contest.get("error") or "Contestação não executada") payload["success"] = False payload["error"] = reason payload["erro"] = "erro_falha_sistema" if contest.get("blocked") else "contestacao_nao_realizada" payload["erro_sistemico"] = bool(contest.get("blocked")) payload["mensagem"] = reason if contest.get("contestation_error_description"): payload["contestation_error_description"] = contest.get("contestation_error_description") return _with_prefetch_events(payload, args) def _workflow_result_dict(result: Any) -> dict[str, Any]: return result.model_dump() if hasattr(result, "model_dump") else dict(result) def _node_output(result: Any, node_id: str) -> dict[str, Any]: data = _workflow_result_dict(result) output = data.get("output") if isinstance(data.get("output"), dict) else {} node = output.get(node_id) return dict(node) if isinstance(node, dict) else {} def _first_invoice_context(invoices: Any) -> tuple[dict[str, Any], dict[str, Any]]: if not isinstance(invoices, dict): return {}, {} billing = invoices.get("billingProfile") or invoices.get("billing_profile") or {} customer = billing.get("customer") if isinstance(billing, dict) and isinstance(billing.get("customer"), dict) else {} items = invoices.get("paymentItems") or invoices.get("payment_items") or [] invoice = next((x for x in items if isinstance(x, dict)), {}) if isinstance(items, list) else {} return dict(customer), dict(invoice) def _value_for_subject(args: dict[str, Any], subject: str) -> Any: for item in args.get("items") or []: if not isinstance(item, dict): continue name = str(item.get("name") or item.get("service") or item.get("subject") or item.get("desc") or "").strip() if name.casefold() == subject.casefold(): return item.get("value") if item.get("value") is not None else item.get("valor") if str(args.get("subject") or "").strip().casefold() == subject.casefold(): return args.get("resolved_value") if args.get("resolved_value") is not None else args.get("valor") return None def _invoice_detail_msisdn(invoice_detail: Any, subject: str) -> str: """Resolve the billed line for an item from invoice detail. The legacy behavior used the invoice evidence to fix LLM/tool arguments that accidentally attached a dependent item to the holder line. This remains a pure deterministic domain normalization before the workflows are invoked. """ needle = str(subject or "").strip().casefold() if not needle or not isinstance(invoice_detail, dict): return "" for line_key, sections in invoice_detail.items(): if not isinstance(sections, dict): continue for rows in sections.values(): if not isinstance(rows, list): continue for row in rows: if not isinstance(row, dict): continue desc = str(row.get("desc") or row.get("name") or row.get("description") or "").strip().casefold() if desc == needle: return str(row.get("msisdn") or line_key or "").strip() return "" def _normalize_cancel_items(args: dict[str, Any], holder_msisdn: str) -> list[dict[str, Any]]: items = [dict(x) for x in (args.get("items") or []) if isinstance(x, dict)] if not items and args.get("subject"): items = [{ "msisdn": args.get("item_msisdn") or holder_msisdn, "name": args.get("subject"), "value": args.get("resolved_value") if args.get("resolved_value") is not None else args.get("valor"), }] invoice_detail = args.get("invoice_detail") for item in items: subject = str(item.get("name") or item.get("service") or item.get("subject") or item.get("desc") or "").strip() resolved_line = _invoice_detail_msisdn(invoice_detail, subject) if resolved_line: item["msisdn"] = resolved_line elif not item.get("msisdn"): item["msisdn"] = holder_msisdn return items def _digits_only(value: Any) -> str: return "".join(ch for ch in str(value or "") if ch.isdigit()) def _money_ptbr(value: Any) -> str: text = str(value if value is not None else "0").strip() try: if "," in text: numeric = float(text.replace(".", "").replace(",", ".")) else: numeric = float(text) except (TypeError, ValueError): return text return f"{numeric:.2f}".replace(".", ",") def _nonzero_money_or(value: Any, fallback: Any) -> Any: try: text = str(value if value is not None else "").strip() if not text: return fallback numeric = float(text.replace(".", "").replace(",", ".") if "," in text else text) return value if abs(numeric) > 1e-9 else fallback except (TypeError, ValueError): return value if value not in (None, "") else fallback async def _run_cancelamento_com_contestacao(args: dict[str, Any]) -> dict[str, Any]: # The session/customer MSISDN is always the holder for the contestation. # Individual cancellation items may belong to dependent lines. holder_msisdn = str(args.get("holder_msisdn") or args.get("session_msisdn") or args.get("msisdn") or "") normalized_args = dict(args) normalized_args["msisdn"] = holder_msisdn # Paridade do backend original: CPF é alias aceito para social_sec_no e # deve chegar ao workflow sem máscara. social_sec_no = _digits_only(args.get("social_sec_no") or args.get("cpf")) if social_sec_no: normalized_args["social_sec_no"] = social_sec_no normalized_args["items"] = _normalize_cancel_items(args, holder_msisdn) # Nova transação de cancelamento = nova execução de workflow. Mesmo que # algum envelope antigo carregue ``workflow_execution_id``, ele não pode ser # reutilizado depois que a transação anterior terminou. Resume é tratado # exclusivamente por ``retomar_workflow``/``aresume``. normalized_args.pop("workflow_execution_id", None) cancel_result = await get_workflow_runtime().arun( "cancelamento_vas_avulso", _workflow_payload("cancelamento_vas_avulso", normalized_args), execution_id=None, ) cancel_data = _workflow_result_dict(cancel_result) if cancel_data.get("status") != "COMPLETED": return _result_payload(cancel_result, workflow_name="cancelamento_vas_avulso") cancel_node = _node_output(cancel_result, "cancelar_vas_avulso") candidate_rows = cancel_node.get("contestation_candidates") if not isinstance(candidate_rows, list): candidate_rows = cancel_node.get("itens_para_contestacao") candidates = [x for x in (candidate_rows or []) if isinstance(x, dict)] raw_results = [x for x in (cancel_node.get("results") or []) if isinstance(x, dict)] successful = [x for x in raw_results if x.get("success")] failed = [x for x in raw_results if not x.get("success")] # Compatibilidade do wrapper original: versões/actions podem reportar o # resultado agregado em cancelados/nao_cancelados sem repetir `success` em # results[]. O wrapper deve preservar esse outcome, não transformá-lo em falha. if not successful: successful = [ {**dict(x), "subject": x.get("subject") or x.get("servico") or x.get("name"), "success": True} for x in (cancel_node.get("cancelados") or []) if isinstance(x, dict) ] if not failed: failed = [ {**dict(x), "subject": x.get("subject") or x.get("servico") or x.get("name"), "success": False} for x in (cancel_node.get("nao_cancelados") or []) if isinstance(x, dict) ] if not candidates: line_protocols = [ str(row.get("protocolo_id") or row.get("protocol") or "").strip() for row in (cancel_node.get("protocolos_por_linha") or []) if isinstance(row, dict) and str(row.get("protocolo_id") or row.get("protocol") or "").strip() ] message = compose_vas_cancellation_message({ "success": bool(successful), "cancelados": [{"servico": x.get("subject")} for x in successful], "nao_encontrados": [{"servico": x.get("subject")} for x in failed if "não encontrado" in str(x.get("error") or "").casefold()], "protocols_for_response": line_protocols, }) cancel_data.setdefault("output", {})["resposta_cancelamento"] = {"mensagem": message} cancel_data.update({ "success": bool(successful), "mensagem": message, "recomenda_finalizacao": bool(successful), "status_finalizacao_sugerido": "resolvido" if successful else "nao_resolvido", "cancelamento_vas_protocol": line_protocols[0] if line_protocols else None, "protocols_for_response": line_protocols, "requires_protocol_in_response": bool(line_protocols), "auto_finalize_on_failure": False, }) cancel_data["metadata"] = { "workflow_name": "cancelamento_vas_avulso", "workflow_execution_id": cancel_data.get("execution_id"), "workflow_status": cancel_data.get("status"), "composite_workflows": ["cancelamento_vas_avulso"], } return cancel_data msisdn = holder_msisdn invoices = args.get("complete_invoices_payload") if not isinstance(invoices, dict): invoices = service.consultar_faturas(msisdn=msisdn) customer, invoice = _first_invoice_context(invoices) items: list[dict[str, Any]] = [] total = 0.0 for candidate in candidates: subject = str(candidate.get("subject") or (candidate.get("service") or {}).get("name") or "").strip() requested_value = _value_for_subject(normalized_args, subject) # Keep the amount claimed by the user separate from the amount validated # against VAS evidence. The cancellation result already carries the # canonical service object, so a missing user amount must never silently # become R$ 0,00. This also preserves mismatch evidence (e.g. user says # 29,98 while the active VAS is 14,99) for CVAL/observation downstream. service_evidence = candidate.get("service") if isinstance(candidate.get("service"), dict) else {} details = service_evidence.get("details") if isinstance(service_evidence.get("details"), dict) else {} validated_value = ( candidate.get("validated_amount") if candidate.get("validated_amount") is not None else candidate.get("validatedAmount") if candidate.get("validatedAmount") is not None else details.get("valor") if details.get("valor") is not None else service_evidence.get("valor") if service_evidence.get("valor") is not None else service_evidence.get("value") if service_evidence.get("value") is not None else service_evidence.get("price") ) if validated_value is None: validated_value = requested_value claimed_value = requested_value if requested_value is not None else validated_value try: text = str(validated_value if validated_value is not None else "0") numeric = float(text.replace(".", "").replace(",", ".")) if "," in text else float(text or 0) except Exception: numeric = 0.0 total += numeric claimed_money = _money_ptbr(claimed_value if claimed_value is not None else 0) validated_money = _money_ptbr(validated_value if validated_value is not None else claimed_value or 0) items.append({ "itemName": subject, "item_name": subject, "claimedAmount": claimed_money, "validatedAmount": validated_money, "claimed_amount": claimed_money, "validated_amount": validated_money, }) contest_payload = { **dict(normalized_args), "msisdn": msisdn, "social_sec_no": normalized_args.get("social_sec_no") or _digits_only(customer.get("document") or customer.get("socialSecNo")) or "", "customer_id": args.get("customer_id") or customer.get("customerId") or customer.get("id") or "", "current_invoice_number": args.get("current_invoice_number") or invoice.get("invoiceId") or invoice.get("invoiceNumber") or "", "invoice_id": args.get("invoice_id") or invoice.get("invoiceId") or "", "current_invoice_due_date": args.get("current_invoice_due_date") or invoice.get("dueDate") or "", "complete_invoices_payload": invoices, "servico": items[0]["itemName"] if items else str(args.get("subject") or ""), "valor": args.get("valor") if args.get("valor") is not None else total, "items": items, "item_msisdn": msisdn, "dependent_invoice_item": any(str(x.get("msisdn") or msisdn) != msisdn for x in candidates), } contest_result = await get_workflow_runtime().arun("contestacao_tool", contest_payload) contest_data = _workflow_result_dict(contest_result) contest_node_raw = _node_output(contest_result, "abrir_contestacao_cliente") protocol_node = _node_output(contest_result, "registrar_protocolo") cancellation_protocols = [ str(row.get("protocolo_id") or row.get("protocol") or "").strip() for row in (cancel_node.get("protocolos_por_linha") or []) if isinstance(row, dict) and str(row.get("protocolo_id") or row.get("protocol") or "").strip() ] partial_protocol = str( protocol_node.get("protocolo_id") or protocol_node.get("protocol_number") or contest_data.get("protocol_number") or "" ).strip() if contest_data.get("status") != "COMPLETED" or (contest_node_raw and contest_node_raw.get("success") is False): details = contest_data.get("error_details") if isinstance(contest_data.get("error_details"), dict) else {} provider_body = details.get("body") if isinstance(details, dict) else None provider_text = str(provider_body or "") contest_error = str( contest_node_raw.get("guardrail_reason") or contest_node_raw.get("contestation_error_description") or contest_data.get("contestation_error_description") or contest_data.get("error") or "Contestação não executada" ) normalized_error = f"{contest_error} {provider_text}".casefold() already_contested = any(token in normalized_error for token in ( "já contestado", "ja contestado", "itens já contestados", "itens ja contestados", "item_ja_contestado", "conflito de itens", )) response_protocols = list(dict.fromkeys([*cancellation_protocols, *([partial_protocol] if partial_protocol else [])])) if already_contested and successful: message = compose_vas_cancellation_message({ "success": True, "cancelados": [{"servico": x.get("subject")} for x in successful], "contestation_error_description": contest_error, "contestacao_protocol": partial_protocol, "protocols_for_response": response_protocols, }) return { **cancel_data, "status": "COMPLETED", "success": True, "error": None, "mensagem": message, "erro_sistemico": False, "auto_finalize_on_failure": False, "protocol_closed": False, "contestacao_protocol": partial_protocol or None, "cancelamento_vas_protocol": cancellation_protocols[0] if cancellation_protocols else None, "protocols_for_response": response_protocols, "requires_protocol_in_response": bool(response_protocols), "contestation_error_description": contest_error, "contestacao_workflow": contest_data, "metadata": { "workflow_name": "cancelamento_vas_avulso", "workflow_execution_id": cancel_data.get("execution_id"), "workflow_status": "COMPLETED", "composite_workflows": ["cancelamento_vas_avulso", "contestacao_tool"], }, } return { **cancel_data, "status": contest_data.get("status") or "FAILED", "error": contest_error, "success": False, "mensagem": contest_error, "erro_sistemico": True, "auto_finalize_on_failure": True, "contestacao_protocol": partial_protocol or None, "cancelamento_vas_protocol": cancellation_protocols[0] if cancellation_protocols else None, "protocols_for_response": response_protocols, "requires_protocol_in_response": bool(response_protocols), "contestation_error_description": contest_node_raw.get("contestation_error_description") or contest_data.get("contestation_error_description") or None, "contestacao_workflow": contest_data, "metadata": { "workflow_name": "cancelamento_vas_avulso", "workflow_execution_id": cancel_data.get("execution_id"), "workflow_status": contest_data.get("status"), "composite_workflows": ["cancelamento_vas_avulso", "contestacao_tool"], }, } contest_node = _node_output(contest_result, "abrir_contestacao_cliente") sms_node = _node_output(contest_result, "enviar_sms") protocol = str(protocol_node.get("protocolo_id") or protocol_node.get("protocol_number") or "") response_protocols = list(dict.fromkeys([*cancellation_protocols, *([protocol] if protocol else [])])) items_response = contest_node.get("items_response") if isinstance(contest_node.get("items_response"), list) else [] normalized_contested = contest_node.get("contested_items") normalized_not_contested = contest_node.get("not_contested_items") contested_items = [x for x in normalized_contested if isinstance(x, dict)] if isinstance(normalized_contested, list) else [x for x in items_response if isinstance(x, dict) and str(x.get("status") or "").upper() in {"CRIAR", "INICIADA", "ENVIADA", "SUCCESS", "SUCESSO"}] not_contested_items = [x for x in normalized_not_contested if isinstance(x, dict)] if isinstance(normalized_not_contested, list) else [x for x in items_response if isinstance(x, dict) and x not in contested_items] already_contested = contest_node.get("itens_ja_contestados") if not isinstance(already_contested, list): already_contested = contest_node.get("already_contested_items") if not isinstance(already_contested, list): already_contested = contest_data.get("itens_ja_contestados") if not isinstance(already_contested, list): already_contested = [] # Ausência de itemsResponse só implica usar os itens solicitados como # contestados quando o provider também não informou explicitamente que eles # já haviam sido contestados. Essa distinção evita afirmar um novo crédito # quando a operação foi recusada por conflito de negócio. if not items_response and not already_contested: contested_items = items message = compose_vas_cancellation_message({ "success": True, "cancelados": [{"servico": x.get("subject")} for x in successful], "nao_encontrados": [{"servico": x.get("subject")} for x in failed if "não encontrado" in str(x.get("error") or "").casefold()], "contested_items": contested_items, "not_contested_items": not_contested_items, "itens_ja_contestados": already_contested, "contested_invoice_amount_open": _money_ptbr(_nonzero_money_or(contest_node.get("contested_invoice_amount_open"), total)), "sms_sent": bool(sms_node), "sms_not_send_error": bool(sms_node and not sms_node.get("success", True)), "contestacao_protocol": protocol, "contestation_error_description": contest_node.get("contestation_error_description") or "", "protocols_for_response": response_protocols, }) combined_output = dict(cancel_data.get("output") or {}) combined_output["contestacao_tool"] = contest_data.get("output") or {} combined_output["resposta_cancelamento"] = {"mensagem": message} return { **cancel_data, "output": combined_output, "contestacao_workflow": contest_data, "success": bool(successful) or bool(contested_items), "contestacao_protocol": protocol, "cancelamento_vas_protocol": cancellation_protocols[0] if cancellation_protocols else None, "protocols_for_response": response_protocols, "requires_protocol_in_response": bool(response_protocols), "recomenda_finalizacao": bool(successful) or bool(contested_items), "status_finalizacao_sugerido": "resolvido" if (successful or contested_items) else "nao_resolvido", "protocol_closed": bool(contest_node.get("protocol_closed") or protocol_node.get("protocol_closed")), "auto_finalize_on_failure": False, "sms_not_send_error": bool(sms_node and not sms_node.get("success", True)), "contestation_error_description": contest_node.get("contestation_error_description") or "", "mensagem": message, "metadata": { "workflow_name": "cancelamento_vas_avulso", "workflow_execution_id": cancel_data.get("execution_id"), "workflow_status": "COMPLETED", "composite_workflows": ["cancelamento_vas_avulso", "contestacao_tool"], }, } def _derive_pro_rata_plans(args: dict[str, Any]) -> list[dict[str, Any]]: """Deriva os dois planos contratuais do PDF parseado, sem LLM.""" detail = args.get("invoice_detail") if not isinstance(detail, dict): return [dict(x) for x in (args.get("planos") or []) if isinstance(x, dict)] payload = detail.get("parsed_content") if isinstance(detail.get("parsed_content"), dict) else detail candidates: list[dict[str, Any]] = [] for _key, section in payload.items() if isinstance(payload, dict) else []: if not isinstance(section, dict): continue plans = section.get("Planos") if not isinstance(plans, dict): continue # A visão por linha possui valores em dict; DANFE-COM possui listas e não # deve ser confundido com os planos contratuais consolidados. if not plans or not all(isinstance(v, dict) for v in plans.values()): continue for name, data in plans.items(): row = {"desc": str(name), **dict(data)} candidates.append(row) if candidates: break return candidates def _prepare_pro_rata(args: dict[str, Any]) -> dict[str, Any] | None: plans = _derive_pro_rata_plans(args) args["planos"] = plans args["has_plano_controle"] = any(bool(p.get("is_controle")) or "controle" in str(p.get("desc") or "").casefold() or "ctrl" in str(p.get("desc") or "").casefold() for p in plans) if len(plans) != 2: return { "success": False, "reason": "requires_exactly_two_plans", "plans_found": len(plans), } return None async def _validate_contestation(args: dict[str, Any]) -> dict[str, Any]: """Business-owned, side-effect-free eligibility validation for contestation.""" await _enrich_invoice_context("contestar_cobranca", args) preflight = _preflight_subject("contestar_cobranca", args) if preflight is not None: status = str(preflight.get("status") or "OUT_OF_SCOPE") if isinstance(preflight, dict) else "OUT_OF_SCOPE" if status == "NEEDS_CLARIFICATION": return {**dict(preflight), "eligible": False} return {**dict(preflight), "eligible": False} invoice_payload = args.get("billing_analysis") if isinstance(args.get("billing_analysis"), dict) else {} requested_value = args.get("valor") if args.get("valor") not in (None, "") else args.get("resolved_value") canonical_items = [{ "item_name": str(args.get("resolved_subject") or args.get("subject") or "").strip(), "claimed_amount": requested_value, "validated_amount": requested_value, }] validated, validation_log, validation_error = validate_contestation_items(canonical_items, invoice_payload) if validation_error: # ``subject`` is an entity reference, not arbitrary free text. If CVAL # proves that the extracted value does not resolve to any concrete item in # the authoritative invoice evidence, request recollection of that single # parameter instead of terminating the transaction. This intentionally # avoids word blacklists: the evidence decides whether the reference is # concrete. Other CVAL failures (amount/category/business rules) remain # terminal and fail closed. unresolved_subject = any( str(entry.get("erro") or "") == "item_nao_encontrado_na_fatura" for entry in (validation_log or []) if isinstance(entry, dict) ) if unresolved_subject: return { "eligible": False, "status": "NEEDS_PARAMETER", "parameter": "subject", "reason": "subject_not_resolved", "subject": args.get("resolved_subject") or args.get("subject"), "resolved_value": requested_value, "validation_log": validation_log, "items": validated, "metadata": { "side_effect_free": True, "target_tool": args.get("target_tool") or "contestar_cobranca", "guardrail_code": "CVAL", "entity_resolution": "invoice_evidence", }, } # A claimed amount larger/different from the authoritative billed amount is # recoverable input, not a reason to emit a generic safety failure. Keep the # subject frozen and recollect only ``valor``, while telling the customer the # concrete amount found in the invoice evidence. amount_mismatch = next(( entry for entry in (validation_log or []) if isinstance(entry, dict) and str(entry.get("erro") or "") in { "valor_ajuste_maior_que_item", "valor_ajuste_divergente_item" } ), None) if amount_mismatch is not None: billed = amount_mismatch.get("valor_item_fatura") canonical_subject = str( amount_mismatch.get("item_fatura_resolvido") or args.get("resolved_subject") or args.get("subject") or "item" ).strip() billed_ptbr = _money_ptbr(billed) if billed not in (None, "") else "" message = ( f"O valor encontrado na fatura para {canonical_subject} é R$ {billed_ptbr}. " f"O valor informado (R$ {_money_ptbr(requested_value)}) não corresponde a essa cobrança. " "Informe o valor da cobrança que deseja contestar." if billed_ptbr else "O valor informado não corresponde à cobrança encontrada na fatura. Informe o valor correto da cobrança que deseja contestar." ) return { "eligible": False, "status": "NEEDS_PARAMETER", "parameter": "valor", "reason": "CVAL", "recoverable_reason": "amount_not_supported_by_invoice", "subject": canonical_subject, "resolved_value": billed, "parameter_message": message, "validation_log": validation_log, "items": validated, "metadata": { "side_effect_free": True, "target_tool": args.get("target_tool") or "contestar_cobranca", "guardrail_code": "CVAL", "entity_resolution": "invoice_evidence", }, } return { "eligible": False, "status": "BLOCKED", "reason": "CVAL", "error": validation_error, "subject": args.get("resolved_subject") or args.get("subject"), "resolved_value": requested_value, "category": args.get("resolved_category"), "validation_log": validation_log, "items": validated, "metadata": {"side_effect_free": True, "target_tool": args.get("target_tool") or "contestar_cobranca", "guardrail_code": "CVAL"}, } return { "eligible": True, "status": "ELIGIBLE", "subject": args.get("resolved_subject") or args.get("subject"), "resolved_value": requested_value, "category": args.get("resolved_category"), "validation_log": validation_log, "items": validated, "metadata": {"side_effect_free": True, "target_tool": args.get("target_tool") or "contestar_cobranca", "guardrail_code": "CVAL"}, } def _norm_entity_reference(value: Any) -> str: return " ".join(_norm_invoice_name(value).split()) def _vas_entity_catalog(args: dict[str, Any]) -> list[dict[str, Any]]: """Build a deduplicated catalog of concrete VAS entities from authorized evidence.""" found: dict[str, dict[str, Any]] = {} def add(name: Any, *, source: str, category: Any = None, cancelable: Any = None) -> None: canonical = str(name or "").strip() key = _norm_entity_reference(canonical) if not key: return row = found.setdefault(key, { "name": canonical, "sources": [], "category": str(category or "").strip(), "cancelable": cancelable, }) if source not in row["sources"]: row["sources"].append(source) if not row.get("category") and category: row["category"] = str(category).strip() if row.get("cancelable") is None and cancelable is not None: row["cancelable"] = bool(cancelable) # Active VAS is authoritative for currently provisioned products. msisdn = str(args.get("msisdn") or "").strip() if msisdn: try: current = service.client.consultar_vas(msisdn) products = current.get("products") or current.get("services") or [] if isinstance(current, dict) else [] for item in products if isinstance(products, list) else []: if not isinstance(item, dict): continue can = item.get("can") if isinstance(item.get("can"), dict) else {} add(item.get("name") or item.get("description"), source="consultar_vas", category="vas_ativo", cancelable=can.get("cancel")) except Exception: # Invoice evidence below still gives a safe, side-effect-free resolver. pass # Billing/invoice evidence covers strategic/bundle items that may not be in the # active-VAS endpoint representation. for item in _invoice_subject_catalog(args): category = str(item.get("category") or "") cat_norm = _norm_entity_reference(category) if any(token in cat_norm for token in ( "servicos contratados de parceiros", "streaming", "avulso", "estrategico", "bundle", "sva" )): add(item.get("name"), source="invoice_evidence", category=category, cancelable=item.get("contestable")) return list(found.values()) def _resolve_catalog_entity(subject: str, catalog: list[dict[str, Any]]) -> tuple[str | None, list[dict[str, Any]]]: needle = _norm_entity_reference(subject) if not needle: return None, [] exact = [item for item in catalog if _norm_entity_reference(item.get("name")) == needle] if len(exact) == 1: return str(exact[0].get("name") or subject), exact partial = [ item for item in catalog if needle in _norm_entity_reference(item.get("name")) or _norm_entity_reference(item.get("name")) in needle ] # Deduplication by canonical name prevents repeated charges of the same named # service from becoming a false ambiguity at the entity-name level. unique: dict[str, dict[str, Any]] = {_norm_entity_reference(i.get("name")): i for i in partial} matches = list(unique.values()) if len(matches) == 1: return str(matches[0].get("name") or subject), matches return None, matches def _vas_domain_policy_from_invoice_detail(canonical: str, args: dict[str, Any]) -> tuple[str, str]: """Resolve the business class from the detailed invoice evidence. ``billing_analysis`` is useful for entity discovery, but it can flatten the same service into broad sections such as ``streaming`` or partner services. The parsed invoice detail carries the domain-owned ``classe`` attribute used by Contas (``avulso``, ``estrategico`` or ``bundle``), so it is the authoritative source for choosing the business treatment after canonicalization. Returns ``(tool_category, item_type)``. An empty pair means that the detail does not provide an unambiguous classification and the specialized resolver should be tried next. """ target = _norm_entity_reference(canonical) if not target: return "", "" detail = args.get("invoice_detail") if not isinstance(detail, dict): return "", "" parsed = detail.get("parsed_content") if isinstance(detail.get("parsed_content"), dict) else detail if not isinstance(parsed, dict): return "", "" roots = list(parsed.values()) if parsed and all(isinstance(v, dict) for v in parsed.values()) else [parsed] classes: set[str] = set() for root in roots: if not isinstance(root, dict): continue for section_name, section_value in root.items(): if not isinstance(section_value, list): continue for item in section_value: if not isinstance(item, dict): continue name = item.get("desc") or item.get("name") if _norm_entity_reference(name) != target: continue raw_class = str(item.get("classe") or "").strip().casefold() if raw_class: classes.add(raw_class) elif _norm_entity_reference(section_name) == _norm_entity_reference("Streamings"): classes.add("estrategico") # Conflicting evidence must not silently change the action. normalized = { "estrategico" if c in {"estrategico", "estrategica", "strategic"} else "bundle" if c == "bundle" else "avulso" if c in {"avulso", "avulsa"} else c for c in classes } normalized.discard("") if len(normalized) != 1: return "", "" item_type = next(iter(normalized)) if item_type in {"estrategico", "bundle"}: return "vas_estrategico", item_type if item_type == "avulso": return "cancelar_vas_avulso", item_type return "", item_type def _resolve_multiple_vas_subjects(subject: str, catalog: list[dict[str, Any]], args: dict[str, Any]) -> list[dict[str, Any]]: """Resolve explicit multi-entity cancellation references against authorized evidence. Named references are split into conversational segments and each segment must resolve uniquely through the same catalog resolver used for single entities. Explicit mass expansion is restricted to "todos os VAS/serviços avulsos"; generic "todos" is deliberately not expanded. """ raw = str(subject or "").strip() norm = _norm_entity_reference(raw) if not norm: return [] explicit_all_avulso = any(token in norm for token in ( "todos os vas avulsos", "todos vas avulsos", "todos os servicos avulsos", "todos servicos avulsos", "todas as assinaturas avulsas", )) selected_names: list[str] = [] if explicit_all_avulso: selected_names = [str(row.get("name") or "").strip() for row in catalog if row.get("name")] else: cleaned = re.sub(r"(?i)\b(cancelar|cancela|cancele|retirar|retire|tirar|tire|desativar|desative)\b", " ", raw) segments = [seg.strip(" .:-") for seg in re.split(r"\s*(?:,|;|/|\be\b|\bmais\b)\s*", cleaned, flags=re.I) if seg.strip(" .:-")] if len(segments) < 2: return [] for segment in segments: canonical, _ = _resolve_catalog_entity(segment, catalog) if canonical: selected_names.append(canonical) selected: list[dict[str, Any]] = [] for name in list(dict.fromkeys(selected_names)): resolved_class, resolved_type = _vas_domain_policy_from_invoice_detail(name, args) if explicit_all_avulso and resolved_class != "cancelar_vas_avulso": continue selected.append({ "name": name, "msisdn": _invoice_detail_msisdn(args.get("invoice_detail"), name) or str(args.get("msisdn") or ""), "tool_category": resolved_class or None, "item_type": resolved_type or None, }) return selected async def _validate_vas_subject(args: dict[str, Any]) -> dict[str, Any]: """Side-effect-free entity resolution used before confirmation/execution.""" await _enrich_invoice_context("validar_vas_subject", args) subject = str(args.get("subject") or "").strip() catalog = _vas_entity_catalog(args) canonical, matches = _resolve_catalog_entity(subject, catalog) requested_tool = str(args.get("target_tool") or "").strip() multi = _resolve_multiple_vas_subjects(subject, catalog, args) if requested_tool == "cancelar_vas_avulso" else [] if len(multi) >= 2 or (multi and any(token in _norm_entity_reference(subject) for token in ("todos os vas avulsos", "todos os servicos avulsos"))): names = [str(item.get("name") or "").strip() for item in multi if item.get("name")] return { "eligible": True, "status": "ELIGIBLE", "subject": subject, "resolved_subject": names[0] if len(names) == 1 else ", ".join(names), "resolved_subjects": names, "entity_resolution": "vas_evidence_multiple", "transaction_decision": { "resolved_arguments": {"subject": ", ".join(names), "items": multi}, "target_tool": requested_tool, "action_changed": False, "requires_reconfirmation": False, "confirmation_message": ( f"Você confirma o cancelamento dos serviços {', '.join(names)}? " "Responda 'sim' para executar ou 'não' para cancelar." ), "domain_policy": {"class": "cancelar_vas_avulso", "item_type": "multiple"}, }, "metadata": {"side_effect_free": True, "target_tool": requested_tool, "catalog_size": len(catalog), "resolved_count": len(names)}, } if canonical: effective_tool = requested_tool resolved_class = "" resolved_type = "" # Domain-owned policy revalidation: canonicalization may reveal that the # requested entity belongs to a different business treatment. The MCP # validator decides that mapping; the framework only applies the generic # transaction_decision contract returned below. try: # Prefer the detailed invoice because it preserves the Contas-owned # ``classe`` attribute. This prevents a canonical entity such as # Youtube Premium from being correctly resolved but then executed by # the previously selected avulso tool. resolved_class, resolved_type = _vas_domain_policy_from_invoice_detail(canonical, args) # Compatibility/fallback for sources that do not expose parsed detail. if not resolved_class: evidence_candidates: list[dict[str, Any]] = [] detail = args.get("invoice_detail") if isinstance(detail, dict): parsed = detail.get("parsed_content") if isinstance(detail.get("parsed_content"), dict) else detail if isinstance(parsed, dict): evidence_candidates.append(parsed) billing_analysis = args.get("billing_analysis") if not isinstance(billing_analysis, dict): billing_analysis = service.client.billing_analysis(str(args.get("msisdn") or "")) if isinstance(billing_analysis, dict): evidence_candidates.append(billing_analysis) for evidence in evidence_candidates: outcome = invoice_resolver.resolve([canonical], evidence) if outcome and len(outcome.resolved) == 1: resolved = outcome.resolved[0] resolved_class = str(getattr(resolved, "tool_category", "") or "") resolved_type = str(getattr(resolved, "item_type", "") or "") break if requested_tool == "cancelar_vas_avulso" and resolved_class == "vas_estrategico": effective_tool = "tratar_vas_estrategico" except Exception: # Resolution already succeeded against authorized VAS evidence. If the # domain classifier is unavailable, preserve the original action rather # than guessing another business operation. effective_tool = requested_tool action_changed = bool(effective_tool and requested_tool and effective_tool != requested_tool) confirmation_message = "" if action_changed: confirmation_message = ( f"Identifiquei o serviço {canonical}. Esse serviço possui tratamento específico. " "Você deseja prosseguir? Responda 'sim' para continuar ou 'não' para cancelar." ) return { "eligible": True, "status": "ELIGIBLE", "subject": canonical, "resolved_subject": canonical, "entity_resolution": "vas_evidence", "transaction_decision": { "resolved_arguments": { "subject": canonical, "_vas_subject_prevalidated": True, **({"type": resolved_type} if resolved_type else {}), }, "target_tool": effective_tool or requested_tool, "action_changed": action_changed, "requires_reconfirmation": action_changed, "confirmation_message": confirmation_message, "domain_policy": { "class": resolved_class or None, "item_type": resolved_type or None, }, }, "metadata": { "side_effect_free": True, "target_tool": requested_tool, "effective_tool": effective_tool or requested_tool, "catalog_size": len(catalog), }, } return { "eligible": False, "status": "NEEDS_PARAMETER", "parameter": "subject", "reason": "subject_not_resolved" if not matches else "subject_ambiguous", "subject": subject, "options": [str(item.get("name") or "") for item in matches if item.get("name")], "entity_resolution": "vas_evidence", "metadata": {"side_effect_free": True, "target_tool": args.get("target_tool"), "catalog_size": len(catalog)}, } def _line_policy_result(name: str, args: dict[str, Any]) -> dict[str, Any] | None: decision = apply_line_policy(name, args) args["_line_policy_name"] = LINE_POLICY_NAME args["_line_policy_authenticated_msisdn"] = str(args.get("msisdn") or "") if decision.requested_reference: args["_line_policy_requested_reference"] = decision.requested_reference if decision.allowed: if decision.effective_msisdn: args["msisdn"] = decision.effective_msisdn args["_line_policy_effective_msisdn"] = decision.effective_msisdn return None return { "success": False, "status": "LINE_POLICY_BLOCKED", "terminal": True, "terminal_action": "block", "reason": decision.reason or "line_policy_blocked", "message": decision.user_message, "user_message": decision.user_message, "metadata": { "line_policy": LINE_POLICY_NAME, "line_policy_description": LINE_POLICY_DESCRIPTION, "side_effect_free": True, "requested_line_reference": decision.requested_reference, }, } async def _invoke(name: str, args: dict[str, Any]) -> Any: requirements = { "consultar_faturas": ("msisdn",), "consultar_plano": ("msisdn",), "invoice_explanation": ("msisdn",), "buscar_informacao": (), "consultar_vas": ("msisdn",), "consultar_linhas_autorizadas": ("msisdn",), "consultar_historico_vas": ("msisdn",), "consultar_historico_descontos": ("msisdn",), "cancelar_vas_avulso": ("msisdn", "subject"), "tratar_vas_estrategico": ("msisdn", "subject"), "validar_vas_subject": ("msisdn", "subject"), "validar_contestacao": ("msisdn", "subject", "valor"), "contestar_cobranca": ("msisdn", "subject", "valor"), "pro_rata": ("msisdn",), "termino_desconto": ("msisdn",), "valor_divergente": ("msisdn",), "enviar_sms": ("msisdn", "message"), "recuperar_fatura_pdf": ("msisdn", "invoice_id"), "retomar_workflow": ("workflow_name", "execution_id", "resposta_usuario"), } _require(args, *requirements.get(name, ())) if name == "buscar_informacao": return service.buscar_informacao(queries=args.get("queries") or []) if name == "consultar_linhas_autorizadas": return authorized_lines_service.consultar_linhas_autorizadas( authenticated_msisdn=str(args.get("msisdn") or ""), customer_key=str(args.get("customer_key") or "") or None, contract_key=str(args.get("contract_key") or "") or None, ) # ALT2 consulta explicitamente a fonte autoritativa de linhas relacionadas. # A referência falada pelo cliente nunca é usada para conceder autorização. if LINE_POLICY_NAME == "authorized_related_lines" and isinstance(args.get("requested_line_reference"), dict): args["authorized_lines_evidence"] = authorized_lines_service.consultar_linhas_autorizadas( authenticated_msisdn=str(args.get("msisdn") or ""), customer_key=str(args.get("customer_key") or "") or None, contract_key=str(args.get("contract_key") or "") or None, ) # Carrega evidência da conta usando a linha autenticada. A política ativa é # aplicada somente depois disso: alt1 bloqueia qualquer outra linha; alt2 # pode resolver uma linha relacionada a partir dessa evidência autorizada. await _enrich_invoice_context(name, args) line_block = _line_policy_result(name, args) if line_block is not None: return _with_prefetch_events(line_block, args) if name == "consultar_historico_descontos": return discount_history_service.consultar_historico_descontos( msisdn=str(args.get("msisdn") or ""), customer_key=str(args.get("customer_key") or "") or None, contract_key=str(args.get("contract_key") or "") or None, nome_plano=str(args.get("nome_plano") or "") or None, ) if name == "termino_desconto": # A própria capability garante a consulta à fonte autoritativa; não # depende de o LLM selecionar uma segunda tool para obter grounding. args["discount_evidence"] = discount_history_service.consultar_historico_descontos( msisdn=str(args.get("msisdn") or ""), customer_key=str(args.get("customer_key") or "") or None, contract_key=str(args.get("contract_key") or "") or None, nome_plano=str(args.get("nome_plano") or "") or None, ) if name == "validar_vas_subject": return await _validate_vas_subject(args) if name == "validar_contestacao": return await _validate_contestation(args) if name == "pro_rata": prepared = _prepare_pro_rata(args) if prepared is not None: return prepared preflight = _preflight_subject(name, args) if preflight is not None: return _with_prefetch_events(preflight, args) domain_redirect = str(args.pop("_domain_redirect", "") or "") if domain_redirect == "tratar_vas_estrategico": redirected = await _run_workflow("vas_estrategico", args) redirected.setdefault("metadata", {})["domain_redirect_from"] = name redirected["metadata"]["domain_redirect_to"] = domain_redirect return _with_prefetch_events(redirected, args) workflow_map = { "invoice_explanation": "invoice_explanation", "cancelar_vas_avulso": "cancelamento_vas_avulso", "tratar_vas_estrategico": "vas_estrategico", "contestar_cobranca": "contestacao_tool", "pro_rata": "pro_rata", "termino_desconto": "termino_desconto", "valor_divergente": "valor_divergente", "finalizar_atendimento": "finalizar_atendimento", } if name == "cancelar_vas_avulso": return _with_prefetch_events(await _run_cancelamento_com_contestacao(args), args) if name in workflow_map: return await _run_workflow(workflow_map[name], args) if name == "retomar_workflow": result = await get_workflow_runtime().aresume( str(args["workflow_name"]), str(args["execution_id"]), {"resposta_usuario": args["resposta_usuario"]}, ) return _with_prefetch_events(_result_payload(result, workflow_name=str(args["workflow_name"])), args) fn: Callable[..., Any] = getattr(service, name) call_args = {k: v for k, v in args.items() if not str(k).startswith("_")} return _with_prefetch_events(fn(**call_args), args) @app.get("/health") async def health() -> dict[str, Any]: return { "status": "ok", "architecture": "framework-native", "legacy_dependency": False, "workflow_engine": "agent_framework.workflows.WorkflowRuntime", "langgraph_direct_import": False, "checkpoint_provider": getattr(settings, "CHECKPOINT_REPOSITORY_PROVIDER", "memory"), "gateway_mode": "mock" if service.client.mock else "real", "line_policy": LINE_POLICY_NAME, "line_policy_description": LINE_POLICY_DESCRIPTION, "tools": len(TOOLS), "env_file": str(PROJECT_ROOT / ".env"), } @app.get("/mcp/tools/list") async def list_tools() -> dict[str, Any]: return {"tools": [{"name": name, **definition} for name, definition in TOOLS.items()]} @app.post("/mcp/tools/call") async def call_tool(call: ToolCall) -> dict[str, Any]: if call.tool_name not in TOOLS: return {"ok": False, "error": f"Tool não encontrada: {call.tool_name}"} try: result = await _invoke(call.tool_name, dict(call.arguments or {})) failed = isinstance(result, dict) and (result.get("success") is False or result.get("status") == "FAILED") metadata = {"server": "contas", "tool": call.tool_name, "framework_native": True} if isinstance(result, dict) and isinstance(result.get("metadata"), dict): metadata.update(result["metadata"]) if failed: return {"ok": False, "error": result.get("error") or "Falha de domínio", "result": result, "metadata": metadata} return {"ok": True, "result": result, "metadata": metadata} except Exception as exc: return {"ok": False, "error": f"{type(exc).__name__}: {exc}", "metadata": {"server": "contas", "tool": call.tool_name, "framework_native": True}}