visual search: harden against adversarial review (7 findings)

app/imgembed.py

@@ -46,7 +46,9 @@ def _embed_sync(data: bytes) -> list[float] | None:
         return None
     model = _get_model()
     for vec in model.embed([img]):  # one image in, one vector out
-        return [float(x) for x in vec]
+        # Round to 6 decimals: keeps cosine effectively identical but roughly halves the
+        # JSON footprint of the stored vector (the shared Neon DB is a scarce resource).
+        return [round(float(x), 6) for x in vec]
     return None
 
 
@@ -68,9 +70,17 @@ async def embed_image_url(url: str) -> list[float] | None:
         return None
     try:
         async with httpx.AsyncClient(timeout=_DOWNLOAD_TIMEOUT, follow_redirects=True) as client:
-            r = await client.get(url)
-            r.raise_for_status()
-            data = r.content
+            # Stream + abort early so a giant/booby-trapped image URL can't balloon memory
+            # (the cap was only checked AFTER the full body had already been buffered).
+            async with client.stream("GET", url) as r:
+                r.raise_for_status()
+                buf = bytearray()
+                async for chunk in r.aiter_bytes():
+                    buf.extend(chunk)
+                    if len(buf) > _MAX_IMAGE_BYTES:
+                        log.warning("imgembed: image too large, aborting %s", url[:120])
+                        return None
+                data = bytes(buf)
     except Exception:  # noqa: BLE001 - network is best-effort
         log.warning("imgembed: download failed for %s", url[:120])
         return None

app/orchestrator.py

@@ -364,7 +364,13 @@ 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)
+                # Error boundary: a tool raising must degrade to a model-readable error,
+                # never 500 the whole chat turn. The model then apologizes / offers help.
+                try:
+                    tool_result = await registry.dispatch(tc.name, tc.arguments, ctx)
+                except Exception:  # noqa: BLE001 - one tool failure never kills the turn
+                    log.warning("tool %s raised; degrading turn", tc.name, exc_info=True)
+                    tool_result = {"error": "tool failed", "status": "error"}
                 payload = json.dumps(tool_result, ensure_ascii=False, default=str)
                 if len(payload) > MAX_TOOL_RESULT_CHARS:  # protect small-window models
                     # keep it VALID JSON (a byte-slice would corrupt the object)

app/products/index.py

@@ -7,6 +7,7 @@ so one store's catalog can never surface in another store's bot.
 
 from __future__ import annotations
 
+import anyio
 from sqlalchemy import select, text
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -58,10 +59,16 @@ async def visual_search(
     if tenant_id is not None:
         stmt = stmt.where(ProductChunk.tenant_id == tenant_id)
     chunks = (await session.execute(stmt)).scalars().all()
-    scored = [
-        (c, embeddings.cosine(image_vec, c.image_embedding))
-        for c in chunks
-        if c.image_embedding
-    ]
-    scored.sort(key=lambda pair: pair[1], reverse=True)
-    return scored[:k]
+
+    # Pure-Python cosine over the whole image corpus can be heavy — run it OFF the event
+    # loop so a photo search never blocks other requests on the single worker.
+    def _score() -> list[tuple[ProductChunk, float]]:
+        scored = [
+            (c, embeddings.cosine(image_vec, c.image_embedding))
+            for c in chunks
+            if c.image_embedding
+        ]
+        scored.sort(key=lambda pair: pair[1], reverse=True)
+        return scored[:k]
+
+    return await anyio.to_thread.run_sync(_score)

app/products/sync.py

@@ -8,6 +8,7 @@ live at answer time. Runs OFF the request path (scheduler), never during a chat.
 
 from __future__ import annotations
 
+import asyncio
 import logging
 
 from sqlalchemy import select
@@ -54,13 +55,21 @@ async def sync_tenant_images(
         todo.append((c, img))
         if len(todo) >= cap:
             break
+
+    # Download + embed a few at a time (overlaps the network I/O), then assign results
+    # sequentially — never mutate ORM rows from concurrent tasks on the shared session.
+    sem = asyncio.Semaphore(4)
+
+    async def _embed(c, img):
+        async with sem:
+            try:
+                return c, img, await imgembed.embed_image_url(img)
+            except Exception:  # noqa: BLE001 - one bad image never blocks the rest
+                log.warning("image embed failed for product %s", c.shopify_product_id, exc_info=True)
+                return c, img, None
+
     done = 0
-    for c, img in todo:
-        try:
-            vec = await imgembed.embed_image_url(img)
-        except Exception:  # noqa: BLE001 - one bad image never blocks the rest
-            log.warning("image embed failed for product %s", c.shopify_product_id, exc_info=True)
-            vec = None
+    for c, img, vec in await asyncio.gather(*(_embed(c, img) for c, img in todo)):
         if vec:
             c.image_embedding = vec
             c.meta = {**(c.meta or {}), "image_emb_src": img}  # reassign so JSON change is tracked

app/routes/chat.py

@@ -271,6 +271,7 @@ async def _run_chat(
     # the conversation history replays coherently). They are never embedded
     # nor written to the knowledge base.
     block, has_image = await _attachment_block(db, tenant.id, ctx.session.id, attachment_ids or [])
+    ctx.image_uploaded = has_image  # visual product search only on the turn the photo rode in
     # intent_text=message: the deterministic grounding gates must read ONLY the
     # customer's own words, never the attachment text riding in the block.
     response = await run_turn(router_, ctx, block + message, brand_name=tenant.brand_name,

app/tools/products_tool.py

@@ -170,17 +170,19 @@ async def run(args: dict[str, Any], ctx: ToolContext) -> dict[str, Any]:
         lim = 4
     lim = max(1, min(lim, MAX_RESULTS))
 
-    # Retrieval arms in parallel: Shopify keyword + semantic corpus + (if the visitor
-    # uploaded a photo this session) VISUAL similarity over the catalog's image vectors.
-    image_vec = await _session_image_vec(ctx)
+    # The Shopify keyword arm is HTTP (independent of the DB session) so it overlaps the
+    # DB work. The semantic + visual arms BOTH use ctx.db — and a single AsyncSession is
+    # NOT safe for concurrent use — so they run sequentially on it, never gathered.
+    # Visual matching only fires on the turn the photo actually rode in (ctx.image_uploaded),
+    # so a stale photo never contaminates later text-only searches in the session.
+    image_vec = await _session_image_vec(ctx) if getattr(ctx, "image_uploaded", False) else None
     kw_task = asyncio.create_task(search_products(ctx.shopify, query))
-    sem_task = asyncio.create_task(
-        product_index.search(ctx.db, query, k=20, tenant_id=ctx.tenant_id)
+    sem_results = await product_index.search(ctx.db, query, k=20, tenant_id=ctx.tenant_id)
+    vis_results = (
+        await product_index.visual_search(ctx.db, image_vec, k=20, tenant_id=ctx.tenant_id)
+        if image_vec else []
     )
-    vis_task = asyncio.create_task(
-        product_index.visual_search(ctx.db, image_vec or [], k=20, tenant_id=ctx.tenant_id)
-    )
-    kw_products, sem_results, vis_results = await asyncio.gather(kw_task, sem_task, vis_task)
+    kw_products = await kw_task
 
     kw_ids = [str(p.get("product_id") or "") for p in kw_products]
     sem_ids = [c.shopify_product_id for c, _ in sem_results]

app/tools/registry.py

@@ -47,6 +47,7 @@ class ToolContext:
     web_urls: list[str] = field(default_factory=list)  # URLs found this turn (link-guard pass)
     customer: Any | None = None  # Customer profile (returning-customer memory), if known
     customer_note: str = ""  # system note injected when this is a RETURNING customer
+    image_uploaded: bool = False  # a photo rode in THIS turn -> enable visual product search
 
 
 SPECS: list[ToolSpec] = [

extension/.shopify/deploy-bundle/288ab506-3769-fe13-0858-10bbb247f01cb0ff7c2a/blocks/chat.liquid

@@ -1,46 +1,20 @@
 {% comment %}
-  App embed block: injects the support chat widget on every storefront page.
-  Enable it in Online Store > Themes > Customize > App embeds.
-{% endcomment %}
+  Atendyo — AI support chat (app embed). The assistant loads from the Atendyo app
+  and talks to your store through the SIGNED Shopify App Proxy (/apps/chat): it
+  reads your catalog, order status and policies, accepts file uploads, supports
+  voice, and can hand off to a human — no links to paste.
 
-<div
-  id="ssb-chat-root"
-  data-backend="{{ block.settings.backend_path }}"
-  data-brand="{{ block.settings.brand_name | escape }}"
-  data-color="{{ block.settings.brand_color }}"
-  data-welcome="{{ block.settings.welcome | escape }}"
-></div>
-<script src="{{ 'widget.js' | asset_url }}" defer></script>
+  Configure it (name, colour, welcome, instructions) from the Atendyo app panel.
+  Enable it here under Online Store > Themes > Customize > App embeds, then Save.
+{% endcomment %}
+<script
+  src="https://victor34593993-flexigo-support-bot.hf.space/widget.js?proxy=1"
+  defer></script>
 
 {% schema %}
 {
   "name": "Chatbot de soporte",
   "target": "body",
-  "settings": [
-    {
-      "type": "text",
-      "id": "backend_path",
-      "label": "Ruta del backend (App Proxy)",
-      "default": "/apps/chat"
-    },
-    {
-      "type": "text",
-      "id": "brand_name",
-      "label": "Nombre del asistente",
-      "default": "Asistente"
-    },
-    {
-      "type": "color",
-      "id": "brand_color",
-      "label": "Color principal",
-      "default": "#e8491d"
-    },
-    {
-      "type": "textarea",
-      "id": "welcome",
-      "label": "Mensaje de bienvenida",
-      "default": "¡Hola! ¿En qué puedo ayudarte?"
-    }
-  ]
+  "settings": []
 }
 {% endschema %}

extension/.shopify/deploy-bundle/manifest.json

@@ -105,6 +105,18 @@
         "api_version": "2026-01",
         "uri": "https://victor34593993-flexigo-support-bot.hf.space/shopify/webhooks/app_subscriptions_update"
       }
+    },
+    {
+      "type": "webhook_subscription",
+      "handle": "c65863898f02bd781a28ed9757c3b563284a6ab2",
+      "uid": "app/uninstalled::undefined::https://victor34593993-flexigo-support-bot.hf.space/shopify/webhooks/app_uninstalled",
+      "assets": "app/uninstalled::undefined::https://victor34593993-flexigo-support-bot.hf.space/shopify/webhooks/app_uninstalled",
+      "target": "",
+      "config": {
+        "topic": "app/uninstalled",
+        "api_version": "2026-01",
+        "uri": "https://victor34593993-flexigo-support-bot.hf.space/shopify/webhooks/app_uninstalled"
+      }
     }
   ]
 }

tests/products/test_visual.py

@@ -68,16 +68,27 @@ async def test_products_tool_blends_uploaded_photo(db_session, monkeypatch):
                  "image": "http://img", "url": "http://u", "variants": [], "tags": [],
                  "description": "", "product_type": ""}]
 
+    async def empty_text_search(session, query, k=8, *, tenant_id=None):
+        return []  # isolate the VISUAL arm: the text/semantic arm matches nothing
+
     monkeypatch.setattr(pt, "search_products", fake_kw)
     monkeypatch.setattr(pt, "fetch_products_by_ids", fake_hydrate)
+    monkeypatch.setattr(product_index, "search", empty_text_search)
 
+    # image_uploaded=True -> the photo rode in THIS turn, so the visual arm fires
     ctx = ToolContext(db=db_session, session=sess, tenant_id=1, shopify=_FakeShopify(),
-                      shop="x.myshopify.com")
+                      shop="x.myshopify.com", image_uploaded=True)
     out = await registry.dispatch("search_products", {"query": "algo así"}, ctx)
     titles = [p["title"] for p in out["products"]]
     assert "Mystery Item" in titles      # surfaced purely by VISUAL similarity
     assert any(c["title"] == "Mystery Item" for c in ctx.cards)
 
+    # ...but a LATER text-only turn (no photo this turn) must NOT re-use the stale photo
+    ctx2 = ToolContext(db=db_session, session=sess, tenant_id=1, shopify=_FakeShopify(),
+                       shop="x.myshopify.com", image_uploaded=False)
+    out2 = await registry.dispatch("search_products", {"query": "otra cosa"}, ctx2)
+    assert "Mystery Item" not in [p["title"] for p in out2["products"]]
+
 
 async def test_imgembed_rejects_bad_input():
     # empty / non-image bytes -> None, without loading the CLIP model

tests/test_orchestrator.py

@@ -493,3 +493,28 @@ async def test_pdf_or_plain_text_does_not_force_product_search(db_session):
     flat = " ".join(c for call in router.seen for c in call)
     # it may force search_knowledge (grounding), but NOT the image->search_products note
     assert "atributos de la imagen" not in flat
+
+
+async def test_tool_exception_degrades_not_500(db_session):
+    """A tool raising must NOT crash the turn — it degrades to a model-readable error."""
+    router = FakeRouter([
+        ChatResult(content=None, tool_calls=[
+            ToolCall(id="t1", name="search_products", arguments={"query": "x"})],
+            finish_reason="tool_calls"),
+        ChatResult(content="Lo siento, ahora no puedo consultarlo.", tool_calls=[], finish_reason="stop"),
+    ])
+    ctx = ToolContext(db=db_session, session=await _session(db_session), tenant_id=1)
+
+    import app.tools.registry as reg
+
+    async def boom(name, args, c):
+        raise RuntimeError("tool blew up")
+
+    # monkeypatch dispatch to raise
+    orig = reg.dispatch
+    reg.dispatch = boom
+    try:
+        resp = await run_turn(router, ctx, "enséñame algo")
+    finally:
+        reg.dispatch = orig
+    assert resp.reply  # a real reply, no exception propagated