Feature 3: abandoned-cart recovery (opt-in)

app/carts.py

@@ -0,0 +1,218 @@
+"""Abandoned-cart recovery.
+
+When the bot builds a Shopify cart permalink in chat but the visitor leaves without
+checking out, we capture the cart (one open row per chat session) and — if we have an
+email — a background job sends ONE friendly recovery email with the link, skipping
+anyone who already completed an order. Strictly per-tenant: every cart row is scoped to
+one tenant_id, so no store ever nudges another's shoppers.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app import mailer
+from app.models import AbandonedCart, Tenant
+
+log = logging.getLogger(__name__)
+
+# A cart is "abandoned" once it has sat this long without checkout; we don't chase
+# anything older than the max age (too stale to feel relevant / risks annoyance).
+RECOVERY_DELAY = timedelta(hours=4)
+RECOVERY_MAX_AGE = timedelta(hours=48)
+
+
+async def _open_cart_for_session(
+    db: AsyncSession, tenant_id: int, session_id: str
+) -> AbandonedCart | None:
+    """The session's current cart that hasn't been recovered yet (one per session)."""
+    return (
+        await db.execute(
+            select(AbandonedCart)
+            .where(
+                AbandonedCart.tenant_id == tenant_id,
+                AbandonedCart.session_id == session_id,
+                AbandonedCart.recovered.is_(False),
+            )
+            .order_by(AbandonedCart.id.desc())
+        )
+    ).scalars().first()
+
+
+async def capture(
+    db: AsyncSession,
+    tenant_id: int,
+    *,
+    session_id: str,
+    checkout_url: str,
+    items_text: str = "",
+    lang: str | None = None,
+    email: str | None = None,
+    name: str | None = None,
+) -> AbandonedCart:
+    """Record (or refresh) the open cart the bot just built for this session. A new
+    cart link in the same session updates the existing row instead of duplicating, so
+    we only ever nudge the LATEST cart. Email/name are filled in when newly known and
+    never overwritten by a blank."""
+    em = (email or "").strip().lower() or None
+    nm = (name or "").strip()
+    cart = await _open_cart_for_session(db, tenant_id, session_id)
+    if cart is None:
+        cart = AbandonedCart(tenant_id=tenant_id, session_id=session_id)
+        db.add(cart)
+    cart.checkout_url = checkout_url or cart.checkout_url
+    if items_text:
+        cart.items_text = items_text[:500]
+    if lang:
+        cart.lang = lang
+    # A fresh cart link means the shopper is active again: reset the nudge so the
+    # background job re-evaluates it (don't keep a stale "already emailed" flag).
+    cart.nudged_at = None
+    if em and not cart.email:
+        cart.email = em
+    if nm and not cart.name:
+        cart.name = nm
+    await db.flush()
+    return cart
+
+
+async def attach_contact(
+    db: AsyncSession,
+    tenant_id: int,
+    *,
+    session_id: str,
+    email: str,
+    name: str | None = None,
+) -> AbandonedCart | None:
+    """Attach the visitor's email (so we can email a reminder) to this session's open
+    cart. Returns the cart, or None if the bot hasn't built a cart in this session
+    yet (nothing to save)."""
+    em = (email or "").strip().lower()
+    if not em:
+        return None
+    cart = await _open_cart_for_session(db, tenant_id, session_id)
+    if cart is None:
+        return None
+    cart.email = em
+    nm = (name or "").strip()
+    if nm and not cart.name:
+        cart.name = nm
+    await db.flush()
+    return cart
+
+
+def _recovery_email(cart: AbandonedCart, brand: str) -> tuple[str, str]:
+    """(subject, body) for the recovery nudge, in the shopper's language (ES default,
+    EN when the chat was in English). Honest + light — a saved cart, not a hard sell."""
+    who = (cart.name or "").strip().split(" ")[0] if cart.name else ""
+    en = (cart.lang or "").lower().startswith("en")
+    if en:
+        hi = f"Hi {who}," if who else "Hi,"
+        subject = f"Your cart is saved at {brand} 🛒"
+        body = (
+            f"{hi}\n\n"
+            f"We saved the cart you were putting together at {brand}. You can pick up "
+            f"right where you left off here:\n\n{cart.checkout_url}\n\n"
+            f"If you have any questions before checking out, just reply to this email "
+            f"and we'll help.\n\n— {brand}"
+        )
+        return subject, body
+    hi = f"¡Hola {who}!" if who else "¡Hola!"
+    subject = f"Te he guardado el carrito en {brand} 🛒"
+    body = (
+        f"{hi}\n\n"
+        f"Te hemos guardado el carrito que estabas preparando en {brand}. Puedes "
+        f"retomarlo justo donde lo dejaste aquí:\n\n{cart.checkout_url}\n\n"
+        f"Si tienes cualquier duda antes de finalizar, responde a este correo y te "
+        f"ayudamos.\n\nUn saludo,\n{brand}"
+    )
+    return subject, body
+
+
+def _aware(dt: datetime) -> datetime:
+    """Treat a naive datetime as UTC. SQLite round-trips TIMESTAMPTZ as naive, so we
+    normalize before comparing to avoid 'can't compare naive and aware' TypeErrors."""
+    return dt if dt.tzinfo is not None else dt.replace(tzinfo=UTC)
+
+
+def _completed_after(orders: list[dict[str, Any]], since: datetime) -> bool:
+    """True if any of the shopper's orders was placed after the cart was built —
+    i.e. they already bought, so we must NOT chase the sale."""
+    floor = _aware(since)
+    for o in orders:
+        raw = o.get("created_at")
+        if not raw:
+            continue
+        try:
+            when = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
+        except ValueError:
+            continue
+        if _aware(when) >= floor:
+            return True
+    return False
+
+
+async def recover_due_carts(db: AsyncSession, settings: Any, *, now: datetime) -> int:
+    """Send ONE recovery email for every due cart of an opted-in, active tenant —
+    skipping (and marking recovered) anyone who already placed an order after building
+    the cart. Returns how many emails were sent. Best-effort per cart: one failure
+    never blocks the rest. Caller commits."""
+    from app.shopify.orders import find_orders_by_email
+    from app.tenancy import tenant_shopify_client
+
+    tenants = (
+        await db.execute(
+            select(Tenant).where(
+                Tenant.allow_cart_recovery.is_(True),
+                Tenant.chatbot_enabled.is_(True),
+                Tenant.is_demo.is_(False),
+            )
+        )
+    ).scalars().all()
+    sent = 0
+    for tenant in tenants:
+        due = (
+            await db.execute(
+                select(AbandonedCart).where(
+                    AbandonedCart.tenant_id == tenant.id,
+                    AbandonedCart.email.is_not(None),
+                    AbandonedCart.nudged_at.is_(None),
+                    AbandonedCart.recovered.is_(False),
+                    AbandonedCart.checkout_url != "",
+                    AbandonedCart.created_at <= now - RECOVERY_DELAY,
+                    AbandonedCart.created_at >= now - RECOVERY_MAX_AGE,
+                )
+            )
+        ).scalars().all()
+        if not due:
+            continue
+        client = tenant_shopify_client(tenant)
+        brand = tenant.brand_name or tenant.name or "la tienda"
+        for cart in due:
+            try:
+                if client is not None:
+                    orders = await find_orders_by_email(client, cart.email or "")
+                    if _completed_after(orders, cart.created_at):
+                        cart.recovered = True  # they bought — don't chase it
+                        continue
+                ok = await mailer.send_email(
+                    settings,
+                    cart.email,
+                    *_recovery_email(cart, brand),
+                    reply_to=tenant.support_email or None,
+                    sender_name=brand,
+                )
+                # Stamp it nudged either way: a transient send failure must not cause
+                # the same shopper to be emailed every tick.
+                cart.nudged_at = now
+                if ok:
+                    sent += 1
+            except Exception:  # noqa: BLE001 - one bad cart never blocks the rest
+                log.warning("cart recovery failed tenant=%s cart=%s", tenant.id, cart.id,
+                            exc_info=True)
+    return sent

app/models.py

@@ -86,6 +86,12 @@ class Tenant(Base):
     # Web search — opt-in per store: lets the bot find/cite REAL external links
     # (e.g. a marketplace listing) when the answer isn't in the store knowledge.
     allow_web_search: Mapped[bool] = mapped_column(default=False)
+    # Abandoned-cart recovery — opt-in per store: when a visitor builds a cart but
+    # leaves without buying, the bot offers to save it + email a reminder, and a
+    # background job sends ONE recovery email (skipping anyone who already bought).
+    allow_cart_recovery: Mapped[bool] = mapped_column(
+        default=False, server_default=false()
+    )
     # Master switch: admin can turn a client's chatbot off (e.g. non-payment)
     # without deleting the account. The widget hides and the chat stops replying.
     chatbot_enabled: Mapped[bool] = mapped_column(default=True)
@@ -287,6 +293,40 @@ class Customer(Base):
     )
 
 
+class AbandonedCart(Base):
+    """A cart the bot helped build in chat that wasn't (yet) checked out — captured
+    so we can email a recovery nudge. Strictly tenant-scoped. One open cart per chat
+    session (the latest one the bot built); the email is attached when the visitor
+    gives it (or is already known for a logged-in customer)."""
+
+    __tablename__ = "abandoned_carts"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    tenant_id: Mapped[int | None] = mapped_column(
+        ForeignKey("tenants.id", ondelete="CASCADE"), nullable=True, index=True
+    )
+    # The chat session this cart belongs to (one open cart per session). Cleared if
+    # the session is purged, but the cart row survives so a pending nudge still fires.
+    session_id: Mapped[str | None] = mapped_column(
+        ForeignKey("chat_sessions.id", ondelete="SET NULL"), nullable=True, index=True
+    )
+    email: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
+    name: Mapped[str] = mapped_column(String(160), default="")
+    lang: Mapped[str | None] = mapped_column(String(10), nullable=True)
+    checkout_url: Mapped[str] = mapped_column(Text, default="")
+    items_text: Mapped[str] = mapped_column(String(500), default="")
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=_utcnow, onupdate=_utcnow
+    )
+    # When the single recovery email was sent (None = not yet). Set once so a cart is
+    # never emailed twice.
+    nudged_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    # True once we detect the customer completed an order after building the cart, so
+    # we never chase a sale that already closed.
+    recovered: Mapped[bool] = mapped_column(default=False, server_default=false())
+
+
 class Event(Base):
     """Per-tenant analytics event (chat -> sales funnel)."""
 

app/orchestrator.py

@@ -258,6 +258,16 @@ async def run_turn(
     # revealed here — identity verification still gates that).
     if getattr(ctx, "customer_note", ""):
         messages.append({"role": "system", "content": ctx.customer_note})
+    # Abandoned-cart recovery (opt-in): once the bot has built a cart, let it offer to
+    # save it and remind the customer by email if they don't finish — only if they
+    # want it (then call save_cart with their email). Never pushy.
+    if getattr(ctx, "allow_cart_recovery", False):
+        messages.append({"role": "system", "content": (
+            "Si has creado un enlace de carrito y el cliente no termina de comprar (duda, "
+            "se va a pensar, o dice que luego), puedes ofrecerle UNA vez guardárselo y "
+            "avisarle por email para retomarlo cuando quiera. Si acepta, pídele el email "
+            "y llama a save_cart con ese email. No insistas si dice que no."
+        )})
     # Deterministic anti-hallucination: asked what the business is/sells/offers
     # or to recommend, weak providers invent a fake store from memory. Force a
     # grounded answer by requiring a tool lookup first.

app/portal_ui/index.html

@@ -170,6 +170,11 @@
           Permitir búsqueda en internet
         </label>
         <p class="muted">Si lo activas, cuando un visitante pregunte por algo externo (p. ej. dónde comprar tu producto en un marketplace), el asistente buscará en la web y dará el enlace real. Desactivado, responde solo con tu conocimiento.</p>
+        <label style="display:flex;align-items:center;gap:8px;font-weight:600;">
+          <input id="b-cartrecovery" type="checkbox" style="width:auto;" />
+          Recuperar carritos abandonados
+        </label>
+        <p class="muted">Si lo activas, cuando un visitante prepare un carrito en el chat pero no termine de comprar, el asistente le ofrecerá guardárselo y le enviará UN email recordatorio con el enlace para retomarlo (saltando a quien ya haya comprado). Necesita tu email de soporte configurado.</p>
         <div class="row" style="margin-top:14px;"><button onclick="saveBusiness()">Guardar</button><span id="biz-state" class="ok"></span></div>
       </div>
     </section>
@@ -426,6 +431,7 @@
     $("b-starters").value = (c.starters || []).join(", ");
     $("b-instructions").value = c.custom_instructions || "";
     $("b-websearch").checked = !!c.allow_web_search;
+    $("b-cartrecovery").checked = !!c.allow_cart_recovery;
     syncColorPick();
     $("sh-shop").value = c.shopify_shop || "";
     $("sh-id").value = c.shopify_client_id || "";
@@ -501,7 +507,8 @@
       avatar_url: $("b-avatar").value.trim(), support_email: $("b-email").value.trim(),
       widget_mode: $("b-mode").value, starters: starters,
       custom_instructions: $("b-instructions").value.trim(),
-      allow_web_search: $("b-websearch").checked }, "biz-state");
+      allow_web_search: $("b-websearch").checked,
+      allow_cart_recovery: $("b-cartrecovery").checked }, "biz-state");
   }
   function saveShopify(){
     var body = { shopify_shop: $("sh-shop").value.trim(), shopify_client_id: $("sh-id").value.trim() };

app/routes/chat.py

@@ -211,6 +211,7 @@ async def _make_context(
         allow_address_change=tenant.allow_address_change,
         allow_returns=tenant.allow_returns,
         allow_web_search=tenant.allow_web_search,
+        allow_cart_recovery=tenant.allow_cart_recovery,
         # tenant's OWN email first (so the output guard's allow-list + any
         # per-tenant use reflects THIS store, not the platform default)
         support_email=tenant.support_email or settings.support_email,

app/routes/portal.py

@@ -53,6 +53,7 @@ class PortalConfigIn(BaseModel):
     starters: list[str] | None = None
     custom_instructions: str | None = None
     allow_web_search: bool | None = None
+    allow_cart_recovery: bool | None = None
     support_email: str | None = None
     shopify_shop: ShopDomain = None
     shopify_client_id: str | None = None
@@ -124,6 +125,7 @@ def _config(tenant: Tenant) -> dict:
         "starters": tenant.starters or [],
         "custom_instructions": tenant.custom_instructions or "",
         "allow_web_search": tenant.allow_web_search,
+        "allow_cart_recovery": tenant.allow_cart_recovery,
         "support_email": tenant.support_email,
         "shopify_shop": tenant.shopify_shop,
         "shopify_client_id": tenant.shopify_client_id,

app/scheduler.py

@@ -30,12 +30,14 @@ _PRODUCT_KEY = "last_product_sync"
 _BILLING_KEY = "last_billing_reconcile"
 _DEMO_PURGE_KEY = "last_demo_purge"
 _TRIAL_KEY = "last_trial_sweep"
+_CART_KEY = "last_cart_recovery"
 REINDEX_EVERY = timedelta(days=7)
 PRODUCT_SYNC_EVERY = timedelta(days=1)  # catalog text changes more often than docs
 BILLING_RECONCILE_EVERY = timedelta(days=1)
 DEMO_PURGE_EVERY = timedelta(hours=6)  # sweep every tick so demos never pile up
 DEMO_MAX_AGE = timedelta(hours=72)  # a demo is deleted 72h after creation, whatever its TTL
 TRIAL_SWEEP_EVERY = timedelta(hours=12)  # pause trials promptly once they expire
+CART_RECOVERY_EVERY = timedelta(hours=3)  # send recovery nudges a few hours after abandonment
 # 29 EUR/mo self-serve link (mirrors app/routes/demo.py MONTHLY_BUY_LINK).
 TRIAL_PAY_LINK = "https://buy.stripe.com/eVq5kC5h6cQS9qa5ar5wI04"
 CHECK_EVERY_SECONDS = 6 * 3600
@@ -288,6 +290,20 @@ async def _maybe_pause_trials() -> None:
             log.info("trial sweep paused %d expired trial(s)", n)
 
 
+async def _maybe_recover_carts() -> None:
+    async with get_sessionmaker()() as db:
+        now = datetime.now(UTC)
+        if not _due(await get_config(db, _CART_KEY) or {}, CART_RECOVERY_EVERY, now):
+            return
+        from app import carts
+
+        n = await carts.recover_due_carts(db, get_settings(), now=now)
+        await upsert_config(db, _CART_KEY, {"at": now.isoformat()})
+        await db.commit()
+        if n:
+            log.info("cart recovery sent %d nudge(s)", n)
+
+
 async def run_periodic_reindex() -> None:
     await asyncio.sleep(FIRST_DELAY_SECONDS)
     while True:
@@ -298,6 +314,7 @@ async def run_periodic_reindex() -> None:
             _maybe_reconcile_billing,
             _maybe_pause_trials,
             _maybe_purge_demos,
+            _maybe_recover_carts,
             _maybe_reindex,
             _maybe_sync_products,
         ):

app/tenancy.py

@@ -15,6 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.crypto import decrypt, encrypt
 from app.models import (
+    AbandonedCart,
     Attachment,
     ChatMessage,
     ChatSession,
@@ -62,6 +63,7 @@ _TENANT_FIELDS = {
     "allow_address_change",
     "allow_returns",
     "allow_web_search",
+    "allow_cart_recovery",
     "chatbot_enabled",
     "brand_name",
     "brand_color",
@@ -167,6 +169,8 @@ async def purge_tenant(db: AsyncSession, tenant: Tenant) -> None:
     ))
     # Visitor uploads (already text-only) — covers admin delete AND demo expiry.
     await db.execute(delete(Attachment).where(Attachment.tenant_id == tenant.id))
+    # Carts reference sessions (SET NULL) — remove them before/with the sessions.
+    await db.execute(delete(AbandonedCart).where(AbandonedCart.tenant_id == tenant.id))
     await db.execute(delete(ChatSession).where(ChatSession.tenant_id == tenant.id))
     await db.execute(delete(Customer).where(Customer.tenant_id == tenant.id))
     await db.execute(delete(HandoffRequest).where(HandoffRequest.tenant_id == tenant.id))

app/tools/cart_tool.py

@@ -46,4 +46,24 @@ async def run(args: dict[str, Any], ctx: ToolContext) -> dict[str, Any]:
         ctx.db, ctx.tenant_id, "cart_link", session_id=ctx.session.id,
         meta={"items": len(parts), "discount": bool(discount)},
     )
+    # Abandoned-cart recovery (opt-in): remember this cart so we can nudge later.
+    # Auto-attaches the email if we already know it (a logged-in/returning customer);
+    # otherwise the bot can ask + call save_cart. Best-effort: never break the cart.
+    if getattr(ctx, "allow_cart_recovery", False):
+        try:
+            from app import carts
+
+            known_email = getattr(getattr(ctx, "customer", None), "email", None)
+            known_name = getattr(getattr(ctx, "customer", None), "name", None)
+            await carts.capture(
+                ctx.db, ctx.tenant_id,
+                session_id=ctx.session.id,
+                checkout_url=url,
+                items_text=f"{len(parts)} artículo(s)",
+                lang=ctx.session.lang,
+                email=known_email,
+                name=known_name,
+            )
+        except Exception:  # noqa: BLE001 - capture is best-effort
+            pass
     return {"status": "ok", "checkout_url": url, "items": len(parts)}

app/tools/registry.py

@@ -18,6 +18,7 @@ from app.tools import (
     knowledge_tool,
     order_tool,
     products_tool,
+    save_cart_tool,
     shipping_tool,
     stock_tool,
     web_search_tool,
@@ -42,6 +43,7 @@ class ToolContext:
     channel: str = "web"  # "web" | "whatsapp" — affects the human-handoff UX
     handoff: bool = False  # set by escalate_to_human to make the widget show its form
     allow_web_search: bool = False  # per-tenant opt-in for the web_search tool
+    allow_cart_recovery: bool = False  # per-tenant opt-in for abandoned-cart recovery
     web_urls: list[str] = field(default_factory=list)  # URLs found this turn (link-guard pass)
     customer: Any | None = None  # Customer profile (returning-customer memory), if known
     customer_note: str = ""  # system note injected when this is a RETURNING customer
@@ -114,6 +116,23 @@ SPECS: list[ToolSpec] = [
             "required": ["items"],
         },
     ),
+    ToolSpec(
+        name="save_cart",
+        description=(
+            "Guarda el carrito que has creado con create_cart_link junto al email del "
+            "cliente para enviarle un recordatorio si no termina la compra. Úsalo SOLO "
+            "cuando el cliente acepte que le avises por email (p. ej. 'sí, guárdamelo' / "
+            "'mándamelo al correo'). Requiere su email; pásalo tal cual lo dé."
+        ),
+        parameters={
+            "type": "object",
+            "properties": {
+                "email": {"type": "string", "description": "Email del cliente"},
+                "name": {"type": "string", "description": "Nombre del cliente (opcional)"},
+            },
+            "required": ["email"],
+        },
+    ),
     ToolSpec(
         name="reorder",
         description=(
@@ -262,6 +281,7 @@ _HANDLERS: dict[str, Callable[[dict[str, Any], ToolContext], Awaitable[dict[str,
     "search_knowledge": knowledge_tool.run,
     "search_products": products_tool.run,
     "create_cart_link": cart_tool.run,
+    "save_cart": save_cart_tool.run,
     "watch_stock": stock_tool.run,
     "lookup_order": order_tool.run,
     "reorder": actions_tool.reorder,

app/tools/save_cart_tool.py

@@ -0,0 +1,32 @@
+"""Tool: save the visitor's in-chat cart + email so we can send a recovery nudge.
+
+The bot calls this when it has built a cart (create_cart_link) and the visitor agrees
+to be reminded by email. Opt-in per store (allow_cart_recovery); a no-op otherwise.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from app.tools.registry import ToolContext
+
+
+async def run(args: dict[str, Any], ctx: ToolContext) -> dict[str, Any]:
+    if not getattr(ctx, "allow_cart_recovery", False):
+        return {"status": "unavailable"}
+    email = str(args.get("email") or "").strip()
+    if "@" not in email:
+        return {"status": "need_info", "need": ["email"]}
+    from app import carts
+
+    cart = await carts.attach_contact(
+        ctx.db, ctx.tenant_id,
+        session_id=ctx.session.id,
+        email=email,
+        name=str(args.get("name") or "").strip() or None,
+    )
+    if cart is None:
+        # No cart built in this session yet — tell the model to create one first.
+        return {"status": "no_cart"}
+    return {"status": "ok", "email": cart.email}

migrations/versions/0029_abandoned_carts.py

@@ -0,0 +1,61 @@
+"""Abandoned-cart recovery.
+
+A per-tenant `abandoned_carts` table (one open cart per chat session, with the
+checkout permalink + an optional captured email) plus a `tenants.allow_cart_recovery`
+opt-in flag, so the bot can save a cart the visitor didn't finish and a background
+job can email a single recovery nudge.
+
+Postgres-only, additive + idempotent. No-op elsewhere (tests/dev use create_all).
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+revision = "0029_abandoned_carts"
+down_revision = "0028_customers"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    if op.get_bind().dialect.name != "postgresql":
+        return
+    op.execute(
+        "ALTER TABLE tenants ADD COLUMN IF NOT EXISTS allow_cart_recovery "
+        "BOOLEAN NOT NULL DEFAULT false"
+    )
+    op.execute(
+        """
+        CREATE TABLE IF NOT EXISTS abandoned_carts (
+            id SERIAL PRIMARY KEY,
+            tenant_id INTEGER REFERENCES tenants(id) ON DELETE CASCADE,
+            session_id VARCHAR(40) REFERENCES chat_sessions(id) ON DELETE SET NULL,
+            email VARCHAR(255),
+            name VARCHAR(160) NOT NULL DEFAULT '',
+            lang VARCHAR(10),
+            checkout_url TEXT NOT NULL DEFAULT '',
+            items_text VARCHAR(500) NOT NULL DEFAULT '',
+            created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+            nudged_at TIMESTAMPTZ,
+            recovered BOOLEAN NOT NULL DEFAULT false
+        )
+        """
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_abandoned_carts_tenant_id ON abandoned_carts (tenant_id)"
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_abandoned_carts_session_id ON abandoned_carts (session_id)"
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_abandoned_carts_email ON abandoned_carts (email)"
+    )
+
+
+def downgrade() -> None:
+    if op.get_bind().dialect.name != "postgresql":
+        return
+    op.execute("DROP TABLE IF EXISTS abandoned_carts")
+    op.execute("ALTER TABLE tenants DROP COLUMN IF EXISTS allow_cart_recovery")

tests/test_carts.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import select
+
+from app import carts, mailer
+from app.models import AbandonedCart, ChatSession, Tenant
+
+
+async def _session(db, tenant_id, lang="es"):
+    s = ChatSession(tenant_id=tenant_id, lang=lang)
+    db.add(s)
+    await db.flush()
+    return s
+
+
+async def test_capture_one_per_session_and_updates(db_session):
+    s = await _session(db_session, 1)
+    c1 = await carts.capture(
+        db_session, 1, session_id=s.id, checkout_url="https://x/cart/1:1", items_text="1 art."
+    )
+    # same session -> SAME row updated (no duplicate), latest link wins
+    c2 = await carts.capture(
+        db_session, 1, session_id=s.id, checkout_url="https://x/cart/2:1", items_text="2 art."
+    )
+    assert c2.id == c1.id
+    assert c2.checkout_url.endswith("2:1") and c2.items_text == "2 art."
+    assert c2.email is None  # nothing captured yet
+
+
+async def test_capture_autofills_known_email(db_session):
+    s = await _session(db_session, 1)
+    c = await carts.capture(
+        db_session, 1, session_id=s.id, checkout_url="https://x/cart/9:1",
+        email="Buyer@X.com", name="Leo",
+    )
+    assert c.email == "buyer@x.com" and c.name == "Leo"  # normalized
+
+
+async def test_attach_contact_requires_existing_cart(db_session):
+    s = await _session(db_session, 1)
+    # no cart built yet in this session -> nothing to attach
+    assert await carts.attach_contact(db_session, 1, session_id=s.id, email="a@b.com") is None
+    await carts.capture(db_session, 1, session_id=s.id, checkout_url="https://x/cart/3:1")
+    c = await carts.attach_contact(db_session, 1, session_id=s.id, email="a@b.com", name="Ann")
+    assert c is not None and c.email == "a@b.com" and c.name == "Ann"
+
+
+async def test_capture_is_tenant_isolated(db_session):
+    sa = await _session(db_session, 1)
+    sb = await _session(db_session, 2)
+    a = await carts.capture(db_session, 1, session_id=sa.id, checkout_url="https://a/cart/1:1")
+    b = await carts.capture(db_session, 2, session_id=sb.id, checkout_url="https://b/cart/1:1")
+    assert a.id != b.id and a.tenant_id == 1 and b.tenant_id == 2
+
+
+def test_recovery_email_language():
+    es_subject, es_body = carts._recovery_email(
+        AbandonedCart(name="María García", lang="es", checkout_url="https://x/cart/1:1"), "MiTienda"
+    )
+    assert "MiTienda" in es_subject and "María" in es_body and "https://x/cart/1:1" in es_body
+    en_subject, en_body = carts._recovery_email(
+        AbandonedCart(name="John", lang="en", checkout_url="https://x/cart/1:1"), "MyShop"
+    )
+    assert "saved" in en_subject.lower() and en_body.startswith("Hi John")
+
+
+async def test_recover_due_carts_sends_and_skips(db_session, monkeypatch):
+    db_session.add(Tenant(slug="cr", name="CR", brand_name="CartShop",
+                          allow_cart_recovery=True, support_email="shop@cr.com"))
+    db_session.add(Tenant(slug="off", name="OFF", allow_cart_recovery=False))
+    await db_session.flush()
+    cr = (await db_session.execute(
+        select(Tenant).where(Tenant.slug == "cr"))).scalar_one()
+    off = (await db_session.execute(
+        select(Tenant).where(Tenant.slug == "off"))).scalar_one()
+    old = datetime.now(UTC) - timedelta(hours=5)  # past the 4h delay, within 48h
+    # due: opted-in tenant, has email, old enough
+    db_session.add(AbandonedCart(tenant_id=cr.id, email="a@b.com", checkout_url="https://x/cart/1:1",
+                                 created_at=old))
+    # not due: no email captured
+    db_session.add(AbandonedCart(tenant_id=cr.id, checkout_url="https://x/cart/2:1", created_at=old))
+    # not due: too recent
+    db_session.add(AbandonedCart(tenant_id=cr.id, email="c@d.com", checkout_url="https://x/cart/3:1",
+                                 created_at=datetime.now(UTC)))
+    # opted-out tenant must NEVER be emailed
+    db_session.add(AbandonedCart(tenant_id=off.id, email="e@f.com", checkout_url="https://x/cart/4:1",
+                                 created_at=old))
+    await db_session.flush()
+
+    sent: list[tuple] = []
+
+    async def fake_send(settings, to, subject, body, **kw):
+        sent.append((to, subject, kw.get("reply_to")))
+        return True
+
+    monkeypatch.setattr(mailer, "send_email", fake_send)
+    n = await carts.recover_due_carts(db_session, object(), now=datetime.now(UTC))
+    assert n == 1
+    assert sent == [("a@b.com", sent[0][1], "shop@cr.com")]  # only the one due cart, with reply-to
+
+
+async def test_recover_due_skips_already_bought(db_session, monkeypatch):
+    import app.shopify.orders as orders_mod
+    import app.tenancy as tenancy_mod
+
+    db_session.add(Tenant(slug="cr2", name="CR2", brand_name="Shop2", allow_cart_recovery=True))
+    await db_session.flush()
+    t = (await db_session.execute(
+        select(Tenant).where(Tenant.slug == "cr2"))).scalar_one()
+    old = datetime.now(UTC) - timedelta(hours=5)
+    cart = AbandonedCart(tenant_id=t.id, email="paid@x.com", checkout_url="https://x/cart/1:1",
+                         created_at=old)
+    db_session.add(cart)
+    await db_session.flush()
+
+    monkeypatch.setattr(tenancy_mod, "tenant_shopify_client", lambda tenant: object())
+
+    async def fake_orders(client, email):
+        # an order placed AFTER the cart was built -> they already bought
+        return [{"created_at": datetime.now(UTC).isoformat()}]
+
+    monkeypatch.setattr(orders_mod, "find_orders_by_email", fake_orders)
+    sent = []
+
+    async def fake_send(*a, **k):
+        sent.append(a)
+        return True
+
+    monkeypatch.setattr(mailer, "send_email", fake_send)
+    n = await carts.recover_due_carts(db_session, object(), now=datetime.now(UTC))
+    assert n == 0 and sent == []  # no email — sale already closed
+    # same identity-mapped row recover_due_carts mutated (caller commits; don't refresh
+    # here or the un-flushed change is discarded)
+    assert cart.recovered is True and cart.nudged_at is None

tests/tools/test_tools.py

@@ -27,6 +27,7 @@ def test_specs_expose_tools():
         "search_knowledge",
         "search_products",
         "create_cart_link",
+        "save_cart",
         "watch_stock",
         "lookup_order",
         "reorder",
@@ -37,6 +38,44 @@ def test_specs_expose_tools():
     } <= names
 
 
+async def test_save_cart_requires_optin_and_cart(db_session):
+    from app.models import AbandonedCart
+    from sqlalchemy import select
+
+    sess = await _session(db_session)
+    # opt-out store: the tool is inert (so the bot never promises an email it can't send)
+    off = ToolContext(db=db_session, session=sess, tenant_id=1, allow_cart_recovery=False)
+    assert (await registry.dispatch("save_cart", {"email": "a@b.com"}, off))["status"] == "unavailable"
+
+    on = ToolContext(db=db_session, session=sess, tenant_id=1, allow_cart_recovery=True)
+    # needs an email
+    assert (await registry.dispatch("save_cart", {}, on))["status"] == "need_info"
+    # no cart built in this session yet
+    assert (await registry.dispatch("save_cart", {"email": "a@b.com"}, on))["status"] == "no_cart"
+
+    # build a cart, then saving attaches the email to it
+    on.shop = "s.myshopify.com"
+    await registry.dispatch("create_cart_link", {"items": [{"variant_id": "9"}]}, on)
+    out = await registry.dispatch("save_cart", {"email": "Buyer@X.com", "name": "Leo"}, on)
+    assert out["status"] == "ok" and out["email"] == "buyer@x.com"
+    cart = (await db_session.execute(
+        select(AbandonedCart).where(AbandonedCart.session_id == sess.id))).scalar_one()
+    assert cart.email == "buyer@x.com" and cart.name == "Leo"
+
+
+async def test_create_cart_link_auto_captures_when_recovery_on(db_session):
+    from app.models import AbandonedCart
+    from sqlalchemy import select
+
+    sess = await _session(db_session)
+    ctx = ToolContext(db=db_session, session=sess, tenant_id=1, shop="s.myshopify.com",
+                      allow_cart_recovery=True)
+    await registry.dispatch("create_cart_link", {"items": [{"variant_id": "5"}]}, ctx)
+    cart = (await db_session.execute(
+        select(AbandonedCart).where(AbandonedCart.session_id == sess.id))).scalar_one()
+    assert cart.checkout_url.endswith("5:1") and cart.email is None  # captured, awaiting email
+
+
 async def test_cart_link_with_discount(db_session):
     ctx = ToolContext(db=db_session, session=await _session(db_session), shop="s.myshopify.com")
     out = await registry.dispatch(