Feature 4: true visual product search (CLIP)

app/imgembed.py

@@ -0,0 +1,77 @@
+"""CLIP image embeddings for VISUAL product search.
+
+A shopper's uploaded photo and each product's primary photo are embedded into the
+same 512-dim CLIP space (fastembed ONNX, CPU, no torch), so we can rank the catalog
+by visual similarity to the photo — "find me products that look like this".
+
+Everything is best-effort: any failure (no model, bad bytes, download error) returns
+None so the caller silently degrades to the text-description search. The model is
+loaded lazily in a worker thread and cached for the process lifetime.
+"""
+
+from __future__ import annotations
+
+import io
+import logging
+
+import anyio
+import httpx
+
+log = logging.getLogger(__name__)
+
+IMAGE_MODEL = "Qdrant/clip-ViT-B-32-vision"
+IMAGE_EMBED_DIM = 512
+_DOWNLOAD_TIMEOUT = 15.0
+_MAX_IMAGE_BYTES = 12 * 1024 * 1024  # ignore anything bigger (defensive)
+
+_model = None
+
+
+def _get_model():
+    global _model
+    if _model is None:
+        from fastembed import ImageEmbedding
+
+        _model = ImageEmbedding(model_name=IMAGE_MODEL, cache_dir=".fastembed_cache")
+    return _model
+
+
+def _embed_sync(data: bytes) -> list[float] | None:
+    from PIL import Image
+
+    try:
+        img = Image.open(io.BytesIO(data)).convert("RGB")
+    except Exception:  # noqa: BLE001 - unreadable/corrupt image
+        log.warning("imgembed: could not decode image bytes")
+        return None
+    model = _get_model()
+    for vec in model.embed([img]):  # one image in, one vector out
+        return [float(x) for x in vec]
+    return None
+
+
+async def embed_image(data: bytes) -> list[float] | None:
+    """512-dim CLIP vector for raw image bytes, or None on any failure."""
+    if not data or len(data) > _MAX_IMAGE_BYTES:
+        return None
+    try:
+        return await anyio.to_thread.run_sync(_embed_sync, data)
+    except Exception:  # noqa: BLE001 - embedding is best-effort by design
+        log.warning("imgembed: embed_image failed", exc_info=True)
+        return None
+
+
+async def embed_image_url(url: str) -> list[float] | None:
+    """Download an image URL (bounded) and embed it. None on any failure."""
+    url = (url or "").strip()
+    if not url.startswith("http"):
+        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
+    except Exception:  # noqa: BLE001 - network is best-effort
+        log.warning("imgembed: download failed for %s", url[:120])
+        return None
+    return await embed_image(data)

app/models.py

@@ -233,6 +233,11 @@ class ProductChunk(Base):
     updated_at: Mapped[str] = mapped_column(String(40), default="")
     text: Mapped[str] = mapped_column(Text)
     embedding: Mapped[list[float]] = mapped_column(EmbeddingType)
+    # CLIP image embedding (512-dim) of the product's primary photo, for visual
+    # search (a shopper's uploaded photo -> visually similar products). Plain JSON
+    # (different dim from the 384-dim text vector); Python cosine, tenant-scoped.
+    # Nullable: filled in lazily by the background sync, best-effort.
+    image_embedding: Mapped[list | None] = mapped_column(JSON, nullable=True)
     meta: Mapped[dict] = mapped_column(JSON, default=dict)
 
 
@@ -396,6 +401,9 @@ class Attachment(Base):
     filename: Mapped[str] = mapped_column(String(255), default="")
     mime: Mapped[str] = mapped_column(String(60), default="")
     text: Mapped[str] = mapped_column(Text, default="")
+    # CLIP image embedding (512-dim) of the uploaded photo — lets the next product
+    # search find visually similar items. Plain JSON, nullable (only for images).
+    image_embedding: Mapped[list | None] = mapped_column(JSON, nullable=True)
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
 
 

app/orchestrator.py

@@ -224,6 +224,7 @@ async def run_turn(
     instructions: str = "",
     on_stage: Callable[[str], None] | None = None,
     intent_text: str | None = None,
+    image_uploaded: bool = False,
 ) -> ChatResponse:
     # intent_text: the customer's OWN words, without any attachment block the
     # route prepended. The deterministic intent gates must match only what the
@@ -288,11 +289,15 @@ async def run_turn(
             "pasar con una persona. NUNCA mandes al cliente a la web del fabricante "
             "ni a un tercero. PROHIBIDO inventar."
         )})
-    # Photo-to-product chain: an attachment rode in with THIS message AND the
-    # customer is product-seeking ("¿tenéis algo como esto?") -> force the
-    # catalog search with the image's attributes. Marker presence ALONE never
-    # forces it: document questions are answered from the attached text itself.
-    if ATTACHMENT_MARKER in user_message and _PHOTO_PRODUCT_INTENT.search(gate_text):
+    # Photo-to-product chain: force a (visually-aware) catalog search when the visitor
+    # uploaded a PHOTO and either (a) said something product-seeking, or (b) sent the
+    # photo with no substantive question of its own — a bare photo IS "do you have
+    # this?". search_products then blends visual similarity (the uploaded photo's CLIP
+    # vector) with text. A PDF or a real non-product question is NOT forced.
+    photo_search = (
+        ATTACHMENT_MARKER in user_message and _PHOTO_PRODUCT_INTENT.search(gate_text)
+    ) or (image_uploaded and not _needs_grounding(gate_text))
+    if photo_search:
         messages.append({"role": "system", "content": PHOTO_GROUNDING_NOTE})
     for m in prior:
         if m.role in ("user", "assistant"):
@@ -379,14 +384,25 @@ async def run_turn(
         # Plain-text reply. ENFORCE grounding in code: a real question answered
         # with NO tool ever called is a memory/hallucination answer (or a redirect)
         # — make the model search first, once, before we accept it.
-        if _needs_grounding(gate_text) and not used_tools and not forced_search:
+        if (_needs_grounding(gate_text) or photo_search) and not used_tools and not forced_search:
             forced_search = True
-            messages.append({"role": "user", "content": (
-                "(Sistema) No respondas todavía. DEBES llamar primero a "
-                "search_knowledge (y a search_products si la pregunta es de "
-                "productos/precio/stock) y responder SOLO con lo que devuelvan. "
-                "No respondas de memoria ni mandes al cliente fuera de la tienda."
-            )})
+            if photo_search:
+                # A photo rode in: the grounded answer comes from the catalog, matched
+                # by the image (visual + its described attributes), never from memory.
+                messages.append({"role": "user", "content": (
+                    "(Sistema) No respondas todavía. DEBES llamar primero a "
+                    "search_products usando como consulta los atributos de la imagen "
+                    "(tipo de objeto, categoría, color, material, medidas, marca) y "
+                    "responder SOLO con los productos que devuelva. Si no hay nada "
+                    "parecido, dilo con sinceridad. No inventes productos."
+                )})
+            else:
+                messages.append({"role": "user", "content": (
+                    "(Sistema) No respondas todavía. DEBES llamar primero a "
+                    "search_knowledge (y a search_products si la pregunta es de "
+                    "productos/precio/stock) y responder SOLO con lo que devuelvan. "
+                    "No respondas de memoria ni mandes al cliente fuera de la tienda."
+                )})
             continue
         reply = result.content or FALLBACK_REPLY
         break

app/products/index.py

@@ -38,3 +38,30 @@ async def search(
     scored = [(c, embeddings.cosine(qvec, c.embedding)) for c in chunks]
     scored.sort(key=lambda pair: pair[1], reverse=True)
     return scored[:k]
+
+
+async def visual_search(
+    session: AsyncSession,
+    image_vec: list[float],
+    k: int = 8,
+    *,
+    tenant_id: int | None = None,
+) -> list[tuple[ProductChunk, float]]:
+    """Rank this tenant's products by VISUAL similarity (CLIP cosine) to an uploaded
+    photo. Only products whose primary image has been embedded participate; returns up
+    to k (ProductChunk, score) pairs, most similar first. Always tenant-scoped, so one
+    store's catalog can never surface in another's. Python cosine full-scan over the
+    tenant's image vectors — bounded by catalog size, and only runs on a photo upload."""
+    if not image_vec:
+        return []
+    stmt = select(ProductChunk).where(ProductChunk.image_embedding.is_not(None))
+    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]

app/products/sync.py

@@ -24,6 +24,50 @@ log = logging.getLogger(__name__)
 # so it can cover the whole catalog — decoupled from the per-request retrieval
 # caps. This avoids a silent ceiling where products past #250 are never indexed.
 CORPUS_MAX_PRODUCTS = 5000
+# Embedding a product photo means downloading it + a CLIP forward pass, so cap how
+# many we (re)embed per sync run; the daily tick fills the rest over a few runs.
+IMAGE_EMBED_CAP = 60
+
+
+async def sync_tenant_images(
+    db: AsyncSession, tenant: Tenant, *, cap: int = IMAGE_EMBED_CAP
+) -> int:
+    """Best-effort: embed the primary photo of products that don't have a current image
+    vector yet (or whose photo URL changed), for visual search. Bounded per run; the
+    CLIP image binary is fetched, embedded, and discarded. Returns how many were
+    embedded. Never raises — visual search degrades to text when a vector is missing."""
+    from app import imgembed
+
+    chunks = (
+        await db.execute(
+            select(ProductChunk).where(ProductChunk.tenant_id == tenant.id)
+        )
+    ).scalars().all()
+    todo = []
+    for c in chunks:
+        img = (c.meta or {}).get("image")
+        if not img:
+            continue
+        # (re)embed when missing, or when the photo URL differs from what we embedded
+        if c.image_embedding is not None and (c.meta or {}).get("image_emb_src") == img:
+            continue
+        todo.append((c, img))
+        if len(todo) >= cap:
+            break
+    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
+        if vec:
+            c.image_embedding = vec
+            c.meta = {**(c.meta or {}), "image_emb_src": img}  # reassign so JSON change is tracked
+            done += 1
+    if done:
+        await db.flush()
+    return done
 
 
 def _embed_text(p: dict) -> str:
@@ -114,6 +158,14 @@ async def sync_tenant(
 
     await db.flush()
 
+    # Visual search: embed product photos (bounded, best-effort). Never let an image
+    # hiccup fail the catalog sync — the text corpus above is what matters most.
+    images_embedded = 0
+    try:
+        images_embedded = await sync_tenant_images(db, tenant)
+    except Exception:  # noqa: BLE001 - image embedding is additive, never fatal
+        log.warning("image embedding pass failed for tenant %s", tenant.id, exc_info=True)
+
     # Also refresh the store's shop policies (returns/shipping/privacy/terms) so
     # the bot ALWAYS knows them — the #1 service gap was "no tengo la política de
     # devoluciones". Best-effort: never let a policy hiccup fail the catalog sync.
@@ -128,6 +180,7 @@ async def sync_tenant(
         "total": len(products),
         "reembedded": len(pending),
         "removed": len(removed),
+        "images_embedded": images_embedded,
         "policies": policies.get("indexed", 0),
         "policies_reason": policies.get("reason") or policies.get("status", ""),
     }

app/routes/chat.py

@@ -221,8 +221,9 @@ async def _make_context(
 
 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.
+) -> tuple[str, bool]:
+    """Context block for the attachments referenced by THIS turn, plus whether any of
+    them is a PHOTO (so the orchestrator can force a visual product search).
 
     Strict isolation invariant (the product's #1 requirement): an id only
     matches when it belongs to the SAME tenant AND the SAME chat session.
@@ -230,7 +231,7 @@ async def _attachment_block(
     is SILENTLY dropped: never an error, so existence is never revealed.
     """
     if not attachment_ids:
-        return ""
+        return "", False
     rows = (
         await db.execute(
             select(Attachment)
@@ -243,7 +244,8 @@ async def _attachment_block(
         )
     ).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]
+    has_image = any((a.mime or "").startswith("image/") for a in rows)
+    return "".join(parts)[:ATTACHMENT_BLOCK_CAP], has_image
 
 
 async def _run_chat(
@@ -268,12 +270,12 @@ async def _run_chat(
     # 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 [])
+    block, has_image = await _attachment_block(db, tenant.id, ctx.session.id, attachment_ids or [])
     # 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,
                             instructions=tenant.custom_instructions, on_stage=on_stage,
-                            intent_text=message)
+                            intent_text=message, image_uploaded=has_image)
     await purge_old_sessions(db, settings.session_retention_days,
                              conversation_hours=settings.conversation_retention_hours)
     await db.commit()
@@ -379,8 +381,16 @@ async def _upload_impl(
         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)
+    image_embedding = None
+    if kind != "pdf":
+        # Vision description (for the LLM's context) AND a CLIP embedding (for visual
+        # product search) in parallel — both from the same bytes, which are then
+        # discarded. Visual embedding is best-effort: None just falls back to text.
+        from app import imgembed
+
+        described, image_embedding = await asyncio.gather(
+            vision.describe_image(data, mime), imgembed.embed_image(data)
+        )
         text = (described or "").strip()[:ATTACHMENT_TEXT_CAP] or VISION_FALLBACK_TEXT
 
     safe_name = SAFE_NAME.sub("_", file.filename or "archivo")[:255]
@@ -389,6 +399,7 @@ async def _upload_impl(
     att = Attachment(
         tenant_id=tenant.id, session_id=session.id,
         filename=safe_name, mime=mime, text=text,
+        image_embedding=image_embedding,
     )
     db.add(att)
     await db.flush()

app/tools/products_tool.py

@@ -17,8 +17,10 @@ import asyncio
 import logging
 from typing import TYPE_CHECKING, Any
 
+from sqlalchemy import select
+
 from app import embeddings
-from app.models import ProductChunk
+from app.models import Attachment, ProductChunk
 from app.products import index as product_index
 from app.products.sync import _embed_text
 from app.shopify.products import fetch_products_by_ids, search_products
@@ -30,18 +32,43 @@ log = logging.getLogger(__name__)
 
 MAX_RESULTS = 8  # never flood the model; surface the few most relevant
 COLD_CATALOG_CAP = 50  # in-request semantic rank cap before the corpus is built
+VISUAL_WEIGHT = 2.0  # boost the photo-similarity arm: an uploaded photo IS the intent
 
 
-def _rrf(*ranked_lists: list[str], k: int = 60) -> list[str]:
-    """Reciprocal Rank Fusion: blend several ranked id lists into one order."""
+def _rrf(
+    *ranked_lists: list[str], k: int = 60, weights: list[float] | None = None
+) -> list[str]:
+    """Reciprocal Rank Fusion: blend several ranked id lists into one order. Optional
+    per-list weights let one arm (e.g. visual similarity) count more."""
     scores: dict[str, float] = {}
-    for lst in ranked_lists:
+    for i, lst in enumerate(ranked_lists):
+        w = weights[i] if weights else 1.0
         for rank, pid in enumerate(lst):
             if pid:
-                scores[pid] = scores.get(pid, 0.0) + 1.0 / (k + rank + 1)
+                scores[pid] = scores.get(pid, 0.0) + w / (k + rank + 1)
     return sorted(scores, key=lambda pid: scores[pid], reverse=True)
 
 
+async def _session_image_vec(ctx: ToolContext) -> list[float] | None:
+    """The CLIP vector of the most recent photo the visitor uploaded THIS session, if
+    any — so search blends in visually-similar products. Strictly session-scoped."""
+    if ctx.tenant_id is None or ctx.session is None:
+        return None
+    row = (
+        await ctx.db.execute(
+            select(Attachment.image_embedding)
+            .where(
+                Attachment.session_id == ctx.session.id,
+                Attachment.tenant_id == ctx.tenant_id,
+                Attachment.image_embedding.is_not(None),
+            )
+            .order_by(Attachment.created_at.desc(), Attachment.id.desc())
+            .limit(1)
+        )
+    ).first()
+    return list(row[0]) if row and row[0] else None
+
+
 async def _persist_cold(
     ctx: ToolContext, catalog: list[dict[str, Any]], vecs: list[list[float]]
 ) -> None:
@@ -143,23 +170,32 @@ async def run(args: dict[str, Any], ctx: ToolContext) -> dict[str, Any]:
         lim = 4
     lim = max(1, min(lim, MAX_RESULTS))
 
-    # Two retrieval arms in parallel: Shopify keyword + semantic corpus.
+    # 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)
     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)
     )
-    kw_products, sem_results = await asyncio.gather(kw_task, sem_task)
+    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_ids = [str(p.get("product_id") or "") for p in kw_products]
     sem_ids = [c.shopify_product_id for c, _ in sem_results]
+    vis_ids = [c.shopify_product_id for c, _ in vis_results]
 
     products: list[dict[str, Any]]
-    if not sem_results and not kw_products:
+    if not sem_results and not kw_products and not vis_results:
         # Corpus not built AND keyword missed → in-request semantic rank so we
         # never wrongly say "no products" (the old arbitrary-slice bug).
         products = await _cold_semantic_rank(ctx, query)
     else:
-        fused = _rrf(kw_ids, sem_ids)[:MAX_RESULTS]
+        # Visual matches count extra: when there's a photo, looking-alike IS the answer.
+        fused = _rrf(
+            kw_ids, sem_ids, vis_ids, weights=[1.0, 1.0, VISUAL_WEIGHT]
+        )[:MAX_RESULTS]
         kw_by_id = {str(p.get("product_id") or ""): p for p in kw_products}
         sem_only = [pid for pid in fused if pid and pid not in kw_by_id]
         # Hydrate live price/stock for the fused shortlist in one batched call.

migrations/versions/0030_image_embeddings.py

@@ -0,0 +1,36 @@
+"""Visual product search: CLIP image embeddings.
+
+Adds nullable `image_embedding` JSON columns to product_chunks (the product's
+primary-photo vector, filled lazily by the background catalog sync) and to
+attachments (a shopper's uploaded-photo vector), so an uploaded photo can be
+matched against the catalog by visual similarity.
+
+Postgres-only, additive + idempotent. No-op elsewhere (tests/dev use create_all).
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+revision = "0030_image_embeddings"
+down_revision = "0029_abandoned_carts"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    if op.get_bind().dialect.name != "postgresql":
+        return
+    op.execute(
+        "ALTER TABLE product_chunks ADD COLUMN IF NOT EXISTS image_embedding JSONB"
+    )
+    op.execute(
+        "ALTER TABLE attachments ADD COLUMN IF NOT EXISTS image_embedding JSONB"
+    )
+
+
+def downgrade() -> None:
+    if op.get_bind().dialect.name != "postgresql":
+        return
+    op.execute("ALTER TABLE product_chunks DROP COLUMN IF EXISTS image_embedding")
+    op.execute("ALTER TABLE attachments DROP COLUMN IF EXISTS image_embedding")

tests/products/test_visual.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+
+from app import imgembed
+from app.models import Attachment, ChatSession, ProductChunk
+from app.products import index as product_index
+from app.tools import registry
+from app.tools.registry import ToolContext
+
+
+def _vec(*xs):  # tiny normalized-ish vectors for cosine
+    return [float(x) for x in xs]
+
+
+async def test_visual_search_ranks_and_is_tenant_scoped(db_session):
+    # tenant 1: a "red" and a "blue" product (toy 3-dim image vectors)
+    db_session.add(ProductChunk(tenant_id=1, shopify_product_id="red", text="r",
+                                embedding=_vec(0, 0, 0), image_embedding=_vec(1, 0, 0)))
+    db_session.add(ProductChunk(tenant_id=1, shopify_product_id="blue", text="b",
+                                embedding=_vec(0, 0, 0), image_embedding=_vec(0, 0, 1)))
+    # tenant 2: a near-identical "red" — must NEVER surface for tenant 1
+    db_session.add(ProductChunk(tenant_id=2, shopify_product_id="red2", text="r2",
+                                embedding=_vec(0, 0, 0), image_embedding=_vec(1, 0, 0)))
+    # a product with NO image vector is ignored
+    db_session.add(ProductChunk(tenant_id=1, shopify_product_id="noimg", text="n",
+                                embedding=_vec(0, 0, 0), image_embedding=None))
+    await db_session.flush()
+
+    out = await product_index.visual_search(db_session, _vec(0.9, 0.1, 0), k=5, tenant_id=1)
+    ids = [c.shopify_product_id for c, _ in out]
+    assert ids[0] == "red"            # most visually similar first
+    assert "red2" not in ids          # tenant isolation
+    assert "noimg" not in ids         # no image vector -> excluded
+    assert "blue" in ids and ids.index("blue") > ids.index("red")
+
+
+async def test_visual_search_empty_vector_returns_nothing(db_session):
+    assert await product_index.visual_search(db_session, [], tenant_id=1) == []
+
+
+class _FakeShopify:
+    """Returns no keyword hits + hydrates the visual-matched product id live."""
+
+    async def execute(self, query, variables=None):
+        return {"products": {"edges": []}}
+
+
+async def test_products_tool_blends_uploaded_photo(db_session, monkeypatch):
+    """End-to-end: a photo uploaded this session makes search_products surface the
+    visually-similar product even when the keyword/semantic text arms miss it."""
+    sess = ChatSession(shop="x", tenant_id=1)
+    db_session.add(sess)
+    await db_session.flush()  # populate sess.id before referencing it
+    db_session.add(ProductChunk(tenant_id=1, shopify_product_id="P9", title="Mystery Item",
+                                text="zzz", embedding=_vec(0, 0, 0), image_embedding=_vec(1, 0, 0)))
+    db_session.add(Attachment(tenant_id=1, session_id=sess.id, mime="image/jpeg",
+                              text="una foto", image_embedding=_vec(0.95, 0.05, 0)))
+    await db_session.flush()
+
+    # keyword arm returns nothing; hydrate returns the live product for the visual id
+    import app.tools.products_tool as pt
+
+    async def fake_kw(client, query, max_products=50):
+        return []
+
+    async def fake_hydrate(client, ids):
+        return [{"product_id": "P9", "title": "Mystery Item", "price": "10", "available": True,
+                 "image": "http://img", "url": "http://u", "variants": [], "tags": [],
+                 "description": "", "product_type": ""}]
+
+    monkeypatch.setattr(pt, "search_products", fake_kw)
+    monkeypatch.setattr(pt, "fetch_products_by_ids", fake_hydrate)
+
+    ctx = ToolContext(db=db_session, session=sess, tenant_id=1, shopify=_FakeShopify(),
+                      shop="x.myshopify.com")
+    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)
+
+
+async def test_imgembed_rejects_bad_input():
+    # empty / non-image bytes -> None, without loading the CLIP model
+    assert await imgembed.embed_image(b"") is None
+    assert await imgembed.embed_image(b"not-an-image") is None
+    assert await imgembed.embed_image_url("ftp://nope") is None
+    assert await imgembed.embed_image_url("") is None

tests/test_orchestrator.py

@@ -450,3 +450,46 @@ async def test_no_cart_email_attach_when_recovery_off(db_session):
     rows = (await db_session.execute(
         select(AbandonedCart).where(AbandonedCart.session_id == sess.id))).scalars().all()
     assert rows == []
+
+
+class _CapturingRouter:
+    def __init__(self, results):
+        self._results = list(results)
+        self.seen = []
+
+    async def chat(self, messages, tools, tier="large", temperature=None, order=None,
+                   tool_choice=None, max_tokens=None):
+        self.seen.append([m.get("content", "") for m in messages])
+        return self._results.pop(0)
+
+
+async def test_bare_photo_forces_product_search(db_session):
+    """A photo uploaded with NO substantive text ('do you have this?') deterministically
+    forces a search_products call even if the model would have just chatted."""
+    sess = await _session(db_session)
+    router = _CapturingRouter([
+        ChatResult(content="Bonita foto.", tool_calls=[], finish_reason="stop"),   # model dodges
+        ChatResult(content="Esto es lo que tenemos.", tool_calls=[], finish_reason="stop"),
+    ])
+    ctx = ToolContext(db=db_session, session=sess, tenant_id=1)
+    await run_turn(
+        router, ctx, ATTACHMENT_MARKER + " foto.jpg]:\nuna zapatilla roja de running",
+        intent_text="", image_uploaded=True,
+    )
+    flat = " ".join(c for call in router.seen for c in call)
+    assert "search_products" in flat  # the deterministic force kicked in
+
+
+async def test_pdf_or_plain_text_does_not_force_product_search(db_session):
+    """No image uploaded + a normal question must NOT trigger the photo->product force."""
+    sess = await _session(db_session)
+    router = _CapturingRouter([
+        ChatResult(content="Te ayudo con eso.", tool_calls=[], finish_reason="stop"),
+        ChatResult(content="Aquí tienes.", tool_calls=[], finish_reason="stop"),
+    ])
+    ctx = ToolContext(db=db_session, session=sess, tenant_id=1)
+    await run_turn(router, ctx, "¿cuánto tarda el envío?", intent_text="¿cuánto tarda el envío?",
+                   image_uploaded=False)
+    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