imag vector generation with dimentionality reduction

app/services/image_embedder.py

@@ -13,7 +13,9 @@ THIS FILE OWNS TWO THINGS AND NOTHING ELSE
 ------------------------------------------
 1. `preprocess()` - bytes to the input tensor. It is the ONLY place the
    resize / crop / scaling decisions live, because those decisions are what
-   make the vectors comparable. See the banner on that function.
+   make the vectors comparable. It is a port of the Nearle Flutter app's
+   OpenCV pipeline (centre crop, INTER_AREA to 224, RGB, 0..1) and uses
+   OpenCV itself so the two agree to a few decimals.
 2. The interpreter - one per process, created lazily on first use, and every
    call into it serialised by one lock. A TFLite interpreter is not
    thread-safe, and this process has exactly two callers: the single
@@ -65,50 +67,81 @@ _warned = False
 # bytes -> input tensor
 # ---------------------------------------------------------------------------
 
+def _flatten_alpha_to_bgr(image_bytes: bytes) -> Optional[np.ndarray]:
+    """Pillow decode for images WITH transparency: EXIF applied, composited on
+    white, returned as uint8 BGR so the OpenCV steps below see exactly what
+    `cv2.imread` would have produced from an opaque file."""
+    from PIL import Image, ImageOps
+
+    im = Image.open(io.BytesIO(image_bytes))
+    im = ImageOps.exif_transpose(im) or im
+    rgba = im.convert("RGBA")
+    white = Image.new("RGBA", rgba.size, (255, 255, 255, 255))
+    rgb = Image.alpha_composite(white, rgba).convert("RGB")
+    return np.ascontiguousarray(np.asarray(rgb, dtype=np.uint8)[:, :, ::-1])
+
+
 def preprocess(image_bytes: bytes) -> Optional[np.ndarray]:
     """Decode `image_bytes` into the model's input tensor, or None.
 
-    ==== PREPROCESSING CONTRACT ==============================================
-    This is the DEFAULT recipe, written before the colleague's own code was
-    available. When theirs arrives, replace the body of this function with a
-    line-for-line port and delete this banner. The three decisions that decide
-    whether two systems' vectors agree are:
-        * resize method       (default: BILINEAR)
-        * squash vs crop      (default: squash the whole image to 224x224,
-                               no aspect-ratio preservation, no centre crop)
-        * value scaling       (default: raw 0..255 as float32 - what a Keras
-                               MobileNetV3 export expects, since the graph
-                               carries its own Rescaling layer)
-    Fixed in any variant: EXIF orientation applied (browsers apply it, so the
-    stored vector describes what a person sees), alpha flattened on white,
-    RGB channel order, NHWC layout, batch of one.
-    ==========================================================================
+    A line-for-line port of the Nearle Flutter app's preprocessing
+    (core/services/image_embed/image_embedder.dart, opencv_dart) and of the
+    colleague's Python reference for it. The steps, in their order:
 
-    No `Image.draft()` here, deliberately: a DCT-downscaled JPEG decode is not
-    bit-comparable with a full decode followed by a resize, and comparability
-    is the entire point. Never raises - pytest runs warnings as errors, and a
+        1. read as BGR            cv2.imdecode(..., IMREAD_COLOR)
+        2. centre square crop     side = min(w, h); x = (w - side) // 2; y = (h - side) // 2
+        3. resize to 224x224      cv2.resize(..., interpolation=INTER_AREA)
+        4. BGR -> RGB             cv2.cvtColor(..., COLOR_BGR2RGB)
+        5. uint8 -> float 0..1    astype(float32) / 255.0   (ONLY that - no mean,
+                                  no std, no -1..1; the model rescales internally)
+        6. batch dimension        [1, 224, 224, 3], HWC
+
+    OpenCV is used for the decode and the resize rather than Pillow on
+    purpose: INTER_AREA and Pillow's BOX filter are not the same filter, and
+    the two JPEG decoders differ by a pixel here and there. Matching the app
+    to 3-4 decimals is the requirement, so the app's library is the tool.
+
+    The two "backend-only extras" from the same spec: an image WITH
+    transparency is flattened onto white before step 1 (IMREAD_COLOR would
+    turn the transparent area black), and a greyscale image comes out of
+    IMREAD_COLOR as three channels already. cv2.imdecode applies EXIF
+    orientation like cv2.imread does, which is what the phone camera path
+    relies on.
+
+    Never raises - pytest runs warnings as errors, and a
     DecompressionBombWarning is one of the things the broad except absorbs.
     """
     if not image_bytes:
         return None
     try:
-        from PIL import Image, ImageOps
+        import cv2
+        from PIL import Image
 
-        im = Image.open(io.BytesIO(image_bytes))
-        if im.width * im.height > IMAGE_VECTOR_MAX_PIXELS:
-            logger.debug("image rejected: %dx%d exceeds pixel cap", im.width, im.height)
+        # Header only: the pixel-bomb guard and the transparency check both
+        # come from the file's metadata, no decode yet.
+        probe = Image.open(io.BytesIO(image_bytes))
+        if probe.width * probe.height > IMAGE_VECTOR_MAX_PIXELS:
+            logger.debug("image rejected: %dx%d exceeds pixel cap", probe.width, probe.height)
             return None
-        im = ImageOps.exif_transpose(im) or im
+        has_alpha = "A" in probe.getbands() or "transparency" in probe.info
 
-        if "A" in im.getbands() or "transparency" in im.info:
-            rgba = im.convert("RGBA")
-            white = Image.new("RGBA", rgba.size, (255, 255, 255, 255))
-            im = Image.alpha_composite(white, rgba)
-        im = im.convert("RGB")
-        im = im.resize((IMAGE_EMBED_INPUT_SIZE, IMAGE_EMBED_INPUT_SIZE), Image.Resampling.BILINEAR)
+        if has_alpha:
+            bgr = _flatten_alpha_to_bgr(image_bytes)
+        else:
+            bgr = cv2.imdecode(np.frombuffer(image_bytes, dtype=np.uint8), cv2.IMREAD_COLOR)   # 1
+        if bgr is None or bgr.ndim != 3 or bgr.shape[2] != 3:
+            logger.debug("image rejected: OpenCV could not decode it to BGR")
+            return None
 
-        arr = np.asarray(im, dtype=np.float32)          # (224, 224, 3), 0..255
-        arr = np.expand_dims(arr, axis=0)                # (1, 224, 224, 3)
+        h, w = bgr.shape[:2]                                                              # 2
+        side = min(w, h)
+        x, y = (w - side) // 2, (h - side) // 2
+        sq = bgr[y:y + side, x:x + side]
+
+        size = (IMAGE_EMBED_INPUT_SIZE, IMAGE_EMBED_INPUT_SIZE)
+        resized = cv2.resize(sq, size, interpolation=cv2.INTER_AREA)                      # 3
+        rgb = cv2.cvtColor(resized, cv2.COLOR_BGR2RGB)                                    # 4
+        arr = (rgb.astype(np.float32) / 255.0)[None]                                       # 5, 6
         if arr.shape != _INPUT_SHAPE:
             logger.debug("image rejected: tensor shape %s", arr.shape)
             return None
@@ -195,6 +228,36 @@ def available() -> bool:
         return _ensure_loaded()
 
 
+def status() -> dict:
+    """Diagnostics for /api/health - NEVER loads the model.
+
+    Exists because the failure this module is built to absorb is silent by
+    design: a container without the .tflite writes NULL vectors and says so
+    once, in a log line nobody reads. This puts the same facts on the health
+    endpoint, where "why are the vectors NULL after the deploy?" can be
+    answered with one curl.
+    """
+    path = IMAGE_EMBED_MODEL_PATH
+    try:
+        import importlib.util
+        runtime = importlib.util.find_spec("ai_edge_litert") is not None
+    except Exception:  # noqa: BLE001 - a broken finder counts as "not importable"
+        runtime = False
+    with _lock:
+        if _interpreter is not None:
+            state = "ready"
+        elif _disabled_reason:
+            state = f"disabled: {_disabled_reason}"
+        else:
+            state = "not loaded yet (loads on first write)"
+    return {
+        "model_path": str(path),
+        "model_present": path.is_file(),
+        "runtime_importable": runtime,
+        "state": state,
+    }
+
+
 def describe() -> str:
     """One line for script banners: where the model is and whether it works."""
     with _lock:

app/services/image_match.py

@@ -0,0 +1,296 @@
+"""Find catalog products from a phone photo: img_vector nearest-neighbour + label text.
+
+(Named image_match, not image_search: app/services/image_search.py is the
+image DISCOVERY module that finds photos for products. This is the reverse.)
+
+WHAT COMES IN
+-------------
+The Nearle app photographs a pack, crops it, embeds it on-device with the same
+MobileNetV3-Small model and preprocessing that filled `img_vector`
+(app/services/image_embedder.py), L2-normalises, and sends the 1024 floats -
+plus whatever OCR read off the label. Alternatively a client sends the photo
+and the API embeds it. Either way this module gets a unit vector and maybe
+some text.
+
+HOW A MATCH IS SCORED
+---------------------
+`score = 1 - (img_vector <=> q)`: cosine similarity, since both sides are unit
+length. This is the app team's convention and is NOT
+`RetrievedProduct.similarity` (`1 - distance/2`, rag_service.py), which is
+the text search's. A photo of a pack against the catalog's render of it
+scored 0.63 in the feasibility test; the app doc's "0.7 = same product" is a
+phone-vs-phone rule of thumb, so the server's default floor is 0.
+
+WHY TEXT IS PART OF IT
+----------------------
+Two reasons, both measured on the catalog:
+
+* Every pack size of one product usually shares one photo, so "Marie Gold
+  89g / 300g / 1kg" tie to the last decimal. The label text is the only thing
+  that can pick the 300g row: size tokens are normalised ("300 g", "300gm",
+  "300G" -> "300g") and weighted above plain words.
+* The brand name, when the OCR caught it, turns a 57-table scan into one
+  table via `query_intent.extract_brand_mention`. If that scope finds nothing
+  above `min_score` the search is retried unscoped and says so
+  (`scope_fallback`), because OCR misreads happen. An EXPLICIT `brand` never
+  falls back - the caller asked for a filter.
+
+The ranking is deterministic on purpose (`rank_key`): tied rows are ordered
+by text overlap, then name, then image_id, so the same request always
+returns the same order and the tests can pin it.
+
+Read-only: nothing here writes, and the pipeline stages are untouched.
+"""
+from __future__ import annotations
+
+import logging
+import math
+import re
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Sequence, Set, Tuple
+
+from app.infrastructure.settings import (
+    IMAGE_SEARCH_DEFAULT_MIN_SCORE,
+    IMAGE_SEARCH_DEFAULT_TOP_K,
+    IMAGE_SEARCH_MAX_FETCH_K,
+    IMAGE_SEARCH_MAX_TOP_K,
+)
+from app.services.image_embedder import EMBED_DIM
+
+logger = logging.getLogger(__name__)
+
+# The floor pgvector's hnsw needs to return LIMIT rows: ef_search < LIMIT
+# silently truncates the candidate list.
+_MIN_EF_SEARCH = 40
+
+_SIZE_RE = re.compile(r"(\d+(?:\.\d+)?)\s*(kg|gms|gm|g|ml|ltr|litre|l|pcs|pc|n)\b", re.I)
+_UNIT_ALIAS = {"gm": "g", "gms": "g", "ltr": "l", "litre": "l", "pc": "pcs", "n": "pcs"}
+_WORD_RE = re.compile(r"[a-z0-9]+")
+_STOP = {
+    "the", "a", "an", "of", "and", "with", "for", "pack", "new", "net", "wt",
+    "weight", "mrp", "rs", "inr", "in", "by", "per", "no", "nos",
+}
+_SIZE_WEIGHT = 3.0
+_WORD_WEIGHT = 1.0
+
+
+class InvalidVectorError(ValueError):
+    """The query vector cannot be searched with: wrong length, NaN, or zero."""
+
+
+@dataclass
+class ImageSearchResult:
+    rows: List[Dict[str, Any]] = field(default_factory=list)   # each carries "score" and "text_overlap"
+    detected_brand: Optional[str] = None                        # explicit brand, else the OCR-derived one
+    scoped_to_brand: bool = False                               # the rows came from one brand table
+    scope_fallback: bool = False                                # OCR scope was empty; retried unscoped
+    min_score: float = 0.0
+    query_text: Optional[str] = None
+    top_k: int = 0
+
+
+# ---------------------------------------------------------------------------
+# the query vector
+# ---------------------------------------------------------------------------
+
+def normalise_vector(vector: Sequence[float]) -> List[float]:
+    """`vector` as a unit-length list of EMBED_DIM floats, or InvalidVectorError.
+
+    Already-unit input (|norm - 1| < 1e-3) is returned as-is so the app's own
+    normalisation is not disturbed by float rounding; anything else is
+    rescaled, because a client that forgot to normalise should still get the
+    right neighbours rather than distances scaled by its norm.
+    """
+    try:
+        values = [float(v) for v in vector]
+    except (TypeError, ValueError) as exc:
+        raise InvalidVectorError(f"vector must be a list of numbers: {exc}") from None
+    if len(values) != EMBED_DIM:
+        raise InvalidVectorError(f"vector must have {EMBED_DIM} values, got {len(values)}")
+    if not all(math.isfinite(v) for v in values):
+        raise InvalidVectorError("vector contains NaN or infinite values")
+    norm = math.sqrt(sum(v * v for v in values))
+    if norm < 1e-6:
+        raise InvalidVectorError("vector is all zeros")
+    if abs(norm - 1.0) < 1e-3:
+        return values
+    return [v / norm for v in values]
+
+
+# ---------------------------------------------------------------------------
+# label text
+# ---------------------------------------------------------------------------
+
+def tokens(text: Optional[str]) -> Tuple[Set[str], Set[str]]:
+    """(words, sizes) from label text.
+
+    Sizes are normalised so the OCR's "300 g", a sheet's "300gm" and a
+    product name's "300G" all become "300g". Words are lowercase, alnum,
+    at least two characters, minus stop-words and minus anything that was
+    part of a size token (so "300" and "g" do not also count as words).
+    """
+    if not text:
+        return set(), set()
+    lowered = text.lower()
+    sizes: Set[str] = set()
+    for num, unit in _SIZE_RE.findall(lowered):
+        unit = _UNIT_ALIAS.get(unit, unit)
+        num = num.rstrip("0").rstrip(".") if "." in num else num
+        sizes.add(f"{num}{unit}")
+    without_sizes = _SIZE_RE.sub(" ", lowered)
+    words = {w for w in _WORD_RE.findall(without_sizes) if len(w) >= 2 and w not in _STOP}
+    return words, sizes
+
+
+def _row_text(row: Dict[str, Any]) -> str:
+    parts = [str(row.get("product_name") or ""), str(row.get("title") or "")]
+    variants = row.get("size_variants") or []
+    if isinstance(variants, (list, tuple)):
+        parts.extend(str(v) for v in variants if v)
+    return " ".join(parts)
+
+
+def text_overlap(words: Set[str], sizes: Set[str], row: Dict[str, Any]) -> float:
+    """How much of the label text this row's name accounts for.
+
+    3.0 per shared size token, 1.0 per shared word. Sizes weigh more because
+    they are what separates the pack sizes of one product, which is the tie
+    this exists to break; brand and product words match every sibling alike.
+    """
+    if not words and not sizes:
+        return 0.0
+    row_words, row_sizes = tokens(_row_text(row))
+    return _SIZE_WEIGHT * len(sizes & row_sizes) + _WORD_WEIGHT * len(words & row_words)
+
+
+def rank_key(row: Dict[str, Any]) -> tuple:
+    """Best first. Score rounded to 3 dp so siblings sharing a photo tie."""
+    return (
+        -round(float(row.get("score", 0.0)), 3),
+        -float(row.get("text_overlap", 0.0)),
+        str(row.get("product_name") or row.get("title") or "").lower(),
+        str(row.get("image_id") or ""),
+    )
+
+
+# ---------------------------------------------------------------------------
+# the search
+# ---------------------------------------------------------------------------
+
+def _candidates(
+    vector: List[float],
+    brand: Optional[str],
+    category: Optional[str],
+    fetch_k: int,
+    ef_search: int,
+    min_score: float,
+) -> List[Dict[str, Any]]:
+    from app.services.vector_store import image_vector_search
+
+    rows = image_vector_search(vector, brand=brand, top_k=fetch_k, category=category, ef_search=ef_search)
+    out: List[Dict[str, Any]] = []
+    seen: Set[Tuple[str, str]] = set()
+    for row in rows:
+        try:
+            score = 1.0 - float(row.get("distance"))
+        except (TypeError, ValueError):
+            continue
+        if score < min_score:
+            continue
+        key = (str(row.get("brand_table") or ""), str(row.get("image_id") or ""))
+        if key in seen:
+            continue
+        seen.add(key)
+        row["score"] = score
+        out.append(row)
+    return out
+
+
+def _hydrate(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """Swap the light candidate rows for full product rows, one read per table.
+
+    Ranking ran on ~100-byte rows (image_id, name, sizes, distance) because
+    the unscoped search pulls candidates from every brand table; only the
+    winners are worth a 7KB product card. `score` and `text_overlap` are
+    carried over. A row whose card cannot be read keeps its light form, so
+    a match is never lost to a hydration hiccup.
+    """
+    from app.services.vector_store import fetch_products_by_image_ids
+
+    by_table: Dict[str, List[str]] = {}
+    for row in rows:
+        by_table.setdefault(str(row.get("brand_table") or ""), []).append(str(row.get("image_id") or ""))
+    cards: Dict[Tuple[str, str], Dict[str, Any]] = {}
+    for table, ids in by_table.items():
+        if not table:
+            continue
+        for card in fetch_products_by_image_ids(table, ids):
+            cards[(table, str(card.get("image_id") or ""))] = card
+
+    out: List[Dict[str, Any]] = []
+    for row in rows:
+        key = (str(row.get("brand_table") or ""), str(row.get("image_id") or ""))
+        card = cards.get(key)
+        if card is None:
+            out.append(row)
+            continue
+        merged = dict(card)
+        merged["brand"] = merged.get("brand") or row.get("brand")
+        merged["brand_table"] = row.get("brand_table")
+        merged["score"] = row["score"]
+        merged["text_overlap"] = row.get("text_overlap", 0.0)
+        out.append(merged)
+    return out
+
+
+def search_by_vector(
+    vector: Sequence[float],
+    text: Optional[str] = None,
+    brand: Optional[str] = None,
+    category: Optional[str] = None,
+    top_k: int = IMAGE_SEARCH_DEFAULT_TOP_K,
+    min_score: float = IMAGE_SEARCH_DEFAULT_MIN_SCORE,
+) -> ImageSearchResult:
+    """The catalog rows most like `vector`, best first, at most `top_k`.
+
+    Raises InvalidVectorError for a vector that cannot be searched with.
+    Everything else degrades to an empty result (no database, no vectors).
+    """
+    unit = normalise_vector(vector)
+    top_k = max(1, min(int(top_k), IMAGE_SEARCH_MAX_TOP_K))
+    fetch_k = min(max(top_k * 3, 30), IMAGE_SEARCH_MAX_FETCH_K)
+    ef_search = max(_MIN_EF_SEARCH, fetch_k)
+    label = (text or "").strip() or None
+
+    explicit = (brand or "").strip() or None
+    detected = explicit
+    if detected is None and label:
+        try:
+            from app.services.query_intent import extract_brand_mention
+            detected = extract_brand_mention(label)
+        except Exception as exc:  # noqa: BLE001 - brand detection is an optimisation
+            logger.debug("brand detection skipped: %s", exc)
+            detected = None
+
+    rows = _candidates(unit, detected, category, fetch_k, ef_search, min_score)
+    scoped = detected is not None
+    fallback = False
+    if not rows and detected and not explicit:
+        rows = _candidates(unit, None, category, fetch_k, ef_search, min_score)
+        scoped, fallback = False, True
+
+    words, sizes = tokens(label)
+    for row in rows:
+        row["text_overlap"] = text_overlap(words, sizes, row)
+    rows.sort(key=rank_key)
+    winners = _hydrate(rows[:top_k])
+
+    return ImageSearchResult(
+        rows=winners,
+        detected_brand=detected,
+        scoped_to_brand=scoped,
+        scope_fallback=fallback,
+        min_score=min_score,
+        query_text=label,
+        top_k=top_k,
+    )

app/services/models/mobilenet/README.md

@@ -0,0 +1,33 @@
+# mobilenet_v3_small_embedder.tflite
+
+The image model behind `img_vector` (`app/services/image_embedder.py`):
+MobileNetV3-Small, input `[1, 224, 224, 3]` float32 in 0..1, output
+`[1, 1024]` (the hard-swish output of the 1024-wide `Conv_2` head layer),
+L2-normalised by the caller. Override the path with `IMAGE_EMBED_MODEL_PATH`.
+
+## Where this file came from
+
+Produced by `scripts/export_mobilenet_embedder.py` on 2026-09-17 (TensorFlow
+2.21.0 / Keras 3.15.1) from Keras' pretrained ImageNet weights, with a
+`Rescaling(2, -1)` layer inside the graph so the app's "divide by 255 and
+nothing else" rule holds. 6.1 MB, float32, builtin TFLite ops only. That
+script's docstring has the design and the exact steps to regenerate it.
+
+It is meant to be equivalent to the Nearle Flutter app's copy at
+`assets/models/mobilenet/mobilenet_v3_small_embedder.tflite`. Same
+architecture, weights, tap and input convention - but whether the numbers
+agree to the last decimal can only be checked against the app: take one
+cropped photo from the app, note the first eight values of its
+`[VECTOR][IMAGE]` log line, run `image_embedder.embedding_for_bytes()` on the
+same file, compare. If they differ, copy the app's file over this one and run
+`python -m scripts.backfill_image_vectors --all --force --apply`.
+
+## Why this directory and not `data/`
+
+`/app/data` is a Docker named volume on every deployment, so a file baked
+into the image under it is invisible on any volume that already exists.
+`app/` is `COPY`d into the image and never mounted.
+
+Commit the file (`.gitattributes` marks `*.tflite` binary). Without it the
+deployed container writes NULL into `img_vector` and reports
+`"model_present": false` under `image_vectors` on `GET /api/health`.

app/services/vector_store.py

@@ -1503,6 +1503,127 @@ def semantic_search(
     return results[:top_k]
 
 
+# The light projection the image search ranks on. Everything the tie-break
+# needs and nothing else: ~100 bytes a row instead of the ~7KB product card,
+# which matters because the unscoped search pulls `top_k` candidates from
+# EVERY brand table (58 x 30 rows) to keep a handful. The winners are then
+# hydrated by fetch_products_by_image_ids.
+_IMAGE_CANDIDATE_COLUMNS = "image_id, product_name, title, size_variants"
+
+
+def image_vector_search(
+    query_vector: List[float],
+    brand: Optional[str] = None,
+    top_k: int = 10,
+    category: Optional[str] = None,
+    ef_search: Optional[int] = None,
+) -> List[Dict[str, Any]]:
+    """Nearest catalog rows to an image embedding, by cosine distance on img_vector.
+
+    The read half of app/services/image_match.py: semantic_search() for the
+    1024-d MobileNetV3 column instead of the 384-d text column, but returning
+    only the LIGHT candidate columns (`_IMAGE_CANDIDATE_COLUMNS`) plus
+    `distance` (pgvector `<=>`, so 1 - distance is cosine similarity for unit
+    vectors), `brand` and `brand_table`. Hydrate the winners with
+    fetch_products_by_image_ids.
+
+    Unlike semantic_search this returns EVERY candidate, `top_k` PER TABLE,
+    merged and sorted by distance, and does not truncate: the pack sizes of
+    one product share a photo and tie exactly, and the caller breaks those
+    ties with the label text before choosing its top_k.
+
+    `ef_search` sets hnsw.ef_search for this connection (one per call, closed
+    below, so a plain SET is per-request). The index returns at most
+    ef_search candidates, so it must be >= LIMIT or the tied siblings are
+    silently dropped. A server that does not know the GUC just logs.
+    """
+    conn = _connect()
+    if not conn:
+        return []
+
+    embedding_str = "[" + ",".join(map(str, query_vector)) + "]"
+    results: List[Dict[str, Any]] = []
+
+    try:
+        with conn, conn.cursor() as cur:
+            if ef_search:
+                try:
+                    cur.execute(f"SET hnsw.ef_search = {int(ef_search)}")
+                except Exception as e:  # noqa: BLE001 - no hnsw, or a recorder cursor: exact scan still works
+                    logger.debug("hnsw.ef_search not applied: %s", e)
+
+            if brand:
+                table_name = _table_name(brand)
+                tables = [(brand, table_name)] if _table_exists(cur, table_name) else []
+            else:
+                # Straight from information_schema a moment ago; a table that
+                # vanishes in between fails inside the per-table try below.
+                tables = [(name, f"brand_{name}") for name in _list_brand_table_suffixes(cur)]
+
+            for brand_label, table_name in tables:
+                sql = (
+                    f"SELECT {_IMAGE_CANDIDATE_COLUMNS}, img_vector <=> %s::vector AS distance "
+                    f"FROM {table_name} WHERE img_vector IS NOT NULL"
+                )
+                params: List[Any] = [embedding_str]
+                if category:
+                    sql += " AND category ILIKE %s"
+                    params.append(f"%{category}%")
+                sql += " ORDER BY distance ASC LIMIT %s"
+                params.append(top_k)
+
+                try:
+                    cur.execute(sql, params)
+                except Exception as e:  # noqa: BLE001 - a table without the column must not break search
+                    logger.warning("Image vector search failed for table %s: %s", table_name, e)
+                    continue
+
+                # The unscoped loop labels by table suffix ("hindustan_unilever");
+                # give every hit the display name the scoped path already has.
+                label = brand_label if brand else display_name_for_suffix(brand_label)
+                colnames = [desc[0] for desc in cur.description]
+                for row in cur.fetchall():
+                    record = dict(zip(colnames, row))
+                    record["brand"] = label
+                    record["brand_table"] = table_name
+                    results.append(record)
+    finally:
+        conn.close()
+
+    results.sort(key=lambda r: r.get("distance", 9.0))
+    return results
+
+
+def fetch_products_by_image_ids(table_name: str, image_ids: List[str]) -> List[Dict[str, Any]]:
+    """Full product rows (vector columns projected out) for `image_ids` in one table.
+
+    The hydration step of the image search: only the rows that survived
+    ranking are read in full. Order is not significant; the caller keys by
+    image_id. An unknown table or a failed read yields [] rather than raising.
+    """
+    if not image_ids:
+        return []
+    conn = _connect()
+    if not conn:
+        return []
+    try:
+        with conn, conn.cursor() as cur:
+            if not _table_exists(cur, table_name):
+                return []
+            try:
+                cur.execute(
+                    f"SELECT {_product_columns(cur, table_name)} FROM {table_name} WHERE image_id = ANY(%s)",
+                    (list(image_ids),),
+                )
+            except Exception as e:  # noqa: BLE001
+                logger.warning("Product hydration failed for table %s: %s", table_name, e)
+                return []
+            colnames = [desc[0] for desc in cur.description]
+            return [dict(zip(colnames, row)) for row in cur.fetchall()]
+    finally:
+        conn.close()
+
+
 def text_search(
     query: str,
     brand: Optional[str] = None,