diff --git a/app/domain/contas/client.py b/app/domain/contas/client.py
index 27a3b46..ac87622 100644
--- a/app/domain/contas/client.py
+++ b/app/domain/contas/client.py
@@ -1,5 +1,6 @@
from __future__ import annotations
+import base64
import json
import os
import time
@@ -41,6 +42,48 @@ class TimApiClient:
path = self.FIXTURES / f"{name}.json"
return json.loads(path.read_text(encoding="utf-8"))
+ @staticmethod
+ def _env_first(*names: str, default: str = "") -> str:
+ """Retorna a primeira variável não vazia, preservando aliases do Contas original."""
+ for name in names:
+ value = os.getenv(name)
+ if value is not None and str(value).strip():
+ return str(value).strip()
+ return default
+
+ @classmethod
+ def _auth_value(
+ cls,
+ *auth_names: str,
+ user_names: tuple[str, ...] = (),
+ password_names: tuple[str, ...] = (),
+ ) -> str:
+ user = cls._env_first(*user_names) if user_names else ""
+ password = cls._env_first(*password_names) if password_names else ""
+ if user and password:
+ token = base64.b64encode(f"{user}:{password}".encode("utf-8")).decode("ascii")
+ return f"Basic {token}"
+ raw = cls._env_first(*auth_names)
+ if not raw:
+ return ""
+ low = raw.lower()
+ if low.startswith("basic ") or low.startswith("bearer "):
+ return raw
+ if ":" in raw and " " not in raw:
+ return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("ascii")
+ return raw
+
+ @staticmethod
+ def _url_with_msisdn(base: str, msisdn: str, *, normalize_country: bool = False) -> str:
+ text = str(base or "").strip()
+ digits = "".join(ch for ch in str(msisdn or "") if ch.isdigit())
+ if normalize_country and digits and not digits.startswith("55"):
+ digits = f"55{digits}"
+ value = digits or str(msisdn or "").strip()
+ if "{msisdn}" in text:
+ return text.replace("{msisdn}", quote(value))
+ return f"{text.rstrip('/')}/{quote(value)}" if text else value
+
def _headers(self, *, client_id: str | None = None, auth: str | None = None, extra: dict[str, str] | None = None) -> dict[str, str]:
h = {"Content-Type": "application/json", "Accept": "application/json"}
if client_id:
@@ -107,22 +150,33 @@ class TimApiClient:
def consultar_faturas(self, msisdn: str) -> Any:
if self.mock:
return self.fixture("complete_invoices")
- url = os.getenv("TIM_COMPLETE_INVOICES_URL", "")
- return self.request("POST", url, payload={"msisdn": msisdn}, headers=self._headers(client_id=None, auth=os.getenv("TIM_COMPLETE_INVOICES_AUTH", ""), extra={"ClientID": os.getenv("TIM_COMPLETE_INVOICES_CLIENT_ID", "AIAGENTCR")}))
+ url = self._env_first("TIM_COMPLETE_INVOICES_URL")
+ return self.request(
+ "POST", url, payload={"msisdn": msisdn},
+ headers=self._headers(
+ client_id=None, auth=self._env_first("TIM_COMPLETE_INVOICES_AUTH"),
+ extra={"ClientID": self._env_first("TIM_COMPLETE_INVOICES_CLIENT_ID", default="AIAGENTCR")},
+ ),
+ timeout=int(self._env_first("TIM_COMPLETE_INVOICES_TIMEOUT", default="30")),
+ )
def billing_analysis(self, msisdn: str, **context: Any) -> Any:
if self.mock:
return self.fixture("divergencia")
- base = os.getenv("TIM_DIVERGENCIA_URL", "").rstrip("/")
+ base = self._env_first("TIM_DIVERGENCIA_URL").rstrip("/")
url = f"{base}/{quote(msisdn)}"
- auth = os.getenv("TIM_DIVERGENCIA_AUTH", "") or os.getenv("TIM_SECURE_PDF_AUTH", "")
+ auth = self._auth_value(
+ "TIM_DIVERGENCIA_AUTH", "TIM_DIVERGENCE_AUTH",
+ user_names=("TIM_DIVERGENCIA_USER", "TIM_DIVERGENCIA_USERNAME", "TIM_DIVERGENCE_USER", "TIM_DIVERGENCE_USERNAME"),
+ password_names=("TIM_DIVERGENCIA_PASSWORD", "TIM_DIVERGENCIA_PASS", "TIM_DIVERGENCE_PASSWORD", "TIM_DIVERGENCE_PASS"),
+ )
headers = self._headers(client_id=None, auth=auth, extra={
- "clientID": os.getenv("TIM_DIVERGENCIA_CLIENT_ID", "AIAGENTCR"),
+ "clientID": self._env_first("TIM_DIVERGENCIA_CLIENT_ID", default="AIAGENTCR"),
})
params = {"channel": context.get("channel") or "AIAGENTCR"}
attempts: list[dict[str, Any]] = []
try:
- result = self.request("GET", url, params=params, headers=headers, timeout=int(os.getenv("TIM_DIVERGENCIA_TIMEOUT", "120")), attempt_log=attempts)
+ result = self.request("GET", url, params=params, headers=headers, timeout=int(self._env_first("TIM_DIVERGENCIA_TIMEOUT", "TIM_DIVERGENCE_TIMEOUT", default="120")), attempt_log=attempts)
return self._attach_transport(result, operation="base_conhecimento", attempts=attempts)
except TimApiError as exc:
if not exc.attempts:
@@ -132,9 +186,16 @@ class TimApiClient:
def consultar_vas(self, msisdn: str) -> Any:
if self.mock:
return self.fixture("query_vas")
- base = os.getenv("TIM_URL_CONSULTA_VAS", "")
- url = base.replace("{msisdn}", quote(msisdn))
- return self.request("GET", url, headers=self._headers(client_id=None, auth=os.getenv("TIM_CONSULTA_AUTH", ""), extra={"clientId": os.getenv("TIM_CONSULTA_CLIENT_ID", "AIAAGENTCR")}))
+ base = self._env_first("TIM_URL_CONSULTA_VAS", "TIM_CONSULTA_URL")
+ url = self._url_with_msisdn(base, msisdn, normalize_country=True)
+ return self.request(
+ "GET", url,
+ headers=self._headers(
+ client_id=None, auth=self._env_first("TIM_QUERY_AUTH", "TIM_CONSULTA_AUTH"),
+ extra={"clientId": self._env_first("TIM_CONSULTA_CLIENT_ID", default="AIAAGENTCR")},
+ ),
+ timeout=int(self._env_first("TIM_QUERY_TIMEOUT", "TIM_CONSULTA_TIMEOUT", default="30")),
+ )
def historico_vas(self, msisdn: str) -> Any:
if self.mock:
@@ -146,7 +207,7 @@ class TimApiClient:
client_id=None,
auth=os.getenv("TIM_VAS_HISTORY_AUTH", ""),
extra={
- "clientId": os.getenv("TIM_VAS_HISTORY_CLIENT_ID", os.getenv("TIM_DEFAULT_CLIENT_ID", "CHAT")),
+ "clientId": self._env_first("TIM_VAS_HISTORY_CLIENT_ID", default="AIAGENTCR"),
"messageId": os.getenv("TIM_VAS_HISTORY_MESSAGE_ID", "") or str(uuid.uuid4()),
},
)
@@ -155,35 +216,35 @@ class TimApiClient:
def bloquear_vas(self, msisdn: str, service: dict[str, Any]) -> Any:
if self.mock:
return self.fixture("block_vas")
- url = os.getenv("TIM_URL_BLOQUEIO_VAS", "")
+ url = self._env_first("TIM_URL_BLOQUEIO_VAS", "TIM_BLOQUEIO_URL")
digits = "".join(ch for ch in str(msisdn) if ch.isdigit())
normalized = digits[2:] if len(digits) == 13 else digits
app_id = str(service.get("appId") or service.get("app_id") or "")
csp_id = str(service.get("cspId") or service.get("csp_id") or os.getenv("TIM_DEFAULT_CSP_ID", "740"))
- op = os.getenv("TIM_BLOQUEIO_OPERATION_TYPE", "block")
+ op = self._env_first("TIM_BLOCK_OPERATION_TYPE", "TIM_BLOQUEIO_OPERATION_TYPE", default="block")
base = {"Customer": {"Msisdn": normalized}, "AppId": app_id, "CspId": csp_id, "TypeOperation": op}
payloads = [
{"customer": {"msisdn": normalized}, "appId": app_id, "cspId": csp_id, "typeOperation": op},
{"input": base, "Input": base},
{"vasBlock": {"msisdn": normalized, "appId": app_id, "cspId": csp_id, "type": op}},
]
- mode = os.getenv("TIM_BLOQUEIO_PAYLOAD_MODE", "auto").strip().lower()
+ mode = self._env_first("TIM_BLOCK_PAYLOAD_MODE", "TIM_BLOQUEIO_PAYLOAD_MODE", default="auto").lower()
if mode == "vasblock": payloads = [payloads[2], payloads[0]]
elif mode == "input": payloads = [payloads[1], payloads[0]]
elif mode == "pmid": payloads = [payloads[0], payloads[1]]
headers = self._headers(
client_id=None,
- auth=os.getenv("TIM_BLOQUEIO_AUTH", ""),
+ auth=self._env_first("TIM_BLOCK_AUTH", "TIM_BLOQUEIO_AUTH"),
extra={
- "Accept-Encoding": os.getenv("TIM_BLOQUEIO_ACCEPT_ENCODING", "gzip,deflate"),
- "clientId": os.getenv("TIM_BLOQUEIO_CLIENT_ID", "AIAGENTCR"),
+ "Accept-Encoding": self._env_first("TIM_BLOCK_ACCEPT_ENCODING", "TIM_BLOQUEIO_ACCEPT_ENCODING", default="gzip,deflate"),
+ "clientId": self._env_first("TIM_BLOCK_CLIENT_ID", "TIM_BLOQUEIO_CLIENT_ID", default="AIAGENTCR"),
"messageId": str(uuid.uuid4()),
},
)
last = None
for index, payload in enumerate(payloads):
try:
- return self.request("POST", url, payload=payload, headers=headers, timeout=int(os.getenv("TIM_BLOQUEIO_TIMEOUT", "30")))
+ return self.request("POST", url, payload=payload, headers=headers, timeout=int(self._env_first("TIM_BLOCK_TIMEOUT", "TIM_BLOQUEIO_TIMEOUT", default="30")))
except TimApiError as exc:
last = exc
if exc.status_code != 400 or index >= len(payloads) - 1:
@@ -193,7 +254,7 @@ class TimApiClient:
def cancelar_vas(self, msisdn: str, service: dict[str, Any], *, protocol: str = "") -> Any:
if self.mock:
return self.fixture("cancel_vas")
- url = os.getenv("TIM_CANCELAMENTO_URL", "")
+ url = self._env_first("TIM_CANCELLATION_URL", "TIM_CANCELAMENTO_URL")
payload = {
"channel": "AIAGENTCR",
"msisdn": msisdn,
@@ -203,17 +264,17 @@ class TimApiClient:
}
headers = self._headers(
client_id=None,
- auth=os.getenv("TIM_CANCELAMENTO_AUTH", ""),
+ auth=self._env_first("TIM_CANCELLATION_AUTH", "TIM_CANCELAMENTO_AUTH"),
extra={
"clientId": os.getenv("TIM_CANCELAMENTO_CLIENT_ID", os.getenv("TIM_DEFAULT_CLIENT_ID", "AIAGENTCR")),
"messageId": str(uuid.uuid4()),
- "AuthorizationOAM": os.getenv("TIM_CANCELAMENTO_AUTHORIZATION_OAM", ""),
- "Cn_field": os.getenv("TIM_CANCELAMENTO_CN_FIELD", ""),
- "Type_field": os.getenv("TIM_CANCELAMENTO_TYPE_FIELD", ""),
+ "AuthorizationOAM": self._env_first("TIM_CANCELLATION_AUTH_OAM", "TIM_CANCELAMENTO_AUTH_OAM", "TIM_CANCELAMENTO_AUTHORIZATION_OAM"),
+ "Cn_field": self._env_first("TIM_CANCELLATION_CN_FIELD", "TIM_CANCELAMENTO_CN_FIELD"),
+ "Type_field": self._env_first("TIM_CANCELLATION_TYPE_FIELD", "TIM_CANCELAMENTO_TYPE_FIELD"),
},
)
attempts: list[dict[str, Any]] = []
- result = self.request("DELETE", url, payload=payload, headers=headers, timeout=int(os.getenv("TIM_CANCELAMENTO_TIMEOUT", "30")), attempt_log=attempts)
+ result = self.request("DELETE", url, payload=payload, headers=headers, timeout=int(self._env_first("TIM_CANCELLATION_TIMEOUT", "TIM_CANCELAMENTO_TIMEOUT", default="30")), attempt_log=attempts)
return self._attach_transport(result, operation="cancela_vas", attempts=attempts)
def contrato(self, msisdn: str) -> Any:
@@ -223,7 +284,7 @@ class TimApiClient:
headers = self._headers(
client_id=None,
auth=os.getenv("TIM_CONTRATO_AUTH", ""),
- extra={"clientId": os.getenv("TIM_CONTRATO_CLIENT_ID", os.getenv("TIM_DEFAULT_CLIENT_ID", "CHAT"))},
+ extra={"clientId": self._env_first("TIM_CONTRATO_CLIENT_ID", default="AIAGENTCR")},
)
return self.request("GET", f"{base}/{quote(msisdn)}", headers=headers, timeout=int(os.getenv("TIM_CONTRATO_TIMEOUT", "30")))
@@ -231,13 +292,13 @@ class TimApiClient:
if self.mock:
# contract fixture carries representative customer identity in local mode.
return self.fixture("contrato")
- url = os.getenv("TIM_PROFILE_FULL_URL", "").replace("{msisdn}", quote(msisdn))
+ url = self._url_with_msisdn(self._env_first("TIM_PROFILE_FULL_URL"), msisdn)
headers = self._headers(
client_id=None,
auth=os.getenv("TIM_PROFILE_FULL_AUTH", ""),
extra={"ClientID": os.getenv("TIM_PROFILE_FULL_CLIENT_ID", "AIAGENTCR")},
)
- return self.request("GET", url, headers=headers)
+ return self.request("GET", url, headers=headers, timeout=int(self._env_first("TIM_PROFILE_FULL_TIMEOUT", default="30")))
def abrir_protocolo(self, payload: dict[str, Any]) -> Any:
"""Registra protocolo V2 preservando o contrato externo do Contas original.
@@ -317,7 +378,7 @@ class TimApiClient:
if self.mock:
return self.fixture("contestacao_tool")
data = dict(payload)
- data.setdefault("userId", "AIAGENTCR")
+ data.setdefault("userId", self._env_first("TIM_CUSTOMER_CONTESTATION_USER_ID", default="AIAGENTCR"))
data.setdefault("customerIdCurrent", data.get("customerId") or "")
data.setdefault("customerType", "2")
data.setdefault("customerStatus", "1")
@@ -445,17 +506,17 @@ class TimApiClient:
"""Perfil de faturamento usando o mesmo contrato CompleteInvoices do original."""
if self.mock:
return self.fixture("profile_bill")
- url = os.getenv("TIM_URL_PERFIL_FATURA", "") or os.getenv("TIM_COMPLETE_INVOICES_URL", "")
+ url = self._env_first("TIM_COMPLETE_INVOICES_URL", "TIM_URL_PERFIL_FATURA")
return self.request(
"POST",
url,
payload={"msisdn": msisdn},
headers=self._headers(
client_id=None,
- auth=os.getenv("TIM_PROFILE_BILL_AUTH", "") or os.getenv("TIM_COMPLETE_INVOICES_AUTH", ""),
- extra={"ClientID": os.getenv("TIM_PROFILE_BILL_CLIENT_ID", "AIAGENTCR")},
+ auth=self._env_first("TIM_COMPLETE_INVOICES_AUTH", "TIM_PROFILE_BILL_AUTH"),
+ extra={"ClientID": self._env_first("TIM_COMPLETE_INVOICES_CLIENT_ID", "TIM_PROFILE_BILL_CLIENT_ID", default="AIAGENTCR")},
),
- timeout=int(os.getenv("TIM_PROFILE_BILL_TIMEOUT", "30")),
+ timeout=int(self._env_first("TIM_COMPLETE_INVOICES_TIMEOUT", "TIM_PROFILE_BILL_TIMEOUT", default="30")),
)
def line_info(self, msisdn: str) -> Any:
@@ -463,7 +524,7 @@ class TimApiClient:
if self.mock:
payload = self.fixture("contrato")
else:
- url = os.getenv("TIM_PROFILE_FULL_URL", "").replace("{msisdn}", quote(msisdn))
+ url = self._url_with_msisdn(self._env_first("TIM_PROFILE_FULL_URL"), msisdn)
payload = self.request(
"GET",
url,
@@ -472,6 +533,7 @@ class TimApiClient:
auth=os.getenv("TIM_PROFILE_FULL_AUTH", ""),
extra={"ClientID": os.getenv("TIM_PROFILE_FULL_CLIENT_ID", "AIAGENTCR")},
),
+ timeout=int(self._env_first("TIM_PROFILE_FULL_TIMEOUT", default="30")),
)
def extract(value: Any) -> str:
if not isinstance(value, dict):
@@ -517,9 +579,9 @@ class TimApiClient:
}
headers = self._headers(
client_id=None,
- auth=os.getenv("TIM_BILL_PDF_AUTH", "") or os.getenv("TIM_SECURE_PDF_AUTH", ""),
+ auth=self._env_first("TIM_BILL_PDF_AUTH", "TIM_SECURE_PDF_AUTH", "TIM_INVOICE_RECOVER_AUTH"),
extra={
- "clientId": os.getenv("TIM_BILL_PDF_CLIENT_ID", "AIAGENTCR"),
+ "clientId": self._env_first("TIM_BILL_PDF_CLIENT_ID", "TIM_SECURE_PDF_CLIENT_ID", default="AIAGENTCR"),
"Accept": "application/pdf",
},
)
@@ -528,7 +590,7 @@ class TimApiClient:
url,
payload=payload,
headers=headers,
- timeout=int(os.getenv("TIM_BILL_PDF_TIMEOUT", os.getenv("TIM_INVOICE_RECOVER_TIMEOUT", "30"))),
+ timeout=int(self._env_first("TIM_BILL_PDF_TIMEOUT", "TIM_SECURE_PDF_TIMEOUT", "TIM_INVOICE_RECOVER_TIMEOUT", default="30")),
)
raw_content = result.get("raw_content") if isinstance(result, dict) else None
parsed = result.get("parsed_content") if isinstance(result, dict) else None
@@ -558,8 +620,8 @@ class TimApiClient:
params=params,
headers=self._headers(
client_id=None,
- auth=os.getenv("TIM_SECURE_PDF_AUTH", ""),
- extra={"clientId": os.getenv("TIM_INVOICE_RECOVER_CLIENT_ID", "AIAGENTCR")},
+ auth=self._env_first("TIM_SECURE_PDF_AUTH", "TIM_BILL_PDF_AUTH", "TIM_INVOICE_RECOVER_AUTH"),
+ extra={"clientid": self._env_first("TIM_INVOICE_RECOVER_CLIENT_ID", "TIM_SECURE_PDF_CLIENT_ID", default="AIAGENTCR")},
),
timeout=int(os.getenv("TIM_INVOICE_RECOVER_TIMEOUT", "30")),
)
diff --git a/app/domain/contas/invoice_resolver.py b/app/domain/contas/invoice_resolver.py
index 0f44e82..61848ef 100644
--- a/app/domain/contas/invoice_resolver.py
+++ b/app/domain/contas/invoice_resolver.py
@@ -296,7 +296,7 @@ class InvoiceResolver:
for mention in mentioned_items:
if not isinstance(mention, str) or not mention.strip():
continue
- resolved.extend(self._resolve_one_mention(mention, invoice_detail))
+ resolved.extend(self._resolve_one_mention(mention, invoice_detail, include_identity_oos=False))
return self._dedupe_exact_matches(resolved)
def resolve_each(
@@ -344,7 +344,7 @@ class InvoiceResolver:
ordered.append((idx, name))
lines[idx] = msisdn
dates[idx] = date
- matches = self._resolve_one_mention(name, invoice_detail)
+ matches = self._resolve_one_mention(name, invoice_detail, include_identity_oos=True)
if matches:
deterministic[idx] = matches
elif (
@@ -537,6 +537,8 @@ class InvoiceResolver:
self,
mention: str,
invoice_detail: dict[str, Any],
+ *,
+ include_identity_oos: bool = False,
) -> list[ResolvedInvoiceItem]:
"""Match determinístico de UMA menção contra todas as combinações
msisdn × seção × entry, com precedência de match exato sobre substring.
@@ -564,20 +566,33 @@ class InvoiceResolver:
casava por substring o prefixo ``"VOD + Canais Abertos"`` — item errado;
a normalização a torna exata ao item ``+Fechados``.)"""
mention_normalized = self._normalize_match_text(mention)
+
+ # Primeiro procure igualdade EXATA em todo o catálogo da fatura, inclusive
+ # seções não acionáveis (Plano, descontos etc.). Isso é um guard de identidade:
+ # se o cliente nomeou precisamente um item conhecido, esse item precisa vencer
+ # antes de qualquer fuzzy matching. Sem isso, um plano explicitamente citado
+ # pode ser removido do universo tratável e o matcher acabar autorizando outro
+ # VAS apenas por similaridade (ex.: TIM CTRL Redes Sociais -> TIM Fashion).
+ if include_identity_oos:
+ identity_candidates = list(self._iter_identity_candidates(invoice_detail))
+ exact_identity = [
+ cand
+ for cand in identity_candidates
+ if mention_normalized
+ and self._normalize_match_text(cand.desc) == mention_normalized
+ ]
+ if exact_identity:
+ return [self._build_item(cand) for cand in exact_identity]
+
+ # Fuzzy/substring continua restrito ao universo de serviços acionáveis +
+ # out-of-scope de seções de serviço. Seções explicitamente não acionáveis
+ # jamais entram no matcher aproximado; elas só podem bloquear por igualdade
+ # exata acima. Isso mantém o comportamento conservador do fluxo.
candidates = list(self._iter_candidates(invoice_detail))
- exact_cands: list[_Candidate] = []
substring_cands: list[_Candidate] = []
for cand in candidates:
- desc_normalized = self._normalize_match_text(cand.desc)
- if mention_normalized and desc_normalized == mention_normalized:
- exact_cands.append(cand)
- elif self._was_mentioned(cand.desc, [mention]):
+ if self._was_mentioned(cand.desc, [mention]):
substring_cands.append(cand)
- # Match exato (chave normalizada) tem precedência absoluta. ``canonical_name``
- # e o payload seguem sempre o ``desc`` CRU da fatura — a normalização é só
- # chave de comparação.
- if exact_cands:
- return [self._build_item(cand) for cand in exact_cands]
guarded = self._apply_prefix_guard(mention, substring_cands, candidates)
return [self._build_item(cand) for cand in guarded]
@@ -828,6 +843,45 @@ class InvoiceResolver:
# ----- construção de candidatos e itens --------------------------------
+ def _iter_identity_candidates(
+ self, invoice_detail: dict[str, Any]
+ ) -> Iterable[_Candidate]:
+ """Itera itens nomeáveis de TODAS as seções da fatura para match exato.
+
+ Diferente de :meth:`_iter_candidates`, este catálogo inclui também seções
+ não acionáveis como ``Plano``/``Planos``. Esses itens são construídos como
+ ``out_of_scope`` e servem somente para preservar a identidade explicitamente
+ citada pelo cliente. Eles nunca participam do fuzzy matcher.
+ """
+ for parent_msisdn, sections in self._iter_msisdn_buckets(invoice_detail):
+ if not isinstance(sections, dict):
+ continue
+ for section, entries in sections.items():
+ if not isinstance(entries, list):
+ continue
+ section_tool, section_type = SECTION_DEFAULTS.get(
+ section, (None, _OUT_OF_SCOPE_TYPE)
+ )
+ for entry in entries:
+ if not isinstance(entry, dict) or self._is_non_service_item(entry):
+ continue
+ desc = str(entry.get("desc") or "").strip()
+ if not desc:
+ continue
+ default_tool, default_type = section_tool, section_type
+ # Se a seção é explicitamente não acionável e a entry não foi
+ # carimbada como tratável pelo parser, force out_of_scope.
+ if section in _NON_SERVICE_SECTIONS and not self._has_treatable_flag(entry):
+ default_tool, default_type = None, _OUT_OF_SCOPE_TYPE
+ yield _Candidate(
+ desc=desc,
+ section=section,
+ default_tool=default_tool,
+ default_type=default_type,
+ parent_msisdn=parent_msisdn,
+ entry=entry,
+ )
+
def _iter_candidates(
self, invoice_detail: dict[str, Any]
) -> Iterable[_Candidate]:
diff --git a/docs/LEGACY_INTEGRATION_PARITY_20260822.md b/docs/LEGACY_INTEGRATION_PARITY_20260822.md
new file mode 100644
index 0000000..ab174e6
--- /dev/null
+++ b/docs/LEGACY_INTEGRATION_PARITY_20260822.md
@@ -0,0 +1,104 @@
+# Paridade de Integração Contas — Mock x Sistemas Reais
+
+Data: 2026-08-22
+
+## Objetivo
+
+Validar o agente Contas migrado contra o código original, usando o legado como fonte de verdade para contratos HTTP, autenticação, headers, payloads, URLs, timeouts e comportamento de integração. O objetivo é manter o funcionamento com mocks locais sem impedir a execução contra sistemas reais quando `TIM_GATEWAY_MODE`/`TIM_USE_MOCK_GATEWAY` forem configurados para modo real.
+
+## Correção crítica — identidade do item transacional
+
+Foi corrigido o caso em que um pedido explícito para `TIM CTRL Redes Sociais 8.0` podia ser reinterpretado pelo matcher fuzzy como `TIM Fashion Mensal`.
+
+### Causa
+
+O `InvoiceResolver` eliminava seções não transacionáveis (por exemplo, planos) antes da resolução de identidade e, em seguida, aplicava similaridade somente sobre VAS. Como `TIM Fashion Mensal` ultrapassava o threshold do matcher, o `subject` era substituído antes da execução.
+
+### Correção
+
+- A resolução exata de identidade agora acontece antes de qualquer fuzzy matching.
+- A busca exata considera também itens fora do escopo transacional, como planos.
+- Um plano encontrado exatamente é classificado como `out_of_scope` para a operação VAS.
+- O fuzzy matching continua restrito aos candidatos realmente tratáveis.
+- `resolve_items()` preserva o comportamento anterior onde necessário para compatibilidade; o fluxo operacional usa a proteção de identidade.
+
+Resultado esperado para o caso:
+
+`TIM CTRL Redes Sociais 8.0` -> exact match -> `plano` -> `out_of_scope` -> não substituir por outro VAS -> não executar cancelamento.
+
+## Comparação com o código original
+
+Foram comparados os comandos do projeto original (`agente_contas_tim/commands`), `factory.py`, `config.py` e o gateway HTTP com o adaptador atual `app/domain/contas/client.py`.
+
+| Serviço/Integração | Contrato encontrado no original | Situação no migrado após revisão |
+|---|---|---|
+| Consulta VAS | GET, URL com `{msisdn}` ou append `/msisdn`, normalização para prefixo 55, clientId | Corrigido: aliases originais, timeout, append e prefixo 55 |
+| Histórico VAS | GET com `?msisdn=`, clientId/messageId/auth | Corrigido default `clientId=AIAGENTCR` |
+| Bloqueio VAS | POST, contratos de payload `pmid`/`input`/`vasBlock`, headers extras | Corrigidos aliases de URL/auth/timeout/clientId/operation/payload/encoding |
+| Cancelamento VAS | DELETE, body com channel/msisdn/appId/cspId/interactionProtocol, OAM/CN/type opcionais | Corrigidos aliases `TIM_CANCELLATION_*` e `TIM_CANCELAMENTO_*` |
+| Divergência / explicação de fatura | GET `/?channel=AIAGENTCR`, Basic opcional user/password, clientID | Corrigidos aliases, Basic auth e timeout |
+| CompleteInvoices | POST `{"msisdn": ...}`, `ClientID=AIAGENTCR` | Compatível; timeout respeitado |
+| Profile bill | Factory original usa configuração de CompleteInvoices | Corrigido para priorizar contrato/config de CompleteInvoices |
+| Profile full | GET com placeholder ou append `/msisdn`, `ClientID=AIAGENTCR` | Corrigido append e timeout |
+| Line info | Mesmo padrão de URL do profile full | Corrigido append e timeout |
+| Contrato | GET `/`, clientId do legado | Corrigido default `AIAGENTCR` |
+| Protocolo V2 | POST serviceRequest/interaction, headers opcionais OAM/CN/type | Mantido no adaptador atual |
+| Contestação do cliente | POST, clientId/messageId/X-Agent-Id, user configurável | Corrigido default via `TIM_CUSTOMER_CONTESTATION_USER_ID` |
+| Atualização de Service Request | POST, channel/serviceRequest, headers de integração | Mantido |
+| Tracking Activities | POST com customer/protocol/invoice/activity/user | Mantido |
+| SMS | POST com msisdn/sender/message/URL e receipt opcional | Mantido |
+| Bill PDF detalhada | POST com invoiceId/customerId e invoiceType `DETALHADA`, retorno PDF | Aliases ampliados |
+| Secure PDF / invoice recover | GET com invoiceId/msisdn/customerId | Aliases/header ajustados |
+
+## Configurações presentes no legado sem uso operacional comprovado
+
+- `status_customer`: configuração encontrada, mas sem comando/runtime consumidor localizado na revisão.
+- configuração OAuth específica de SMS: declarada no config original, mas sem consumidor runtime localizado.
+
+Esses itens não foram tratados como requisito ativo sem evidência de uso no código original.
+
+## Mock x modo real
+
+O mock continua suportado. Em mock, o cliente retorna fixtures locais para as operações previstas. Em modo real, o mesmo adaptador segue os contratos HTTP reconstruídos a partir do código original.
+
+A principal diferença de risco é que um mock tende a responder `200/OK` para cenários preparados. Por isso, a validação de identidade deve ocorrer antes do gateway — como agora ocorre — para impedir que um erro de resolução de entidade seja mascarado pelo mock e, principalmente, que chegue a um backend real.
+
+## Testes executados
+
+### Regressão + contratos existentes
+
+- 101 testes passaram no conjunto de contratos, paridade, idempotência e resolução.
+
+### Novos testes de compatibilidade legado/real
+
+- 7 testes passaram cobrindo:
+ - prefixo 55 e composição da URL de consulta VAS;
+ - aliases originais e timeout;
+ - aliases de cancelamento;
+ - append de MSISDN em profile full;
+ - Basic auth de divergência via usuário/senha;
+ - client IDs de histórico VAS e contrato;
+ - uso de CompleteInvoices no profile bill.
+
+### Suite `tests/migration`
+
+Resultado observado após as mudanças:
+
+- 660 passed
+- 4 failed
+
+As quatro falhas remanescentes são de configuração/contexto de guardrails (`conversation_history` e FRASEOLOGIA) e não estão relacionadas ao `InvoiceResolver` nem aos contratos de integração revisados.
+
+## Limite desta validação
+
+A revisão comprova paridade de contrato em nível de código-fonte e testes locais. Ela não é uma certificação de conectividade real porque não foram usados endpoints, credenciais ou rede dos sistemas legados neste ambiente.
+
+Para homologação real, recomenda-se executar testes de contrato contra um ambiente não produtivo dos serviços TIM, verificando status HTTP, schemas reais, autenticação, timeouts, headers obrigatórios e respostas de erro.
+
+## Gaps de endurecimento recomendados
+
+1. Adicionar validação de readiness no startup quando `mock=false`, falhando cedo se endpoint/auth obrigatórios estiverem ausentes.
+2. Criar testes de contrato contra ambiente de homologação para cada integração ativa.
+3. Comparar periodicamente fixtures mock com schemas/respostas reais para evitar drift.
+4. Manter invariantes transacionais: item solicitado, item resolvido e item executado nunca podem divergir silenciosamente.
+5. Evoluir mascaramento/observabilidade do cliente migrado para o mesmo nível do `HttpGateway` original, sem registrar secrets.
diff --git a/tests/migration/test_mcp_contestation_subject_guard.py b/tests/migration/test_mcp_contestation_subject_guard.py
index 0cf4fc4..dffec5e 100644
--- a/tests/migration/test_mcp_contestation_subject_guard.py
+++ b/tests/migration/test_mcp_contestation_subject_guard.py
@@ -70,3 +70,25 @@ def test_correct_subject_is_preserved():
assert result is not None
assert result["subject"] == "TIM CTRL Redes Sociais 8.0"
assert args.get("_subject_corrected_from") is None
+
+
+def test_cancel_preflight_exact_plan_is_out_of_scope_and_never_rebound_to_vas():
+ args = _args(subject="TIM CTRL Redes Sociais 8.0")
+ result = _preflight_subject("cancelar_vas_avulso", args)
+ assert result is not None
+ assert result["status"] == "OUT_OF_SCOPE"
+ assert args["subject"] == "TIM CTRL Redes Sociais 8.0"
+ assert "TIM Fashion" not in str(result)
+
+
+def test_invoice_resolver_exact_plan_identity_wins_before_similarity_matcher():
+ from app.domain.contas.invoice_resolver import InvoiceResolver
+ from app.domain.contas.item_matcher import SimilarityItemMatcher
+
+ detail = _args(subject="TIM CTRL Redes Sociais 8.0")["billing_analysis"]
+ outcome = InvoiceResolver(matcher=SimilarityItemMatcher()).resolve(
+ ["TIM CTRL Redes Sociais 8.0"], detail
+ )
+ assert outcome.resolved == []
+ assert len(outcome.out_of_scope) == 1
+ assert outcome.out_of_scope[0].matches[0].canonical_name == "TIM CTRL Redes Sociais 8.0"
diff --git a/tests/migration/test_real_legacy_integration_compat.py b/tests/migration/test_real_legacy_integration_compat.py
new file mode 100644
index 0000000..96d0fbc
--- /dev/null
+++ b/tests/migration/test_real_legacy_integration_compat.py
@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import base64
+
+from app.domain.contas.client import TimApiClient
+
+
+def _real(monkeypatch) -> TimApiClient:
+ monkeypatch.setenv("TIM_USE_MOCK_GATEWAY", "false")
+ monkeypatch.setenv("TIM_GATEWAY_MODE", "real")
+ return TimApiClient()
+
+
+def test_query_vas_legacy_base_url_appends_country_code_and_msisdn(monkeypatch):
+ c = _real(monkeypatch)
+ monkeypatch.setenv("TIM_URL_CONSULTA_VAS", "http://tim/access/v1")
+ calls = []
+ monkeypatch.setattr(c, "request", lambda *a, **kw: calls.append((a, kw)) or {})
+ c.consultar_vas("11999999999")
+ assert calls[0][0] == ("GET", "http://tim/access/v1/5511999999999")
+
+
+def test_query_vas_accepts_original_aliases_and_timeout(monkeypatch):
+ c = _real(monkeypatch)
+ monkeypatch.delenv("TIM_URL_CONSULTA_VAS", raising=False)
+ monkeypatch.setenv("TIM_CONSULTA_URL", "http://tim/query")
+ monkeypatch.setenv("TIM_QUERY_AUTH", "Basic old")
+ monkeypatch.setenv("TIM_QUERY_TIMEOUT", "44")
+ calls = []
+ monkeypatch.setattr(c, "request", lambda *a, **kw: calls.append((a, kw)) or {})
+ c.consultar_vas("5511999999999")
+ _, kw = calls[0]
+ assert kw["headers"]["Authorization"] == "Basic old"
+ assert kw["timeout"] == 44
+
+
+def test_cancelamento_accepts_original_cancellation_aliases(monkeypatch):
+ c = _real(monkeypatch)
+ monkeypatch.delenv("TIM_CANCELAMENTO_URL", raising=False)
+ monkeypatch.setenv("TIM_CANCELLATION_URL", "http://tim/cancel")
+ monkeypatch.setenv("TIM_CANCELLATION_AUTH", "Basic legacy")
+ monkeypatch.setenv("TIM_CANCELLATION_AUTH_OAM", "OAM")
+ monkeypatch.setenv("TIM_CANCELLATION_CN_FIELD", "CN")
+ monkeypatch.setenv("TIM_CANCELLATION_TYPE_FIELD", "TYPE")
+ calls = []
+ monkeypatch.setattr(c, "request", lambda *a, **kw: calls.append((a, kw)) or {})
+ c.cancelar_vas("119", {"appId": "1", "cspId": "740"})
+ args, kw = calls[0]
+ assert args[:2] == ("DELETE", "http://tim/cancel")
+ assert kw["headers"]["Authorization"] == "Basic legacy"
+ assert kw["headers"]["AuthorizationOAM"] == "OAM"
+ assert kw["headers"]["Cn_field"] == "CN"
+ assert kw["headers"]["Type_field"] == "TYPE"
+
+
+def test_profile_full_appends_msisdn_when_original_url_has_no_placeholder(monkeypatch):
+ c = _real(monkeypatch)
+ monkeypatch.setenv("TIM_PROFILE_FULL_URL", "http://tim/profile")
+ calls = []
+ monkeypatch.setattr(c, "request", lambda *a, **kw: calls.append((a, kw)) or {})
+ c.profile_full("11999999999")
+ assert calls[0][0] == ("GET", "http://tim/profile/11999999999")
+
+
+def test_divergencia_preserves_original_user_password_basic_auth(monkeypatch):
+ c = _real(monkeypatch)
+ monkeypatch.setenv("TIM_DIVERGENCIA_URL", "http://tim/div")
+ monkeypatch.setenv("TIM_DIVERGENCIA_USER", "user")
+ monkeypatch.setenv("TIM_DIVERGENCIA_PASSWORD", "pass")
+ calls = []
+ monkeypatch.setattr(c, "request", lambda *a, **kw: calls.append((a, kw)) or {})
+ c.billing_analysis("119")
+ auth = calls[0][1]["headers"]["Authorization"]
+ assert auth == "Basic " + base64.b64encode(b"user:pass").decode("ascii")
+
+
+def test_vas_history_and_contrato_default_client_id_match_original(monkeypatch):
+ c = _real(monkeypatch)
+ monkeypatch.setenv("TIM_VAS_HISTORY_URL", "http://tim/history")
+ monkeypatch.setenv("TIM_CONTRATO_URL", "http://tim/contract")
+ monkeypatch.delenv("TIM_VAS_HISTORY_CLIENT_ID", raising=False)
+ monkeypatch.delenv("TIM_CONTRATO_CLIENT_ID", raising=False)
+ calls = []
+ monkeypatch.setattr(c, "request", lambda *a, **kw: calls.append((a, kw)) or {})
+ c.historico_vas("119")
+ c.contrato("119")
+ assert calls[0][1]["headers"]["clientId"] == "AIAGENTCR"
+ assert calls[1][1]["headers"]["clientId"] == "AIAGENTCR"
+
+
+def test_profile_bill_uses_complete_invoices_contract_like_original_factory(monkeypatch):
+ c = _real(monkeypatch)
+ monkeypatch.setenv("TIM_COMPLETE_INVOICES_URL", "http://tim/complete")
+ monkeypatch.setenv("TIM_URL_PERFIL_FATURA", "http://tim/legacy-profile")
+ monkeypatch.setenv("TIM_COMPLETE_INVOICES_CLIENT_ID", "CID")
+ calls = []
+ monkeypatch.setattr(c, "request", lambda *a, **kw: calls.append((a, kw)) or {})
+ c.profile_bill("119")
+ args, kw = calls[0]
+ assert args[:2] == ("POST", "http://tim/complete")
+ assert kw["headers"]["ClientID"] == "CID"