deep knowledge + guards

app/lang.py

@@ -201,19 +201,54 @@ _LATIN_LETTERS = re.compile(r"[A-Za-zÀ-ÿ]")
 
 
 def needs_language_repair(reply: str, pinned_name: str | None) -> bool:
-    """True when a non-Latin language was pinned but the reply drifted to a
-    mostly-Latin script (e.g. Korean pinned, answer came back in Spanish)."""
+    """True when the reply drifted off the pinned language.
+
+    Non-Latin languages are verified by script (cheap, reliable). Latin-script
+    languages are verified by re-detecting the (long) reply and comparing — this
+    catches e.g. an English question answered in Spanish.
+    """
     if not reply or not pinned_name:
         return False
     code = _NAME_TO_CODE.get(pinned_name)
     script = _SCRIPT_BY_CODE.get(code or "")
-    if script is None:  # Latin or unmapped → skip (can't verify reliably by script)
+    if script is not None:
+        target = len(script.findall(reply))
+        latin = len(_LATIN_LETTERS.findall(reply))
+        # A correct non-Latin answer is dominated by its script; product names
+        # and links add a little Latin. Drift = Latin letters outnumber script.
+        return target < latin
+    # Latin-script pinned language: re-detect the reply and compare. Only act on
+    # a confident, different result (detect_language returns None when unsure),
+    # so we never loop on ambiguous text.
+    if len(reply) < 20:
         return False
-    target = len(script.findall(reply))
-    latin = len(_LATIN_LETTERS.findall(reply))
-    # A correct non-Latin answer is dominated by its script; product names and
-    # links add a little Latin. Drift = Latin letters outnumber target script.
-    return target < latin
+    detected = detect_language(reply)
+    return detected is not None and detected != pinned_name
+
+
+# Strip emoji/pictographs the model may add despite the no-emoji rule. The
+# product must never show emojis (hard requirement), so we enforce it
+# deterministically on output instead of trusting the model.
+_EMOJI = re.compile(
+    "["
+    "🀀-\U0001FAFF"      # emoji, pictographs, symbols, supplemental
+    "✀-➿"               # dingbats
+    "─-⯿"   # misc symbols, arrows-block emoji, stars (☀ ⭐ ✅ ⌚ …)
+    "\U0001F1E6-\U0001F1FF"  # regional indicators (flags)
+    "︀-️‍⃣"  # variation selectors, ZWJ, keycap
+    "]+",
+    flags=re.UNICODE,
+)
+
+
+def strip_emojis(text: str) -> str:
+    if not text:
+        return text
+    cleaned = _EMOJI.sub("", text)
+    # tidy up double spaces / space-before-punct left by removed emojis
+    cleaned = re.sub(r"[ \t]{2,}", " ", cleaned)
+    cleaned = re.sub(r" ([,.;:!?])", r"\1", cleaned)
+    return cleaned.strip()
 
 
 def detect_language(text: str) -> str | None:

app/orchestrator.py

@@ -131,7 +131,7 @@ async def run_turn(
     # Language guard: if a non-Latin language was pinned but the model drifted
     # (e.g. answered Korean question in Spanish), repair with one translation
     # pass. Guarantees the output language regardless of which provider answered.
-    from app.lang import needs_language_repair
+    from app.lang import needs_language_repair, strip_emojis
 
     if reply != FALLBACK_REPLY and needs_language_repair(reply, language):
         repair_msgs = [
@@ -154,6 +154,9 @@ async def run_turn(
         except ProviderError:
             pass  # keep the original reply rather than fail the turn
 
+    # Hard no-emoji guarantee: strip any emoji the model added despite the rule.
+    reply = strip_emojis(reply) or reply
+
     ctx.db.add(ChatMessage(session_id=ctx.session.id, role="assistant", content=reply))
 
     from app import analytics

app/rag/extract.py

@@ -1,31 +1,46 @@
-"""Extract plain text from uploaded files and from URLs."""
+"""Extract plain text from uploaded files and from URLs (with site crawling)."""
 
 from __future__ import annotations
 
 import io
+import logging
+import re
 from pathlib import Path
+from urllib.parse import urldefrag, urljoin, urlparse
 
 import httpx
 from selectolax.parser import HTMLParser
 
+log = logging.getLogger(__name__)
 
-def extract_html(html: bytes | str) -> str:
-    """Strip tags/scripts/styles and return visible text."""
+_UA = {"User-Agent": "shopify-support-bot/0.1"}
+# Bound the crawl so indexing a site stays fast and free.
+CRAWL_MAX_PAGES = 25
+
+
+def _text_and_links(html: bytes | str) -> tuple[str, list[str]]:
+    """Return (visible_text, hrefs) from an HTML document."""
     tree = HTMLParser(html if isinstance(html, str) else html.decode("utf-8", errors="replace"))
+    links = [a.attributes.get("href") for a in tree.css("a")]
+    links = [h for h in links if h]
     for tag in tree.css("script, style, noscript"):
         tag.decompose()
     body = tree.body or tree.root
-    if body is None:
-        return ""
-    text = body.text(separator=" ", strip=True)
-    return " ".join(text.split())
+    text = " ".join(body.text(separator=" ", strip=True).split()) if body is not None else ""
+    return text, links
+
+
+def extract_html(html: bytes | str) -> str:
+    """Strip tags/scripts/styles and return visible text."""
+    return _text_and_links(html)[0]
 
 
 async def extract_url(url: str, *, client: httpx.AsyncClient | None = None) -> str:
+    """Fetch a single page's visible text."""
     owns = client is None
     client = client or httpx.AsyncClient(timeout=20.0, follow_redirects=True)
     try:
-        resp = await client.get(url, headers={"User-Agent": "shopify-support-bot/0.1"})
+        resp = await client.get(url, headers=_UA)
         resp.raise_for_status()
         return extract_html(resp.content)
     finally:
@@ -33,6 +48,67 @@ async def extract_url(url: str, *, client: httpx.AsyncClient | None = None) -> s
             await client.aclose()
 
 
+async def _sitemap_urls(client: httpx.AsyncClient, base: str, host: str) -> list[str]:
+    try:
+        resp = await client.get(urljoin(base, "/sitemap.xml"), headers=_UA)
+        if resp.status_code != 200:
+            return []
+        locs = re.findall(r"<loc>\s*([^<\s]+)\s*</loc>", resp.text)
+        return [u for u in locs if urlparse(u).netloc == host]
+    except Exception:  # noqa: BLE001 - sitemap is best-effort
+        return []
+
+
+async def crawl_url(
+    start_url: str, *, max_pages: int = CRAWL_MAX_PAGES, client: httpx.AsyncClient | None = None
+) -> str:
+    """Crawl same-domain pages from ``start_url`` and return the combined text.
+
+    Follows internal links (and sitemap.xml) breadth-first up to ``max_pages``
+    so the whole site is indexed, not just the single landing page. Stays on the
+    starting host and only parses HTML responses.
+    """
+    owns = client is None
+    client = client or httpx.AsyncClient(timeout=20.0, follow_redirects=True)
+    host = urlparse(start_url).netloc
+    seen: set[str] = set()
+    queue: list[str] = [urldefrag(start_url)[0]]
+    texts: list[str] = []
+    try:
+        for u in await _sitemap_urls(client, start_url, host):
+            u = urldefrag(u)[0]
+            if u not in queue:
+                queue.append(u)
+        while queue and len(seen) < max_pages:
+            url = queue.pop(0)
+            if url in seen:
+                continue
+            seen.add(url)
+            try:
+                resp = await client.get(url, headers=_UA)
+                resp.raise_for_status()
+                if "html" not in resp.headers.get("content-type", "").lower():
+                    continue
+                text, links = _text_and_links(resp.content)
+            except Exception as exc:  # noqa: BLE001 - skip a bad page, keep crawling
+                log.debug("crawl skip %s: %s", url, exc)
+                continue
+            if text:
+                texts.append(text)
+            if len(seen) + len(queue) >= max_pages:
+                continue
+            for href in links:
+                nxt = urldefrag(urljoin(url, href))[0]
+                p = urlparse(nxt)
+                if p.scheme in ("http", "https") and p.netloc == host:
+                    if nxt not in seen and nxt not in queue:
+                        queue.append(nxt)
+    finally:
+        if owns:
+            await client.aclose()
+    return "\n\n".join(texts)
+
+
 def _extract_pdf(data: bytes) -> str:
     from pypdf import PdfReader
 

app/rag/index.py

@@ -18,7 +18,9 @@ log = logging.getLogger(__name__)
 
 async def _source_text(source: KnowledgeSource) -> str:
     if source.kind == "url":
-        return await extract.extract_url(source.location)
+        # Crawl the whole site (internal links + sitemap), not just one page,
+        # so nothing on the merchant's site is left out of the knowledge base.
+        return await extract.crawl_url(source.location)
     return extract.extract_file(source.location)
 
 

app/shopify/products.py

@@ -8,26 +8,33 @@ from typing import Any
 from app.shopify.client import ShopifyGraphQLClient
 
 PRODUCTS_QUERY = """
-query ProductSearch($q: String!) {
-  products(first: 6, query: $q) {
+query ProductSearch($q: String!, $after: String) {
+  products(first: 50, query: $q, after: $after) {
     edges {
       node {
         title
         handle
         onlineStoreUrl
         description
+        productType
+        tags
         featuredImage { url }
-        variants(first: 25) {
-          edges { node { id title price availableForSale } }
+        variants(first: 50) {
+          edges { node { id title price availableForSale sku } }
         }
       }
     }
+    pageInfo { hasNextPage endCursor }
   }
 }
 """
 
+# Walk through at most this many products (paginated). Covers full catalogs for
+# typical stores while bounding cost; queries page until this cap is reached.
+MAX_PRODUCTS = 250
 
-def _truncate(text: str | None, n: int = 300) -> str:
+
+def _truncate(text: str | None, n: int = 1500) -> str:
     text = (text or "").strip()
     return text if len(text) <= n else text[:n].rstrip() + "…"
 
@@ -61,6 +68,7 @@ def _parse_product(node: dict[str, Any]) -> dict[str, Any]:
             }
         )
     image = (node.get("featuredImage") or {}).get("url")
+    tags = node.get("tags") or []
     return {
         "title": node.get("title", ""),
         "price": min(prices) if prices else None,
@@ -68,11 +76,31 @@ def _parse_product(node: dict[str, Any]) -> dict[str, Any]:
         "url": node.get("onlineStoreUrl"),
         "image": image,
         "description": _truncate(node.get("description")),
+        "product_type": node.get("productType") or "",
+        "tags": tags if isinstance(tags, list) else [],
         "variants": variants,
     }
 
 
-async def search_products(client: ShopifyGraphQLClient, query: str) -> list[dict[str, Any]]:
-    data = await client.execute(PRODUCTS_QUERY, {"q": query})
-    edges = data.get("products", {}).get("edges", [])
-    return [_parse_product(e["node"]) for e in edges]
+async def search_products(
+    client: ShopifyGraphQLClient, query: str, *, max_products: int = MAX_PRODUCTS
+) -> list[dict[str, Any]]:
+    """Search the catalog, paginating until exhausted or ``max_products`` hit.
+
+    Shopify caps a page at a handful of results; without pagination a store with
+    more than a page of matches would be invisible past the first page.
+    """
+    out: list[dict[str, Any]] = []
+    after: str | None = None
+    while len(out) < max_products:
+        data = await client.execute(PRODUCTS_QUERY, {"q": query, "after": after})
+        products = data.get("products", {})
+        for e in products.get("edges", []):
+            out.append(_parse_product(e["node"]))
+        page = products.get("pageInfo", {})
+        if not page.get("hasNextPage"):
+            break
+        after = page.get("endCursor")
+        if not after:
+            break
+    return out[:max_products]

app/tools/knowledge_tool.py

@@ -14,7 +14,7 @@ async def run(args: dict[str, Any], ctx: ToolContext) -> dict[str, Any]:
     query = (args.get("query") or "").strip()
     if not query:
         return {"context": "", "sources": [], "note": "empty query"}
-    results = await index.search(ctx.db, query, k=6, tenant_id=ctx.tenant_id)
+    results = await index.search(ctx.db, query, k=8, tenant_id=ctx.tenant_id)
     if not results:
         # Log the gap so the merchant sees what the bot couldn't answer and can
         # add it to their knowledge. Tells the model to NOT invent an answer.

app/tools/products_tool.py

@@ -17,6 +17,14 @@ async def run(args: dict[str, Any], ctx: ToolContext) -> dict[str, Any]:
     if not query:
         return {"status": "ok", "products": []}
     products = await search_products(ctx.shopify, query)
+    fallback = False
+    # Keyword search misses when the query language differs from the catalog
+    # (e.g. "garden hose" vs a Spanish catalog) or matches nothing. Rather than
+    # tell the customer "no products" when products exist, fall back to the full
+    # catalog so the model can match semantically across languages.
+    if not products:
+        products = await search_products(ctx.shopify, "", max_products=50)
+        fallback = bool(products)
     # Surface as visual cards to the widget (de-duplicated by title).
     seen = {c.get("title") for c in ctx.cards}
     for p in products:
@@ -29,20 +37,30 @@ async def run(args: dict[str, Any], ctx: ToolContext) -> dict[str, Any]:
                 p["add_url"] = f"https://{ctx.shop}/cart/{variant['variant_id']}:1"
             ctx.cards.append(p)
             seen.add(p["title"])
-    # Compact view for the model (the rich cards already went to the widget):
-    # cap to 5 products + 10 variants, drop descriptions — keeps requests small
-    # so we stay under Groq's per-minute token limit.
+    # Compact view for the model (rich cards already went to the widget). Include
+    # a short description + type/tags so the model can match the customer's need
+    # across languages; cap products/variants to stay within token limits.
     compact = [
         {
             "title": p["title"],
             "price": p["price"],
             "available": p["available"],
+            "type": p.get("product_type") or None,
+            "tags": (p.get("tags") or [])[:6] or None,
+            "description": (p.get("description") or "")[:200] or None,
             "variants": [
                 {"variant_id": v["variant_id"], "title": v["title"], "price": v["price"]}
-                for v in p["variants"][:10]
+                for v in p["variants"][:12]
                 if v["available"]
             ],
         }
-        for p in products[:5]
+        for p in products[:10]
     ]
-    return {"status": "ok", "products": compact}
+    out: dict[str, Any] = {"status": "ok", "products": compact}
+    if fallback:
+        out["note"] = (
+            "La búsqueda por palabra clave no coincidió; estos son productos del "
+            "catálogo. Elige los que encajen con lo que pide el cliente (puede "
+            "estar en otro idioma); si ninguno encaja, dilo con sinceridad."
+        )
+    return out

tests/rag/test_extract.py

@@ -86,3 +86,73 @@ def test_extract_unsupported_raises(tmp_path):
 
     with pytest.raises(ValueError):
         extract_file(p)
+
+
+class _Resp:
+    def __init__(self, *, content=b"", text="", status_code=200, content_type="text/html"):
+        self.content = content
+        self.text = text
+        self.status_code = status_code
+        self.headers = {"content-type": content_type}
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise RuntimeError(f"http {self.status_code}")
+
+
+class _FakeClient:
+    """Maps URLs to responses; unknown URLs return 404."""
+
+    def __init__(self, pages):
+        self.pages = pages
+        self.requested = []
+
+    async def get(self, url, headers=None):
+        self.requested.append(url)
+        return self.pages.get(url, _Resp(status_code=404))
+
+
+async def test_crawl_follows_internal_links_and_sitemap():
+    from app.rag.extract import crawl_url
+
+    base = "https://shop.example"
+    pages = {
+        f"{base}/": _Resp(content=(
+            f'<html><body><h1>Inicio</h1>'
+            f'<a href="/faq">FAQ</a> <a href="{base}/envios">Envios</a>'
+            f'<a href="https://otro.com/x">externo</a></body></html>'
+        ).encode()),
+        f"{base}/faq": _Resp(content=b"<html><body>Garantia de dos anios</body></html>"),
+        f"{base}/envios": _Resp(content=b"<html><body>Envio gratis desde 199</body></html>"),
+        # discovered only via sitemap, not linked from any page
+        f"{base}/oculta": _Resp(content=b"<html><body>Pagina secreta del sitemap</body></html>"),
+        f"{base}/sitemap.xml": _Resp(
+            text=f"<urlset><url><loc>{base}/oculta</loc></url></urlset>",
+            content_type="application/xml",
+        ),
+    }
+    client = _FakeClient(pages)
+    text = await crawl_url(base + "/", client=client)
+    # every internal page is indexed (landing + linked + sitemap-only)
+    assert "Garantia de dos anios" in text
+    assert "Envio gratis desde 199" in text
+    assert "Pagina secreta del sitemap" in text
+    # external domain is never fetched
+    assert "https://otro.com/x" not in client.requested
+
+
+async def test_crawl_respects_max_pages():
+    from app.rag.extract import crawl_url
+
+    base = "https://big.example"
+    pages = {f"{base}/sitemap.xml": _Resp(status_code=404)}
+    # landing links to 10 pages; cap at 3
+    links = " ".join(f'<a href="/p{i}">p{i}</a>' for i in range(10))
+    pages[f"{base}/"] = _Resp(content=f"<html><body>{links}</body></html>".encode())
+    for i in range(10):
+        pages[f"{base}/p{i}"] = _Resp(content=f"<html><body>page {i}</body></html>".encode())
+    client = _FakeClient(pages)
+    await crawl_url(base + "/", max_pages=3, client=client)
+    # sitemap probe + at most 3 page fetches
+    page_fetches = [u for u in client.requested if "sitemap" not in u]
+    assert len(page_fetches) <= 3

tests/shopify/test_products.py

@@ -47,3 +47,45 @@ async def test_search_products_maps_nodes_with_variants_and_image():
 
 async def test_search_products_empty():
     assert await search_products(FakeClient({"products": {"edges": []}}), "nada") == []
+
+
+class PagingClient:
+    """Returns successive pages, honouring the cursor pagination contract."""
+
+    def __init__(self, pages):
+        self.pages = pages
+        self.calls = 0
+
+    async def execute(self, query, variables=None):
+        page = self.pages[min(self.calls, len(self.pages) - 1)]
+        self.calls += 1
+        return page
+
+
+def _node(title):
+    return {"node": {"title": title, "handle": title, "onlineStoreUrl": None,
+                     "description": "", "variants": {"edges": []}}}
+
+
+async def test_search_products_paginates_until_exhausted():
+    pages = [
+        {"products": {"edges": [_node("A"), _node("B")],
+                      "pageInfo": {"hasNextPage": True, "endCursor": "c1"}}},
+        {"products": {"edges": [_node("C")],
+                      "pageInfo": {"hasNextPage": False, "endCursor": None}}},
+    ]
+    client = PagingClient(pages)
+    products = await search_products(client, "x")
+    assert [p["title"] for p in products] == ["A", "B", "C"]  # both pages walked
+    assert client.calls == 2
+
+
+async def test_long_description_not_truncated_to_300():
+    desc = "x" * 900  # was capped at 300; full descriptions must reach the model
+    data = {"products": {"edges": [{"node": {
+        "title": "T", "handle": "t", "onlineStoreUrl": None,
+        "description": desc, "variants": {"edges": []},
+    }}]}}
+    products = await search_products(FakeClient(data), "t")
+    assert products[0]["description"] == desc
+    assert "…" not in products[0]["description"]

tests/test_lang.py

@@ -111,11 +111,24 @@ def test_language_guard_flags_drift():
     # correct: answer in the pinned script → no repair (Latin product names allowed)
     assert needs_language_repair("네, 정원용 호스가 있습니다. Gobeflat Medium 모델입니다.", ko) is False
     assert needs_language_repair("我们有花园用的水管。PVC 平面水管。", zh) is False
-    # Latin-script pinned languages are not script-verifiable → never force repair
-    assert needs_language_repair("Sí, tenemos productos.", "français") is False
+    # Latin drift is also caught: a French-pinned reply that came back in Spanish
+    assert needs_language_repair("Sí, tenemos varias mangueras de jardín disponibles para ti.", "français") is True
+    # ...but a genuinely French reply pinned to French is left alone
+    assert needs_language_repair("Bonjour, nous avons plusieurs tuyaux d'arrosage disponibles.", "français") is False
     assert needs_language_repair("anything", None) is False
 
 
+def test_strip_emojis_removes_pictographs_keeps_text():
+    from app.lang import strip_emojis
+
+    assert strip_emojis("¡Hola! 😊 ¿En qué te ayudo?") == "¡Hola! ¿En qué te ayudo?"
+    assert strip_emojis("Tenemos mangueras 🌿✅ disponibles") == "Tenemos mangueras disponibles"
+    # non-Latin text must survive untouched (emoji ranges don't overlap scripts)
+    assert strip_emojis("정원용 호스 😀 있습니다") == "정원용 호스 있습니다"
+    assert strip_emojis("我们有水管 🚀") == "我们有水管"
+    assert strip_emojis("normal text") == "normal text"
+
+
 def test_prompt_forbids_emojis():
     from app.prompts import build_system_prompt
 

tests/tools/test_tools.py

@@ -247,6 +247,38 @@ async def test_order_tool_rate_limited_across_sessions(db_session, monkeypatch):
     assert last["status"] == "locked"  # 4th attempt across fresh sessions is throttled
 
 
+class QueryAwareShopify:
+    """Empty for keyword queries, returns the catalog for the broad ('') query."""
+
+    def __init__(self, catalog):
+        self.catalog = catalog
+
+    async def execute(self, query, variables=None):
+        q = (variables or {}).get("q", "")
+        edges = self.catalog if q == "" else []
+        return {"products": {"edges": edges, "pageInfo": {"hasNextPage": False}}}
+
+
+async def test_products_tool_falls_back_to_catalog_when_keyword_misses(db_session):
+    catalog = [{"node": {
+        "title": "Manguera Plana Gobeflat", "handle": "gf", "onlineStoreUrl": None,
+        "description": "Manguera para riego y achique", "productType": "Mangueras",
+        "tags": ["jardin"], "variants": {"edges": [
+            {"node": {"id": "gid://shopify/ProductVariant/9", "title": "10m",
+                      "price": "16.99", "availableForSale": True}}]}}}]
+    ctx = ToolContext(
+        db=db_session, session=await _session(db_session),
+        shop="s.myshopify.com", shopify=QueryAwareShopify(catalog),
+    )
+    # customer asked in another language; keyword "garden hose" matches nothing
+    out = await registry.dispatch("search_products", {"query": "garden hose"}, ctx)
+    assert out["status"] == "ok"
+    assert out["products"], "fallback must surface catalog products, not empty"
+    assert out["products"][0]["title"] == "Manguera Plana Gobeflat"
+    assert "note" in out  # tells the model these came from the catalog fallback
+    assert ctx.cards  # visual cards still surfaced
+
+
 async def test_escalate_tool_stores_request(db_session):
     from sqlalchemy import select