agent_contas/contas_mcp/servers/contas_mcp_server/main.py

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}}