bugfix: route stickness precedences (transaction in the same intent)

contas_mcp/servers/contas_mcp_server/main.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import sys
+import re
+import unicodedata
 from pathlib import Path
 from typing import Any, Callable
 
@@ -70,6 +72,7 @@ class ToolCall(BaseModel):
 
 TOOLS: dict[str, dict[str, Any]] = {
     "consultar_faturas": {"description": "Consulta faturas do cliente.", "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"}},
     "consultar_vas": {"description": "Consulta VAS ativos.", "input_schema": {"msisdn": "string"}},
     "consultar_historico_vas": {"description": "Consulta histórico de VAS.", "input_schema": {"msisdn": "string"}},
@@ -111,6 +114,97 @@ def _clarification_option(item: Any) -> dict[str, Any]:
     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``. O ``motivo``/
+    ``descricao`` preserva a fala que abriu a transação e é uma fonte mais forte
+    que um ``subject`` residual. Só corrige quando há exatamente um nome canônico
+    da fatura explicitamente presente; não faz fuzzy matching.
+    """
+    original = " ".join(str(args.get(k) or "") for k in ("motivo", "descricao")).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.
 
@@ -125,6 +219,42 @@ def _preflight_subject(name: str, args: dict[str, Any]) -> dict[str, Any] | None
     subject = str(args.get("subject") or "").strip()
     if not msisdn or not subject:
         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):
@@ -176,7 +306,7 @@ def _preflight_subject(name: str, args: dict[str, Any]) -> dict[str, Any] | None
     return None
 
 async def _enrich_invoice_context(name: str, args: dict[str, Any]) -> None:
-    if name not in {"invoice_explanation", "cancelar_vas_avulso", "tratar_vas_estrategico", "contestar_cobranca"}:
+    if name not in {"consultar_plano", "invoice_explanation", "cancelar_vas_avulso", "tratar_vas_estrategico", "contestar_cobranca"}:
         return
     msisdn = str(args.get("msisdn") or "").strip()
     if not msisdn:
@@ -690,6 +820,7 @@ async def _run_cancelamento_com_contestacao(args: dict[str, Any]) -> dict[str, A
 async def _invoke(name: str, args: dict[str, Any]) -> Any:
     requirements = {
         "consultar_faturas": ("msisdn",),
+        "consultar_plano": ("msisdn",),
         "invoice_explanation": ("msisdn",),
         "consultar_vas": ("msisdn",),
         "consultar_historico_vas": ("msisdn",),