safety: confirm gate for write actions

app/confirm.py

@@ -0,0 +1,53 @@
+"""Detect an explicit affirmative confirmation before irreversible order actions.
+
+Deliberately conservative: a false negative just makes the bot ask again (safe),
+while a false positive could fire a non-reversible action (cancel/return/address)
+the customer never approved. So we only accept clear yes/confirm words, and we
+tokenize on word boundaries — "incorrecto" must NEVER match "correcto".
+"""
+
+from __future__ import annotations
+
+import re
+import unicodedata
+
+# Single-word affirmatives (accent-stripped, lowercase), matched as whole tokens.
+_AFFIRM_WORDS = {
+    # español
+    "si", "sip", "sii", "siii", "siiii", "claro", "vale", "ok", "oka", "okay",
+    "okey", "correcto", "confirmo", "confirmado", "confirmar", "adelante",
+    "hazlo", "hazla", "procede", "proceder", "dale", "perfecto", "afirmativo",
+    "exacto", "venga",
+    # português
+    "sim", "certo", "isso", "avanca", "avancar", "podes", "pode",
+    # english
+    "yes", "yeah", "yep", "yup", "confirm", "confirmed", "sure", "proceed",
+    "correct",
+}
+
+# Multi-word affirmatives, matched as substrings on the normalized text.
+_AFFIRM_PHRASES = (
+    "de acuerdo", "eso es", "esta bien", "estoy de acuerdo", "go ahead",
+    "do it", "hazlo ya", "por favor hazlo", "que si", "adelante con",
+)
+
+
+def _normalize(text: str) -> str:
+    text = unicodedata.normalize("NFKD", text or "").encode("ascii", "ignore").decode()
+    return text.lower().strip()
+
+
+def is_affirmative(text: str) -> bool:
+    """True only when the message is a clear yes/confirm.
+
+    Examples:
+        "sí" / "vale, hazlo" / "confirmo" -> True
+        "incorrecto" / "no" / "quiero devolver" / "" -> False
+    """
+    norm = _normalize(text)
+    if not norm:
+        return False
+    if any(phrase in norm for phrase in _AFFIRM_PHRASES):
+        return True
+    tokens = set(re.findall(r"[a-z]+", norm))
+    return bool(tokens & _AFFIRM_WORDS)

app/orchestrator.py

@@ -49,6 +49,7 @@ async def run_turn(
     brand_name: str = "Asistente",
 ) -> ChatResponse:
     prior = await _load_history(ctx)
+    ctx.user_message = user_message  # for the deterministic write-action confirm gate
 
     from app.lang import detect_language
 

app/prompts.py

@@ -24,7 +24,7 @@ VENDER: recomienda por necesidad/presupuesto buscando con search_products y expl
 
 PEDIDOS (privacidad): para localizar un pedido pide email + UN segundo dato (nº de pedido, código postal o nombre). Pásalos a lookup_order. Revela SOLO estado de envío y seguimiento (transportista, nº, enlace, fecha); NUNCA dirección completa ni pago. Si no localiza, dilo de forma genérica (sin decir qué campo falla).
 
-ACCIONES sobre el pedido (reorder/cancel_order/change_shipping_address/start_return): solo sobre un pedido YA verificado con lookup_order; CONFIRMA con el cliente antes de ejecutar. País en código ISO (ES, PT). Si devuelve `not_allowed` o `error`, discúlpate y deriva con escalate_to_human. Los REEMBOLSOS no los haces tú: inicia la devolución y deriva al equipo.
+ACCIONES sobre el pedido (cancel_order/change_shipping_address/start_return): solo sobre un pedido YA verificado con lookup_order. CONFIRMACIÓN OBLIGATORIA: antes de ejecutar, pregunta y ESPERA un "sí" CLARO del cliente. Un motivo de devolución, un "no", un "incorrecto" o cualquier respuesta ambigua NO es confirmación: vuelve a preguntar, no ejecutes. Nunca metas el motivo como si fuera el "sí". País en código ISO (ES, PT). Si una herramienta devuelve `needs_confirmation`, pide el "sí" y no la repitas hasta tenerlo. Si devuelve `not_allowed` o `error`, discúlpate y deriva con escalate_to_human. Los REEMBOLSOS no los haces tú: inicia la devolución y deriva al equipo.
 
 ESCALADO: si no puedes resolver, o si el cliente se frustra/enfada, discúlpate y ofrece pasar con una persona (pide email, usa escalate_to_human).
 """

app/tools/actions_tool.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 from typing import TYPE_CHECKING, Any
 
 from app import analytics
+from app.confirm import is_affirmative
 from app.shopify import actions
 
 if TYPE_CHECKING:
@@ -25,6 +26,23 @@ def _need_verify() -> dict[str, Any]:
     return {"status": "need_verification", "hint": "verifica el pedido con lookup_order primero"}
 
 
+def _need_confirm(ctx: ToolContext) -> dict[str, Any] | None:
+    """Deterministic safety gate for irreversible actions: only proceed if the
+    customer's latest message is a clear affirmative. Returns a block result
+    otherwise so the model has to ask for an explicit 'sí' first. This does NOT
+    depend on the prompt (which can drift across providers)."""
+    if is_affirmative(ctx.user_message):
+        return None
+    return {
+        "status": "needs_confirmation",
+        "hint": (
+            "El cliente AÚN no ha confirmado. Pídele que confirme de forma "
+            "EXPLÍCITA con un 'sí' antes de ejecutar. Un motivo o un 'no' NO "
+            "es confirmación. No vuelvas a llamar a esta herramienta hasta el 'sí'."
+        ),
+    }
+
+
 async def reorder(args: dict[str, Any], ctx: ToolContext) -> dict[str, Any]:
     if ctx.shopify is None or not ctx.shop:
         return {"status": "unavailable"}
@@ -49,6 +67,8 @@ async def cancel_order(args: dict[str, Any], ctx: ToolContext) -> dict[str, Any]
     gid = _verified_order(ctx)
     if not gid:
         return _need_verify()
+    if (block := _need_confirm(ctx)) is not None:
+        return block
     result = await actions.cancel_order(ctx.shopify, gid, reason=args.get("reason"))
     if result.get("ok"):
         analytics.record(ctx.db, ctx.tenant_id, "order_cancelled", session_id=ctx.session.id)
@@ -64,6 +84,8 @@ async def change_address(args: dict[str, Any], ctx: ToolContext) -> dict[str, An
     gid = _verified_order(ctx)
     if not gid:
         return _need_verify()
+    if (block := _need_confirm(ctx)) is not None:
+        return block
     address = {
         k: v
         for k, v in {
@@ -96,6 +118,8 @@ async def start_return(args: dict[str, Any], ctx: ToolContext) -> dict[str, Any]
     gid = _verified_order(ctx)
     if not gid:
         return _need_verify()
+    if (block := _need_confirm(ctx)) is not None:
+        return block
     result = await actions.start_return(ctx.shopify, gid, reason=args.get("reason"))
     if result.get("ok"):
         analytics.record(ctx.db, ctx.tenant_id, "return_started", session_id=ctx.session.id)

app/tools/registry.py

@@ -36,6 +36,7 @@ class ToolContext:
     escalation_sender: Callable[[str, str, str], Awaitable[bool]] | None = None
     now: datetime = field(default_factory=lambda: datetime.now(UTC))
     cards: list[dict[str, Any]] = field(default_factory=list)  # product cards for the widget
+    user_message: str = ""  # message that triggered this turn (for the confirm gate)
 
 
 SPECS: list[ToolSpec] = [

tests/test_actions.py

@@ -39,7 +39,7 @@ class FakeShopify:
         return {}
 
 
-async def _verified_ctx(db, **kw):
+async def _verified_ctx(db, *, user_message="sí, confirmo", **kw):
     t = Tenant(slug="t", **kw)
     db.add(t)
     await db.flush()
@@ -51,6 +51,7 @@ async def _verified_ctx(db, **kw):
         shopify=FakeShopify(), allow_cancel=kw.get("allow_cancel", False),
         allow_address_change=kw.get("allow_address_change", False),
         allow_returns=kw.get("allow_returns", False),
+        user_message=user_message,  # customer's latest message (the confirm gate reads it)
     )
 
 
@@ -97,3 +98,33 @@ async def test_start_return_succeeds_when_allowed(db_session):
     ctx = await _verified_ctx(db_session, allow_returns=True)
     out = await actions_tool.start_return({"reason": "llegó defectuoso"}, ctx)
     assert out["status"] == "return_requested"
+
+
+# --- confirmation gate: irreversible actions need an explicit "sí" -----------
+
+async def test_return_blocked_without_confirmation(db_session):
+    # The real incident: user typed "INCORRECTO" (meaning "no"); it must NOT fire.
+    ctx = await _verified_ctx(db_session, allow_returns=True, user_message="INCORRECTO")
+    out = await actions_tool.start_return({"reason": "INCORRECTO"}, ctx)
+    assert out["status"] == "needs_confirmation"
+    assert not ctx.shopify.calls  # never reached Shopify
+
+
+async def test_cancel_blocked_without_confirmation(db_session):
+    ctx = await _verified_ctx(db_session, allow_cancel=True, user_message="quiero cancelar")
+    out = await actions_tool.cancel_order({}, ctx)
+    assert out["status"] == "needs_confirmation"
+    assert not ctx.shopify.calls
+
+
+async def test_change_address_blocked_without_confirmation(db_session):
+    ctx = await _verified_ctx(db_session, allow_address_change=True, user_message="no")
+    out = await actions_tool.change_address({"address1": "C/1", "city": "Madrid", "zip": "28013"}, ctx)
+    assert out["status"] == "needs_confirmation"
+    assert not ctx.shopify.calls
+
+
+async def test_cancel_proceeds_after_yes(db_session):
+    ctx = await _verified_ctx(db_session, allow_cancel=True, user_message="sí, cancélalo")
+    out = await actions_tool.cancel_order({}, ctx)
+    assert out["status"] == "cancelled"

tests/test_confirm.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import pytest
+
+from app.confirm import is_affirmative
+
+
+@pytest.mark.parametrize(
+    "text",
+    [
+        "sí",
+        "si",
+        "Sí, confirmo",
+        "vale, hazlo",
+        "ok",
+        "okey",
+        "dale",
+        "adelante",
+        "perfecto, procede",
+        "correcto",
+        "de acuerdo",
+        "sim",  # portugués
+        "yes",  # inglés
+        "sí, cancélalo por favor",
+    ],
+)
+def test_affirmatives(text):
+    assert is_affirmative(text) is True
+
+
+@pytest.mark.parametrize(
+    "text",
+    [
+        "INCORRECTO",  # the real incident — must NOT count as yes
+        "incorrecto",
+        "no",
+        "no, mejor no",
+        "para",
+        "quiero devolver",
+        "el producto llegó roto",
+        "talla incorrecta",
+        "",
+        "   ",
+        "mmm no sé",
+    ],
+)
+def test_non_affirmatives(text):
+    assert is_affirmative(text) is False