widget UX batch

app/demo_ui/index.html

@@ -229,14 +229,23 @@ function setFormError(msg){
 }
 function clearFormError(){ document.getElementById("form-error").hidden = true; }
 
+// Founder bypass: /demo?admin=<token> stores the token for this tab only and
+// sends it as X-Admin-Token on /demo/start (the backend skips the caps).
+function adminToken(){
+  try { return sessionStorage.getItem("demo_admin_token") || ""; } catch(e){ return ""; }
+}
+
 function startDemo(url, email){
   clearFormError();
   document.getElementById("expired-note").hidden = true;
   var btn = document.getElementById("start-btn");
   btn.disabled = true; btn.textContent = tr("Creando tu demo…");
+  var headers = { "Content-Type": "application/json" };
+  var admin = adminToken();
+  if (admin) headers["X-Admin-Token"] = admin;
   fetch(API + "/demo/start", {
     method: "POST",
-    headers: { "Content-Type": "application/json" },
+    headers: headers,
     body: JSON.stringify({ url: url, email: email || "" })
   })
   .then(function(r){ return r.json().then(function(d){ return { ok: r.ok, d: d }; }); })
@@ -381,6 +390,13 @@ document.getElementById("copy-btn").addEventListener("click", function(){
 document.addEventListener("DOMContentLoaded", function(){
   var l = currentLang(); if (l === "en") applyLang("en");
   var qs = new URLSearchParams(location.search);
+  var adm = qs.get("admin");
+  if (adm) {
+    try { sessionStorage.setItem("demo_admin_token", adm); } catch(e){}
+    qs.delete("admin");
+    var rest = qs.toString();
+    history.replaceState(null, "", location.pathname + (rest ? "?" + rest : ""));
+  }
   var d = qs.get("d");
   var u = qs.get("url");
   if (d) {

app/models.py

@@ -275,6 +275,26 @@ class HandoffRequest(Base):
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
 
 
+class Attachment(Base):
+    """A file a VISITOR uploaded mid-conversation (PDF or photo), already
+    converted to text (PDF extraction / vision description). The binary is
+    NEVER persisted (ephemeral disk on the free host) and the text is scoped
+    to exactly one tenant + one chat session: it is never embedded, never
+    written to the knowledge base, never visible to any other conversation."""
+
+    __tablename__ = "attachments"
+
+    id: Mapped[str] = mapped_column(String(40), primary_key=True, default=lambda: uuid.uuid4().hex)
+    tenant_id: Mapped[int | None] = mapped_column(
+        ForeignKey("tenants.id", ondelete="CASCADE"), nullable=True, index=True
+    )
+    session_id: Mapped[str] = mapped_column(String(40), index=True)
+    filename: Mapped[str] = mapped_column(String(255), default="")
+    mime: Mapped[str] = mapped_column(String(60), default="")
+    text: Mapped[str] = mapped_column(Text, default="")
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
+
+
 class ChatMessage(Base):
     __tablename__ = "chat_messages"
 

app/orchestrator.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import json
 import logging
 import re
+from collections.abc import Callable
 from typing import Any
 
 from sqlalchemy import select
@@ -46,6 +47,17 @@ NO_TOOLS_NOTE = (
 )
 
 
+def _emit_stage(on_stage: Callable[[str], None] | None, key: str) -> None:
+    """Best-effort UI progress signal. A broken callback must NEVER break a
+    chat turn, so every invocation is fully shielded."""
+    if on_stage is None:
+        return
+    try:
+        on_stage(key)
+    except Exception:
+        log.warning("on_stage callback failed for %r", key, exc_info=True)
+
+
 def _toolcall_to_openai(tc: ToolCall) -> dict[str, Any]:
     return {
         "id": tc.id,
@@ -72,7 +84,9 @@ async def run_turn(
     *,
     brand_name: str = "Asistente",
     instructions: str = "",
+    on_stage: Callable[[str], None] | None = None,
 ) -> ChatResponse:
+    _emit_stage(on_stage, "thinking")
     prior = await _load_history(ctx)
     ctx.user_message = user_message  # for the deterministic write-action confirm gate
 
@@ -157,6 +171,7 @@ async def run_turn(
             )
             for tc in result.tool_calls:
                 used_tools.append(tc.name)
+                _emit_stage(on_stage, "tool:" + tc.name)
                 tool_result = await registry.dispatch(tc.name, tc.arguments, ctx)
                 payload = json.dumps(tool_result, ensure_ascii=False, default=str)
                 if len(payload) > MAX_TOOL_RESULT_CHARS:  # protect small-window models

app/ratelimit.py

@@ -30,6 +30,24 @@ class RateLimiter:
         self._hits[key] = hits
         return True
 
+    def would_allow(self, key: str, now: datetime | None = None) -> bool:
+        """Peek: like allow() but WITHOUT consuming quota. Lets callers gate a
+        request up front and only record() once the work actually succeeds, so
+        rejected attempts don't burn the caller's budget."""
+        ts = (now or datetime.now(UTC)).timestamp()
+        cutoff = ts - self.window
+        # .get(): peeking must not materialize defaultdict entries for keys
+        # that never get a recorded hit.
+        return len([h for h in self._hits.get(key, []) if h > cutoff]) < self.max_hits
+
+    def record(self, key: str, now: datetime | None = None) -> None:
+        """Consume one slot unconditionally (pair with would_allow())."""
+        ts = (now or datetime.now(UTC)).timestamp()
+        cutoff = ts - self.window
+        hits = [h for h in self._hits[key] if h > cutoff]
+        hits.append(ts)
+        self._hits[key] = hits
+
     def reset(self, key: str) -> None:
         self._hits.pop(key, None)
 

app/retention.py

@@ -7,7 +7,7 @@ from datetime import UTC, datetime, timedelta
 from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.models import ChatMessage, ChatSession, Event
+from app.models import Attachment, ChatMessage, ChatSession, Event
 
 
 async def purge_old_sessions(db: AsyncSession, days: int) -> int:
@@ -28,6 +28,9 @@ async def purge_old_sessions(db: AsyncSession, days: int) -> int:
         await db.flush()
         return 0
     await db.execute(delete(ChatMessage).where(ChatMessage.session_id.in_(ids)))
+    # Visitor uploads die with their conversation (GDPR: they may contain
+    # personal documents, so they never outlive the session that owns them).
+    await db.execute(delete(Attachment).where(Attachment.session_id.in_(ids)))
     await db.execute(delete(ChatSession).where(ChatSession.id.in_(ids)))
     await db.flush()
     return len(ids)

app/routes/chat.py

@@ -2,38 +2,69 @@
 
 - POST /chat?t=<slug>   — simple mode: public, CORS-enabled, no signature. The
                           backend-served widget uses this (one <script> to embed).
+- POST /chat/stream     — same contract as /chat but answers as an SSE stream:
+                          live "stage" progress events while the bot works, then
+                          one "final" event with the complete ChatResponse JSON.
 - POST /apps/chat       — secure mode: behind the Shopify App Proxy (HMAC); tenant
                           resolved from the signed `shop` param (or ?t).
 
-Both share one handler. Order disclosure always requires email+order#
+All share one core. Order disclosure always requires email+order#
 verification + per-email rate limiting regardless of entry point.
 """
 
 from __future__ import annotations
 
+import asyncio
+import contextlib
+import json
+import logging
+from collections.abc import AsyncIterator
 from datetime import UTC, datetime
 
-from fastapi import APIRouter, Depends, HTTPException, Request
-from sqlalchemy import select
+from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile
+from fastapi.responses import StreamingResponse
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app import vision
 from app.config import Settings, get_settings
 from app.db import get_session
 from app.deps import get_router
 from app.handoff import create_handoff
-from app.models import Tenant
+from app.models import Attachment, Tenant
 from app.orchestrator import run_turn
+from app.rag import extract
 from app.retention import purge_old_sessions
 from app.schemas import ChatRequest, ChatResponse, HandoffIn
 from app.shopify.proxy import verify_proxy_signature
+from app.sources import SAFE_NAME
 from app.store import get_or_create_session
 from app.tenancy import DEFAULT_SLUG, get_tenant_by_slug, tenant_shopify_client
 from app.tools.registry import ToolContext
 from app.verification import mark_trusted
 
+log = logging.getLogger(__name__)
+
 router = APIRouter()
 
 PROXY_TIMESTAMP_TOLERANCE = 300  # seconds; reject replayed/stale signed requests
+STREAM_PING_SECONDS = 10.0  # SSE keep-alive comment cadence while the turn runs
+
+# --- visitor uploads (Phase 3) ---------------------------------------------
+# Binaries are NEVER persisted (ephemeral disk): a PDF becomes its extracted
+# text and a photo becomes a vision description, stored per tenant + session.
+MAX_ATTACHMENT_BYTES = 5 * 1024 * 1024  # hard cap, enforced while reading
+MAX_ATTACHMENTS_PER_SESSION = 3
+ATTACHMENT_TEXT_CAP = 6000  # chars stored per attachment
+ATTACHMENT_BLOCK_CAP = 12000  # chars of attachment context prepended per turn
+VISION_FALLBACK_TEXT = "(imagen recibida pero no se ha podido analizar)"
+
+MSG_TOO_MANY_ATTACHMENTS = "Solo se pueden adjuntar 3 archivos por conversación."
+MSG_ATTACHMENT_TOO_BIG = "El archivo supera el tamaño máximo de 5 MB."
+MSG_BAD_ATTACHMENT_TYPE = (
+    "Tipo de archivo no admitido. Adjunta un PDF o una imagen (PNG, JPG o WEBP)."
+)
+MSG_PDF_NO_TEXT = "El PDF no tiene texto legible."
 
 
 def _collect_params(request: Request) -> dict[str, object]:
@@ -80,27 +111,28 @@ async def _resolve_tenant(db: AsyncSession, slug: str | None, *, shop: str = "")
     return tenant
 
 
-async def _run_chat(
+def _ensure_chatbot_enabled(tenant: Tenant) -> None:
+    if not tenant.chatbot_enabled:
+        # Admin turned this store's chatbot off (e.g. non-payment): refuse
+        # politely without spending an LLM call. The widget also self-hides.
+        raise HTTPException(status_code=403, detail="chatbot disabled for this store")
+
+
+async def _make_context(
     *,
     db: AsyncSession,
-    router_,
     settings: Settings,
     tenant: Tenant,
-    message: str,
     session_id: str | None,
     trusted: bool,
     now: datetime,
-) -> ChatResponse:
-    if not tenant.chatbot_enabled:
-        # Admin turned this store's chatbot off (e.g. non-payment): refuse
-        # politely without spending an LLM call. The widget also self-hides.
-        raise HTTPException(status_code=403, detail="chatbot disabled for this store")
+) -> ToolContext:
     session = await get_or_create_session(
         db, session_id, shop=tenant.shopify_shop, tenant_id=tenant.id
     )
     if trusted:
         mark_trusted(session)
-    ctx = ToolContext(
+    return ToolContext(
         db=db,
         session=session,
         shopify=tenant_shopify_client(tenant),
@@ -113,8 +145,59 @@ async def _run_chat(
         support_email=settings.support_email,
         now=now,
     )
-    response = await run_turn(router_, ctx, message, brand_name=tenant.brand_name,
-                            instructions=tenant.custom_instructions)
+
+
+async def _attachment_block(
+    db: AsyncSession, tenant_id: int, session_id: str, attachment_ids: list[str]
+) -> str:
+    """Context block for the attachments referenced by THIS turn.
+
+    Strict isolation invariant (the product's #1 requirement): an id only
+    matches when it belongs to the SAME tenant AND the SAME chat session.
+    Any other id — another tenant's, another session's, or simply unknown —
+    is SILENTLY dropped: never an error, so existence is never revealed.
+    """
+    if not attachment_ids:
+        return ""
+    rows = (
+        await db.execute(
+            select(Attachment)
+            .where(
+                Attachment.id.in_(attachment_ids),
+                Attachment.tenant_id == tenant_id,
+                Attachment.session_id == session_id,
+            )
+            .order_by(Attachment.created_at, Attachment.id)
+        )
+    ).scalars().all()
+    parts = [f"[El cliente ha adjuntado {a.filename}]:\n{a.text}\n\n" for a in rows]
+    return "".join(parts)[:ATTACHMENT_BLOCK_CAP]
+
+
+async def _run_chat(
+    *,
+    db: AsyncSession,
+    router_,
+    settings: Settings,
+    tenant: Tenant,
+    message: str,
+    session_id: str | None,
+    trusted: bool,
+    now: datetime,
+    on_stage=None,
+    attachment_ids: list[str] | None = None,
+) -> ChatResponse:
+    _ensure_chatbot_enabled(tenant)
+    ctx = await _make_context(
+        db=db, settings=settings, tenant=tenant,
+        session_id=session_id, trusted=trusted, now=now,
+    )
+    # Attachments ride INSIDE the user message (also persisted that way, so
+    # the conversation history replays coherently). They are never embedded
+    # nor written to the knowledge base.
+    block = await _attachment_block(db, tenant.id, ctx.session.id, attachment_ids or [])
+    response = await run_turn(router_, ctx, block + message, brand_name=tenant.brand_name,
+                            instructions=tenant.custom_instructions, on_stage=on_stage)
     await purge_old_sessions(db, settings.session_retention_days)
     await db.commit()
     return response
@@ -139,6 +222,181 @@ async def chat_simple(
         session_id=body.session_id,
         trusted=False,
         now=datetime.now(UTC),
+        attachment_ids=body.attachment_ids,
+    )
+
+
+def _sniff_attachment(data: bytes) -> tuple[str, str] | None:
+    """(kind, mime) by MAGIC BYTES — extension and Content-Type are attacker
+    controlled and ignored. Only PDF, PNG, JPEG and WEBP are accepted."""
+    if data.startswith(b"%PDF"):
+        return ("pdf", "application/pdf")
+    if data.startswith(b"\x89PNG\r\n\x1a\n"):
+        return ("image", "image/png")
+    if data.startswith(b"\xff\xd8\xff"):
+        return ("image", "image/jpeg")
+    if len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP":
+        return ("image", "image/webp")
+    return None
+
+
+@router.post("/chat/upload")
+async def chat_upload(
+    request: Request,
+    file: UploadFile = File(...),
+    session_id: str | None = Form(None),
+    db: AsyncSession = Depends(get_session),
+) -> dict:
+    """Visitor mid-conversation upload (PDF or photo). Same gates as /chat —
+    unknown explicit slug 404, paused tenant 403 — and the attachment binds to
+    exactly the session the next chat message will use. The binary is turned
+    into text here (PDF extraction / vision description) and discarded."""
+    tenant = await _resolve_tenant(db, request.query_params.get("t"))
+    _ensure_chatbot_enabled(tenant)
+    session = await get_or_create_session(
+        db, session_id, shop=tenant.shopify_shop, tenant_id=tenant.id
+    )
+    existing = (
+        await db.execute(
+            select(func.count()).select_from(Attachment).where(
+                Attachment.tenant_id == tenant.id,
+                Attachment.session_id == session.id,
+            )
+        )
+    ).scalar_one()
+    if existing >= MAX_ATTACHMENTS_PER_SESSION:
+        raise HTTPException(status_code=422, detail=MSG_TOO_MANY_ATTACHMENTS)
+
+    # Size cap enforced WHILE reading: an oversized body is rejected as soon
+    # as it crosses the limit, without buffering the whole thing.
+    buf = bytearray()
+    while True:
+        chunk = await file.read(1024 * 1024)
+        if not chunk:
+            break
+        buf.extend(chunk)
+        if len(buf) > MAX_ATTACHMENT_BYTES:
+            raise HTTPException(status_code=413, detail=MSG_ATTACHMENT_TOO_BIG)
+    data = bytes(buf)
+
+    sniffed = _sniff_attachment(data)
+    if sniffed is None:
+        raise HTTPException(status_code=422, detail=MSG_BAD_ATTACHMENT_TYPE)
+    kind, mime = sniffed
+
+    if kind == "pdf":
+        try:
+            # Same extractor as the knowledge pipeline (app/rag/extract.py).
+            text = await asyncio.to_thread(extract.extract_bytes, data, "pdf")
+        except Exception:  # noqa: BLE001 - corrupt/unreadable PDF -> friendly 422
+            log.warning("attachment pdf extraction failed", exc_info=True)
+            text = ""
+        text = (text or "").strip()[:ATTACHMENT_TEXT_CAP]
+        if not text:
+            raise HTTPException(status_code=422, detail=MSG_PDF_NO_TEXT)
+    else:
+        described = await vision.describe_image(data, mime)
+        text = (described or "").strip()[:ATTACHMENT_TEXT_CAP] or VISION_FALLBACK_TEXT
+
+    safe_name = SAFE_NAME.sub("_", file.filename or "archivo")[:255]
+    if safe_name in (".", "..", ""):
+        safe_name = "archivo"
+    att = Attachment(
+        tenant_id=tenant.id, session_id=session.id,
+        filename=safe_name, mime=mime, text=text,
+    )
+    db.add(att)
+    await db.flush()
+    out = {
+        "attachment_id": att.id,
+        "filename": safe_name,
+        "kind": kind,
+        "preview": text[:200],
+        # The widget MUST adopt this id so the next message lands on the same
+        # session the attachment is bound to (first contact may be an upload).
+        "session_id": session.id,
+    }
+    await db.commit()
+    return out
+
+
+def _sse(event: str, data: str) -> str:
+    return f"event: {event}\ndata: {data}\n\n"
+
+
+async def _stream_events(
+    *,
+    db: AsyncSession,
+    router_,
+    settings: Settings,
+    tenant: Tenant,
+    message: str,
+    session_id: str | None,
+    attachment_ids: list[str] | None = None,
+) -> AsyncIterator[str]:
+    """Yield SSE frames: live 'stage' events while run_turn works, then one
+    'final' event carrying the complete ChatResponse JSON (same shape as
+    POST /chat). On failure: an 'error' event (the widget falls back to /chat)."""
+    queue: asyncio.Queue[tuple[str, str] | None] = asyncio.Queue()
+
+    def on_stage(key: str) -> None:
+        queue.put_nowait(("stage", json.dumps({"key": key})))
+
+    async def _runner() -> None:
+        try:
+            response = await _run_chat(
+                db=db, router_=router_, settings=settings, tenant=tenant,
+                message=message, session_id=session_id, trusted=False,
+                now=datetime.now(UTC), on_stage=on_stage,
+                attachment_ids=attachment_ids,
+            )
+            queue.put_nowait(("final", response.model_dump_json()))
+        except Exception:
+            log.exception("chat stream turn failed tenant=%s", tenant.slug)
+            queue.put_nowait(("error", json.dumps({"detail": "stream failed"})))
+        finally:
+            queue.put_nowait(None)  # sentinel: nothing more will arrive
+
+    task = asyncio.create_task(_runner())
+    try:
+        while True:
+            try:
+                item = await asyncio.wait_for(queue.get(), timeout=STREAM_PING_SECONDS)
+            except TimeoutError:
+                yield ": ping\n\n"  # keep proxies from killing the idle stream
+                continue
+            if item is None:
+                break
+            yield _sse(*item)
+    finally:
+        if not task.done():
+            task.cancel()  # client went away mid-turn: stop the LLM work
+        # surface unexpected runner crashes instead of swallowing them
+        with contextlib.suppress(asyncio.CancelledError):
+            await task
+
+
+@router.post("/chat/stream")
+async def chat_stream(
+    body: ChatRequest,
+    request: Request,
+    db: AsyncSession = Depends(get_session),
+    router_=Depends(get_router),
+    settings: Settings = Depends(get_settings),
+) -> StreamingResponse:
+    """Streaming variant of /chat (SSE over POST). Tenant resolution and the
+    kill-switch run BEFORE the stream starts so a deleted slug is a plain 404
+    and a paused tenant a plain 403 — never a stream, never an LLM call."""
+    tenant = await _resolve_tenant(db, request.query_params.get("t"))
+    _ensure_chatbot_enabled(tenant)
+    return StreamingResponse(
+        _stream_events(
+            db=db, router_=router_, settings=settings, tenant=tenant,
+            message=body.message, session_id=body.session_id,
+            attachment_ids=body.attachment_ids,
+        ),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
     )
 
 

app/routes/demo.py

@@ -15,11 +15,17 @@ limit today, and adding one is out of this feature's scope):
 - at most DEMO_MAX_ACTIVE live demo tenants at any moment.
 Together these cap the worst-case number of bots that can be burning free-tier
 LLM calls even if a demo link goes viral.
+
+Two refinements: only SUCCESSFUL starts consume quota (a rejected URL never
+burns the visitor's hourly budget nor the daily counter), and a request whose
+X-Admin-Token header matches settings.admin_token bypasses all three caps so
+the founder can test from his own IP without eating visitor quota.
 """
 
 from __future__ import annotations
 
 import asyncio
+import hmac
 import ipaddress
 import logging
 import secrets
@@ -174,6 +180,16 @@ def _client_ip(request: Request) -> str:
     return request.client.host if request.client else "unknown"
 
 
+def _is_admin_request(request: Request) -> bool:
+    """True when the request carries X-Admin-Token matching the configured
+    admin token (constant-time compare, same as deps.require_admin). The
+    founder demos from his own IP, so an admin start skips the per-IP, daily
+    and active-demo caps — and is not counted against any of them."""
+    token = request.headers.get("x-admin-token", "")
+    admin_token = get_settings().admin_token
+    return bool(token) and bool(admin_token) and hmac.compare_digest(token, admin_token)
+
+
 # --- progress registry ("teatro de progreso") -----------------------------
 
 _PROGRESS_MAX = 200
@@ -336,8 +352,14 @@ async def _unique_demo_slug(db: AsyncSession) -> str:
 async def demo_start(
     body: DemoStartIn, request: Request, db: AsyncSession = Depends(get_session)
 ) -> dict:
+    # X-Admin-Token: the founder testing from his own IP bypasses every cap.
+    is_admin = _is_admin_request(request)
+    client_ip = _client_ip(request)
     # Cheapest gate first: the per-IP limiter also shields the DNS resolver.
-    if not _ip_limiter.allow(_client_ip(request)):
+    # PEEK only — quota is consumed (record) just after the demo tenant is
+    # durably created, so SSRF-rejected/invalid attempts don't burn the
+    # visitor's hourly budget.
+    if not is_admin and not _ip_limiter.would_allow(client_ip):
         raise HTTPException(status_code=429, detail=MSG_IP_LIMIT)
     url = await validate_demo_url(body.url)
 
@@ -345,15 +367,16 @@ async def demo_start(
     today = now.date().isoformat()
     daily = await get_config(db, _DAILY_KEY) or {}
     used_today = int(daily.get("count", 0)) if daily.get("date") == today else 0
-    if used_today >= DEMO_DAILY_CAP:
-        raise HTTPException(status_code=429, detail=MSG_DAILY_LIMIT)
-    active = (
-        await db.execute(
-            select(func.count()).select_from(Tenant).where(Tenant.is_demo.is_(True))
-        )
-    ).scalar_one()
-    if active >= DEMO_MAX_ACTIVE:
-        raise HTTPException(status_code=429, detail=MSG_ACTIVE_LIMIT)
+    if not is_admin:
+        if used_today >= DEMO_DAILY_CAP:
+            raise HTTPException(status_code=429, detail=MSG_DAILY_LIMIT)
+        active = (
+            await db.execute(
+                select(func.count()).select_from(Tenant).where(Tenant.is_demo.is_(True))
+            )
+        ).scalar_one()
+        if active >= DEMO_MAX_ACTIVE:
+            raise HTTPException(status_code=429, detail=MSG_ACTIVE_LIMIT)
 
     domain = (urlsplit(url).hostname or "").removeprefix("www.")
     slug = await _unique_demo_slug(db)
@@ -382,8 +405,14 @@ async def demo_start(
     except Exception:  # noqa: BLE001 - lead capture must never break the demo start
         log.warning("demo handoff capture failed (non-fatal)", exc_info=True)
 
-    await upsert_config(db, _DAILY_KEY, {"date": today, "count": used_today + 1})
+    # Consumption is recorded ONLY for real (non-admin) starts that got this
+    # far: failed validations/caps above never reach these lines, so they
+    # never burn the visitor's hourly quota nor the global daily counter.
+    if not is_admin:
+        await upsert_config(db, _DAILY_KEY, {"date": today, "count": used_today + 1})
     await db.commit()
+    if not is_admin:
+        _ip_limiter.record(client_ip)
 
     _progress_start(slug)
     _spawn(_run_demo_index(slug, src.id))

app/schemas.py

@@ -31,6 +31,14 @@ ShopDomain = Annotated[str | None, AfterValidator(_normalize_shop)]
 class ChatRequest(BaseModel):
     message: str = Field(min_length=1, max_length=4000)
     session_id: str | None = None
+    # Ids of files the visitor uploaded via /chat/upload for THIS conversation.
+    # Server-side they are matched strictly by (id, tenant, session); anything
+    # else is silently dropped, so a foreign id can never leak content. Real
+    # ids are 32-hex (column is 40): the per-item cap keeps the only otherwise
+    # unbounded client string on this endpoint out of the SQL IN clause.
+    attachment_ids: list[Annotated[str, Field(max_length=64)]] = Field(
+        default_factory=list, max_length=10
+    )
 
 
 class ChatResponse(BaseModel):

app/static/widget.js

@@ -32,6 +32,18 @@
     hfThanks: "\u00a1Gracias! Un compa\u00f1ero del equipo te responder\u00e1 en breve por email.",
     hfError: "No he podido enviar el formulario. Int\u00e9ntalo de nuevo en un momento.",
     open: "Abrir chat de soporte", close: "Cerrar", send: "Enviar",
+    mute: "Silenciar avisos", unmute: "Activar avisos",
+    newMsg: "Nuevo mensaje",
+    stThinking: "Pensando\u2026",
+    stCatalog: "Buscando en el cat\u00e1logo\u2026",
+    stOrder: "Consultando el pedido\u2026",
+    stDocs: "Leyendo la documentaci\u00f3n\u2026",
+    stWait: "Un momento\u2026",
+    attach: "Adjuntar archivo",
+    uploading: "Subiendo\u2026",
+    tooBig: "Archivo demasiado grande (m\u00e1x. 5 MB)",
+    upFail: "No se pudo subir el archivo",
+    attached: "Adjunto",
     starters: ["\u00bfQu\u00e9 ofrec\u00e9is?", "Quiero m\u00e1s informaci\u00f3n", "Hablar con una persona"]
   } : {
     status: "Online \u00b7 Replies in seconds",
@@ -46,6 +58,18 @@
     hfThanks: "Thank you! Someone from the team will reply to you by email shortly.",
     hfError: "The form could not be sent. Please try again in a moment.",
     open: "Open support chat", close: "Close", send: "Send",
+    mute: "Mute notifications", unmute: "Unmute notifications",
+    newMsg: "New message",
+    stThinking: "Thinking\u2026",
+    stCatalog: "Searching the catalog\u2026",
+    stOrder: "Checking your order\u2026",
+    stDocs: "Reading the documentation\u2026",
+    stWait: "One moment\u2026",
+    attach: "Attach a file",
+    uploading: "Uploading\u2026",
+    tooBig: "File too large (max 5 MB)",
+    upFail: "The file could not be uploaded",
+    attached: "Attachment",
     starters: ["What do you offer?", "I want more information", "Talk to a person"]
   };
 
@@ -74,6 +98,13 @@
   var ICON_SEND = '<svg viewBox="0 0 24 24"><path d="M3 11l18-8-8 18-2.5-7.5L3 11z"/></svg>';
   var ICON_WA = '<svg viewBox="0 0 24 24"><path d="M17.472 14.382c-.297-.149-1.758-.867-2.03-.967-.273-.099-.471-.148-.67.15-.197.297-.767.966-.94 1.164-.173.199-.347.223-.644.075-.297-.15-1.255-.463-2.39-1.475-.883-.788-1.48-1.761-1.653-2.059-.173-.297-.018-.458.13-.606.134-.133.297-.347.446-.52.149-.174.198-.298.298-.497.099-.198.05-.371-.025-.52-.075-.149-.669-1.612-.916-2.207-.242-.579-.487-.5-.669-.51-.173-.008-.371-.01-.57-.01-.198 0-.52.074-.792.372-.272.297-1.04 1.016-1.04 2.479 0 1.462 1.065 2.875 1.213 3.074.149.198 2.096 3.2 5.077 4.487.709.306 1.262.489 1.694.625.712.227 1.36.195 1.871.118.571-.085 1.758-.719 2.006-1.413.248-.694.248-1.289.173-1.413-.074-.124-.272-.198-.57-.347m-5.421 7.403h-.004a9.87 9.87 0 01-5.031-1.378l-.361-.214-3.741.982.998-3.648-.235-.374a9.86 9.86 0 01-1.51-5.26c.001-5.45 4.436-9.884 9.888-9.884 2.64 0 5.122 1.03 6.988 2.898a9.825 9.825 0 012.893 6.994c-.003 5.45-4.437 9.884-9.885 9.884m8.413-18.297A11.815 11.815 0 0012.05 0C5.495 0 .16 5.335.157 11.892c0 2.096.547 4.142 1.588 5.945L.057 24l6.305-1.654a11.882 11.882 0 005.683 1.448h.005c6.554 0 11.89-5.335 11.893-11.893a11.821 11.821 0 00-3.48-8.413z"/></svg>';
 
+  // Speaker on / speaker muted (Material-style paths) for the sound toggle.
+  var ICON_SND_ON = '<svg viewBox="0 0 24 24"><path d="M3 9v6h4l5 5V4L7 9H3zm13.5 3c0-1.77-1.02-3.29-2.5-4.03v8.05c1.48-.73 2.5-2.25 2.5-4.02zM14 3.23v2.06c2.89.86 5 3.54 5 6.71s-2.11 5.85-5 6.71v2.06c4.01-.91 7-4.49 7-8.77s-2.99-7.86-7-8.77z"/></svg>';
+  var ICON_SND_OFF = '<svg viewBox="0 0 24 24"><path d="M16.5 12c0-1.77-1.02-3.29-2.5-4.03v2.21l2.45 2.45c.03-.2.05-.41.05-.63zm2.5 0c0 .94-.2 1.82-.54 2.64l1.51 1.51C20.63 14.91 21 13.5 21 12c0-4.28-2.99-7.86-7-8.77v2.06c2.89.86 5 3.54 5 6.71zM4.27 3L3 4.27 7.73 9H3v6h4l5 5v-6.73l4.25 4.25c-.67.52-1.42.93-2.25 1.18v2.06c1.38-.31 2.63-.95 3.69-1.81L19.73 21 21 19.73l-9-9L4.27 3zM12 4L9.91 6.09 12 8.18V4z"/></svg>';
+
+  // Paperclip (attach a PDF/photo to the conversation).
+  var ICON_CLIP = '<svg viewBox="0 0 24 24"><path d="M16.5 6v11.5c0 2.21-1.79 4-4 4s-4-1.79-4-4V5c0-1.38 1.12-2.5 2.5-2.5s2.5 1.12 2.5 2.5v10.5c0 .55-.45 1-1 1s-1-.45-1-1V6H10v9.5c0 1.38 1.12 2.5 2.5 2.5s2.5-1.12 2.5-2.5V5c0-2.21-1.79-4-4-4S7 2.79 7 5v12.5c0 3.04 2.46 5.5 5.5 5.5s5.5-2.46 5.5-5.5V6h-1.5z"/></svg>';
+
   function fillAvatar(el) {
     if (cfg.avatar_url) {
       el.style.background = shade(cfg.brand_color, 40) + " url('" + cfg.avatar_url + "') center/cover no-repeat";
@@ -136,6 +167,10 @@
       "#ssb-bubble{position:fixed;bottom:92px;right:20px;max-width:230px;background:#fff;color:#1f2937;border-radius:14px;padding:12px 14px;font-family:system-ui,-apple-system,Segoe UI,sans-serif;font-size:14px;line-height:1.35;box-shadow:0 10px 30px rgba(0,0,0,.22);cursor:pointer;z-index:99998;animation:ssbpop .3s ease}" +
       "@keyframes ssbpop{from{opacity:0;transform:translateY(8px)}to{opacity:1;transform:none}}" +
       "#ssb-head .ssb-x{margin-left:auto;background:transparent;border:0;color:#fff;font-size:24px;cursor:pointer;opacity:.85;line-height:1}" +
+      "#ssb-mute{margin-left:auto;width:32px;height:32px;background:transparent;border:0;cursor:pointer;opacity:.85;display:flex;align-items:center;justify-content:center;padding:0;flex:none}" +
+      "#ssb-mute:hover{opacity:1}#ssb-mute svg{width:20px;height:20px;fill:#fff}" +
+      "#ssb-head .ssb-x{margin-left:6px}" +
+      "#ssb-launch .ssb-badge{position:absolute;top:-4px;right:-4px;min-width:20px;height:20px;box-sizing:border-box;padding:0 5px;border-radius:10px;background:#e11d48;border:2px solid #fff;color:#fff;font-family:system-ui,-apple-system,Segoe UI,sans-serif;font-size:11px;font-weight:700;line-height:16px;text-align:center;display:none}" +
       "#ssb-msgs{flex:1;overflow-y:auto;padding:16px 14px;background:#f4f5f7;display:flex;flex-direction:column;gap:10px}" +
       ".ssb-row{display:flex;align-items:flex-end;gap:8px;max-width:88%}" +
       ".ssb-row.user{align-self:flex-end;flex-direction:row-reverse}.ssb-row.bot{align-self:flex-start}" +
@@ -147,8 +182,18 @@
       ".ssb-b ul{margin:4px 0;padding-left:18px}.ssb-b li{margin:2px 0}.ssb-b code{background:#eee;padding:1px 4px;border-radius:4px;font-size:13px}" +
       ".ssb-typing{display:flex;gap:4px;padding:12px 14px}.ssb-typing span{width:7px;height:7px;border-radius:50%;background:#bbb;animation:ssbbounce 1.2s infinite}" +
       ".ssb-typing span:nth-child(2){animation-delay:.2s}.ssb-typing span:nth-child(3){animation-delay:.4s}" +
+      ".ssb-b.ssb-stage{color:#8b8b94;font-style:italic}" +
       "@keyframes ssbbounce{0%,60%,100%{transform:translateY(0);opacity:.5}30%{transform:translateY(-5px);opacity:1}}" +
       "#ssb-form{display:flex;align-items:center;gap:8px;padding:10px;border-top:1px solid #ececef;background:#fff}" +
+      "#ssb-att{display:none;flex-wrap:wrap;gap:6px;padding:8px 12px;border-top:1px solid #ececef;background:#fff}" +
+      ".ssb-att-chip{display:inline-flex;align-items:center;gap:6px;max-width:100%;background:#f1f5f9;border:1px solid #e2e8f0;border-radius:14px;padding:4px 10px;font-family:system-ui,-apple-system,Segoe UI,sans-serif;font-size:12px;color:#334155}" +
+      ".ssb-att-chip .ssb-att-name{overflow:hidden;text-overflow:ellipsis;white-space:nowrap;max-width:200px}" +
+      ".ssb-att-chip button{border:0;background:transparent;cursor:pointer;color:#64748b;font-size:15px;line-height:1;padding:0;flex:none}" +
+      ".ssb-att-line{font-size:12px;font-style:italic;opacity:.85}" +
+      "#ssb-attach{width:38px;height:38px;border-radius:50%;border:0;background:transparent;cursor:pointer;display:flex;align-items:center;justify-content:center;flex:none;padding:0}" +
+      "#ssb-attach svg{width:20px;height:20px;fill:#9aa0a6}#ssb-attach:hover svg{fill:" + c + "}" +
+      "#ssb-file{display:none}" +
+      ".ssb-toast{position:absolute;left:12px;right:12px;bottom:78px;background:#1f2937;color:#fff;border-radius:10px;padding:10px 12px;font-family:system-ui,-apple-system,Segoe UI,sans-serif;font-size:13px;text-align:center;box-shadow:0 6px 18px rgba(0,0,0,.25);z-index:2;animation:ssbpop .25s ease}" +
       "#ssb-input{flex:1;border:1px solid #e0e0e4;border-radius:22px;padding:11px 15px;font-size:14px;outline:none}" +
       "#ssb-input:focus{border-color:" + c + "}" +
       "#ssb-send{width:42px;height:42px;border-radius:50%;border:0;background:" + c + ";color:#fff;cursor:pointer;display:flex;align-items:center;justify-content:center;flex:none}#ssb-send svg{width:19px;height:19px;fill:#fff}" +
@@ -208,7 +253,7 @@
 
     var launch = document.createElement("button");
     launch.id = "ssb-launch"; launch.setAttribute("aria-label", T.open);
-    launch.innerHTML = '<span class="ssb-pulse"></span>' + ICON_AGENT + '<span class="ssb-dot"></span>';
+    launch.innerHTML = '<span class="ssb-pulse"></span>' + ICON_AGENT + '<span class="ssb-dot"></span><span class="ssb-badge"></span>';
 
     var panel = document.createElement("div");
     panel.id = "ssb-panel";
@@ -216,9 +261,14 @@
       '<div id="ssb-head"><div class="ssb-av"></div><div>' +
       '<div class="ssb-name">' + esc(name) + '</div>' +
       '<div class="ssb-status"><i></i>' + T.status + '</div></div>' +
+      '<button id="ssb-mute" type="button"></button>' +
       '<button class="ssb-x" aria-label="' + T.close + '">&times;</button></div>' +
       '<div id="ssb-msgs"></div>' +
-      '<form id="ssb-form"><input id="ssb-input" type="text" autocomplete="off" placeholder="' + T.input + '" />' +
+      '<div id="ssb-att"></div>' +
+      '<form id="ssb-form">' +
+      '<button id="ssb-attach" type="button" aria-label="' + T.attach + '" title="' + T.attach + '">' + ICON_CLIP + '</button>' +
+      '<input id="ssb-file" type="file" accept="image/*,application/pdf" />' +
+      '<input id="ssb-input" type="text" autocomplete="off" placeholder="' + T.input + '" />' +
       '<button id="ssb-send" type="submit" aria-label="' + T.send + '">' + ICON_SEND + '</button></form>' +
       '<div id="ssb-foot">' + T.footer + '</div>';
 
@@ -231,6 +281,201 @@
     var input = panel.querySelector("#ssb-input");
     var greeted = false;
 
+    // ---- Visitor uploads (Phase 3): paperclip -> chip -> ids on next send ----
+    var attBox = panel.querySelector("#ssb-att");
+    var attachBtn = panel.querySelector("#ssb-attach");
+    var fileInput = panel.querySelector("#ssb-file");
+    var attachments = [];  // [{id, name}] uploaded + ready for the next message
+
+    function toast(msg) {
+      try {
+        var old = panel.querySelector(".ssb-toast"); if (old) old.remove();
+        var el = document.createElement("div"); el.className = "ssb-toast"; el.textContent = msg;
+        panel.appendChild(el);
+        setTimeout(function () { try { el.remove(); } catch (e) {} }, 3500);
+      } catch (e) {}
+    }
+    function paintAtt() {
+      try { attBox.style.display = attBox.children.length ? "flex" : "none"; } catch (e) {}
+    }
+    function addChip(label) {
+      var ch = document.createElement("span"); ch.className = "ssb-att-chip";
+      var nm = document.createElement("span"); nm.className = "ssb-att-name"; nm.textContent = label;
+      ch.appendChild(nm);
+      attBox.appendChild(ch); paintAtt();
+      return ch;
+    }
+    function chipReady(ch, id, name) {
+      ch.querySelector(".ssb-att-name").textContent = name;
+      var x = document.createElement("button"); x.type = "button";
+      x.textContent = "×"; x.setAttribute("aria-label", T.close);
+      x.addEventListener("click", function () {
+        for (var i = attachments.length - 1; i >= 0; i--) {
+          if (attachments[i].id === id) attachments.splice(i, 1);
+        }
+        try { ch.remove(); } catch (e) {}
+        paintAtt();
+      });
+      ch.appendChild(x);
+      attachments.push({ id: id, name: name });
+    }
+    function takeAttachments() {
+      var ids = [], names = [];
+      for (var i = 0; i < attachments.length; i++) { ids.push(attachments[i].id); names.push(attachments[i].name); }
+      attachments = [];
+      try { while (attBox.firstChild) attBox.removeChild(attBox.firstChild); } catch (e) {}
+      paintAtt();
+      return { ids: ids, names: names };
+    }
+    function uploadFile(f) {
+      if (!f || !window.FormData) return;
+      if (attachments.length >= 3) { toast(T.upFail); return; }
+      if (f.size > 5 * 1024 * 1024) { toast(T.tooBig); return; }
+      var chip = addChip(T.uploading);
+      var fd = new FormData();
+      fd.append("file", f, f.name || "archivo");
+      var sid = getSid(); if (sid) fd.append("session_id", sid);
+      fetch(BACKEND + "/chat/upload" + QS, { method: "POST", body: fd })
+        .then(function (r) { if (!r.ok) throw new Error("up"); return r.json(); })
+        .then(function (d) {
+          if (!d || !d.attachment_id) throw new Error("up");
+          // Adopt the server's session id: the next message must land on the
+          // SAME session the attachment is bound to.
+          setSid(d.session_id);
+          chipReady(chip, d.attachment_id, d.filename || f.name || "archivo");
+        })
+        .catch(function () {
+          try { chip.remove(); } catch (e) {}
+          paintAtt();
+          toast(T.upFail);
+        });
+    }
+    if (attachBtn) attachBtn.addEventListener("click", function () {
+      try { fileInput.click(); } catch (e) {}
+    });
+    if (fileInput) fileInput.addEventListener("change", function () {
+      try {
+        var f = fileInput.files && fileInput.files[0];
+        fileInput.value = "";
+        uploadFile(f);
+      } catch (e) {}
+    });
+
+    // ---- Notifications: subtle pop sound + unread badge + tab title flash ----
+    var soundOn = true;
+    try { soundOn = localStorage.getItem("ssb_sound") !== "off"; } catch (e) {}
+
+    // Autoplay policy: only create/resume the AudioContext after the first
+    // real user gesture on the page; the one-time listeners remove themselves.
+    var audioCtx = null;
+    var GESTURE_OPTS = { passive: true };
+    function armAudio() {
+      try {
+        document.removeEventListener("pointerdown", armAudio, GESTURE_OPTS);
+        document.removeEventListener("keydown", armAudio, GESTURE_OPTS);
+        var AC = window.AudioContext || window.webkitAudioContext;
+        if (!AC) return;
+        if (!audioCtx) audioCtx = new AC();
+        if (audioCtx.state === "suspended") {
+          var p = audioCtx.resume();
+          if (p && p.catch) p.catch(function () {});
+        }
+      } catch (e) {}
+    }
+    try {
+      document.addEventListener("pointerdown", armAudio, GESTURE_OPTS);
+      document.addEventListener("keydown", armAudio, GESTURE_OPTS);
+    } catch (e) {}
+
+    // Short quiet "pop": sine sweep with a fast gain envelope (~120ms).
+    function playPop() {
+      try {
+        if (!soundOn || !audioCtx || audioCtx.state !== "running") return;
+        var t0 = audioCtx.currentTime;
+        var osc = audioCtx.createOscillator();
+        var gain = audioCtx.createGain();
+        osc.type = "sine";
+        osc.frequency.setValueAtTime(740, t0);
+        osc.frequency.exponentialRampToValueAtTime(420, t0 + 0.11);
+        gain.gain.setValueAtTime(0.0001, t0);
+        gain.gain.exponentialRampToValueAtTime(0.07, t0 + 0.02);
+        gain.gain.exponentialRampToValueAtTime(0.0001, t0 + 0.12);
+        osc.connect(gain); gain.connect(audioCtx.destination);
+        osc.start(t0); osc.stop(t0 + 0.13);
+      } catch (e) {}
+    }
+
+    var muteBtn = panel.querySelector("#ssb-mute");
+    function paintMute() {
+      try {
+        if (!muteBtn) return;
+        muteBtn.innerHTML = soundOn ? ICON_SND_ON : ICON_SND_OFF;
+        muteBtn.setAttribute("aria-label", soundOn ? T.mute : T.unmute);
+      } catch (e) {}
+    }
+    if (muteBtn) muteBtn.addEventListener("click", function () {
+      soundOn = !soundOn;
+      try { localStorage.setItem("ssb_sound", soundOn ? "on" : "off"); } catch (e) {}
+      paintMute();
+    });
+    paintMute();
+
+    // Unread badge on the launcher: bot messages that arrive while closed.
+    var unread = 0;
+    var badgeEl = launch.querySelector(".ssb-badge");
+    function setBadge(n) {
+      unread = n;
+      try {
+        if (!badgeEl) return;
+        if (n > 0) { badgeEl.textContent = n > 9 ? "9+" : String(n); badgeEl.style.display = "block"; }
+        else badgeEl.style.display = "none";
+      } catch (e) {}
+    }
+
+    // Tab title flash while the document is hidden. The original title is
+    // re-read each time flashing starts (so merchant SPA title changes are
+    // respected) and restored EXACTLY when flashing stops.
+    var hiddenN = 0, flashTimer = null, flashOrig = null;
+    function startFlash() {
+      try {
+        if (flashTimer) return;
+        flashOrig = document.title;
+        var shown = false;
+        var swap = function () {
+          try {
+            shown = !shown;
+            document.title = shown ? "(" + hiddenN + ") " + T.newMsg : flashOrig;
+          } catch (e) {}
+        };
+        swap();
+        flashTimer = setInterval(swap, 1500);
+      } catch (e) {}
+    }
+    function stopFlash() {
+      try {
+        hiddenN = 0;
+        if (!flashTimer) return;
+        clearInterval(flashTimer); flashTimer = null;
+        if (flashOrig !== null) document.title = flashOrig;
+        flashOrig = null;
+      } catch (e) {}
+    }
+    try {
+      document.addEventListener("visibilitychange", function () { if (!document.hidden) stopFlash(); });
+      window.addEventListener("focus", stopFlash);
+    } catch (e) {}
+
+    // A bot message landed: badge / title / sound only when not being watched.
+    function notify() {
+      try {
+        var closed = !panel.classList.contains("open");
+        var hidden = !!document.hidden;
+        if (closed) setBadge(unread + 1);
+        if (hidden) { hiddenN++; startFlash(); }
+        if (closed || hidden) playPop();
+      } catch (e) {}
+    }
+
     function row(who) {
       var r = document.createElement("div"); r.className = "ssb-row " + who;
       if (who === "bot") { var a = document.createElement("div"); a.className = "ssb-av"; fillAvatar(a); r.appendChild(a); }
@@ -255,7 +500,21 @@
       });
       msgs.appendChild(wrap); msgs.scrollTop = msgs.scrollHeight;
     }
-    function userMsg(t) { row("user").textContent = t; msgs.scrollTop = msgs.scrollHeight; }
+    function userMsg(t, attNames) {
+      var b = row("user");
+      if (attNames && attNames.length) {
+        for (var i = 0; i < attNames.length; i++) {
+          var ln = document.createElement("div"); ln.className = "ssb-att-line";
+          ln.textContent = "[" + T.attached + ": " + attNames[i] + "]";
+          b.appendChild(ln);
+        }
+        var tx = document.createElement("div"); tx.textContent = t;
+        b.appendChild(tx);
+      } else {
+        b.textContent = t;
+      }
+      msgs.scrollTop = msgs.scrollHeight;
+    }
     function typing() { var b = row("bot"); b.classList.add("ssb-typing"); b.innerHTML = "<span></span><span></span><span></span>"; msgs.scrollTop = msgs.scrollHeight; return b; }
 
     // Suggested starter chips — per tenant (a services store needs different
@@ -278,6 +537,7 @@
       var bubble = document.getElementById("ssb-bubble"); if (bubble) bubble.remove();
       panel.classList.toggle("open");
       if (panel.classList.contains("open")) {
+        setBadge(0); stopFlash();
         if (!greeted) { botMsg(cfg.welcome); renderStarters(); greeted = true; }
         input.focus();
       }
@@ -285,22 +545,162 @@
     launch.addEventListener("click", toggle);
     panel.querySelector(".ssb-x").addEventListener("click", toggle);
 
-    function sendText(text) {
-      text = (text || "").trim(); if (!text) return;
-      userMsg(text); input.value = "";
-      var t = typing();
-      fetch(BACKEND + "/chat" + QS, {
-        method: "POST", headers: { "Content-Type": "application/json" },
-        body: JSON.stringify({ message: text, session_id: getSid() })
-      }).then(function (r) { return r.json(); }).then(function (d) {
-        setSid(d.session_id); t.classList.remove("ssb-typing"); t.innerHTML = md(d.reply || "…");
+    // ---- Perceived speed: live stage line while the bot works (Phase 2) ----
+    function stageLabel(key) {
+      if (key === "thinking") return T.stThinking;
+      if (key.indexOf("tool:") === 0) {
+        var tool = key.slice(5);
+        if (tool === "search_products") return T.stCatalog;
+        if (tool === "search_knowledge") return T.stDocs;
+        if (tool.indexOf("order") !== -1 || tool === "change_shipping_address" || tool === "start_return") return T.stOrder;
+        return T.stWait;
+      }
+      return T.stWait;
+    }
+    function showStage(t, key) {
+      try {
+        t.classList.remove("ssb-typing");
+        t.classList.add("ssb-stage");
+        t.textContent = stageLabel(key);
+        msgs.scrollTop = msgs.scrollHeight;
+      } catch (e) {}
+    }
+    function showTypingDots(t) {
+      t.classList.remove("ssb-stage");
+      t.classList.add("ssb-typing");
+      t.innerHTML = "<span></span><span></span><span></span>";
+    }
+
+    // Typewriter render (~4 chars per animation frame). Instant-finish on a
+    // user click/scroll gesture or prefers-reduced-motion; always ends with
+    // the full markdown-rendered reply (the typewriter is presentation only).
+    function typewrite(el, text, done) {
+      var finished = false, i = 0;
+      function finish() {
+        if (finished) return; finished = true;
+        try {
+          document.removeEventListener("pointerdown", finish, GESTURE_OPTS);
+          msgs.removeEventListener("wheel", finish, GESTURE_OPTS);
+          msgs.removeEventListener("touchmove", finish, GESTURE_OPTS);
+        } catch (e) {}
+        try { el.innerHTML = md(text); } catch (e) { el.textContent = text; }
+        msgs.scrollTop = msgs.scrollHeight;
+        if (done) done();
+      }
+      var reduce = false;
+      try { reduce = window.matchMedia && window.matchMedia("(prefers-reduced-motion: reduce)").matches; } catch (e) {}
+      if (reduce || !window.requestAnimationFrame) { finish(); return; }
+      try {
+        document.addEventListener("pointerdown", finish, GESTURE_OPTS);
+        msgs.addEventListener("wheel", finish, GESTURE_OPTS);
+        msgs.addEventListener("touchmove", finish, GESTURE_OPTS);
+      } catch (e) {}
+      function step() {
+        if (finished) return;
+        i += 4;
+        if (i >= text.length) { finish(); return; }
+        el.textContent = text.slice(0, i);
+        msgs.scrollTop = msgs.scrollHeight;
+        window.requestAnimationFrame(step);
+      }
+      window.requestAnimationFrame(step);
+    }
+
+    // Single arrival point for a bot reply (streamed or classic): Phase 1
+    // notify() hook, typewriter text, then cards/handoff exactly as before.
+    function deliverReply(t, d) {
+      setSid(d.session_id);
+      t.classList.remove("ssb-typing"); t.classList.remove("ssb-stage");
+      notify();
+      typewrite(t, d.reply || "…", function () {
         renderCards(d.products);
         if (d.handoff) renderHandoffForm();
         msgs.scrollTop = msgs.scrollHeight;
+      });
+    }
+
+    // Classic non-streaming path: the source of truth for correctness, and
+    // the transparent fallback whenever streaming is unavailable or breaks.
+    function sendClassic(text, t, attIds) {
+      fetch(BACKEND + "/chat" + QS, {
+        method: "POST", headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ message: text, session_id: getSid(), attachment_ids: attIds || [] })
+      }).then(function (r) { return r.json(); }).then(function (d) {
+        deliverReply(t, d);
       }).catch(function () {
-        t.classList.remove("ssb-typing"); t.textContent = T.error;
+        t.classList.remove("ssb-typing"); t.classList.remove("ssb-stage");
+        t.textContent = T.error;
+        notify();
       });
     }
+
+    // SSE-over-POST: consume /chat/stream with fetch + ReadableStream (not
+    // EventSource). Any problem before the "final" event retries ONCE via the
+    // classic /chat path, reusing the same bubble — never a duplicate message.
+    function sendStream(text, t, attIds) {
+      var supported = false;
+      try { supported = !!(window.fetch && window.ReadableStream && window.TextDecoder); } catch (e) {}
+      if (!supported) { sendClassic(text, t, attIds); return; }
+      var settled = false;
+      function fallback() {
+        if (settled) return;
+        settled = true;
+        showTypingDots(t);
+        sendClassic(text, t, attIds);
+      }
+      fetch(BACKEND + "/chat/stream" + QS, {
+        method: "POST", headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ message: text, session_id: getSid(), attachment_ids: attIds || [] })
+      }).then(function (r) {
+        if (!r.ok || !r.body || !r.body.getReader) throw new Error("no-stream");
+        var reader = r.body.getReader();
+        var dec = new TextDecoder();
+        var buf = "";
+        function frame(raw) {
+          if (settled) return;
+          var ev = "", data = "", lines = raw.split("\n");
+          for (var i = 0; i < lines.length; i++) {
+            var ln = lines[i].replace(/\r$/, "");
+            if (ln.indexOf("event:") === 0) ev = ln.slice(6).replace(/^\s+/, "");
+            else if (ln.indexOf("data:") === 0) data += ln.slice(5).replace(/^\s+/, "");
+          }
+          if (ev === "stage") {
+            var key = "";
+            try { key = JSON.parse(data).key || ""; } catch (e) {}
+            showStage(t, key);
+          } else if (ev === "final") {
+            var d = JSON.parse(data);  // a broken payload throws -> fallback
+            settled = true;
+            deliverReply(t, d);
+            try { reader.cancel(); } catch (e) {}
+          } else if (ev === "error") {
+            throw new Error("stream-error");
+          }
+        }
+        function pump() {
+          return reader.read().then(function (res) {
+            if (settled) return;
+            if (res.done) throw new Error("eof-before-final");
+            buf += dec.decode(res.value, { stream: true });
+            var cut;
+            while (!settled && (cut = buf.indexOf("\n\n")) !== -1) {
+              var raw = buf.slice(0, cut); buf = buf.slice(cut + 2);
+              if (raw) frame(raw);
+            }
+            if (!settled) return pump();
+          });
+        }
+        return pump();
+      }).catch(fallback);
+    }
+
+    function sendText(text) {
+      text = (text || "").trim(); if (!text) return;
+      var att = takeAttachments();  // chips are consumed by THIS message
+      userMsg(text, att.names); input.value = "";
+      var t = typing();
+      sendStream(text, t, att.ids);
+    }
     form.addEventListener("submit", function (e) { e.preventDefault(); sendText(input.value); });
 
     // Proactive nudge: after a bit, invite the visitor (once) if chat unopened.
@@ -310,6 +710,7 @@
       b.textContent = T.bubble;
       b.addEventListener("click", toggle);
       document.body.appendChild(b);
+      notify();  // the proactive bubble counts as one unread message
       setTimeout(function () { var x = document.getElementById("ssb-bubble"); if (x) x.remove(); }, 18000);
     }, 22000);
 
@@ -338,8 +739,10 @@
         }).then(function (r) { return r.json(); }).then(function () {
           box.remove();
           botMsg(T.hfThanks);
+          notify();
         }).catch(function () {
           botMsg(T.hfError);
+          notify();
         });
       });
     }

app/tenancy.py

@@ -14,6 +14,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.crypto import decrypt, encrypt
 from app.models import (
+    Attachment,
     ChatMessage,
     ChatSession,
     Event,
@@ -133,6 +134,8 @@ async def purge_tenant(db: AsyncSession, tenant: Tenant) -> None:
             select(ChatSession.id).where(ChatSession.tenant_id == tenant.id)
         )
     ))
+    # Visitor uploads (already text-only) — covers admin delete AND demo expiry.
+    await db.execute(delete(Attachment).where(Attachment.tenant_id == tenant.id))
     await db.execute(delete(ChatSession).where(ChatSession.tenant_id == tenant.id))
     await db.execute(delete(HandoffRequest).where(HandoffRequest.tenant_id == tenant.id))
     await db.execute(delete(Event).where(Event.tenant_id == tenant.id))

app/vision.py

@@ -0,0 +1,85 @@
+"""Describe customer-uploaded photos with a free Groq vision model.
+
+The visitor's image binary is NEVER persisted: at upload time it is turned
+into a detailed Spanish text description here, and only that text is stored
+(scoped to one tenant + one chat session). Any failure — no key, network
+error, empty answer — returns None so the upload endpoint can degrade to a
+friendly placeholder instead of failing the request.
+"""
+
+from __future__ import annotations
+
+import base64
+import logging
+
+import httpx
+
+from app.config import get_settings
+
+log = logging.getLogger(__name__)
+
+VISION_MODEL = "meta-llama/llama-4-scout-17b-16e-instruct"
+VISION_TIMEOUT = 30.0
+VISION_MAX_TOKENS = 500
+_PROMPT = (
+    "Describe esta imagen en español de forma detallada y factual para el "
+    "asistente de una tienda: objetos que aparecen, texto visible, daños o "
+    "defectos, marca, modelo o referencias visibles. Describe SOLO lo que se "
+    "ve, sin inventar nada."
+)
+
+
+def _groq_keys() -> list[str]:
+    """Same key sources as the LLM router (app/llm/router.py): GROQ_API_KEY
+    plus GROQ_API_KEYS, de-duplicated, preserving order."""
+    settings = get_settings()
+    keys: list[str] = []
+    for key in [settings.groq_api_key, *settings.groq_api_keys]:
+        key = (key or "").strip()
+        if key and key not in keys:
+            keys.append(key)
+    return keys
+
+
+async def describe_image(data: bytes, mime: str) -> str | None:
+    """A detailed factual Spanish description of the image, or None on ANY
+    failure (the caller stores a placeholder and the bot apologizes naturally)."""
+    keys = _groq_keys()
+    if not keys:
+        return None
+    payload = {
+        "model": VISION_MODEL,
+        "temperature": 0.2,
+        "max_tokens": VISION_MAX_TOKENS,
+        "messages": [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": _PROMPT},
+                    {
+                        "type": "image_url",
+                        "image_url": {
+                            "url": "data:"
+                            + mime
+                            + ";base64,"
+                            + base64.b64encode(data).decode("ascii")
+                        },
+                    },
+                ],
+            }
+        ],
+    }
+    url = get_settings().groq_base_url.rstrip("/") + "/chat/completions"
+    for key in keys:  # one try per key: a rate-limited key fails over to the next
+        try:
+            async with httpx.AsyncClient(timeout=VISION_TIMEOUT) as client:
+                resp = await client.post(
+                    url, json=payload, headers={"Authorization": f"Bearer {key}"}
+                )
+                resp.raise_for_status()
+                content = (resp.json()["choices"][0]["message"]["content"] or "").strip()
+                if content:
+                    return content
+        except Exception:  # noqa: BLE001 - vision is best-effort by design
+            log.warning("vision describe_image attempt failed", exc_info=True)
+    return None

migrations/versions/0020_attachments.py

@@ -0,0 +1,45 @@
+"""attachments table: per-session visitor uploads converted to text
+
+Revision ID: 0020_attachments
+Revises: 0019_demo_tenants
+Create Date: 2026-06-11
+
+Additive, idempotent on Postgres (CREATE TABLE IF NOT EXISTS). On other
+dialects it is a no-op (tests/dev use create_all which already has the table).
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+revision = "0020_attachments"
+down_revision = "0019_demo_tenants"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    if op.get_bind().dialect.name != "postgresql":
+        return
+    op.execute(
+        """
+        CREATE TABLE IF NOT EXISTS attachments (
+            id VARCHAR(40) PRIMARY KEY,
+            tenant_id INTEGER REFERENCES tenants(id) ON DELETE CASCADE,
+            session_id VARCHAR(40),
+            filename VARCHAR(255),
+            mime VARCHAR(60),
+            text TEXT,
+            created_at TIMESTAMPTZ
+        )
+        """
+    )
+    op.execute("CREATE INDEX IF NOT EXISTS ix_attachments_tenant_id ON attachments (tenant_id)")
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_attachments_session_id ON attachments (session_id)"
+    )
+
+
+def downgrade() -> None:
+    if op.get_bind().dialect.name == "postgresql":
+        op.execute("DROP TABLE IF EXISTS attachments")

tests/routes/test_chat_stream.py

@@ -0,0 +1,250 @@
+"""POST /chat/stream — SSE progress events + final ChatResponse.
+
+The stream is sugar on top of the classic POST /chat: same tenant resolution,
+same kill-switch gates, and a 'final' event that carries the exact ChatResponse
+JSON. Every test proves a POSITIVE outcome first (the stream/classic path works
+through the very fake router under test) before asserting the refusal/fallback.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+import pytest
+from sqlalchemy import func, select
+
+from app.config import get_settings
+from app.deps import get_router
+from app.llm.base import ChatResult, ToolCall
+from app.models import ChatSession
+from app.orchestrator import run_turn
+from app.routes.chat import _stream_events
+from app.tools.registry import ToolContext
+
+
+class ScriptRouter:
+    """Plays a fixed script of ChatResults (same shape as other route tests)."""
+
+    def __init__(self, results):
+        self._results = list(results)
+        self.calls = 0
+
+    async def chat(self, messages, tools, tier="large", temperature=None, order=None, tool_choice=None, max_tokens=None):
+        self.calls += 1
+        return self._results.pop(0)
+
+
+class BoomRouter:
+    """Unexpected (non-ProviderError) crash inside the turn."""
+
+    async def chat(self, messages, tools, tier="large", temperature=None, order=None, tool_choice=None, max_tokens=None):
+        raise RuntimeError("boom")
+
+
+def _script():
+    """One knowledge lookup, then a text answer — identical for both endpoints."""
+    return [
+        ChatResult(
+            content=None,
+            tool_calls=[ToolCall(id="t1", name="search_knowledge", arguments={"query": "envíos"})],
+            finish_reason="tool_calls",
+        ),
+        ChatResult(content="Hacemos envíos gratis desde 199€.", tool_calls=[], finish_reason="stop"),
+    ]
+
+
+@pytest.fixture(autouse=True)
+def no_real_search(monkeypatch):
+    async def fake_search(session, query, k=4, *, tenant_id=None):
+        return []
+
+    from app.rag import index
+
+    monkeypatch.setattr(index, "search", fake_search)
+
+
+def _parse_sse(text: str) -> list[tuple[str, str]]:
+    """Parse standard SSE framing into (event, data) pairs; comments ignored."""
+    events = []
+    for raw in text.split("\n\n"):
+        ev, data = "", ""
+        for line in raw.split("\n"):
+            if line.startswith("event:"):
+                ev = line[6:].strip()
+            elif line.startswith("data:"):
+                data += line[5:].strip()
+        if ev:
+            events.append((ev, data))
+    return events
+
+
+async def _count_sessions(db) -> int:
+    return (await db.execute(select(func.count()).select_from(ChatSession))).scalar_one()
+
+
+async def test_stream_emits_stages_then_final_equal_to_classic_chat(app_client, tenant):
+    app, client = app_client
+
+    # classic path first: the source of truth for this exact conversation
+    app.dependency_overrides[get_router] = lambda: ScriptRouter(_script())
+    classic = await client.post("/chat?t=t1", json={"message": "hola"})
+    assert classic.status_code == 200
+    expected = classic.json()
+    assert expected["reply"] == "Hacemos envíos gratis desde 199€."
+    assert expected["used_tools"] == ["search_knowledge"]
+
+    # same fake-router conversation through the stream
+    app.dependency_overrides[get_router] = lambda: ScriptRouter(_script())
+    r = await client.post("/chat/stream?t=t1", json={"message": "hola"})
+    assert r.status_code == 200
+    assert r.headers["content-type"].startswith("text/event-stream")
+
+    events = _parse_sse(r.text)
+    names = [e for e, _ in events]
+    # real stage events, in execution order, before the final payload
+    stage_keys = [json.loads(d)["key"] for e, d in events if e == "stage"]
+    assert stage_keys == ["thinking", "tool:search_knowledge"]
+    assert names[-1] == "final"
+    assert "error" not in names
+
+    final = json.loads(events[-1][1])
+    # identical ChatResponse shape + identical conversation outcome
+    assert final["reply"] == expected["reply"]
+    assert final["used_tools"] == expected["used_tools"]
+    assert final["products"] == expected["products"]
+    assert final["handoff"] == expected["handoff"]
+    assert final["lang"] == expected["lang"]
+    assert final["provider"] == expected["provider"]
+    assert final["session_id"]  # its own (new) session, but a real one
+
+
+async def test_stream_persists_session_like_classic(app_client, db_session, tenant):
+    app, client = app_client
+    app.dependency_overrides[get_router] = lambda: ScriptRouter(_script())
+
+    before = await _count_sessions(db_session)
+    r = await client.post("/chat/stream?t=t1", json={"message": "hola"})
+    assert r.status_code == 200
+    final = json.loads(_parse_sse(r.text)[-1][1])
+
+    assert await _count_sessions(db_session) == before + 1
+    session = await db_session.get(ChatSession, final["session_id"])
+    assert session is not None and session.tenant_id == tenant.id
+
+
+async def test_stream_unknown_slug_404_no_stream_no_llm(app_client, db_session, default_tenant):
+    """KILL-SWITCH: an explicit unknown slug must 404 — never fall back to the
+    'default' tenant (which exists here, so a fallback bug would pass a 200)."""
+    app, client = app_client
+    fake = ScriptRouter(_script())
+    app.dependency_overrides[get_router] = lambda: fake
+
+    r = await client.post("/chat/stream?t=ghost", json={"message": "hola"})
+    assert r.status_code == 404
+    assert "text/event-stream" not in r.headers.get("content-type", "")
+    assert fake.calls == 0
+    assert await _count_sessions(db_session) == 0
+
+
+async def test_stream_paused_tenant_403_before_any_llm_call(app_client, db_session, tenant):
+    app, client = app_client
+    fake = ScriptRouter(_script() + _script())
+    app.dependency_overrides[get_router] = lambda: fake
+
+    # alive: the stream works through this very fake router
+    ok = await client.post("/chat/stream?t=t1", json={"message": "hola"})
+    assert ok.status_code == 200
+    assert fake.calls == 2  # tool turn + answer turn
+
+    tenant.chatbot_enabled = False
+    await db_session.commit()
+
+    # paused: plain 403 (no SSE body), zero additional LLM calls or sessions
+    r = await client.post("/chat/stream?t=t1", json={"message": "hola"})
+    assert r.status_code == 403
+    assert "text/event-stream" not in r.headers.get("content-type", "")
+    assert fake.calls == 2
+    assert await _count_sessions(db_session) == 1
+
+
+async def test_stream_error_inside_run_turn_emits_error_event(app_client, tenant):
+    """An unexpected crash mid-turn must end the stream with an 'error' event
+    (the widget falls back to POST /chat) — not a hang and not a 'final'."""
+    app, client = app_client
+    app.dependency_overrides[get_router] = lambda: BoomRouter()
+
+    r = await client.post("/chat/stream?t=t1", json={"message": "hola"})
+    assert r.status_code == 200
+    events = _parse_sse(r.text)
+    names = [e for e, _ in events]
+    assert names[0] == "stage"  # it really started working ("thinking")
+    assert "error" in names
+    assert "final" not in names
+
+
+async def test_stream_client_disconnect_cancels_turn_no_orphan(db_session, tenant):
+    """If the visitor closes the tab mid-turn, Starlette closes the SSE
+    generator: that close must CANCEL the in-flight run_turn task (no orphaned
+    LLM work burning quota) and must complete without deadlocking on the queue."""
+
+    class HangRouter:
+        """Blocks forever inside the turn; records whether it was cancelled."""
+
+        def __init__(self):
+            self.started = asyncio.Event()
+            self.cancelled = False
+
+        async def chat(self, messages, tools, tier="large", temperature=None, order=None, tool_choice=None, max_tokens=None):
+            self.started.set()
+            try:
+                await asyncio.Event().wait()  # never set: hangs until cancelled
+            except asyncio.CancelledError:
+                self.cancelled = True
+                raise
+
+    router = HangRouter()
+    agen = _stream_events(
+        db=db_session, router_=router, settings=get_settings(),
+        tenant=tenant, message="hola", session_id=None,
+    )
+    first = await asyncio.wait_for(agen.__anext__(), timeout=5)
+    assert "thinking" in first  # the turn really started and streamed a stage
+    await asyncio.wait_for(router.started.wait(), timeout=5)  # LLM in flight
+
+    await asyncio.wait_for(agen.aclose(), timeout=5)  # client disconnects
+
+    assert router.cancelled is True  # the LLM task was cancelled, not orphaned
+
+
+async def test_run_turn_emits_stages_in_order(db_session):
+    router = ScriptRouter(_script())
+    s = ChatSession(shop="x")
+    db_session.add(s)
+    await db_session.flush()
+    stages: list[str] = []
+
+    ctx = ToolContext(db=db_session, session=s)
+    resp = await run_turn(router, ctx, "hola", on_stage=stages.append)
+
+    assert resp.reply == "Hacemos envíos gratis desde 199€."
+    assert stages == ["thinking", "tool:search_knowledge"]
+
+
+async def test_run_turn_raising_callback_does_not_break_reply(db_session):
+    router = ScriptRouter(_script())
+    s = ChatSession(shop="x")
+    db_session.add(s)
+    await db_session.flush()
+    seen: list[str] = []
+
+    def bad_callback(key: str) -> None:
+        seen.append(key)  # prove it WAS invoked for every stage...
+        raise RuntimeError("ui hook exploded")  # ...and always blew up
+
+    ctx = ToolContext(db=db_session, session=s)
+    resp = await run_turn(router, ctx, "hola", on_stage=bad_callback)
+
+    assert resp.reply == "Hacemos envíos gratis desde 199€."
+    assert resp.used_tools == ["search_knowledge"]
+    assert seen == ["thinking", "tool:search_knowledge"]

tests/routes/test_demo.py

@@ -183,6 +183,73 @@ async def test_per_ip_rate_limit_blocks_fourth_start(app_client):
     assert other.status_code == 200
 
 
+async def test_rejected_start_does_not_consume_ip_quota(app_client):
+    """Rejections are free: 3 invalid/SSRF attempts from one IP must not burn
+    the visitor's 3/hour budget — three VALID starts still succeed after them,
+    and only then does the cap kick in."""
+    _app, client = app_client
+    ip = "198.51.100.30"
+    for _ in range(3):
+        rejected = await _start(client, url="http://localhost:8000/admin", ip=ip)
+        assert rejected.status_code == 422
+    for _ in range(3):
+        ok = await _start(client, ip=ip)
+        assert ok.status_code == 200  # the full quota was still available
+    fourth = await _start(client, ip=ip)
+    assert fourth.status_code == 429  # successful starts stay capped at 3
+    assert "hora" in fourth.json()["detail"]
+
+
+async def test_admin_token_bypasses_per_ip_daily_and_active_caps(
+    app_client, db_session, monkeypatch
+):
+    """X-Admin-Token == settings.admin_token skips every cap (founder testing
+    from his own IP) and consumes no visitor quota."""
+    from app.ratelimit import RateLimiter
+    from app.store import get_config, upsert_config
+
+    _app, client = app_client
+    # Every cap at "full": zero per-IP slots, daily counter at the cap, and an
+    # active-demo ceiling of zero.
+    monkeypatch.setattr(demo, "_ip_limiter", RateLimiter(0, 3600))
+    monkeypatch.setattr(demo, "DEMO_DAILY_CAP", 5)
+    monkeypatch.setattr(demo, "DEMO_MAX_ACTIVE", 0)
+    today = datetime.now(UTC).date().isoformat()
+    await upsert_config(db_session, "demo_daily", {"date": today, "count": 5})
+    await db_session.commit()
+
+    # Positive control: a normal visitor is blocked outright.
+    blocked = await _start(client, ip="198.51.100.40")
+    assert blocked.status_code == 429
+
+    r = await client.post(
+        "/demo/start",
+        json={"url": "https://mitienda.com", "email": ""},
+        headers={**_xff("198.51.100.40"), "X-Admin-Token": "test-admin-token"},
+    )
+    assert r.status_code == 200
+    slug = r.json()["slug"]
+    await demo._drain_bg()
+    status = (await client.get(f"/demo/status?d={slug}")).json()
+    assert status["stage"] == "listo"  # a real, fully indexed demo
+    # The founder's test did not burn the visitors' daily budget either.
+    assert await get_config(db_session, "demo_daily") == {"date": today, "count": 5}
+
+
+async def test_wrong_admin_token_does_not_bypass(app_client, monkeypatch):
+    from app.ratelimit import RateLimiter
+
+    _app, client = app_client
+    monkeypatch.setattr(demo, "_ip_limiter", RateLimiter(0, 3600))
+    r = await client.post(
+        "/demo/start",
+        json={"url": "https://mitienda.com", "email": ""},
+        headers={**_xff("198.51.100.50"), "X-Admin-Token": "wrong-token"},
+    )
+    assert r.status_code == 429
+    assert "hora" in r.json()["detail"]
+
+
 async def test_global_daily_cap(app_client, db_session):
     from app.store import get_config, upsert_config
 

tests/routes/test_killswitch.py

@@ -120,6 +120,115 @@ async def test_apps_chat_deleted_slug_404_even_with_valid_signature(app_client,
     assert await _count(db_session, ChatSession) == 0
 
 
+async def test_chat_stream_deleted_slug_404_no_stream_no_llm_call(app_client, db_session):
+    """/chat/stream shares _resolve_tenant: a deleted slug is a plain 404
+    BEFORE any SSE stream starts — never a fallback to 'default'."""
+    app, client = app_client
+    fake = RecordingRouter()
+    app.dependency_overrides[get_router] = lambda: fake
+    await _setup_default_and(client, "borrada")
+
+    # alive: the stream really completes through this very fake router
+    ok = await client.post("/chat/stream?t=borrada", json={"message": "hola"})
+    assert ok.status_code == 200
+    assert "event: final" in ok.text
+    assert len(fake.calls) == 1
+
+    await _delete(client, "borrada")
+
+    r = await client.post("/chat/stream?t=borrada", json={"message": "hola"})
+    assert r.status_code == 404
+    assert "text/event-stream" not in r.headers.get("content-type", "")
+    assert len(fake.calls) == 1  # the LLM was never invoked again
+    assert await _count(db_session, ChatSession) == 0
+
+
+async def test_chat_stream_paused_tenant_403_no_llm_call(app_client, db_session):
+    app, client = app_client
+    fake = RecordingRouter()
+    app.dependency_overrides[get_router] = lambda: fake
+    r = await client.post("/admin/tenants", headers=AUTH, json={"slug": "morosa"})
+    assert r.status_code == 201
+
+    # enabled: the stream works
+    ok = await client.post("/chat/stream?t=morosa", json={"message": "hola"})
+    assert ok.status_code == 200 and "event: final" in ok.text
+    assert len(fake.calls) == 1
+
+    r = await client.put(
+        "/admin/tenants/morosa", headers=AUTH, json={"chatbot_enabled": False}
+    )
+    assert r.status_code == 200 and r.json()["chatbot_enabled"] is False
+
+    # paused: plain 403, no SSE body, zero additional LLM calls
+    r = await client.post("/chat/stream?t=morosa", json={"message": "hola"})
+    assert r.status_code == 403
+    assert "text/event-stream" not in r.headers.get("content-type", "")
+    assert len(fake.calls) == 1
+
+
+def _pdf() -> bytes:
+    from fpdf import FPDF
+
+    pdf = FPDF()
+    pdf.add_page()
+    pdf.set_font("helvetica", size=12)
+    pdf.cell(0, 10, text="documento del visitante")
+    return bytes(pdf.output())
+
+
+async def test_chat_upload_deleted_slug_404_and_attachments_wiped(app_client, db_session):
+    """/chat/upload shares the same gates AND a deleted tenant's visitor
+    uploads are wiped with the rest of its data (admin delete -> purge)."""
+    from app.models import Attachment
+
+    _app, client = app_client
+    await _setup_default_and(client, "borrada")
+
+    # alive: the upload really lands
+    ok = await client.post(
+        "/chat/upload?t=borrada", files={"file": ("doc.pdf", _pdf(), "application/pdf")}
+    )
+    assert ok.status_code == 200
+    assert await _count(db_session, Attachment) == 1
+
+    await _delete(client, "borrada")
+    assert await _count(db_session, Attachment) == 0  # died with the tenant
+
+    # deleted: 404 (NOT a fallback to 'default'), nothing stored anywhere
+    r = await client.post(
+        "/chat/upload?t=borrada", files={"file": ("doc.pdf", _pdf(), "application/pdf")}
+    )
+    assert r.status_code == 404
+    assert await _count(db_session, Attachment) == 0
+    assert await _count(db_session, ChatSession) == 0
+
+
+async def test_chat_upload_paused_tenant_403_nothing_stored(app_client, db_session):
+    from app.models import Attachment
+
+    _app, client = app_client
+    r = await client.post("/admin/tenants", headers=AUTH, json={"slug": "morosa"})
+    assert r.status_code == 201
+
+    ok = await client.post(
+        "/chat/upload?t=morosa", files={"file": ("doc.pdf", _pdf(), "application/pdf")}
+    )
+    assert ok.status_code == 200  # enabled: it works
+    assert await _count(db_session, Attachment) == 1
+
+    r = await client.put(
+        "/admin/tenants/morosa", headers=AUTH, json={"chatbot_enabled": False}
+    )
+    assert r.status_code == 200 and r.json()["chatbot_enabled"] is False
+
+    r = await client.post(
+        "/chat/upload?t=morosa", files={"file": ("doc2.pdf", _pdf(), "application/pdf")}
+    )
+    assert r.status_code == 403
+    assert await _count(db_session, Attachment) == 1  # only the pre-pause one
+
+
 async def test_handoff_deleted_slug_404_zero_rows(app_client, db_session):
     _app, client = app_client
     await _setup_default_and(client, "borrada")

tests/routes/test_uploads.py

@@ -0,0 +1,457 @@
+"""POST /chat/upload + attachment_ids on /chat — STRICT tenant+session isolation.
+
+The point of this battery is the founder's #1 requirement: customer data must
+never mix across tenants (or across conversations of the same tenant). Every
+test proves a POSITIVE outcome first — the upload/chat really works through the
+very fake router under test — before asserting the refusal or the silent drop.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+import pytest_asyncio
+from sqlalchemy import func, select
+
+from app.deps import get_router
+from app.llm.base import ChatResult
+from app.models import Attachment, ChatMessage, ChatSession, Tenant
+
+PDF_SECRET = "Factura 78901 garantia dos anios producto TOORX"
+VISION_DESC = "Una zapatilla deportiva roja talla 42 con la caja visiblemente danada"
+VISION_FALLBACK = "(imagen recibida pero no se ha podido analizar)"
+
+# Real magic bytes; only the prefix matters (vision is mocked in image tests).
+PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 64
+EXE_BYTES = b"MZ\x90\x00\x03\x00" + b"\x00" * 64
+
+
+class CaptureRouter:
+    """Replies with fixed text and records every prompt run_turn sends."""
+
+    def __init__(self, reply="Entendido, he leido el adjunto."):
+        self.reply = reply
+        self.prompts: list[list[dict]] = []
+
+    async def chat(self, messages, tools, tier="large", temperature=None, order=None, tool_choice=None, max_tokens=None):
+        self.prompts.append([dict(m) for m in messages])
+        return ChatResult(content=self.reply, tool_calls=[], finish_reason="stop")
+
+    def all_user_text(self) -> str:
+        return "\n".join(
+            m["content"] for p in self.prompts for m in p if m["role"] == "user"
+        )
+
+
+@pytest_asyncio.fixture
+async def tenant_b(db_session):
+    """A second committed tenant (slug 't2') for cross-tenant isolation tests."""
+    t = Tenant(slug="t2", name="Tienda 2", brand_name="Asistente T2")
+    db_session.add(t)
+    await db_session.commit()
+    return t
+
+
+def _pdf_bytes(text: str = PDF_SECRET) -> bytes:
+    from fpdf import FPDF
+
+    pdf = FPDF()
+    pdf.add_page()
+    pdf.set_font("helvetica", size=12)
+    pdf.cell(0, 10, text=text)
+    return bytes(pdf.output())
+
+
+async def _upload(client, slug, filename, content, mime="application/octet-stream", session_id=None):
+    data = {} if session_id is None else {"session_id": session_id}
+    return await client.post(
+        f"/chat/upload?t={slug}", files={"file": (filename, content, mime)}, data=data
+    )
+
+
+async def _count_attachments(db) -> int:
+    return (await db.execute(select(func.count()).select_from(Attachment))).scalar_one()
+
+
+# --- (a) happy path: PDF -> text -> injected into the LLM prompt -----------
+
+
+async def test_pdf_upload_extracts_text_and_binds_to_tenant_session(app_client, db_session, tenant):
+    app, client = app_client
+
+    r = await _upload(client, "t1", "manual.pdf", _pdf_bytes(), "application/pdf")
+    assert r.status_code == 200
+    up = r.json()
+    assert up["kind"] == "pdf"
+    assert up["filename"] == "manual.pdf"
+    assert "78901" in up["preview"]
+    assert up["session_id"]
+
+    att = await db_session.get(Attachment, up["attachment_id"])
+    assert att is not None
+    assert att.tenant_id == tenant.id
+    assert att.session_id == up["session_id"]
+    assert "78901" in att.text
+    assert att.mime == "application/pdf"
+
+    # the session really exists and belongs to the tenant (same one /chat uses)
+    session = await db_session.get(ChatSession, up["session_id"])
+    assert session is not None and session.tenant_id == tenant.id
+
+
+async def test_chat_with_attachment_id_injects_text_into_prompt(app_client, db_session, tenant):
+    app, client = app_client
+    fake = CaptureRouter()
+    app.dependency_overrides[get_router] = lambda: fake
+
+    up = (await _upload(client, "t1", "manual.pdf", _pdf_bytes(), "application/pdf")).json()
+    sid = up["session_id"]
+
+    r = await client.post(
+        "/chat?t=t1",
+        json={
+            "message": "que numero de factura aparece?",
+            "session_id": sid,
+            "attachment_ids": [up["attachment_id"]],
+        },
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["reply"] == fake.reply  # the turn really flowed
+    assert body["session_id"] == sid  # same session the attachment is bound to
+
+    prompt = fake.all_user_text()
+    assert "[El cliente ha adjuntado manual.pdf]" in prompt
+    assert "78901" in prompt
+    assert "que numero de factura aparece?" in prompt
+
+    # history replays coherently: the stored user message keeps the marker
+    msgs = (
+        await db_session.execute(
+            select(ChatMessage).where(
+                ChatMessage.session_id == sid, ChatMessage.role == "user"
+            )
+        )
+    ).scalars().all()
+    assert any("[El cliente ha adjuntado manual.pdf]" in m.content for m in msgs)
+    assert any("78901" in m.content for m in msgs)
+
+
+async def test_chat_stream_carries_attachment_too(app_client, tenant):
+    app, client = app_client
+    fake = CaptureRouter()
+    app.dependency_overrides[get_router] = lambda: fake
+
+    up = (await _upload(client, "t1", "manual.pdf", _pdf_bytes(), "application/pdf")).json()
+
+    r = await client.post(
+        "/chat/stream?t=t1",
+        json={
+            "message": "resume el adjunto",
+            "session_id": up["session_id"],
+            "attachment_ids": [up["attachment_id"]],
+        },
+    )
+    assert r.status_code == 200
+    assert "event: final" in r.text  # the stream really completed
+    assert fake.reply in r.text
+    prompt = fake.all_user_text()
+    assert "[El cliente ha adjuntado manual.pdf]" in prompt
+    assert "78901" in prompt
+
+
+# --- (b) image path: vision mocked, incl. graceful failure ------------------
+
+
+async def test_image_upload_stores_vision_description(app_client, db_session, tenant, monkeypatch):
+    app, client = app_client
+    fake = CaptureRouter()
+    app.dependency_overrides[get_router] = lambda: fake
+    seen = {}
+
+    async def fake_vision(data, mime):
+        seen["mime"] = mime
+        seen["data"] = data
+        return VISION_DESC
+
+    monkeypatch.setattr("app.vision.describe_image", fake_vision)
+
+    r = await _upload(client, "t1", "foto.png", PNG_BYTES, "image/png")
+    assert r.status_code == 200
+    up = r.json()
+    assert up["kind"] == "image"
+    assert "zapatilla" in up["preview"]
+    assert seen["mime"] == "image/png"  # mime came from MAGIC bytes
+    assert seen["data"] == PNG_BYTES  # vision got the real binary
+
+    att = await db_session.get(Attachment, up["attachment_id"])
+    assert att is not None and att.text == VISION_DESC
+
+    # and the description reaches the LLM on the next message
+    r2 = await client.post(
+        "/chat?t=t1",
+        json={
+            "message": "este es el producto que recibi",
+            "session_id": up["session_id"],
+            "attachment_ids": [up["attachment_id"]],
+        },
+    )
+    assert r2.status_code == 200 and r2.json()["reply"] == fake.reply
+    assert VISION_DESC in fake.all_user_text()
+
+
+async def test_image_vision_failure_falls_back_and_still_succeeds(app_client, db_session, tenant, monkeypatch):
+    app, client = app_client
+    fake = CaptureRouter()
+    app.dependency_overrides[get_router] = lambda: fake
+
+    async def broken_vision(data, mime):
+        return None
+
+    monkeypatch.setattr("app.vision.describe_image", broken_vision)
+
+    r = await _upload(client, "t1", "foto.jpg", b"\xff\xd8\xff\xe0" + b"\x00" * 32, "image/jpeg")
+    assert r.status_code == 200  # the upload still succeeds
+    up = r.json()
+    assert up["kind"] == "image"
+
+    att = await db_session.get(Attachment, up["attachment_id"])
+    assert att is not None and att.text == VISION_FALLBACK
+
+    r2 = await client.post(
+        "/chat?t=t1",
+        json={"message": "mira la foto", "session_id": up["session_id"],
+              "attachment_ids": [up["attachment_id"]]},
+    )
+    assert r2.status_code == 200 and r2.json()["reply"] == fake.reply
+    assert VISION_FALLBACK in fake.all_user_text()  # the bot can apologize naturally
+
+
+# --- (c) ISOLATION: the whole point --------------------------------------
+
+
+async def test_other_tenants_attachment_id_is_silently_dropped(app_client, db_session, tenant, tenant_b):
+    app, client = app_client
+    fake = CaptureRouter()
+    app.dependency_overrides[get_router] = lambda: fake
+
+    # tenant A uploads a document with a private marker
+    up = (await _upload(client, "t1", "privado.pdf", _pdf_bytes("Datos privados 555777"), "application/pdf")).json()
+    att = await db_session.get(Attachment, up["attachment_id"])
+    assert att is not None and "555777" in att.text  # the secret IS on the server
+
+    # tenant B (its own fresh session) references tenant A's attachment id
+    r = await client.post(
+        "/chat?t=t2",
+        json={"message": "que pone en el documento?", "attachment_ids": [up["attachment_id"]]},
+    )
+    assert r.status_code == 200  # NEVER an error: existence is not revealed
+    assert r.json()["reply"] == fake.reply  # the conversation flows normally
+
+    prompt = fake.all_user_text()
+    assert "que pone en el documento?" in prompt  # the message DID reach the LLM
+    assert "555777" not in prompt  # ...but tenant A's content did NOT
+    assert "[El cliente ha adjuntado" not in prompt
+
+
+async def test_cross_tenant_with_stolen_session_id_still_drops(app_client, db_session, tenant, tenant_b):
+    """WORST-CASE attacker: knows BOTH the victim's attachment id AND the
+    victim's session id, and replays them through another tenant's widget.
+    The foreign session id must not be adopted and the content must not leak."""
+    app, client = app_client
+    fake = CaptureRouter()
+    app.dependency_overrides[get_router] = lambda: fake
+
+    up = (await _upload(client, "t1", "privado.pdf", _pdf_bytes("Datos privados 555777"), "application/pdf")).json()
+    att = await db_session.get(Attachment, up["attachment_id"])
+    assert att is not None and "555777" in att.text  # the secret IS on the server
+
+    r = await client.post(
+        "/chat?t=t2",
+        json={"message": "lee el adjunto", "session_id": up["session_id"],
+              "attachment_ids": [up["attachment_id"]]},
+    )
+    assert r.status_code == 200  # NEVER an error: existence is not revealed
+    assert r.json()["reply"] == fake.reply  # the conversation flows normally
+    assert r.json()["session_id"] != up["session_id"]  # foreign sid NOT adopted
+
+    prompt = fake.all_user_text()
+    assert "lee el adjunto" in prompt  # the message DID reach the LLM
+    assert "555777" not in prompt  # ...but the victim's content did NOT
+    assert "[El cliente ha adjuntado" not in prompt
+
+
+async def test_stream_drops_cross_tenant_attachment_too(app_client, db_session, tenant, tenant_b):
+    """/chat/stream applies the exact same (id, tenant, session) filter: the
+    stolen pair injects NOTHING there either, while the stream itself works."""
+    app, client = app_client
+    fake = CaptureRouter()
+    app.dependency_overrides[get_router] = lambda: fake
+
+    up = (await _upload(client, "t1", "privado.pdf", _pdf_bytes("Datos privados 555777"), "application/pdf")).json()
+
+    r = await client.post(
+        "/chat/stream?t=t2",
+        json={"message": "lee el adjunto", "session_id": up["session_id"],
+              "attachment_ids": [up["attachment_id"]]},
+    )
+    assert r.status_code == 200
+    assert "event: final" in r.text  # the stream really completed
+    assert fake.reply in r.text  # ...with a normal reply
+
+    prompt = fake.all_user_text()
+    assert "lee el adjunto" in prompt
+    assert "555777" not in prompt
+    assert "[El cliente ha adjuntado" not in prompt
+
+
+async def test_same_tenant_different_session_is_silently_dropped(app_client, db_session, tenant):
+    app, client = app_client
+    fake = CaptureRouter()
+    app.dependency_overrides[get_router] = lambda: fake
+
+    up = (await _upload(client, "t1", "privado.pdf", _pdf_bytes("Datos privados 555777"), "application/pdf")).json()
+
+    # same tenant, but NO session_id -> a brand new conversation
+    r = await client.post(
+        "/chat?t=t1",
+        json={"message": "que pone en el documento?", "attachment_ids": [up["attachment_id"]]},
+    )
+    assert r.status_code == 200
+    assert r.json()["reply"] == fake.reply
+    assert r.json()["session_id"] != up["session_id"]  # truly another session
+
+    prompt = fake.all_user_text()
+    assert "que pone en el documento?" in prompt
+    assert "555777" not in prompt
+    assert "[El cliente ha adjuntado" not in prompt
+
+
+# --- (d) gates --------------------------------------------------------------
+
+
+async def test_upload_unknown_slug_404_never_falls_back(app_client, db_session, default_tenant):
+    app, client = app_client
+
+    r = await _upload(client, "ghost", "a.pdf", _pdf_bytes(), "application/pdf")
+    assert r.status_code == 404  # KILL-SWITCH: no fallback to 'default'
+    assert await _count_attachments(db_session) == 0
+
+
+async def test_upload_paused_tenant_403(app_client, db_session, tenant):
+    app, client = app_client
+
+    ok = await _upload(client, "t1", "a.pdf", _pdf_bytes(), "application/pdf")
+    assert ok.status_code == 200  # alive first: the endpoint really works
+
+    tenant.chatbot_enabled = False
+    await db_session.commit()
+
+    r = await _upload(client, "t1", "b.pdf", _pdf_bytes(), "application/pdf")
+    assert r.status_code == 403
+    assert await _count_attachments(db_session) == 1  # only the pre-pause one
+
+
+async def test_upload_fourth_attachment_rejected(app_client, db_session, tenant):
+    app, client = app_client
+
+    first = (await _upload(client, "t1", "a.pdf", _pdf_bytes(), "application/pdf")).json()
+    sid = first["session_id"]
+    for name in ("b.pdf", "c.pdf"):
+        ok = await _upload(client, "t1", name, _pdf_bytes(), "application/pdf", session_id=sid)
+        assert ok.status_code == 200  # 3 in total are accepted
+
+    r = await _upload(client, "t1", "d.pdf", _pdf_bytes(), "application/pdf", session_id=sid)
+    assert r.status_code == 422
+    assert "3 archivos" in r.json()["detail"]  # Spanish, human-readable
+    assert await _count_attachments(db_session) == 3
+
+
+async def test_upload_oversize_rejected_without_storing(app_client, db_session, tenant):
+    app, client = app_client
+
+    blob = b"%PDF-1.4 " + b"0" * (6 * 1024 * 1024)  # valid magic, 6 MB body
+    r = await _upload(client, "t1", "big.pdf", blob, "application/pdf")
+    assert r.status_code in (413, 422)
+    assert await _count_attachments(db_session) == 0
+
+
+async def test_upload_exe_magic_bytes_rejected(app_client, db_session, tenant):
+    app, client = app_client
+
+    # extension and Content-Type LIE; the magic bytes (MZ) tell the truth
+    r = await _upload(client, "t1", "factura.pdf", EXE_BYTES, "application/pdf")
+    assert r.status_code == 422
+    assert await _count_attachments(db_session) == 0
+
+
+async def test_upload_pdf_without_text_rejected(app_client, db_session, tenant):
+    from fpdf import FPDF
+
+    app, client = app_client
+    pdf = FPDF()
+    pdf.add_page()  # a page with no text at all
+
+    r = await _upload(client, "t1", "vacio.pdf", bytes(pdf.output()), "application/pdf")
+    assert r.status_code == 422
+    assert "texto legible" in r.json()["detail"]
+    assert await _count_attachments(db_session) == 0
+
+
+# --- (e) purge: admin delete + retention -----------------------------------
+
+
+async def test_purge_tenant_removes_its_attachments(app_client, db_session, tenant):
+    from app.tenancy import purge_tenant
+
+    app, client = app_client
+    up = (await _upload(client, "t1", "a.pdf", _pdf_bytes(), "application/pdf")).json()
+    assert await db_session.get(Attachment, up["attachment_id"]) is not None  # exists BEFORE
+
+    await purge_tenant(db_session, tenant)
+    await db_session.commit()
+
+    assert await _count_attachments(db_session) == 0  # gone AFTER
+
+
+async def test_retention_purge_removes_attachments_of_old_sessions(db_session, tenant):
+    from app.retention import purge_old_sessions
+
+    now = datetime.now(UTC)
+    old = ChatSession(shop="x", tenant_id=tenant.id, last_seen=now - timedelta(days=60))
+    recent = ChatSession(shop="x", tenant_id=tenant.id, last_seen=now)
+    db_session.add_all([old, recent])
+    await db_session.flush()
+    old_att = Attachment(tenant_id=tenant.id, session_id=old.id, filename="viejo.pdf",
+                         mime="application/pdf", text="viejo")
+    new_att = Attachment(tenant_id=tenant.id, session_id=recent.id, filename="nuevo.pdf",
+                         mime="application/pdf", text="nuevo")
+    db_session.add_all([old_att, new_att])
+    await db_session.flush()
+    assert await _count_attachments(db_session) == 2  # both exist BEFORE
+
+    purged = await purge_old_sessions(db_session, days=30)
+    assert purged == 1
+
+    remaining = (await db_session.execute(select(Attachment))).scalars().all()
+    assert [a.filename for a in remaining] == ["nuevo.pdf"]  # old one died with its session
+
+
+# --- (f) demo tenants work out of the box ----------------------------------
+
+
+async def test_demo_tenant_can_upload(app_client, db_session):
+    app, client = app_client
+    demo = Tenant(
+        slug="demo-cafe1234", name="Demo cafe.com", brand_name="cafe.com",
+        is_demo=True, demo_expires_at=datetime.now(UTC) + timedelta(hours=48),
+    )
+    db_session.add(demo)
+    await db_session.commit()
+
+    r = await _upload(client, "demo-cafe1234", "carta.pdf", _pdf_bytes(), "application/pdf")
+    assert r.status_code == 200
+    up = r.json()
+    att = await db_session.get(Attachment, up["attachment_id"])
+    assert att is not None and att.tenant_id == demo.id
+    assert "78901" in att.text

tests/test_ratelimit.py

@@ -35,3 +35,27 @@ def test_keys_independent():
     now = datetime(2026, 6, 9, tzinfo=UTC)
     assert rl.allow("a", now) is True
     assert rl.allow("b", now) is True
+
+
+def test_would_allow_peeks_without_consuming():
+    rl = RateLimiter(max_hits=2, window_seconds=60)
+    now = datetime(2026, 6, 9, tzinfo=UTC)
+    # Peeking any number of times never spends quota...
+    for _ in range(10):
+        assert rl.would_allow("k", now) is True
+    # ...so both real slots are still available afterwards.
+    assert rl.allow("k", now) is True
+    assert rl.allow("k", now) is True
+    assert rl.would_allow("k", now) is False  # and the peek sees the limit
+    assert rl.allow("k", now) is False
+
+
+def test_record_consumes_a_slot_and_window_expires():
+    rl = RateLimiter(max_hits=1, window_seconds=60)
+    now = datetime(2026, 6, 9, tzinfo=UTC)
+    assert rl.would_allow("k", now) is True
+    rl.record("k", now)
+    assert rl.would_allow("k", now) is False
+    assert rl.allow("k", now) is False
+    # The recorded hit ages out of the window like any allow() hit.
+    assert rl.would_allow("k", now + timedelta(seconds=61)) is True