from __future__ import annotations

import asyncio
import os

import calendar
import json
import re
from datetime import datetime
from decimal import Decimal, InvalidOperation
from difflib import SequenceMatcher
from typing import Any

from agent_framework.workflows import WorkflowActionRegistry
from agent_framework.idempotency import InMemoryIdempotencyStore
from app.domain.contas.contestation_validation import validate_contestation_items
from app.domain.contas.client import get_tim_integration_value

from .client import TimApiError
from .contestation_rules import (
    billing_cutoff_reference, complete_invoices_context, due_day, manual_conta_certa_from_evidence,
    resolve_refund_option,
)
from .ic_tags import CVNTag, MPITag, SADTag, VEBTag, VAATag
from .invoice_resolver import InvoiceResolver
from .protocol_triplets import resolve_protocol_triplet
from .rct_policy import rct_tags_for_attempt
from .pro_rata_rules import calculate_refund as calculate_pro_rata_refund, payment_message as pro_rata_payment_message
from .service import ContasDomainService
from .vas_variation import varied_vas_charges


_INFORMATIONAL_INVOICE_NOTES = "Explicação dos valores da fatura"
_STRATEGIC_SERVICE_ALIASES = (
    "apple music", "deezer", "disney", "fuze", "forge", "hbo", "looke",
    "max mensal", "netflix", "paramount", "paramount+", "paramount plus",
    "tim cloud gaming", "youtube", "youtube premium", "globoplay", "amazon prime",
)
_INFORMATIONAL_AVULSO_SERVICE_ALIASES = ("tim fashion", "tim fashion mensal")

def _normalize_service_alias(value: Any) -> str:
    import unicodedata
    text = unicodedata.normalize("NFKD", str(value or "").lower())
    text = "".join(ch for ch in text if not unicodedata.combining(ch))
    return re.sub(r"[^a-z0-9]+", "", text)

def _service_alias_matches(value: Any, aliases: tuple[str, ...]) -> bool:
    normalized = _normalize_service_alias(value)
    return bool(normalized) and any(_normalize_service_alias(alias) in normalized for alias in aliases)


def _events(*codes: Any, **payload: Any) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    for code in codes:
        text = str(getattr(code, "value", code) or "").strip()
        if text:
            out.append({"code": text, "payload": dict(payload)})
    return out


def _event_context(params: dict[str, Any], state: dict[str, Any], **overrides: Any) -> dict[str, Any]:
    """Build the TIM business-event metadata expected by the historical contract.

    Domain actions define business meaning; the framework observer remains
    responsible for event transport, sequencing and publication.  This helper
    only enriches the event payload with TIM's domain metadata.
    """
    source = state.get("input", {}) if isinstance(state, dict) else {}
    vars_ = state.get("vars", {}) if isinstance(state, dict) else {}
    if not isinstance(source, dict):
        source = {}
    if not isinstance(vars_, dict):
        vars_ = {}

    def first(*keys: str, default: Any = "") -> Any:
        for key in keys:
            if overrides.get(key) not in (None, ""):
                return overrides.get(key)
            if params.get(key) not in (None, ""):
                return params.get(key)
            if source.get(key) not in (None, ""):
                return source.get(key)
            if vars_.get(key) not in (None, ""):
                return vars_.get(key)
        return default

    protocol = str(first("agentProtocolId", "agent_protocol_id", "protocol_number", "protocolo_id", "protocol") or "")
    adjusted = str(first("adjustedProtocol", "adjusted_protocol", default=protocol) or "")
    billing_id = str(first("billingId", "billing_id", "invoice_id", "invoiceId") or "")
    gsm = str(first("gsm", "msisdn", "titular_msisdn") or "")
    channel_id = str(first("channelId", "channel_id", "channel", default="URA") or "URA")
    ura_call_id = str(first("uraCallId", "ura_call_id", "call_id") or "")
    session_id = str(first("sessionId", "session_id") or "")
    message_id = str(first("messageId", "message_id") or "")
    ura_protocol = str(first("uraProtocolId", "ura_protocol_id", "interactionProtocol") or "")

    specific = overrides.get("agentSpecificData")
    if specific is None:
        specific = {}
        if billing_id:
            specific["billingId"] = billing_id
    if isinstance(specific, dict):
        specific_wire: Any = json.dumps(specific, ensure_ascii=False, default=str) if specific else ""
    else:
        specific_wire = specific

    payload: dict[str, Any] = {
        "agentId": str(first("agentId", "agent_id", default=os.getenv("TIM_AGENT_ID", "contas")) or "contas"),
        "channelId": channel_id,
        "gsm": gsm,
        "uraCallId": ura_call_id,
        "sessionId": session_id,
        "messageId": message_id,
    }
    if protocol or "agentProtocolId" in overrides:
        payload["agentProtocolId"] = protocol
    if adjusted or "adjustedProtocol" in overrides:
        payload["adjustedProtocol"] = adjusted
    if ura_protocol:
        payload["uraProtocolId"] = ura_protocol
    if billing_id:
        payload["billingId"] = billing_id
    if specific_wire not in (None, ""):
        payload["agentSpecificData"] = specific_wire

    # Additional TIM fields are copied verbatim when the domain action knows them.
    # Conversational families (MPI/VEB/SAD/CVN) historically carry the text of
    # the customer turn and the assistant answer in the IC metadata. Keep those
    # fields in the domain event and let the framework mapper normalize the wire
    # contract.
    passthrough = (
        "adjustedItems", "adjustedItemsAmount", "apiUrl", "apiStatusCode",
        "apiResponsePayload", "latencyMs", "customerCode", "ani", "status",
        "customerMessage", "llmResponse", "documentsRetrieved",
        "documentsSelected", "noMatchRag", "sessionEndAt",
    )
    conversational_aliases = {
        "customerMessage": ("customerMessage", "customer_message", "resposta_usuario", "text"),
        "llmResponse": ("llmResponse", "llm_response", "mensagem_base", "mensagem", "response"),
        "documentsRetrieved": ("documentsRetrieved", "ragRetrievedDocuments", "rag_retrieved_documents"),
        "documentsSelected": ("documentsSelected", "ragSelectedDocuments", "rag_selected_documents"),
        "noMatchRag": ("noMatchRag", "no_match_rag"),
        "sessionEndAt": ("sessionEndAt", "session_end_at"),
    }
    for wire_key, aliases in conversational_aliases.items():
        if overrides.get(wire_key) not in (None, ""):
            payload[wire_key] = overrides[wire_key]
            continue
        value = first(*aliases, default=None)
        if value not in (None, ""):
            if wire_key == "sessionEndAt" and isinstance(value, str):
                try:
                    value = int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() * 1000)
                except ValueError:
                    pass
            payload[wire_key] = value

    for key in passthrough:
        if key in payload:
            continue
        value = overrides.get(key, params.get(key))
        if value is not None and value != "":
            payload[key] = value
    return payload


def _events_ctx(codes: Any, params: dict[str, Any], state: dict[str, Any], **overrides: Any) -> list[dict[str, Any]]:
    seq = codes if isinstance(codes, (list, tuple, set)) else (codes,)
    payload = _event_context(params, state, **overrides)
    return _events(*seq, **payload)



def _transport_rct_events(value: Any) -> list[dict[str, Any]]:
    """Converte metadados de retry HTTP em business_events RCT.

    O client de domínio registra apenas fatos de transporte. O catálogo RCT
    escolhe a tag; a publicação continua no AgentObserver do framework.
    """
    out: list[dict[str, Any]] = []
    if isinstance(value, dict):
        transport = value.get("_transport")
        if isinstance(transport, dict):
            operation = transport.get("rct_operation")
            for attempt in transport.get("attempts") or []:
                if not isinstance(attempt, dict):
                    continue
                for code in rct_tags_for_attempt(
                    operation, int(attempt.get("attempt") or 0), success=bool(attempt.get("success"))
                ):
                    response_payload = transport.get("api_response_payload")
                    out.extend(_events(
                        code,
                        apiUrl=attempt.get("api_url") or "",
                        apiStatusCode=attempt.get("status_code") or 0,
                        apiResponsePayload=json.dumps(response_payload, ensure_ascii=False, default=str) if response_payload not in (None, "") else "",
                        latencyMs=attempt.get("latency_ms") or 0,
                        error=attempt.get("error") or "",
                        rctSource=operation or "",
                    ))
        for key, nested in value.items():
            if key != "_transport":
                out.extend(_transport_rct_events(nested))
    elif isinstance(value, list):
        for nested in value:
            out.extend(_transport_rct_events(nested))
    return out

def _transport_event_metadata(value: Any) -> dict[str, Any]:
    if not isinstance(value, dict):
        return {}
    transport = value.get("_transport")
    if not isinstance(transport, dict):
        return {}
    attempts = transport.get("attempts") or []
    last = attempts[-1] if isinstance(attempts, list) and attempts else {}
    out: dict[str, Any] = {}
    if isinstance(last, dict):
        out.update({
            "apiUrl": last.get("api_url") or "",
            "apiStatusCode": last.get("status_code") or 0,
            "latencyMs": last.get("latency_ms") or 0,
        })
    if transport.get("api_response_payload") not in (None, ""):
        out["apiResponsePayload"] = json.dumps(transport.get("api_response_payload"), ensure_ascii=False, default=str)
    return out


def _first(params: dict[str, Any], state: dict[str, Any], *keys: str, default: Any = "") -> Any:
    source = state.get("input", {}) if isinstance(state, dict) else {}
    for key in keys:
        if params.get(key) not in (None, ""):
            return params.get(key)
        if isinstance(source, dict) and source.get(key) not in (None, ""):
            return source.get(key)
    return default


def _money(value: Any) -> str:
    try:
        d = Decimal(str(value).replace(".", "").replace(",", ".") if isinstance(value, str) and "," in value else str(value))
    except (InvalidOperation, ValueError):
        return str(value or "0,00")
    return f"{d:.2f}".replace(".", ",")


def _invoice_focus_normalize(value: Any) -> str:
    """Normalize customer/item wording only for conservative invoice focus matching.

    This does not infer a charge.  It is used only to narrow an already fetched
    current-invoice payload to an item whose meaningful name tokens are all present
    in the customer's utterance.  The backend data remains the authority.
    """
    import unicodedata
    raw = unicodedata.normalize("NFKD", str(value or "").lower().replace("+", " mais "))
    raw = "".join(ch for ch in raw if not unicodedata.combining(ch))
    return " ".join(re.sub(r"[^a-z0-9]+", " ", raw).split())


def _focused_invoice_explanation(user_text: str, billing_analysis: Any) -> str:
    """Return a factual explanation focused on invoice items named by the user.

    The legacy Billing Analysis ``invoiceExplanation`` is a global variation summary
    and can omit a specifically questioned item even when that item exists in
    ``currentInvoice``.  In that case the generic summary is the wrong customer
    answer and can trigger AOFERTA for mentioning unrelated services.

    We narrow only when the user's utterance contains every meaningful token of an
    authoritative current-invoice item name (e.g. ``VOD + Canais abertos``).
    Otherwise the existing global explanation remains untouched.
    """
    if not isinstance(billing_analysis, dict):
        return ""
    user_norm = _invoice_focus_normalize(user_text)
    if not user_norm:
        return ""
    user_tokens = set(user_norm.split())
    stop = {"de", "do", "da", "dos", "das", "e", "mais", "a", "o", "um", "uma"}
    matches: list[dict[str, Any]] = []
    for section in billing_analysis.get("currentInvoice") or []:
        if not isinstance(section, dict):
            continue
        for item in section.get("items") or []:
            if not isinstance(item, dict):
                continue
            desc = str(item.get("desc") or "").strip()
            if not desc:
                continue
            desc_tokens = [t for t in _invoice_focus_normalize(desc).split() if t not in stop and len(t) > 1]
            # Require at least two meaningful tokens.  This avoids narrowing on
            # generic one-word coincidences such as "plano" or "servico".
            if len(desc_tokens) < 2 or not all(t in user_tokens for t in desc_tokens):
                continue
            matches.append(item)
    if not matches:
        return ""

    # Keep a single canonical item family. If the utterance happens to fully match
    # more than one distinct name, do not guess which one the customer meant.
    names = {}
    for item in matches:
        key = _invoice_focus_normalize(item.get("desc"))
        names.setdefault(key, []).append(item)
    if len(names) != 1:
        return ""
    items = next(iter(names.values()))
    canonical = str(items[0].get("desc") or "cobrança").strip()

    def fmt_date(value: Any) -> str:
        raw = str(value or "").strip()
        m = re.match(r"^(\d{4})-(\d{2})-(\d{2})", raw)
        if m:
            return f"{m.group(3)}/{m.group(2)}/{m.group(1)}"
        return raw

    details = []
    for item in items:
        value = _money(item.get("value"))
        date = fmt_date(item.get("date"))
        details.append(f"R$ {value}" + (f" em {date}" if date else ""))
    count = len(details)
    if count == 1:
        prefix = f"Na fatura atual, identifiquei uma cobrança de {canonical}: "
    else:
        prefix = f"Na fatura atual, identifiquei {count} cobranças de {canonical}: "
    if count == 1:
        detail_text = details[0]
    else:
        detail_text = ", ".join(details[:-1]) + " e " + details[-1]
    return (
        prefix + detail_text + ". "
        "Os dados consultados comprovam esses lançamentos na fatura, mas não informam "
        "a origem ou contratação além do item faturado."
    )


def _normalize_contestation_name(value: Any) -> str:
    import unicodedata
    text = unicodedata.normalize("NFKD", str(value or "").strip())
    text = "".join(ch for ch in text if not unicodedata.combining(ch))
    text = re.sub(r"[^a-zA-Z0-9]+", " ", text)
    return re.sub(r"\s+", " ", text).upper().strip()


def _same_contestation_name(left: Any, right: Any) -> bool:
    a, b = _normalize_contestation_name(left), _normalize_contestation_name(right)
    return bool(a and b and (a == b or a in b or b in a))


def _already_contested_item(item: dict[str, Any]) -> bool:
    message = _normalize_contestation_name(item.get("message"))
    return "ITEM JA FOI CONTESTADO" in message or "JA CONTESTADO" in message


def _successful_contestation_item(item: dict[str, Any]) -> bool:
    statuses = (item.get("status"), item.get("correctAccountStatus"), item.get("correctAccountstatus"), item.get("correct_account_status"))
    return any(str(x or "").strip().upper() in {"ENVIADA", "CRIAR", "INICIADA"} for x in statuses)


def _money_decimal(value: Any) -> Decimal:
    """Parse TIM monetary values without turning 14.99 into 1499.

    Accepts canonical decimal-dot values (14.99), pt-BR decimal-comma values
    (14,99), and values with thousands separators (1.234,56 / 1,234.56).
    """
    text = str(value if value is not None else "0").strip()
    if not text:
        return Decimal("0")
    text = re.sub(r"[^0-9,.-]", "", text)
    if "," in text and "." in text:
        if text.rfind(",") > text.rfind("."):
            text = text.replace(".", "").replace(",", ".")
        else:
            text = text.replace(",", "")
    elif "," in text:
        text = text.replace(".", "").replace(",", ".")
    # Dot-only input is already the canonical decimal representation used by
    # the migrated agent/MCP contract. Do not strip it as a thousands marker.
    try:
        return Decimal(text)
    except InvalidOperation:
        return Decimal("0")


def _classify_contestation_items(requested: list[dict[str, Any]], response_items: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[str], str, str]:
    contested = [x for x in response_items if _successful_contestation_item(x) and not _already_contested_item(x)]
    already = [str(x.get("itemName") or x.get("item_name") or "").strip() for x in response_items if _already_contested_item(x)]
    already = [x for x in already if x]
    not_contested = [x for x in response_items if not _successful_contestation_item(x) and not _already_contested_item(x)]

    returned = [str(x.get("itemName") or x.get("item_name") or "") for x in contested + not_contested] + already
    for req in requested:
        name = str(req.get("item_name") or req.get("itemName") or req.get("name") or "").strip()
        if not name or any(_same_contestation_name(name, x) for x in returned):
            continue
        not_contested.append({
            "itemName": name,
            "itemType": req.get("item_type") or req.get("itemType") or "VAS_AVULSO",
            "claimedAmount": req.get("claimed_amount", req.get("claimedAmount", "0")),
            "validatedAmount": req.get("validated_amount", req.get("validatedAmount", "0")),
        })
        returned.append(name)

    contested_names = [str(x.get("itemName") or x.get("item_name") or "") for x in contested]
    claimed = Decimal("0")
    validated = Decimal("0")
    for req in requested:
        name = str(req.get("item_name") or req.get("itemName") or req.get("name") or "")
        if not any(_same_contestation_name(name, x) for x in contested_names):
            continue
        claimed += _money_decimal(req.get("claimed_amount", req.get("claimedAmount", "0")))
        validated += _money_decimal(
            req.get("validated_amount", req.get("validatedAmount", req.get("claimed_amount", req.get("claimedAmount", "0"))))
        )
    return contested, not_contested, already, f"{claimed:.2f}", f"{validated:.2f}"


def _normalize_final_status(value: Any) -> str:
    valid = {
        "resolvido", "nao_resolvido", "resolvido_outros_assuntos",
        "outros_assuntos", "erro_falha_sistema", "erro_no_match", "erro_no_input",
    }
    aliases = {
        "0": "resolvido", "1": "nao_resolvido",
        "2": "resolvido_outros_assuntos", "3": "outros_assuntos",
        "final": "resolvido",
    }
    text = str(value or "").strip().lower()
    text = aliases.get(text, text)
    return text if text in valid else "erro_falha_sistema"

def _final_msisdn(value: str) -> str:
    digits = re.sub(r"\D", "", str(value or ""))
    return digits[-4:] if digits else ""


def _format_list(values: list[str]) -> str:
    values = [v for v in values if v]
    if not values:
        return ""
    if len(values) == 1:
        return values[0]
    return ", ".join(values[:-1]) + " e " + values[-1]




def _normalize_service_for_notes(value: Any) -> str:
    return re.sub(r"[^a-z0-9]+", "", str(value or "").casefold())


def _service_matches_for_notes(invoice_service: str, mentioned_service: str) -> bool:
    a = _normalize_service_for_notes(invoice_service)
    b = _normalize_service_for_notes(mentioned_service)
    if not a or not b:
        return False
    if a == b or a in b or b in a:
        return True
    if min(len(a), len(b)) < 7 or a[:3] != b[:3]:
        return False
    return SequenceMatcher(None, a, b).ratio() >= 0.90


def _infer_informational_vas_types_with_count(services: list[str], invoice_detail: Any) -> tuple[set[str], int]:
    if isinstance(invoice_detail, str):
        try:
            invoice_detail = json.loads(invoice_detail)
        except Exception:
            return set(), 0
    if not services or not isinstance(invoice_detail, dict):
        return set(), 0
    found: set[str] = set()
    matched_services: set[int] = set()
    try:
        snapshot = InvoiceResolver().build_snapshot(invoice_detail)
    except Exception:
        snapshot = ()
    for item in snapshot:
        item_type = str(getattr(item, "item_type", "") or "").strip().lower()
        name = str(getattr(item, "canonical_name", "") or "")
        for idx, service in enumerate(services):
            if item_type in {"bundle", "estrategico", "avulso"} and _service_matches_for_notes(name, service):
                found.add(item_type)
                matched_services.add(idx)

    # Billing Analysis também pode vir no formato currentInvoice/invoiceVariation.
    for root_key in ("currentInvoice", "invoiceVariation"):
        sections = invoice_detail.get(root_key)
        if not isinstance(sections, list):
            continue
        for section in sections:
            if not isinstance(section, dict):
                continue
            section_name = str(section.get("desc") or section.get("type") or "").casefold()
            for entry in section.get("items") or []:
                if not isinstance(entry, dict):
                    continue
                name = str(entry.get("desc") or entry.get("name") or "")
                matched_indexes = [idx for idx, service in enumerate(services) if _service_matches_for_notes(name, service)]
                if not matched_indexes:
                    continue
                matched_services.update(matched_indexes)
                classe = str(entry.get("classe") or "").casefold()
                typ = str(entry.get("type") or "").casefold()
                if entry.get("estrategico") is True or classe in {"estrategico", "estrategica", "strategic"} or "streaming" in section_name or typ == "streaming":
                    found.add("estrategico")
                elif classe == "bundle":
                    found.add("bundle")
                elif classe in {"avulso", "avulsa"} or bool(entry.get("contestable")):
                    found.add("avulso")
                elif "serviços de valor adicionado" in section_name or "servicos de valor adicionado" in section_name:
                    found.add("estrategico")
    return found, len(matched_services)


def _infer_informational_vas_types(services: list[str], invoice_detail: Any) -> set[str]:
    found, _ = _infer_informational_vas_types_with_count(services, invoice_detail)
    return found

def _format_informational_vas_notes(types: set[str]) -> str:
    labels = {
        "bundle": "VAS Bundle",
        "estrategico": "VAS Estratégico",
        "avulso": "VAS Avulso",
    }
    ordered = [labels[k] for k in ("bundle", "estrategico", "avulso") if k in types]
    return f"Explicação de {', '.join(ordered)}" if ordered else ""


def _build_vas_lines(items: list[dict[str, Any]], lines: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]:
    if lines:
        return [dict(x) for x in lines if isinstance(x, dict)]
    grouped: dict[str, dict[str, Any]] = {}
    for item in items or []:
        if not isinstance(item, dict):
            continue
        msisdn = str(item.get("msisdn") or item.get("line") or "").strip()
        name = str(item.get("name") or item.get("desc") or item.get("subject") or "").strip()
        typ = str(item.get("type") or item.get("item_type") or "").strip().lower()
        if not msisdn or not name:
            continue
        if typ not in {"bundle", "estrategico"}:
            typ = "estrategico"
        row = grouped.setdefault(msisdn, {"msisdn": msisdn, "items": [], "bundle_names": [], "estrategico_names": []})
        row["items"].append({"type": typ, "msisdn": msisdn, "name": name})
        row["bundle_names" if typ == "bundle" else "estrategico_names"].append(name)
    return list(grouped.values())


def _vas_initial_message(lines: list[dict[str, Any]]) -> str:
    sections: list[str] = []
    for line in lines:
        bundles = [str(x).strip() for x in line.get("bundle_names", []) if str(x).strip()]
        strategic = [str(x).strip() for x in line.get("estrategico_names", []) if str(x).strip()]
        if bundles:
            names = _format_list(bundles)
            sections.append(
                (f"O serviço {names} está incluso" if len(bundles) == 1 else f"Os serviços {names} estão inclusos")
                + " no plano e não gera cobrança adicional, não podendo ser cancelado."
                + (" Com essa explicação, sanei sua dúvida?" if not strategic else "")
            )
        if strategic:
            names = _format_list(strategic)
            sections.append(
                f"{'O' if len(strategic)==1 else 'Os'} {names} só {'é' if len(strategic)==1 else 'são'} ativado"
                f"{' ' if len(strategic)==1 else 's '}depois da confirmação da contratação por código de verificação. "
                "Como a validação foi concluída, a cobrança é considerada válida e não pode ser retirada ou ressarcida pela TIM. "
                "Com essa explicação, sanei sua dúvida?"
            )
    return " ".join(sections) or "Não encontrei itens válidos para a tratativa de VAS estratégico."


def _pro_rata_control_message(plans: list[dict[str, Any]]) -> str:
    control = next((p for p in plans if "controle" in str(p.get("desc") or p.get("name") or "").lower()), None)
    other = next((p for p in plans if p is not control), None)
    if not control or not other:
        return (
            "Houve uma troca de plano e, por isso, na sua fatura apareceram duas cobranças. "
            "Uma é proporcional ao período de uso e a outra é do plano Controle. No Controle, a renovação disponibiliza "
            "todos os benefícios imediatamente e por isso o valor integral é cobrado. Consegui esclarecer sua dúvida?"
        )
    c_name = re.sub(r"\s*\([^)]*\)\s*$", "", str(control.get("desc") or control.get("name") or "Controle")).strip()
    o_name = re.sub(r"\s*\([^)]*\)\s*$", "", str(other.get("desc") or other.get("name") or "plano anterior")).strip()
    c_value = _money(control.get("valor_final") or control.get("value") or control.get("valor_bruto") or 0)
    o_value = _money(other.get("valor_final") or other.get("value") or other.get("valor_bruto") or 0)
    days = other.get("days") or other.get("dias")
    period = f"referente a {days} {'dia' if str(days)=='1' else 'dias'} de uso" if days else "referente ao período de uso na fatura"
    return (
        "Houve uma troca de plano e, por isso, na sua fatura apareceram duas cobranças. "
        f"Uma é cobrança proporcional do plano {o_name}, no valor de R$ {o_value}, {period}. "
        f"A outra é do plano {c_name}, no valor de R$ {c_value}. No plano Controle, sempre que a franquia é renovada, "
        "o valor do plano é cobrado por completo porque todos os benefícios ficam disponíveis imediatamente. "
        "Consegui esclarecer sua dúvida?"
    )


def _extract_invoice_text(payload: Any) -> str:
    if not isinstance(payload, dict):
        return ""
    for key in ("invoiceExplanation", "invoice_explanation", "explanation", "answer", "text"):
        if isinstance(payload.get(key), str) and payload[key].strip():
            return payload[key].strip()
    return ""


def _extract_invoice_item(invoices: Any, invoice_id: str = "") -> dict[str, Any]:
    if not isinstance(invoices, dict):
        return {}
    items = invoices.get("paymentItems") or invoices.get("payment_items") or []
    if not isinstance(items, list):
        return {}
    if invoice_id:
        for item in items:
            if isinstance(item, dict) and str(item.get("invoiceId") or item.get("invoice_id") or "") == str(invoice_id):
                return item
    return next((x for x in items if isinstance(x, dict)), {})


def _next_month(year: int, month: int) -> tuple[int, int]:
    return (year + 1, 1) if month == 12 else (year, month + 1)


def _workflow_var(state: dict[str, Any], name: str) -> dict[str, Any]:
    vars_ = state.get("vars", {}) if isinstance(state, dict) else {}
    value = vars_.get(name, {}) if isinstance(vars_, dict) else {}
    return value if isinstance(value, dict) else {}


def _iso_tracking_date(value: Any) -> str:
    text = str(value or "").strip()
    if not text:
        return ""
    # TIM tracking expects yyyy-mm-ddT00:00:00.00Z. Accept ISO and dd/mm/yyyy.
    for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%Y-%m-%dT%H:%M:%S"):
        try:
            dt = datetime.strptime(text[:19] if fmt.endswith("%S") else text, fmt)
            return dt.strftime("%Y-%m-%dT00:00:00.00Z")
        except ValueError:
            continue
    return text


def _selected_invoice_for_tracking(state: dict[str, Any], invoice_id: str) -> dict[str, Any]:
    check = _workflow_var(state, "check_invoice_status")
    candidates: list[dict[str, Any]] = []
    for key in ("open_invoices", "closed_invoices", "paymentItems", "payment_items"):
        raw = check.get(key)
        if isinstance(raw, (list, tuple)):
            candidates.extend(x for x in raw if isinstance(x, dict))
    selected = check.get("invoice")
    if isinstance(selected, dict):
        candidates.insert(0, selected)
    if invoice_id:
        for item in candidates:
            if str(item.get("invoiceId") or item.get("invoice_id") or item.get("id") or item.get("number") or "") == invoice_id:
                return item
        return {}
    return candidates[0] if candidates else {}


def build_contas_workflow_actions(service: ContasDomainService, *, idempotency_store=None) -> WorkflowActionRegistry:
    reg = WorkflowActionRegistry()
    idempotency_store = idempotency_store or InMemoryIdempotencyStore(namespace="contas")

    @reg.action("no_op")
    def no_op(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        return {"ok": True}

    @reg.action("montar_resposta_texto")
    def montar_resposta_texto(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        data = params.get("dados") if isinstance(params.get("dados"), dict) else params
        return {"mensagem": str(data.get("texto_usuario") or data.get("mensagem") or "")}

    @reg.action("buscar_fatura")
    def buscar_fatura(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        msisdn = str(_first(params, state, "msisdn"))
        invoice_id = str(params.get("invoice_id") or _first(params, state, "invoice_id"))
        customer_id = str(params.get("customer_id") or _first(params, state, "customer_id"))
        # O workflow original busca o PDF detalhado quando possui os identificadores.
        # CompleteInvoices continua sendo fallback para consultas sem invoice/customer.
        if invoice_id and customer_id:
            detail = service.buscar_fatura_detalhada(
                msisdn=msisdn,
                invoice_id=invoice_id,
                customer_id=customer_id,
                include_danfe=bool(params.get("include_danfe", False)),
                output=str(params.get("output") or ""),
            )
            ok = bool(detail.get("parsed_content") is not None or detail.get("file_content")) if isinstance(detail, dict) else False
            return {
                "success": ok,
                **(detail if isinstance(detail, dict) else {"result": detail}),
                "business_events": _events("RCT.073" if ok else "RCT.074"),
            }
        invoices = service.consultar_faturas(msisdn=msisdn)
        item = _extract_invoice_item(invoices, invoice_id)
        return {"success": bool(item), "invoice": item, "complete_invoices": invoices}

    @reg.action("buscar_informacao_rag")
    def buscar_informacao_rag(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        query = str(params.get("query") or "").strip()
        queries = params.get("queries") if isinstance(params.get("queries"), list) else ([query] if query else [])
        return {
            "queries": queries,
            "documents": [],
            "answer": "",
            "noMatchRag": True,
            "ragRetrievedDocuments": [],
            "ragSelectedDocuments": [],
            "delegate_to_framework_rag": True,
            "business_events": _events("RCT.080"),
        }

    @reg.action("reescrever_resposta_buscar_informacao")
    def reescrever_resposta_buscar_informacao(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        answer = str(params.get("answer") or "").strip()
        return {**params, "mensagem": answer, "delegate_to_framework_llm": True}

    @reg.action("preparar_invoice_explanation")
    def preparar_invoice_explanation(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        msisdn = str(_first(params, state, "msisdn"))
        cached = str(params.get("explicacao_base") or "").strip()
        if not msisdn and not cached:
            attempt = int(params.get("tentativa_anterior") or 0) + 1
            return {"success": False, "service_failed": False, "tentativa": attempt, "business_events": _events_ctx("CVN.002", params, state)}
        if cached:
            text = cached
            evidence = {}
        else:
            try:
                call_args = dict(params)
                call_args.pop("msisdn", None)
                evidence = service.invoice_explanation(msisdn=msisdn, **call_args)
                billing_analysis = evidence.get("billing_analysis") if isinstance(evidence, dict) else None
                user_text = str(_first(params, state, "message", "text", "customer_message", "user_text") or "").strip()
                text = _focused_invoice_explanation(user_text, billing_analysis) or _extract_invoice_text(billing_analysis)
            except TimApiError as exc:
                transport = {"_transport": {"rct_operation": "base_conhecimento", "attempts": list(exc.attempts or [])}}
                rct_events = _transport_rct_events(transport)
                return {
                    "success": False,
                    "service_failed": True,
                    "error": str(exc),
                    "api_status_code": exc.status_code or 0,
                    "api_response_payload": exc.body,
                    "business_events": _events_ctx("CVN.006", params, state) + (rct_events or _events_ctx("RCT.080", params, state)),
                }
            except Exception as exc:
                return {"success": False, "service_failed": True, "error": str(exc), "business_events": _events_ctx(("CVN.006", "RCT.080"), params, state)}
        transport_events = _transport_rct_events(evidence) if isinstance(evidence, dict) else []
        if not text:
            return {"success": False, "service_failed": True, "business_events": _events_ctx("CVN.006", params, state) + (transport_events or _events_ctx("RCT.080", params, state))}
        return {"success": True, "service_failed": False, "explicacao_base": text, "tentativa": 0, "evidence": evidence, "business_events": _events_ctx(("CVN.001", "CVN.003"), {**params, "llmResponse": text}, state) + (transport_events or _events_ctx("RCT.079", params, state))}

    @reg.action("formatar_invoice_explanation")
    def formatar_invoice_explanation(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        base = str(params.get("explicacao_base") or "").strip()
        trailer = str(params.get("trailer_override") or "Com essa explicação, sanei sua dúvida?").strip()
        message = base
        if trailer and trailer.lower() not in message.lower():
            message = f"{message.rstrip()} {trailer}".strip()
        return {
            "mensagem": message,
            "await_user_input": True,
            "requires_llm_composition": True,
            "response_instruction": (
                "Componha a resposta somente com a evidência fornecida; não invente cobranças, "
                "causas, políticas ou valores ausentes. Preserve a pergunta final prevista pelo workflow."
            ),
        }

    @reg.action("checar_tentativa_cvn")
    def checar_tentativa_cvn(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        attempt = int(params.get("tentativa") or 1)
        exceeded = attempt > 2
        return {"tentativa": attempt, "limite_excedido": exceeded, "pode_tentar_novamente": not exceeded, "business_events": _events_ctx("CVN.004" if exceeded else "CVN.005", params, state)}

    @reg.action("registrar_atendimento_invoice_explanation")
    def registrar_atendimento_invoice_explanation(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        path = str(params.get("caminho") or params.get("resposta_usuario") or "").upper()
        accepted = path in {"SIM", "ACEITOU", "OK"}
        event_params = {**params, "customerMessage": str(_first(params, state, "resposta_usuario", "customer_message") or path)}
        return {"success": True, "accepted": accepted, "business_events": _events_ctx(MPITag.EXPLICACAO_SIM if accepted else MPITag.EXPLICACAO_NAO, event_params, state)}

    @reg.action("preparar_handoff_invoice_explanation")
    def preparar_handoff_invoice_explanation(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        """Materializa a decisão de handoff declarada no workflow do domínio.

        A action não decide quando transferir: ela apenas transforma a configuração
        do nó em um resultado estrutural consumível pelo agente/framework.
        """
        message = str(params.get("mensagem") or "Para continuar com a sua solicitação, aguarde um instante.").strip()
        reason = str(params.get("reason") or "invoice_explanation_not_resolved").strip()
        return {
            "success": True,
            "mensagem": message,
            "session_control": "HUMAN_HANDOFF",
            "human_handoff_requested": True,
            "handoff": True,
            "session_ended": True,
            "terminal_status": "human_handoff",
            "handoff_reason": reason,
        }

    @reg.action("registrar_protocolo_inicio")
    @reg.action("registrar_protocolo")
    def registrar_protocolo(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        msisdn = str(_first(params, state, "msisdn"))
        scenario = str(params.get("scenario") or "atendimento_geral")
        triplet = resolve_protocol_triplet(scenario, stage="open", context=params)
        payload = {
            "msisdn": msisdn,
            "socialSecNo": _first(params, state, "social_sec_no", "socialSecNo"),
            "reason1": triplet.reason1,
            "reason2": triplet.reason2,
            "reason3": triplet.reason3,
            "requestStatus": params.get("request_status") or "Aberto",
            "status": params.get("status") or "OPENED",
            "serviceRequestNotes": params.get("service_request_notes") or params.get("notes") or "",
            "messageId": params.get("message_id") or "",
            "rct_operation": params.get("rct_operation") or "",
        }
        result = service.client.abrir_protocolo(payload)
        protocol = str(
            (result or {}).get("interactionProtocol")
            or (result or {}).get("protocolNumber")
            or (result or {}).get("protocol")
            or (result or {}).get("protocolo")
            or ""
        ) if isinstance(result, dict) else ""
        response = {"success": bool(protocol or result), "protocolo_id": protocol, "protocol_number": protocol, "result": result}
        # A protocol-opening action is reused by several workflows. Only nodes
        # that explicitly request a final workflow response opt into this
        # presentation contract; session terminality remains independent.
        if bool(params.get("workflow_response_final")):
            response["workflow_response_final"] = True
            configured = str(params.get("mensagem_final") or "").strip()
            if configured:
                response["mensagem"] = configured.replace("{protocol}", protocol)
            elif protocol:
                response["mensagem"] = f"Seu número de protocolo é {protocol}."
        return response

    @reg.action("checar_vas_variado")
    def checar_vas_variado(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        # A regra de variação é determinística e preserva multiplicidade, direção
        # passada→atual e os dois buckets VAS do Billing Analysis. Não usa busca
        # textual aproximada porque um falso positivo pode levar a cancelamento.
        vars_state = state.get("vars") or {} if isinstance(state, dict) else {}
        prep = vars_state.get("preparar", {}) if isinstance(vars_state, dict) else {}
        evidence = prep.get("evidence", {}) if isinstance(prep, dict) else {}
        billing = evidence.get("billing_analysis", {}) if isinstance(evidence, dict) else {}
        invoice_variation = billing.get("invoiceVariation") if isinstance(billing, dict) else None
        charges = varied_vas_charges(
            invoice_variation=invoice_variation if isinstance(invoice_variation, dict) else billing,
            invoice_detail=billing if isinstance(billing, dict) else None,
        )
        serialized = [{"desc": c.desc, "value": str(c.value)} for c in charges]
        return {"tem_vas_variado": bool(charges), "variations": serialized}

    @reg.action("preparar_pro_rata")
    def preparar_pro_rata(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        plans = params.get("planos") if isinstance(params.get("planos"), list) else []
        has_control = bool(params.get("has_plano_controle")) or any("controle" in str(p.get("desc") or p.get("name") or "").lower() for p in plans if isinstance(p, dict))
        if has_control:
            message = _pro_rata_control_message(plans)
            return {
                "mensagem": message,
                "mensagem_reperguntar_esclarecimento": message,
                "mensagem_pos_aceite": "Perfeito. A cobrança foi esclarecida e não será solicitado ajuste.",
                "await_user_input": True,
                "has_plano_controle": True,
                "planos": plans,
                "business_events": _events_ctx(MPITag.PRO_RATA, params, state),
            }
        rendered = [f"{str(p.get('desc') or p.get('name') or 'plano')} (R$ {_money(p.get('valor_final') or p.get('value') or 0)}, linha final {_final_msisdn(str(p.get('msisdn') or ''))})" for p in plans if isinstance(p, dict)]
        return {
            "mensagem": "Identifiquei cobrança proporcional na sua fatura" + (f" envolvendo {_format_list(rendered)}" if rendered else "") + ". Como não houve troca para Plano Controle, trata-se da cobrança proporcional padrão pelos dias usados.",
            "await_user_input": False,
            "has_plano_controle": False,
            "planos": plans,
            "business_events": _events_ctx(MPITag.PRO_RATA, params, state),
        }

    @reg.action("formatar_pro_rata")
    def formatar_pro_rata(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        return {"mensagem": str(params.get("mensagem_base") or "")}

    @reg.action("registrar_atendimento_pro_rata")
    def registrar_atendimento_pro_rata(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        msisdn = str(_first(params, state, "msisdn"))
        protocol = registrar_protocolo({"msisdn": msisdn, "scenario": "pro_rata", "request_status": "Fechado", "status": "CLOSED"}, state)
        ic = params.get("ic") or ("VEB.003" if params.get("caminho") == "aceitou" else "VEB.002")
        message = str(params.get("mensagem_base") or "")
        return {"success": True, "mensagem": message, "protocol": protocol, "business_events": _events_ctx(ic, {**params, "llmResponse": message}, state)}

    @reg.action("definir_devolucao_ajuste_pro_rata")
    def definir_devolucao_ajuste_pro_rata(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        plans = params.get("planos") if isinstance(params.get("planos"), list) else _first(params, state, "planos", default=[])
        if not isinstance(plans, list):
            plans = []
        invoice_detail = params.get("invoice_detail") or _first(params, state, "invoice_detail", "invoiceDetail", default={})
        if not isinstance(invoice_detail, dict) or not invoice_detail.get("DANFE-COM"):
            invoice_id = str(params.get("invoice_id") or _first(params, state, "invoice_id", "current_invoice_number") or "")
            customer_id = str(params.get("customer_id") or _first(params, state, "customer_id") or "")
            msisdn = str(params.get("msisdn") or _first(params, state, "msisdn") or "")
            if invoice_id and customer_id and msisdn:
                fetched = service.buscar_fatura_detalhada(
                    msisdn=msisdn, invoice_id=invoice_id, customer_id=customer_id, include_danfe=True
                )
                if isinstance(fetched, dict):
                    parsed = fetched.get("parsed_content")
                    if isinstance(parsed, dict):
                        invoice_detail = parsed
        try:
            result = calculate_pro_rata_refund(
                planos=[x for x in plans if isinstance(x, dict)],
                invoice_detail=invoice_detail if isinstance(invoice_detail, dict) else {},
                invoice_period=str(params.get("invoice_period") or _first(params, state, "invoice_period", default="") or ""),
                invoice_emissao=str(params.get("invoice_emissao") or _first(params, state, "invoice_emissao", default="") or ""),
            )
        except ValueError as exc:
            return {"success": False, "error": str(exc), "business_events": _events_ctx("RCT.033", params, state)}
        return {"success": True, **result, "business_events": _events_ctx("RCT.031", params, state)}

    @reg.action("executar_contestacao_plano_controle")
    def executar_contestacao_plano_controle(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        devolucao = params.get("devolucao") if isinstance(params.get("devolucao"), dict) else {}
        items = params.get("items") if isinstance(params.get("items"), list) else devolucao.get("items")
        if not isinstance(items, list) or not items:
            return {"success": False, "error": "items obrigatorio para contestacao de pro_rata"}
        msisdn = str(params.get("msisdn") or _first(params, state, "msisdn") or "")
        customer_id = str(params.get("customer_id") or _first(params, state, "customer_id") or "")
        invoice_number = str(params.get("current_invoice_number") or params.get("invoice_id") or _first(params, state, "current_invoice_number", "invoice_id") or "")
        if not msisdn or not customer_id or not invoice_number:
            return {"success": False, "error": "msisdn, customer_id e invoice_number obrigatorios para contestacao de pro_rata"}
        total = str(params.get("invoice_amount") or devolucao.get("invoice_amount") or sum(Decimal(str(i.get("validatedAmount") or i.get("validated_amount") or 0)) for i in items if isinstance(i, dict)))
        payload = {
            "msisdn": msisdn,
            "socialSecNo": re.sub(r"\D", "", str(params.get("social_sec_no") or params.get("cpf") or _first(params, state, "social_sec_no", "cpf") or "")),
            "customerId": customer_id,
            "invoiceNumber": invoice_number,
            "customerType": str(params.get("customer_type") or "2"),
            "customerStatus": str(params.get("customer_status") or "1"),
            "invoiceStatus": str(params.get("invoice_status") or "1"),
            "invoiceAmountOpen": str(params.get("invoice_amount_open") or devolucao.get("invoice_amount_open") or total),
            "invoiceAmount": total,
            "invoiceDueDate": str(params.get("current_invoice_due_date") or ""),
            "contestationType": str(params.get("contestation_type") or "0"),
            "adjustReason": str(params.get("adjust_reason") or "SERVICO_NAO_SOLICITADO"),
            "observation": str(params.get("observation") or params.get("descricao") or ""),
            "refundOption": str(params.get("refund_option") or ""),
            "manualContaCertaIndicator": bool(params.get("manual_conta_certa_indicator", False)),
            "doubleRefund": bool(params.get("double_refund", False)),
            "userId": str(params.get("user_id") or os.getenv("TIM_DEFAULT_CLIENT_ID")),
            "messageId": str(params.get("message_id") or _first(params, state, "message_id") or ""),
            "clientId": str(params.get("client_id") or os.getenv("TIM_DEFAULT_CLIENT_ID")),
            "tipo_atendimento": "pro_rata",
            "skip_invoice_item_validation": True,
            "items": items,
        }
        try:
            result = service.client.contestar(payload)
        except Exception as exc:
            return {"success": False, "error": str(exc), "business_events": _events_ctx("RCT.033", params, state)}
        response = result if isinstance(result, dict) else {"result": result}
        protocol = str(response.get("protocol_number") or response.get("protocolo_id") or response.get("sr") or "")
        format_text = "sms" if bool(response.get("sms_enviado") or response.get("sms_sent")) else "conta_futura"
        merged = dict(devolucao)
        merged.update({
            "items": items,
            "invoice_amount_open": payload["invoiceAmountOpen"],
            "invoice_amount": payload["invoiceAmount"],
            "format_text": str(response.get("format_text") or format_text),
            "resolution_type": str(response.get("resolution_type") or ("new_boleto" if format_text == "sms" else "credit_bill")),
            "sms_enviado": bool(response.get("sms_enviado") or response.get("sms_sent")),
            "sms_sent": bool(response.get("sms_enviado") or response.get("sms_sent")),
        })
        for key in ("decision_reason", "data_credito_proxima_fatura", "barcode", "items_response", "sr_conta_certa_id"):
            if key in response:
                merged[key] = response[key]
        if protocol:
            merged["protocolo_id"] = protocol
        message = str(response.get("mensagem") or f"Contestacao de pro rata registrada com sucesso. Protocolo: {protocol or 'n/a'}.")
        audit = {
            "plano_cliente": ", ".join(str(i.get("itemName") or i.get("item_name") or "") for i in items if isinstance(i, dict)),
            "valor_calculado_credito": payload["invoiceAmount"],
            "transaction_id_ajuste": str(response.get("contestation_id") or response.get("sr") or protocol),
            "mensagem_confirmacao": message,
        }
        return {
            "success": True, "mensagem": message, "protocolo_id": protocol,
            "contestation_id": str(response.get("contestation_id") or ""), "sr": str(response.get("sr") or protocol),
            "items": items, "invoice_amount_open": payload["invoiceAmountOpen"], "invoice_amount": payload["invoiceAmount"],
            "devolucao": merged, "audit": audit, "response": response,
            "business_events": _events_ctx("RCT.031", params, state),
        }

    @reg.action("orientar_pagamento_pro_rata")
    def orientar_pagamento_pro_rata(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        devolucao = params.get("devolucao") if isinstance(params.get("devolucao"), dict) else {}
        refund_option = str(params.get("refund_option") or devolucao.get("format_text") or "proxima_fatura")
        message = pro_rata_payment_message(devolucao)
        instruction = (
            "Use a mensagem determinística calculada pelo domínio como fonte de verdade. Preserve valor, forma de devolução, "
            "prazo, código de barras e protocolo; apenas torne a fala natural, sem prometer crédito diferente do registrado."
        )
        return {
            "mensagem": message, "devolucao": devolucao, "refund_option": refund_option,
            "requires_llm_composition": True, "response_instruction": instruction,
        }

    @reg.action("preparar_vas_estrategico")
    def preparar_vas_estrategico(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        items = params.get("items") if isinstance(params.get("items"), list) else []
        lines = _build_vas_lines(items, params.get("linhas") if isinstance(params.get("linhas"), list) else None)
        has_strategic = any(line.get("estrategico_names") for line in lines)
        has_bundle = any(line.get("bundle_names") for line in lines)
        bundle_names = [name for line in lines for name in line.get("bundle_names", [])]
        return {
            "linhas": lines,
            "mensagem": _vas_initial_message(lines),
            "mensagem_pos_aceite": "",
            "mensagem_bundle_fechamento": ("Como expliquei, o serviço é incluso no seu plano e não há como remover." if len(bundle_names) <= 1 else "Como expliquei, os serviços são inclusos no seu plano e não há como remover."),
            "await_user_input": bool(has_strategic or has_bundle),
            "has_estrategico_items": bool(has_strategic),
            "has_bundle_items": bool(has_bundle),
            "business_events": _events_ctx((MPITag.VAS_ESTRATEGICO, VEBTag.VAS_ESTRATEGICO_INICIO), params, state),
        }

    @reg.action("montar_explicacao_cancelamento_vas_estrategico")
    def montar_explicacao_cancelamento_vas_estrategico(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        sections: list[str] = []
        for line in params.get("linhas") or []:
            if not isinstance(line, dict):
                continue
            for name in line.get("estrategico_names") or []:
                sections.append(f"Para cancelar {name} na linha final {_final_msisdn(str(line.get('msisdn') or ''))}, acesse o app ou site oficial do parceiro e solicite o cancelamento da assinatura.")
        message = " ".join(sections) or "Não há serviços estratégicos com procedimento adicional de cancelamento."
        return {
            "mensagem": message,
            "business_events": _events_ctx(VEBTag.INFO_CANCELAMENTO, {**params, "mensagem": message}, state),
        }

    @reg.action("registrar_atendimento_vas_estrategico")
    def registrar_atendimento_vas_estrategico(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        lines = params.get("linhas") if isinstance(params.get("linhas"), list) else []
        source = state.get("input") if isinstance(state.get("input"), dict) else {}
        answer_raw = str(source.get("resposta_usuario") or source.get("customer_message") or "")
        answer = answer_raw.strip().upper()
        message = str(params.get("mensagem_base") or "")

        informational_services: list[str] = []
        informational_types: list[str] = []
        for line in lines:
            if not isinstance(line, dict):
                continue
            bundles = [str(x).strip() for x in (line.get("bundle_names") or []) if str(x).strip()]
            strategic = [str(x).strip() for x in (line.get("estrategico_names") or []) if str(x).strip()]
            informational_services.extend(bundles + strategic)
            if bundles and "bundle" not in informational_types:
                informational_types.append("bundle")
            if strategic and "estrategico" not in informational_types:
                informational_types.append("estrategico")

        bundle_close_path = answer == "NAO" and "bundle" in informational_types
        event_params = {**params, "customerMessage": answer_raw, "llmResponse": message}
        events: list[dict[str, Any]] = []
        if answer == "SIM":
            events += _events_ctx(VEBTag.EXPLICACAO_ACEITA, event_params, state)
        elif answer == "NAO":
            events += _events_ctx(VEBTag.EXPLICACAO_REJEITADA, event_params, state)
            events += _events_ctx(
                VEBTag.INFO_CANCELAMENTO if bundle_close_path else VEBTag.DEVOLUCAO_URA,
                event_params, state,
            )

        # Bundle não pode ser removido. O original não abre protocolo aqui;
        # registra a passagem pelo atendimento e deixa o RT-15 para finalização.
        if bundle_close_path:
            events += _events_ctx(VEBTag.REGISTRA_ATENDIMENTO_OK, event_params, state)
            deduped_services = list(dict.fromkeys(informational_services))
            strategic_services = list(dict.fromkeys(
                str(name).strip()
                for line in lines if isinstance(line, dict)
                for name in (line.get("estrategico_names") or [])
                if str(name).strip()
            ))
            out = {
                "success": True,
                "mensagem": message,
                "protocolos_por_linha": [],
                "erros": [],
                "protocol_closed": False,
                "informational_only": True,
                "protocol_deferred_to_finalization": True,
                "vas_estrategico_protocol_deferred_to_finalization": True,
                "suppress_cvn_protocol_ic": True,
                "informational_service_names": deduped_services,
                "informational_vas_types": informational_types,
                "conversation_explanation_resolved": answer == "SIM",
                "recomenda_finalizacao": True,
                "status_finalizacao_sugerido": "resolvido",
                "business_events": events,
            }
            # Em fluxo misto Bundle + Estratégico, o protocolo continua deferido,
            # mas as assinaturas estratégicas ainda precisam da orientação do
            # parceiro. O domínio declara as queries; RagService permanece no framework.
            if strategic_services:
                out["business_workflow"] = "vas_estrategico"
                out["requires_rag"] = True
                out["rag_queries"] = [
                    f"Como cancelar o serviço {name} no parceiro? Procedimento oficial de cancelamento."
                    for name in strategic_services
                ]
            return out

        protocols: list[dict[str, str]] = []
        errors: list[dict[str, Any]] = []
        for line in lines:
            if not isinstance(line, dict):
                continue
            msisdn = str(line.get("msisdn") or "").strip()
            if not msisdn:
                continue
            p = registrar_protocolo({
                "msisdn": msisdn,
                "scenario": "vas_estrategico",
                "request_status": params.get("request_status") or "Fechado",
                "status": params.get("status") or "CLOSED",
                "service_request_notes": _format_informational_vas_notes(set(informational_types)),
                "message_id": params.get("message_id") or source.get("message_id") or "",
                "rct_operation": "reg_atend_vas_estrat",
            }, state)
            line_event_params = {**event_params, "msisdn": msisdn}
            if p.get("success") and p.get("protocolo_id"):
                protocols.append({"msisdn": msisdn, "protocolo_id": str(p["protocolo_id"])})
                events += _events_ctx(VEBTag.REGISTRA_ATENDIMENTO_OK, line_event_params, state)
            else:
                errors.append({"msisdn": msisdn, "stage": "register_protocol", "erro": p.get("error") or "Falha ao registrar protocolo"})
                events += _events_ctx(VEBTag.REGISTRA_ATENDIMENTO_FAIL, line_event_params, state)

        out: dict[str, Any] = {
            "success": bool(protocols),
            "mensagem": message,
            "protocolos_por_linha": protocols,
            "erros": errors,
            "protocol_closed": bool(protocols),
            "business_events": events,
        }
        if answer == "SIM" and protocols:
            out.update({
                "informational_only": True,
                "informational_service_names": list(dict.fromkeys(informational_services)),
                "informational_vas_types": informational_types,
                "conversation_explanation_resolved": True,
            })
        elif answer != "SIM":
            out["business_workflow"] = "vas_estrategico"
            strategic_services = list(dict.fromkeys(
                str(name).strip()
                for line in lines if isinstance(line, dict)
                for name in (line.get("estrategico_names") or [])
                if str(name).strip()
            ))
            if strategic_services:
                # O domínio apenas declara a necessidade e as consultas. A busca
                # documental é executada pelo RagService/AgentRuntimeMixin do
                # framework, preservando a arquitetura framework-native.
                out["requires_rag"] = True
                out["rag_queries"] = [
                    f"Como cancelar o serviço {name} no parceiro? Procedimento oficial de cancelamento."
                    for name in strategic_services
                ]
        if protocols:
            out["requires_protocol_in_response"] = True
            out["protocols_for_response"] = [x["protocolo_id"] for x in protocols]
            out["suppress_cvn_protocol_ic"] = True
        return out

    @reg.action("cancelamento_vas_avulso_batch")
    async def cancelamento_vas_avulso_batch(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        items = params.get("items") if isinstance(params.get("items"), list) else []
        if not items:
            subject = str(_first(params, state, "subject"))
            msisdn = str(_first(params, state, "msisdn"))
            if subject and msisdn:
                items = [{"name": subject, "msisdn": msisdn}]

        idempotency_root = str(
            params.get("idempotency_key")
            or _first(params, state, "interaction_key", "message_id", "session_id")
            or state.get("execution_id")
            or "contas-cancelamento"
        )
        concurrency = max(1, int(os.getenv("TIM_CANCELAMENTO_BATCH_CONCURRENCY", "5")))
        semaphore = asyncio.Semaphore(concurrency)

        async def _process_item_impl(item: dict[str, Any]) -> dict[str, Any]:
            async with semaphore:
                msisdn = str(item.get("msisdn") or _first(params, state, "msisdn"))
                subject = str(item.get("name") or item.get("desc") or item.get("subject") or "")
                key = idempotency_store.canonical_key(idempotency_root, msisdn, subject.casefold())
                cached = await idempotency_store.get(key)
                if isinstance(cached, dict):
                    return {**cached, "idempotent_replay": True}

                # O original registrava um protocolo fechado por linha antes do
                # side effect. Se esse registro falhar, não podemos cancelar.
                protocol = str(item.get("protocol") or params.get("protocol") or "").strip()
                protocol_result: dict[str, Any] = {}
                social_sec_no = str(item.get("social_sec_no") or "").strip()
                holder_msisdn = str(params.get("msisdn") or _first(params, state, "msisdn") or "").strip()
                # Plano família: o protocolo de cancelamento é por linha. Quando
                # o item pertence ao dependente, recupere o CPF daquela linha em
                # vez de reutilizar silenciosamente o CPF do titular.
                if not social_sec_no and msisdn and holder_msisdn and msisdn != holder_msisdn and hasattr(service, "client") and hasattr(service.client, "line_info"):
                    try:
                        line = await asyncio.to_thread(service.client.line_info, msisdn)
                        if isinstance(line, dict):
                            social_sec_no = str(line.get("social_sec_no") or "").strip()
                    except Exception:
                        social_sec_no = ""
                if not social_sec_no:
                    social_sec_no = str(params.get("social_sec_no") or "").strip()
                if not protocol and hasattr(service, "client"):
                    protocol_result = await asyncio.to_thread(
                        registrar_protocolo,
                        {
                            "msisdn": msisdn,
                            "social_sec_no": social_sec_no,
                            "scenario": "cancelamento_vas_avulso",
                            "request_status": "Fechado",
                            "status": "CLOSED",
                            "service_request_notes": f"Cancelamento do serviço {subject}",
                            "message_id": params.get("message_id") or "",
                        },
                        state,
                    )
                    protocol = str(protocol_result.get("protocolo_id") or protocol_result.get("protocol_number") or "").strip()
                    if not protocol:
                        return {
                            "success": False, "msisdn": msisdn, "subject": subject,
                            "reason": "protocol_registration_failed",
                            "error": "Não foi possível registrar o protocolo antes do cancelamento.",
                            "protocol_result": protocol_result,
                        }
                try:
                    result = await asyncio.to_thread(
                        service.cancelar_vas_avulso, msisdn=msisdn, subject=subject, protocol=protocol
                    )
                    item_result = {
                        "msisdn": msisdn, "subject": subject, "protocol": protocol,
                        "protocol_result": protocol_result, **result,
                    }
                    if item_result.get("success"):
                        await idempotency_store.set(key, item_result)
                    return item_result
                except Exception as exc:
                    return {
                        "success": False,
                        "msisdn": msisdn,
                        "subject": subject,
                        "reason": "cancel_vas_failed",
                        "error": str(exc),
                        "protocol": protocol,
                        "recoverable": True,
                    }

        async def _process_item(item: dict[str, Any]) -> dict[str, Any]:
            """Boundary de robustez: nenhuma falha técnica de um item derruba o batch/request."""
            try:
                return await _process_item_impl(item)
            except Exception as exc:
                msisdn = str(item.get("msisdn") or _first(params, state, "msisdn"))
                subject = str(item.get("name") or item.get("desc") or item.get("subject") or "")
                logger.exception(
                    "Falha não tratada no cancelamento VAS; convertendo em resultado recuperável msisdn=%s subject=%s",
                    msisdn,
                    subject,
                )
                return {
                    "success": False,
                    "msisdn": msisdn,
                    "subject": subject,
                    "reason": "internal_processing_failed",
                    "error": str(exc),
                    "recoverable": True,
                    "fallback_applied": True,
                }

        valid_items = [dict(x) for x in items if isinstance(x, dict)]
        results = list(await asyncio.gather(*[_process_item(item) for item in valid_items])) if valid_items else []
        errors = [
            {"msisdn": x.get("msisdn"), "subject": x.get("subject"), "error": x.get("error")}
            for x in results if x.get("error") and not x.get("success")
        ]
        cancelados = [x for x in results if x.get("success")]
        nao_encontrados = [
            x for x in results
            if not x.get("success") and str(x.get("reason") or "") == "service_not_found"
        ]
        nao_cancelados = [x for x in results if not x.get("success")]
        # O fluxo original ainda encaminha para contestação quando o item foi
        # identificado na fatura mas o bloqueio/cancelamento falhou. Apenas
        # no-match/serviço já inativo não deve gerar RT-02 automaticamente.
        itens_para_contestacao = [
            x for x in results
            if x.get("success") or bool(x.get("eligible_for_contestation"))
        ]
        batch_gsm = str(
            params.get("msisdn")
            or _first(params, state, "msisdn", "ani")
            or next((x.get("msisdn") for x in results if isinstance(x, dict) and x.get("msisdn")), "")
            or ""
        )
        vaa_events = _events_ctx(VAATag.INICIO_FLUXO, {**params, "msisdn": batch_gsm}, state)
        if valid_items:
            vaa_events += _events_ctx(VAATag.BLOQUEIO_INICIO, {**params, "msisdn": batch_gsm}, state)
        operational_failures = [
            x for x in nao_cancelados
            if str(x.get("reason") or "") in {
                "block_vas_failed", "cancel_vas_failed", "protocol_registration_failed"
            }
        ]
        if operational_failures:
            vaa_events += _events_ctx(VAATag.ITEM_CANCELADO_FAIL, {**params, "msisdn": batch_gsm}, state)
        elif cancelados:
            vaa_events += _events_ctx(VAATag.ITEM_CANCELADO_OK, {**params, "msisdn": batch_gsm}, state)
        return {
            "success": bool(cancelados),
            "results": results,
            "errors": errors,
            "cancelados": cancelados,
            "nao_encontrados": nao_encontrados,
            "nao_cancelados": nao_cancelados,
            "contestation_candidates": itens_para_contestacao,
            "itens_para_contestacao": itens_para_contestacao,
            "protocolos_por_linha": [
                {"msisdn": x.get("msisdn"), "protocolo_id": x.get("protocol")}
                for x in cancelados if x.get("protocol")
            ],
            "business_events": _events_ctx(MPITag.CANCELAR_VAS_AVULSO, params, state) + vaa_events + _transport_rct_events(results)
            + ([] if cancelados else _events("RCT.002", rctSource="cancela_vas")),
        }

    @reg.action("check_invoice_status")
    def check_invoice_status(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        payload = params.get("complete_invoices_payload")
        # Paridade do original: contestação reaproveita exclusivamente o prefetch;
        # não dispara uma segunda CompleteInvoices silenciosamente.
        if not isinstance(payload, dict) or not payload:
            raise RuntimeError(
                "Payload do CompleteInvoices não encontrado no prefetch; nova chamada remota não permitida"
            )
        invoice_id = str(params.get("invoice_id") or params.get("current_invoice_number") or "")
        ctx = complete_invoices_context(payload, invoice_id)
        return {"success": True, "complete_invoices_payload": payload, **ctx}

    @reg.action("abrir_contestacao_cliente")
    def abrir_contestacao_cliente(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        items = params.get("items") if isinstance(params.get("items"), list) else []
        if not items and params.get("servico"):
            items = [{"itemName": params.get("servico"), "claimedAmount": params.get("valor"), "validatedAmount": params.get("valor")}]

        vars_state = state.get("vars") if isinstance(state.get("vars"), dict) else {}
        check = vars_state.get("check_invoice_status") if isinstance(vars_state, dict) and isinstance(vars_state.get("check_invoice_status"), dict) else {}
        complete_payload = params.get("complete_invoices_payload") if isinstance(params.get("complete_invoices_payload"), dict) else check.get("complete_invoices_payload")
        invoice_id = str(params.get("current_invoice_number") or params.get("invoice_id") or "")
        invoice_ctx = complete_invoices_context(complete_payload or {}, invoice_id)
        selected_invoice = invoice_ctx.get("invoice") if isinstance(invoice_ctx.get("invoice"), dict) else {}
        status_value = str(params.get("invoice_status") or invoice_ctx.get("invoice_status") or "")
        statuses = list(invoice_ctx.get("invoice_statuses") or [])
        if status_value and status_value not in statuses:
            statuses.append(status_value)
        refund_option = str(params.get("refund_option") or "").strip()
        if not refund_option:
            refund_option = resolve_refund_option(
                payment_method=str(invoice_ctx.get("payment_method") or check.get("payment_method") or ""),
                has_open_bill=bool(invoice_ctx.get("has_open_bill") or check.get("has_open_bill")),
                invoice_statuses=statuses,
            )["refund_option"]
        manual_indicator = bool(params.get("manual_conta_certa_indicator"))
        if not manual_indicator:
            contract_payload = None
            try:
                contract_payload = service.client.contrato(str(_first(params, state, "msisdn")))
            except Exception:
                contract_payload = None
            manual_indicator = manual_conta_certa_from_evidence(
                dependent_invoice_item=bool(params.get("dependent_invoice_item")),
                invoice_item=selected_invoice,
                contract=contract_payload,
            )
        invoice_due = str(params.get("current_invoice_due_date") or selected_invoice.get("dueDate") or selected_invoice.get("invoiceDueDate") or "")
        payload = {
            "msisdn": str(_first(params, state, "msisdn")),
            "sr": params.get("protocolo_id") or "",
            "socialSecNo": params.get("social_sec_no") or "",
            "customerId": params.get("customer_id") or "",
            "customerIdCurrent": params.get("customer_id_current") or params.get("customer_id") or "",
            "invoiceNumber": params.get("current_invoice_number") or params.get("invoice_id") or "",
            "invoiceStatus": status_value,
            "invoiceAmountOpen": params.get("invoice_amount_open"),
            "invoiceAmount": params.get("invoice_amount"),
            "invoiceDueDate": invoice_due,
            "contestationType": params.get("contestation_type") or "0",
            "adjustReason": params.get("adjust_reason") or "SERVICO_NAO_SOLICITADO",
            "refundOption": refund_option or "0",
            "manualContaCertaIndicator": manual_indicator,
            "doubleRefund": bool(params.get("double_refund")),
            "description": params.get("descricao") or "",
            "items": items,
            "userId": params.get("user_id") or os.getenv("TIM_DEFAULT_CLIENT_ID"),
            "messageId": params.get("message_id") or "",
            "clientId": params.get("client_id") or os.getenv("TIM_DEFAULT_CLIENT_ID"),
        }
        # CVAL é capability do framework: antes do side effect, valida que os
        # itens existem na fatura, pertencem a VAS, não são estratégicos e que o
        # valor solicitado não excede o cobrado. O domínio apenas fornece a fatura.
        if not bool(params.get("skip_invoice_item_validation")) and items:
            try:
                invoice_payload = service.client.billing_analysis(payload["msisdn"])
                canonical_items = []
                for item in items:
                    if not isinstance(item, dict):
                        continue
                    canonical_items.append({
                        "item_name": item.get("item_name") or item.get("itemName") or item.get("name") or item.get("desc") or "",
                        "claimed_amount": item.get("claimed_amount") if item.get("claimed_amount") is not None else item.get("claimedAmount") or item.get("valor") or 0,
                        "validated_amount": item.get("validated_amount") if item.get("validated_amount") is not None else item.get("validatedAmount") or item.get("claimedAmount") or item.get("valor") or 0,
                    })
                validated, validation_log, validation_error = validate_contestation_items(
                    canonical_items, invoice_payload if isinstance(invoice_payload, dict) else {}
                )
                if validation_error:
                    return {
                        "success": False,
                        "blocked": True,
                        "guardrail_code": "CVAL",
                        "guardrail_reason": validation_error,
                        "validation_log": validation_log,
                        "items": validated,
                        "business_events": _events(VAATag.PROTOCOLO_INICIO, VAATag.CONTESTA_VAS_AVULSO_FAIL, gsm=payload["msisdn"]),
                    }
            except Exception as exc:
                # Fail closed: contestação é efeito financeiro. Sem evidência da
                # fatura a validação não pode ser pulada silenciosamente.
                return {
                    "success": False,
                    "blocked": True,
                    "guardrail_code": "CVAL",
                    "guardrail_reason": f"Falha ao validar itens da contestação: {exc}",
                    "business_events": _events(VAATag.PROTOCOLO_INICIO, VAATag.CONTESTA_VAS_AVULSO_FAIL, gsm=payload["msisdn"]),
                }

        protocol_id = str(payload.get("sr") or "")
        adjusted_items = [
            str(x.get("itemName") or x.get("item_name") or x.get("name") or x.get("desc") or "")
            for x in items if isinstance(x, dict)
        ]
        adjusted_amounts = [
            str(x.get("validatedAmount") if x.get("validatedAmount") is not None else x.get("claimedAmount") if x.get("claimedAmount") is not None else x.get("valor") or "")
            for x in items if isinstance(x, dict)
        ]
        event_overrides = {
            "agentProtocolId": protocol_id,
            "adjustedProtocol": protocol_id,
            "billingId": payload.get("invoiceNumber") or "",
            "adjustedItems": adjusted_items,
            "adjustedItemsAmount": adjusted_amounts,
            "agentSpecificData": {
                "adjustedItems": adjusted_items,
                "adjustedItemsAmount": adjusted_amounts,
            },
        }
        try:
            result = service.client.contestar(payload)
        except Exception as exc:
            return {
                "success": False,
                "error": str(exc),
                "sr": payload["sr"],
                "business_events": _events_ctx(
                    (VAATag.PROTOCOLO_INICIO, VAATag.CONTESTA_VAS_AVULSO_FAIL), params, state, **event_overrides
                ),
            }
        items_response = []
        if isinstance(result, dict):
            items_response = result.get("itemsResponse") or result.get("items_response") or result.get("items") or []
        items_response = [x for x in items_response if isinstance(x, dict)] if isinstance(items_response, list) else []
        contested_items, not_contested_items, already_contested_items, contested_open, contested_total = _classify_contestation_items(
            [x for x in items if isinstance(x, dict)], items_response
        )
        barcode = ""
        manual = bool(params.get("manual_conta_certa_indicator"))
        if isinstance(result, dict):
            barcode = str(result.get("barcode") or result.get("codigo_boleto") or result.get("barCode") or "")
            digits = re.sub(r"\D", "", barcode)
            if digits and set(digits) == {"0"}:
                barcode = ""
            manual = bool(result.get("manualContaCertaIndicator", manual))
        events = _events_ctx((VAATag.PROTOCOLO_INICIO, VAATag.PROTOCOLO_OK), params, state, **event_overrides)
        events += _events_ctx(
            VAATag.CODIGO_BARRAS_ELEGIVEL if barcode else VAATag.CODIGO_BARRAS_NAO_ELEGIVEL,
            params, state, **event_overrides
        )
        conta_certa_status = "NAO_SE_APLICA"
        for item in items_response:
            value = str(item.get("correctAccountStatus") or item.get("correctAccounStatus") or "").strip().upper()
            if value:
                conta_certa_status = value
                if value in {"CRIAR", "ENVIADA", "INICIADA"}:
                    break
        events += _transport_rct_events(result) or _events("RCT.007", rctSource="contestacao")
        return {
            "success": True,
            "result": result,
            "items_response": items_response,
            "contested_items": contested_items,
            "not_contested_items": not_contested_items,
            "itens_ja_contestados": already_contested_items,
            "contested_invoice_amount_open": contested_open,
            "contested_invoice_amount": contested_total,
            "barcode": barcode,
            "manual_conta_certa_indicator": manual,
            "conta_certa_status": conta_certa_status,
            "contestation_registered": True,
            "sr": payload["sr"],
            "business_events": events,
        }

    @reg.action("enviar_sms")
    def enviar_sms(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        # Boleto/SMS sempre vai para o titular. Barcode vem do resultado da
        # contestação quando não foi explicitamente mapeado pelo workflow.
        msisdn = str(_first(params, state, "titular_msisdn", "msisdn"))
        contest = _workflow_var(state, "abrir_contestacao_cliente")
        barcode = str(params.get("barcode") or contest.get("barcode") or "").strip()
        message = str(params.get("message") or (f"Codigo do boleto: {barcode}" if barcode else ""))
        try:
            if hasattr(service, "client") and hasattr(service.client, "sms"):
                result = service.client.sms(
                    msisdn, message, sender_name=get_tim_integration_value("TIM_SMS_SENDER_NAME"), long_url=get_tim_integration_value("TIM_SMS_LONG_URL"),
                    notify_url=get_tim_integration_value("TIM_SMS_NOTIFY_URL"),
                )
            else:
                result = service.enviar_sms(msisdn=msisdn, message=message)
            sms_meta = _transport_event_metadata(result)
            sms_params = {**params, "msisdn": msisdn, "channelId": "SMS"}
            events = _events_ctx((VAATag.SMS_OK, VAATag.STATUS_SR_COM_SMS), sms_params, state, **sms_meta)
            events += _transport_rct_events(result) or _events_ctx("RCT.013", sms_params, state, rctSource="sgr_codbar", **sms_meta)
            resource_url = str(result.get("resource_url") or result.get("resourceUrl") or "") if isinstance(result, dict) else ""
            return {"success": True, "result": result, "resource_url": resource_url, "sms_enviado": True, "sms_sent": True, "sms_not_send_error": False, "business_events": events}
        except Exception as exc:
            return {
                "success": True, "result": None, "error": str(exc),
                "sms_enviado": False, "sms_sent": False, "sms_not_send_error": True,
                "sms_nao_enviado_erro": True, "resource_url": "",
                "business_events": _events_ctx((VAATag.SMS_FAIL, VAATag.STATUS_SR_SEM_SMS), {**params, "msisdn": msisdn, "channelId": "SMS"}, state, apiResponsePayload=str(exc)),
            }

    @reg.action("consultar_contrato_corte")
    def consultar_contrato_corte(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        msisdn = str(_first(params, state, "msisdn"))
        contract = service.client.contrato(msisdn)
        billing = {}
        if isinstance(contract, dict):
            billing = contract.get("billing_profile") or contract.get("billingProfile") or {}
        date_info = billing.get("date") if isinstance(billing, dict) else {}
        cutoff_day = 0
        if isinstance(date_info, dict):
            try: cutoff_day = int(date_info.get("cutoffDay") or 0)
            except (TypeError, ValueError): cutoff_day = 0
        after_cutoff, reference = billing_cutoff_reference(cutoff_day)
        dependent = bool(params.get("dependent_invoice_item"))
        vars_state = state.get("vars") or {}
        contest = vars_state.get("abrir_contestacao_cliente", {}) if isinstance(vars_state, dict) else {}
        registered = bool(contest.get("success")) if isinstance(contest, dict) else False
        status = "NAO_SE_APLICA"
        if isinstance(contest, dict):
            for item in contest.get("items_response") or []:
                if isinstance(item, dict):
                    status = str(item.get("correctAccountStatus") or item.get("correctAccounStatus") or status).upper()
                    if status in {"CRIAR", "ENVIADA", "INICIADA"}:
                        registered = True
                        break
        due = str(params.get("current_invoice_due_date") or "")
        due_day_value = due_day(due)
        return {
            "apos_data_corte": after_cutoff,
            "data_corte_flag": "0" if after_cutoff else "1",
            "abrir_conta_certa_manual": bool(after_cutoff or dependent),
            "contestation_success": registered,
            "contestation_registered": registered,
            "conta_certa_status": status,
            "dependent_invoice_item": dependent,
            "dia_corte": cutoff_day,
            "dia_vencimento": due_day_value,
            "data_corte_referencia": reference.strftime("%Y-%m-%d") if reference else "",
        }

    @reg.action("abrir_sr_conta_certa_manual")
    def abrir_sr_conta_certa_manual(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        cutoff = _workflow_var(state, "consultar_contrato_corte")
        day = params.get("dia_vencimento") or cutoff.get("dia_vencimento") or ""
        triplet = resolve_protocol_triplet("conta_certa_manual", stage="open", context={"dia_vencimento": day})
        payload = {
            "msisdn": str(_first(params, state, "msisdn")),
            "socialSecNo": params.get("social_sec_no") or "",
            "reason1": triplet.reason1, "reason2": triplet.reason2, "reason3": triplet.reason3,
            "requestStatus": params.get("request_status") or "Encaminhado", "status": params.get("status") or "Encaminhado",
        }
        try:
            result = service.client.abrir_protocolo(payload)
            pid = str(result.get("interactionProtocol") or result.get("protocolNumber") or "") if isinstance(result, dict) else ""
            return {"success": True, "status": "Encaminhado", "sr_conta_certa_id": pid, "sr_conta_certa_aberta": bool(pid), "conta_certa_manual_error": False, "dia_vencimento": day, "result": result}
        except Exception as exc:
            # Conta Certa Manual é best-effort no original: a falha da API não
            # derruba a contestação principal.
            return {"success": True, "status": "Encaminhado", "sr_conta_certa_id": "", "sr_conta_certa_aberta": False, "conta_certa_manual_error": True, "dia_vencimento": day, "error": str(exc), "result": None}

    @reg.action("atualizar_status_sr")
    def atualizar_status_sr(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        msisdn = str(_first(params, state, "msisdn"))
        protocol_var = _workflow_var(state, "registrar_protocolo")
        manual_var = _workflow_var(state, "abrir_sr_conta_certa_manual")
        sms_var = _workflow_var(state, "enviar_sms")
        protocol_id = str(params.get("protocolo_id") or params.get("protocol_number") or protocol_var.get("protocolo_id") or protocol_var.get("protocol_number") or "")
        scenario = str(params.get("scenario") or "contestacao")
        status = str(params.get("status") or "Fechado")
        status_payload = {
            "msisdn": msisdn, "socialSecNo": re.sub(r"\D", "", str(params.get("social_sec_no") or "")),
            "protocolNumber": protocol_id, "status": status, "channel": os.getenv("TIM_DEFAULT_CHANNEL"),
            "clientId": params.get("client_id") or os.getenv("TIM_DEFAULT_CLIENT_ID"),
            "reason1": "Reclamação", "reason2": "Conta", "reason3": "Valor", "notes": "",
        }
        try:
            result = service.client.status_sr(status_payload)
        except Exception as exc:
            return {
                "success": True, "result": None, "error": str(exc), "protocol_closed": False,
                "business_events": _events_ctx(VAATag.STATUS_SR_FAIL, params, state, agentProtocolId=protocol_id, adjustedProtocol=protocol_id, apiResponsePayload=str(exc)),
            }

        # O original registra TrackingActivities em cenários de contestação,
        # enriquecido com os dados da fatura já presentes no prefetch/check.
        tracking_result = None
        if scenario in {"contestacao", "valor_divergente", "conta_certa_manual"}:
            invoice_id = str(params.get("current_invoice_number") or params.get("invoice_id") or "")
            selected = _selected_invoice_for_tracking(state, invoice_id)
            if selected or not invoice_id:
                raw_status = str(selected.get("invoiceStatus") or selected.get("status") or "") if selected else ""
                normalized_status = "Aberto" if raw_status.strip().lower() in {"a vencer", "em aberto", "aberto", "open"} else (raw_status or "Fechada")
                tracking_payload = {
                    "msisdn": msisdn,
                    "socialSecNo": re.sub(r"\D", "", str(params.get("social_sec_no") or "")),
                    "protocolNumber": protocol_id,
                    "invoiceEmissionDate": _iso_tracking_date(selected.get("issueDate") or selected.get("emissionDate") or "") if selected else "",
                    "invoiceExpirationDate": _iso_tracking_date(selected.get("dueDate") or selected.get("expirationDate") or "") if selected else "",
                    "invoiceNumber": invoice_id or str(selected.get("invoiceId") or selected.get("id") or "") if selected else invoice_id,
                    "invoiceStatus": normalized_status,
                    "invoiceOpenAmount": str(params.get("invoice_amount_open") or selected.get("openAmount") or selected.get("amountOpen") or "") if selected else str(params.get("invoice_amount_open") or ""),
                    "invoiceTotalAmount": str(params.get("invoice_amount") or selected.get("amount") or selected.get("totalAmount") or "") if selected else str(params.get("invoice_amount") or ""),
                    "activityType": "Contestação", "activityStatus": "Aberto", "activityId": "",
                }
                try:
                    tracking_result = service.client.tracking(tracking_payload)
                except Exception:
                    tracking_result = None

        sent = bool(sms_var.get("sms_sent") or sms_var.get("sms_enviado"))
        sms_error = bool(sms_var.get("sms_not_send_error") or sms_var.get("sms_nao_enviado_erro"))
        format_text = "sms" if sent else "conta_futura"
        credit_date = str(params.get("data_credito_proxima_fatura") or _first(params, state, "data_credito_proxima_fatura") or "")
        status_meta = _transport_event_metadata(result)
        return {
            "success": True, "result": result, "tracking_result": tracking_result, "protocol_closed": True,
            "format_text": format_text, "sms_enviado": sent, "sms_sent": sent, "sms_not_send_error": sms_error,
            "resolution_type": "new_boleto" if sent else "credit_bill",
            "data_credito_proxima_fatura": credit_date,
            "sr_conta_certa_id": str(manual_var.get("sr_conta_certa_id") or ""),
            "business_events": _events_ctx(VAATag.STATUS_SR_OK, params, state, agentProtocolId=protocol_id, adjustedProtocol=protocol_id, **status_meta),
        }

    def _discount_evidence_context(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        source = state.get("input") if isinstance(state.get("input"), dict) else {}
        return {**source, **params}

    def _first_explicit_discount_reason(value: Any) -> tuple[str, str]:
        """Return (reason, source_key) only from explicit backend evidence.

        This deliberately does not infer expiration from installment counters,
        absence of a discount, plan names or the customer's wording. Causal
        claims must be supplied by a system of record/mock contract.
        """
        reason_keys = {
            "discount_reason", "discountReason", "motivo_desconto", "motivoDesconto",
            "termination_reason", "terminationReason", "motivo_termino", "motivoTermino",
            "promotion_end_reason", "promotionEndReason",
        }
        status_keys = {"discount_status", "discountStatus", "promotion_status", "promotionStatus"}
        end_date_keys = {"promotion_end_date", "promotionEndDate", "discount_end_date", "discountEndDate"}

        def walk(obj: Any) -> tuple[str, str]:
            if isinstance(obj, dict):
                for key, item in obj.items():
                    if key in reason_keys and isinstance(item, (str, int, float)) and str(item).strip():
                        return str(item).strip(), key
                for key, item in obj.items():
                    if key in status_keys and str(item or "").strip().upper() in {
                        "EXPIRED", "ENDED", "TERMINATED", "ENCERRADO", "EXPIRADO", "FINALIZADO"
                    }:
                        return f"status explícito: {str(item).strip()}", key
                for key, item in obj.items():
                    if key in end_date_keys and isinstance(item, (str, int, float)) and str(item).strip():
                        return f"data de término registrada: {str(item).strip()}", key
                for item in obj.values():
                    found = walk(item)
                    if found[0]:
                        return found
            elif isinstance(obj, list):
                for item in obj:
                    found = walk(item)
                    if found[0]:
                        return found
            return "", ""

        return walk(value)

    def _discount_record_from_evidence(value: Any) -> dict[str, Any]:
        """Select the most relevant discount record from authoritative evidence."""
        if not isinstance(value, dict):
            return {}
        rows = value.get("discounts") if isinstance(value.get("discounts"), list) else []
        candidates = [row for row in rows if isinstance(row, dict)]
        if not candidates:
            return {}
        terminal_statuses = {"EXPIRED", "ENDED", "TERMINATED", "ENCERRADO", "EXPIRADO", "FINALIZADO"}
        for row in candidates:
            if str(row.get("discount_status") or row.get("status") or "").strip().upper() in terminal_statuses:
                return row
        return candidates[0]

    @staticmethod
    def _format_brl(value: Any) -> str:
        try:
            number = float(value)
        except (TypeError, ValueError):
            return ""
        return f"R$ {number:,.2f}".replace(",", "X").replace(".", ",").replace("X", ".")

    def _format_iso_date_br(value: Any) -> str:
        text = str(value or "").strip()
        if re.fullmatch(r"\d{4}-\d{2}-\d{2}", text):
            year, month, day = text.split("-")
            return f"{day}/{month}/{year}"
        return text

    def _plan_names_from_invoice_detail(invoice_detail: Any) -> list[str]:
        names: list[str] = []
        if not isinstance(invoice_detail, dict):
            return names
        for bucket in invoice_detail.values():
            if not isinstance(bucket, dict):
                continue
            planos = bucket.get("Planos")
            if isinstance(planos, dict):
                for name in planos:
                    text = str(name or "").strip()
                    if text and text not in names:
                        names.append(text)
        return names

    @reg.action("formatar_capability_resposta")
    def formatar_capability_resposta(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        typ = str(params.get("tipo") or "")
        if typ == "termino_desconto":
            context = _discount_evidence_context(params, state)
            evidence = {
                "discount_evidence": context.get("discount_evidence"),
                "invoice_detail": context.get("invoice_detail"),
                "billing_analysis": context.get("billing_analysis"),
                "plan_data": context.get("plan_data"),
            }
            reason, reason_source = _first_explicit_discount_reason(evidence)
            record = _discount_record_from_evidence(context.get("discount_evidence"))
            requested_plan = str(context.get("nome_plano") or "").strip()
            discovered_plans = _plan_names_from_invoice_detail(context.get("invoice_detail"))
            plan = str(record.get("plan_name") or requested_plan or (discovered_plans[0] if len(discovered_plans) == 1 else "")).strip()
            plan_text = f" do plano {plan}" if plan else ""

            # Prefer the human-readable causal description from the system of
            # record. The code remains audit metadata, not customer-facing prose.
            reason_description = str(
                record.get("termination_reason_description")
                or record.get("discount_reason_description")
                or ""
            ).strip()
            if reason_description:
                reason = reason_description
                reason_source = "termination_reason_description"

            if reason:
                discount_name = str(record.get("discount_name") or "").strip()
                previous_value = _format_brl(record.get("previous_value"))
                end_date = _format_iso_date_br(record.get("end_date") or record.get("discount_end_date"))
                subject = f"do desconto {discount_name}" if discount_name else "do desconto"
                details: list[str] = []
                if previous_value:
                    details.append(f"no valor de {previous_value}")
                if end_date:
                    details.append(f"em {end_date}")
                detail_text = (", " + ", ".join(details)) if details else ""
                clean_reason = reason.rstrip(" .")
                msg = f"Identifiquei o término {subject}{plan_text}{detail_text}. Motivo informado pelo sistema: {clean_reason}."

                # O histórico de desconto descreve a situação contratual em uma
                # data de referência, enquanto a última fatura pode cobrir um
                # período anterior. Quando o backend fornece ambas as referências
                # temporais, explicite a diferença sem inferir causalidade.
                last_billed_value = _format_brl(record.get("last_billed_discount_value"))
                last_billed_period = str(record.get("last_billed_period") or "").strip()
                last_invoice_issue_date = _format_iso_date_br(record.get("last_invoice_issue_date"))
                as_of_date = _format_iso_date_br(record.get("as_of_date"))
                current_value = record.get("current_value")
                current_value_reference = str(record.get("current_value_reference") or "").strip()
                if (
                    current_value_reference == "contract_as_of_date"
                    and current_value in (0, 0.0, "0", "0.0", "0.00")
                    and last_billed_value
                    and last_billed_period
                ):
                    billed_parts = [
                        f"O último período faturado com esse desconto foi {last_billed_period}",
                        f"com {last_billed_value} de desconto",
                    ]
                    if last_invoice_issue_date:
                        billed_parts.append(f"na fatura emitida em {last_invoice_issue_date}")
                    contract_ref = f"; a situação contratual em {as_of_date} já consta como encerrada" if as_of_date else "; a situação contratual atual já consta como encerrada"
                    msg += " " + ", ".join(billed_parts) + contract_ref + "."
                grounded = True
            else:
                msg = (
                    f"Identifiquei dados de desconto{plan_text}, mas os dados disponíveis não informam "
                    "o motivo da retirada ou do término do desconto."
                )
                grounded = False

            return {
                "mensagem": msg,
                "business_events": _events(MPITag.TERMINO_DESCONTO),
                "discount_reason_grounded": grounded,
                "discount_reason": reason or None,
                "discount_reason_source": reason_source or None,
                "discount_record": record or None,
                "evidence_policy": "explicit_reason_only",
                "epistemic_status": "grounded_fact" if grounded else "insufficient_evidence",
            }
        elif typ == "valor_divergente":
            msisdn = str(params.get("msisdn") or "").strip()
            suffix = msisdn[-2:] if len(msisdn) >= 2 else msisdn
            line = f" na linha final {suffix}" if suffix else ""
            msg = f"Identifiquei uma alteração no valor do plano{line}."
            ev = [MPITag.VALOR_DIVERGENTE]
        else:
            msg = str(params.get("mensagem") or "")
            ev = []
        return {"mensagem": msg, "business_events": _events(*ev)}

    @reg.action("finalizar_atendimento_action")
    def finalizar_atendimento_action(params: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
        state_input = state.get("input") if isinstance(state.get("input"), dict) else {}
        context = {**state_input, **params}
        customer_protocols = context.get("customer_protocols") if isinstance(context.get("customer_protocols"), list) else []
        customer_protocol_number = next((str(x.get("number") or "").strip() for x in customer_protocols if isinstance(x, dict) and str(x.get("number") or "").strip()), "")
        line_protocols = context.get("protocolos_por_linha") if isinstance(context.get("protocolos_por_linha"), list) else []
        line_protocol_number = next((str(x.get("protocolo_id") or x.get("protocol_number") or "").strip() for x in line_protocols if isinstance(x, dict) and str(x.get("protocolo_id") or x.get("protocol_number") or "").strip()), "")
        force_rt15 = bool(context.get("force_rt15_finalization_protocol"))
        existing_protocol = str(
            context.get("protocol_number")
            or context.get("protocolo_id")
            or context.get("contestacao_protocol")
            or customer_protocol_number
            or (line_protocol_number if not force_rt15 else "")
            or ""
        ).strip()
        informational_types: set[str] = set()
        for key in ("informational_vas_types", "informational_rag_vas_types"):
            values = context.get(key) if isinstance(context.get(key), list) else []
            informational_types.update(str(v).strip() for v in values if str(v).strip() in {"bundle", "estrategico", "avulso"})
        saved_informational_types = set(informational_types)
        informational_services = [
            str(x).strip() for x in (context.get("informational_service_names") or [])
            if str(x).strip()
        ] if isinstance(context.get("informational_service_names"), list) else []
        invoice_detail = context.get("invoice_detail")
        invoice_detail_available = bool(
            invoice_detail if isinstance(invoice_detail, dict) else str(invoice_detail or "").strip()
        )
        if informational_services and invoice_detail_available:
            inferred, matched_count = _infer_informational_vas_types_with_count(informational_services, invoice_detail)
            # Mesma precedência do original: sem match, tipos salvos são descartados;
            # com match total, a fatura define tudo; em match parcial, preservamos
            # também os tipos salvos dos serviços ainda não representados no detalhe.
            if matched_count == 0:
                informational_types = set()
            elif matched_count == len(informational_services):
                informational_types = set(inferred)
            else:
                informational_types = set(inferred) | saved_informational_types
        if not informational_types and informational_services:
            if any(_service_alias_matches(name, _STRATEGIC_SERVICE_ALIASES) for name in informational_services):
                informational_types.add("estrategico")
            elif any(_service_alias_matches(name, _INFORMATIONAL_AVULSO_SERVICE_ALIASES) for name in informational_services):
                informational_types.add("avulso")
        notes = _format_informational_vas_notes(informational_types)

        executed = {str(x) for x in (context.get("business_workflows_executed") or [])}
        invoice_notes = str(context.get("invoice_explanation_base") or "").strip()
        status = _normalize_final_status(context.get("status") or "resolvido")
        summary_text = str(context.get("summary") or "").strip()

        suppress_lookup = bool(context.get("suppress_cvn_lookup_on_finalize") or context.get("suppressCvnLookupOnFinalize"))
        suppress_negative = bool(context.get("suppress_cvn_009_on_finalize") or context.get("suppressCvn009OnFinalize"))
        suppress_protocol = bool(context.get("suppress_informational_protocol_on_finalize"))
        suppress_protocol_ic = bool(context.get("suppress_cvn_protocol_ic") or context.get("suppressCvnProtocolIc"))
        unresolved_transition = bool(context.get("conversation_unresolved_transition_emitted"))
        explicit_resolution = bool(context.get("conversation_explanation_resolved"))
        pending_resolution = bool(context.get("conversation_explanation_pending_resolution"))

        informational_flows = {"invoice_explanation", "buscar_informacao", "vas_estrategico"}
        transactional_flows = {
            "cancelar_vas_avulso", "cancelamento_vas_avulso", "contestacao_tool",
            "contestar_cobranca", "pro_rata", "termino_desconto", "valor_divergente",
        }
        business_or_transition = bool(executed) or unresolved_transition
        transactional_or_transition = bool(executed & transactional_flows) or unresolved_transition
        invoice_flow = "invoice_explanation" in executed

        existing_before_notes = bool(existing_protocol)
        suppress_conversational = bool(
            context.get("vas_estrategico_protocol_deferred_to_finalization")
            or suppress_protocol
            or suppress_protocol_ic
            or (bool(executed & transactional_flows) and existing_before_notes)
        )

        # Historical behaviour: prefetch/informational context may justify RT-15 even
        # when invoice_explanation was not explicitly latched, as long as finalization
        # semantics indicate an informational resolution.
        summary_l = summary_text.casefold()
        info_summary = bool(summary_l) and "fora do escopo" not in summary_l and any(
            token in summary_l for token in ("explica", "entendeu", "entendi", "agradec", "obrigado", "obrigada", "encerr")
        )
        if not notes and invoice_notes and (
            bool(executed & informational_flows) or pending_resolution or info_summary
        ):
            notes = _INFORMATIONAL_INVOICE_NOTES

        events: list[dict[str, Any]] = []
        should_lookup = bool(
            pending_resolution or invoice_notes or informational_services or informational_types
            or bool(executed & informational_flows)
        )
        if not business_or_transition and not suppress_conversational and not suppress_lookup and should_lookup:
            events += _events(CVNTag.PRE_VALIDACAO_OK, CVNTag.SERVICO_OK)

        resolved_status = status in {"resolvido", "resolvido_outros_assuntos", "outros_assuntos"}
        terminal_veb = bool(
            explicit_resolution
            and (
                context.get("veb_protocol_closed_requires_rt15_finalization")
                or context.get("vas_estrategico_protocol_deferred_to_finalization")
                or (suppress_protocol_ic and bool(informational_types & {"bundle", "estrategico"}))
            )
        )
        can_emit_acceptance = bool(
            resolved_status
            and not context.get("conversation_final_acceptance_cvn_emitted")
            and not terminal_veb
            and (not transactional_or_transition or explicit_resolution)
            and (not suppress_conversational or explicit_resolution)
        )
        conversational_context = bool(
            invoice_flow or invoice_notes or pending_resolution or explicit_resolution
            or informational_services or informational_types or context.get("invoice_id")
        )
        can_emit_rejection = bool(
            status == "nao_resolvido"
            and conversational_context
            and not transactional_or_transition
            and not suppress_conversational
            and not suppress_negative
        )
        if can_emit_acceptance:
            events += _events_ctx(CVNTag.CLIENTE_CONCORDOU, context, state)
        elif can_emit_rejection:
            events += _events_ctx(CVNTag.CLIENTE_NAO_CONCORDOU, context, state)

        informational_protocol = ""
        protocol_failed = False
        if notes and not existing_protocol and not suppress_protocol:
            deferred_veb = bool(context.get("vas_estrategico_protocol_deferred_to_finalization"))
            try:
                protocol_result = registrar_protocolo({
                    "msisdn": context.get("msisdn") or "",
                    "social_sec_no": context.get("social_sec_no") or context.get("cpf") or "",
                    "scenario": "vas_estrategico" if deferred_veb else "finalizacao_informacional_fechada",
                    "request_status": "Fechado",
                    "status": "CLOSED",
                    "service_request_notes": notes,
                    "message_id": context.get("message_id") or context.get("messageId") or "",
                    "rct_operation": "reg_chamado_bo_rt15",
                }, state)
                informational_protocol = str(protocol_result.get("protocolo_id") or "")
                protocol_failed = not bool(protocol_result.get("success"))
            except Exception:
                protocol_result = {"success": False, "protocolo_id": ""}
                protocol_failed = True
            protocol_transport_events = _transport_rct_events(protocol_result)
            events += protocol_transport_events or _events_ctx(
                "RCT.085" if informational_protocol and not protocol_failed else "RCT.086",
                context, state,
                agentProtocolId=informational_protocol, adjustedProtocol=informational_protocol,
            )
            if not suppress_protocol_ic:
                protocol_meta = {
                    "agentProtocolId": informational_protocol,
                    "adjustedProtocol": informational_protocol,
                    "uraProtocolId": context.get("ura_protocol_id") or context.get("uraProtocolId") or "",
                }
                transport = (protocol_result.get("result") or {}).get("_transport") if isinstance(protocol_result, dict) and isinstance(protocol_result.get("result"), dict) else {}
                if isinstance(transport, dict):
                    attempts = transport.get("attempts") or []
                    last_attempt = attempts[-1] if isinstance(attempts, list) and attempts else {}
                    if isinstance(last_attempt, dict):
                        protocol_meta.update({
                            "apiUrl": last_attempt.get("api_url") or "",
                            "apiStatusCode": last_attempt.get("status_code") or 0,
                            "latencyMs": last_attempt.get("latency_ms") or 0,
                        })
                    if transport.get("api_response_payload") not in (None, ""):
                        protocol_meta["apiResponsePayload"] = json.dumps(transport.get("api_response_payload"), ensure_ascii=False, default=str)
                events += _events_ctx(
                    CVNTag.REGISTRA_ATENDIMENTO_OK if informational_protocol and not protocol_failed else CVNTag.REGISTRA_ATENDIMENTO_FAIL,
                    context, state, **protocol_meta
                )

        # MPI mirrors the historical informational acceptance contract.
        can_emit_mpi_positive = bool(
            resolved_status
            and not context.get("conversation_final_acceptance_mpi_emitted")
            and not terminal_veb
            and (not suppress_conversational or explicit_resolution)
            and (not unresolved_transition or explicit_resolution)
            and (not bool(executed & transactional_flows) or explicit_resolution)
        )
        if can_emit_mpi_positive and (invoice_flow or invoice_notes or pending_resolution or explicit_resolution):
            events += _events_ctx(MPITag.EXPLICACAO_SIM, context, state)
        elif can_emit_rejection and (invoice_flow or invoice_notes or pending_resolution):
            events += _events_ctx(MPITag.EXPLICACAO_NAO, context, state)

        # Optional detailed SAD decision tree from the historical contract.
        events += _events_ctx(SADTag.FINALIZACAO, context, state)
        if bool(context.get("sad_exit_decision_enabled")):
            technical = status in {"erro_falha_sistema", "erro_no_match", "erro_no_input"}
            events += _events_ctx(SADTag.ERRO_TECNICO_SIM if technical else SADTag.ERRO_TECNICO_NAO, context, state)
            if not technical:
                handoff = bool(context.get("human_handoff"))
                events += _events_ctx(SADTag.INTENCAO_ATH_SIM if handoff else SADTag.INTENCAO_ATH_NAO, context, state)
                if handoff:
                    first_time = bool(context.get("ath_first_time", True))
                    events += _events_ctx(SADTag.ATH_1X_SIM if first_time else SADTag.ATH_1X_NAO, context, state)

        protocol_to_close = "" if bool(context.get("protocol_closed")) else existing_protocol
        result = service.finalizar_atendimento(
            status=status,
            summary=str(context.get("summary") or ""),
            protocol=protocol_to_close,
            msisdn=context.get("msisdn") or "",
        )
        if existing_protocol and bool(context.get("protocol_closed")):
            result["protocol_number"] = existing_protocol
            result["protocolo_id"] = existing_protocol
        if informational_protocol:
            result["protocol_number"] = informational_protocol
            result["protocolo_id"] = informational_protocol
            result["finalizacao_protocol"] = informational_protocol
            result["informational_protocol_notes"] = notes
        result["business_events"] = events
        return result

    return reg