new feature: Multi-Intent

contas_mcp/servers/contas_mcp_server/main.py

@@ -1356,6 +1356,74 @@ def _resolve_multiple_vas_subjects(subject: str, catalog: list[dict[str, Any]],
     return selected
 
 
+def _select_relative_invoice_charges(subject: str, args: dict[str, Any]) -> list[dict[str, Any]]:
+    """Resolve bounded quantity/superlative references over invoice evidence.
+
+    This is deliberately deterministic: the LLM may preserve the customer's
+    phrase, but ordering and cardinality are calculated from authoritative rows.
+    When no focused name is supplied, selection is allowed only if exactly one
+    repeated service exists, preventing a global "most expensive" guess.
+    """
+    raw = str(subject or "").strip()
+    normalized = _norm_entity_reference(raw)
+    quantity_match = re.search(r"\b(?:as|os)?\s*(duas|dois|2)\s+mais\s+car[oa]s?\b", normalized)
+    single_match = re.search(r"\b(?:a|o)?\s*mais\s+car[oa]\b", normalized)
+    if not quantity_match and not single_match:
+        return []
+    limit = 2 if quantity_match else 1
+    detail = args.get("invoice_detail")
+    if not isinstance(detail, dict):
+        return []
+    parsed = detail.get("parsed_content") if isinstance(detail.get("parsed_content"), dict) else detail
+    if not isinstance(parsed, dict):
+        return []
+    focus = str(args.get("focused_subject") or args.get("context_subject") or "").strip()
+    focus_norm = _norm_entity_reference(focus)
+    rows: list[dict[str, Any]] = []
+    roots = list(parsed.items()) if parsed and all(isinstance(v, dict) for v in parsed.values()) else [(str(args.get("msisdn") or ""), parsed)]
+    for line, sections in roots:
+        if not isinstance(sections, dict):
+            continue
+        for section, values in sections.items():
+            if not isinstance(values, list):
+                continue
+            for index, row in enumerate(values):
+                if not isinstance(row, dict):
+                    continue
+                name = str(row.get("desc") or row.get("name") or row.get("description") or "").strip()
+                if not name:
+                    continue
+                value = row.get("value") if row.get("value") is not None else row.get("valor")
+                try:
+                    amount = float(str(value).replace(".", "").replace(",", ".")) if "," in str(value) else float(value)
+                except (TypeError, ValueError):
+                    continue
+                rows.append({
+                    "name": name, "msisdn": str(row.get("msisdn") or line or ""),
+                    "value": amount, "charge_date": row.get("date") or row.get("charge_date"),
+                    "section": str(section), "_order": len(rows),
+                })
+    if focus_norm:
+        rows = [row for row in rows if focus_norm in _norm_entity_reference(row["name"]) or _norm_entity_reference(row["name"]) in focus_norm]
+    else:
+        groups: dict[str, list[dict[str, Any]]] = {}
+        for row in rows:
+            groups.setdefault(_norm_entity_reference(row["name"]), []).append(row)
+        repeated = [group for group in groups.values() if len(group) >= limit]
+        if len(repeated) != 1:
+            return []
+        rows = repeated[0]
+    if len(rows) < limit:
+        return []
+    selected = sorted(rows, key=lambda row: (-row["value"], row["_order"]))[:limit]
+    for row in selected:
+        row.pop("_order", None)
+        resolved_class, resolved_type = _vas_domain_policy_from_invoice_detail(row["name"], args)
+        row["tool_category"] = resolved_class or "cancelar_vas_avulso"
+        row["item_type"] = resolved_type or "avulso"
+    return selected
+
+
 async def _validate_vas_subject(args: dict[str, Any]) -> dict[str, Any]:
     """Side-effect-free entity resolution used before confirmation/execution."""
     await _enrich_invoice_context("validar_vas_subject", args)
@@ -1363,6 +1431,27 @@ async def _validate_vas_subject(args: dict[str, Any]) -> dict[str, Any]:
     catalog = _vas_entity_catalog(args)
     canonical, matches = _resolve_catalog_entity(subject, catalog)
     requested_tool = str(args.get("target_tool") or "").strip()
+    relative = _select_relative_invoice_charges(subject, args) if requested_tool == "cancelar_vas_avulso" else []
+    if relative:
+        names = [str(item["name"]) for item in relative]
+        summary = ", ".join(f"{item['name']} de R$ {_money_ptbr(item['value'])}" for item in relative)
+        return {
+            "eligible": True,
+            "status": "ELIGIBLE",
+            "subject": subject,
+            "resolved_subject": names[0] if len(set(names)) == 1 else ", ".join(names),
+            "resolved_subjects": names,
+            "entity_resolution": "relative_invoice_charge_selection",
+            "transaction_decision": {
+                "resolved_arguments": {"subject": names[0], "items": relative},
+                "target_tool": requested_tool,
+                "action_changed": False,
+                "requires_reconfirmation": True,
+                "confirmation_message": f"Você confirma o tratamento destas cobranças: {summary}? Responda 'sim' ou 'não'.",
+                "domain_policy": {"class": "cancelar_vas_avulso", "item_type": "multiple_charges"},
+            },
+            "metadata": {"side_effect_free": True, "selection": "highest", "limit": len(relative)},
+        }
     multi = _resolve_multiple_vas_subjects(subject, catalog, args) if requested_tool == "cancelar_vas_avulso" else []
     if len(multi) >= 2 or (multi and any(token in _norm_entity_reference(subject) for token in ("todos os vas avulsos", "todos os servicos avulsos"))):
         names = [str(item.get("name") or "").strip() for item in multi if item.get("name")]