feat(matching): Jev entity matching phases 2-5 (per-field fan-out, variants, candidate ambiguity buttons)

Phase 2 — jev.py client: POST /v1/systemone, PINNED model jev-1.13.0 (drift warning when the response model differs), one retry on 429/5xx, fail-soft (None on any error), usage/latency returned for observability. Config adds TYPESAFE_API_KEY / JEV_* (thresholds 0.85/0.60, 15 questions per field); no daily cap by owner decision. bot/.env is now mounted read-only into the container so the key never enters compose/git.

Phase 3 — matching.py: برون‌گروهی (one request per field, parallel), درون‌گروهی (one choice question per mention sharing the same full candidate list + «_none»), درون‌واریانسی (all STT spellings in one question). Routing: conf>=exact → bind; floor..exact → ambiguity; else/_none → new; missing/failed field → difflib fallback (bug fixed during testing: a failed request used to record 'new' instead of falling back). Resolver gains full cached lists for suppliers/materials (paged) plus brands (API+DB) and specs (EAV values + draft specs). NLU prompt now emits entity_mentions (6 fields) with the voice-variant grouping rule; sanitize_mentions guards the shape. handle_message runs matching only for register_quote/update_draft/add_material behind JEV_ENABLED; jev_map threads into quote/material resolve_payload (Jev hit skips difflib; exact writes ids; ambiguity skips difflib). Spec effect per owner: spelling normalization only (normalize_brand_spec).

Phase 4 — entity-match ambiguity sub-messages: create_ambiguities() builds deduped entries {kind: entity_match, field, variants, candidates, details with Persian percentages}; new keyboards.ambiguity_candidates_keyboard renders one button per candidate + «هیچ‌کدام — ایجاد جدید»; callbacks handle amb⛏️{prefix}:{amb_id}:{idx} (applies the candidate via matching.apply_choice then re-resolves/renders) and amb:new (rejected → auto-create path at commit). Every user choice lands in observability (ambiguity.user_choice with arg+candidates).

Phase 5 — voice variants: NLU groups different STT spellings of one entity into one variants[] (verified live: مصطفوی/مستفدی → one supplier mention); tools/test_matching.py live section (RUN_LIVE=1) exercises real DeepSeek + real Jev without touching Bale.

Tests: test_jev.py 5/5 (mock HTTP: no-key, happy path, 429-retry, 400, exception), test_matching.py 11/11 mock + 3/3 live (real NLU produced 7 mentions with correct fields; real Jev matched «50 کیلویی»→«۵۰ کیلویی» 0.98 and «کیسه» 1.0 and correctly returned new against the emptied catalog); observability regression 8/8.

bot/app/api/client.py

@@ -67,8 +67,8 @@ class ErpClient:
 
     # ---------------------------------------------------------------- lookup
 
-    async def list_suppliers(self, q: str | None = None, page_size: int = 20) -> list[dict]:
-        params: dict = {"page_size": page_size}
+    async def list_suppliers(self, q: str | None = None, page_size: int = 20, page: int = 1) -> list[dict]:
+        params: dict = {"page_size": page_size, "page": page}
         if q:
             params["q"] = q
         return (await self._request("GET", "/suppliers", params=params)).get("data") or []
@@ -76,8 +76,8 @@ class ErpClient:
     async def create_supplier(self, payload: dict, actor_id: str | None = None) -> dict:
         return (await self._request("POST", "/suppliers", json=payload, actor_id=actor_id)).get("data") or {}
 
-    async def list_materials(self, q: str | None = None, page_size: int = 20) -> list[dict]:
-        params: dict = {"page_size": page_size}
+    async def list_materials(self, q: str | None = None, page_size: int = 20, page: int = 1) -> list[dict]:
+        params: dict = {"page_size": page_size, "page": page}
         if q:
             params["q"] = q
         return (await self._request("GET", "/materials", params=params)).get("data") or []
@@ -100,6 +100,9 @@ class ErpClient:
     async def list_categories(self, page_size: int = 100) -> list[dict]:
         return (await self._request("GET", "/material-categories", params={"page_size": page_size})).get("data") or []
 
+    async def list_brands(self, page: int = 1, page_size: int = 100) -> list[dict]:
+        return (await self._request("GET", "/brands", params={"page": page, "page_size": page_size})).get("data") or []
+
     # ---------------------------------------------------------------- quotes
 
     async def list_quotes(self, *, status: str | None = None, q: str | None = None,

bot/app/config.py

@@ -139,6 +139,17 @@ class Settings(BaseSettings):
     OBS_VOICE_DIR: str = "data/voices"   # persistent volume (./bot/data) — keep forever
     OBS_PROMPT_VERSION: str = ""         # manual label for the NLU prompt in use
 
+    # Jev (TypeSafe) entity matching — phases 2-5. Model version is PINNED
+    # (phase 0 resolved jev-1.13.0) so upstream updates cannot silently change
+    # decisions; every response's actual model is recorded for drift detection.
+    TYPESAFE_API_KEY: str = ""
+    JEV_MODEL: str = "jev-1.13.0"
+    JEV_ENABLED: bool = True
+    JEV_TIMEOUT: float = 8.0
+    JEV_CONF_EXACT: float = 0.85
+    JEV_CONF_FLOOR: float = 0.60
+    JEV_MAX_QUESTIONS_PER_FIELD: int = 15
+
     LOG_LEVEL: str = "INFO"
 
     @property

bot/app/handlers/__init__.py

@@ -224,7 +224,20 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
     result = await understand(settings, input_payload, draft_summary, trace_id=trace_id)
     intent, patch = result.get("intent"), result.get("patch") or {}
     consensus_text = result.get("consensus_text") or text
-    await obs.event(trace_id, "entities.extracted", payload={"patch": patch, "intent": intent})
+    mentions = result.get("entity_mentions") or []
+    await obs.event(trace_id, "entities.extracted",
+                    payload={"patch": patch, "intent": intent, "entity_mentions": mentions})
+
+    # Phase 3: Jev per-field matching — only when it can be applied (flag-gated;
+    # a failure leaves jev_map empty and the difflib path runs unchanged).
+    jev_map: dict = {}
+    if settings.JEV_ENABLED and mentions and intent in ("register_quote", "update_draft", "add_material"):
+        try:
+            from app import matching
+            jev_map = await matching.resolve_mentions(ctx, patch, mentions, trace_id)
+        except Exception:
+            logger.exception("jev matching failed")
+            jev_map = {}
 
     if source == "voice":
         await ctx._log(chat_id, "in", "voice", consensus_text)
@@ -238,6 +251,7 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
             incoming_msg_id=message_id,
             target_draft=target_draft,
             trace_id=trace_id,
+            jev_map=jev_map,
         )
     elif intent == "commit_draft":
         active = target_draft or await draft_store.get_active(chat_id)
@@ -280,7 +294,7 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
     elif intent == "add_supplier":
         await supplier.handle_entities(ctx, chat_id, user, patch, source, trace_id=trace_id)
     elif intent == "add_material":
-        await material.handle_entities(ctx, chat_id, user, patch, source, trace_id=trace_id)
+        await material.handle_entities(ctx, chat_id, user, patch, source, trace_id=trace_id, jev_map=jev_map)
     elif intent == "add_project":
         await project.handle_entities(ctx, chat_id, user, patch, source, trace_id=trace_id)
     elif intent == "status_query":

bot/app/handlers/callbacks.py

@@ -82,6 +82,8 @@ async def _dispatch(ctx, chat_id: int, data: str, cq: dict) -> None:
         await obs.event(tid, "ambiguity.user_choice", payload={
             "amb_id": amb_id, "action": action, "kind": target_amb.get("kind"),
             "title": target_amb.get("title"),
+            "arg": parts[4] if len(parts) > 4 else None,
+            "candidates": target_amb.get("candidates"),
         })
 
         if action == "ok":
@@ -107,6 +109,38 @@ async def _dispatch(ctx, chat_id: int, data: str, cq: dict) -> None:
                     await ctx.delete(chat_id, target_amb["message_id"])
                 except Exception:
                     pass
+        elif action == "pick" and len(parts) >= 5:
+            # Phase 4: user picked one of the Jev candidates.
+            try:
+                idx = int(parts[4])
+            except ValueError:
+                return
+            cands = target_amb.get("candidates") or []
+            if not (0 <= idx < len(cands)):
+                return
+            from app import matching
+            matching.apply_choice(payload, target_amb.get("field"), cands[idx],
+                                  target_amb.get("variants") or [])
+            target_amb["status"] = "applied"
+            await quote.resolve_payload(ctx, payload)
+            await draft_store.save_payload(draft, mark_bale_synced=True)
+            # In-place edit main review card
+            await quote.render(ctx, chat_id, draft, fresh=False)
+            # Delete ONLY this ambiguity message
+            if target_amb.get("message_id"):
+                try:
+                    await ctx.delete(chat_id, target_amb["message_id"])
+                except Exception:
+                    pass
+        elif action == "new":
+            # «هیچ‌کدام» — keep the raw name; the auto-create path handles it at commit.
+            target_amb["status"] = "rejected"
+            await draft_store.save_payload(draft, mark_bale_synced=True)
+            if target_amb.get("message_id"):
+                try:
+                    await ctx.delete(chat_id, target_amb["message_id"])
+                except Exception:
+                    pass
         return
 
     # ---- pick a candidate for a draft field -------------------------------

bot/app/handlers/material.py

@@ -2,6 +2,7 @@
 import logging
 
 from app import drafts as draft_store
+from app import matching
 from app import observability as obs
 from app.api.client import ApiError
 from app.drafts import (
@@ -15,7 +16,8 @@ logger = logging.getLogger("bot.material")
 
 
 async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None,
-                          trace_id: str | None = None) -> None:
+                          trace_id: str | None = None,
+                          jev_map: dict | None = None) -> None:
     draft = await draft_store.get_active(chat_id, KIND_MATERIAL)
     if draft is not None:
         payload = draft.payload
@@ -25,25 +27,36 @@ async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None,
         merge_material_patch(payload, patch)
         draft = await draft_store.create(chat_id, KIND_MATERIAL, payload, source=source)
     await obs.link_draft(trace_id, draft.id, draft.kind)
-    await resolve_payload(ctx, payload, trace_id)
+    await resolve_payload(ctx, payload, trace_id, jev_map)
     await draft_store.save_payload(draft, missing={"fields": material_missing(payload)})
     await render(ctx, chat_id, draft, fresh=draft.review_message_id is None, trace_id=trace_id)
 
 
-async def resolve_payload(ctx, payload: dict, trace_id: str | None = None) -> None:
+async def resolve_payload(ctx, payload: dict, trace_id: str | None = None,
+                          jev_map: dict | None = None) -> None:
     options = payload.setdefault("options", {})
     if not payload.get("canonical_unit_id") and payload.get("unit_name"):
-        res = await safe_resolve(ctx.resolver, "unit", payload["unit_name"])
-        await obs.log_resolution(trace_id, "unit", payload["unit_name"], res.status,
-                                 [{"id": c.id, "name": c.name} for c in (res.candidates or [])[:4]])
-        if res.resolved:
-            payload["canonical_unit_id"] = res.candidates[0].id
-            payload["unit_name"] = res.candidates[0].name
-            options.pop("unit", None)
-        elif res.candidates:
-            options["unit"] = [{"id": c.id, "name": c.name} for c in res.candidates]
+        hit = matching.lookup(jev_map, payload["unit_name"])
+        if hit is not None:
+            if hit["status"] == "exact":
+                payload["canonical_unit_id"] = hit["id"]
+                payload["unit_name"] = hit["name"]
+                options.pop("unit", None)
+            elif hit["status"] == "ambiguity" and hit.get("candidates"):
+                options["unit"] = [{"id": c["id"], "name": c["name"]}
+                                   for c in hit["candidates"][:6]]
         else:
-            options.pop("unit", None)
+            res = await safe_resolve(ctx.resolver, "unit", payload["unit_name"])
+            await obs.log_resolution(trace_id, "unit", payload["unit_name"], res.status,
+                                     [{"id": c.id, "name": c.name} for c in (res.candidates or [])[:4]])
+            if res.resolved:
+                payload["canonical_unit_id"] = res.candidates[0].id
+                payload["unit_name"] = res.candidates[0].name
+                options.pop("unit", None)
+            elif res.candidates:
+                options["unit"] = [{"id": c.id, "name": c.name} for c in res.candidates]
+            else:
+                options.pop("unit", None)
 
 
 async def apply_pick(ctx, chat_id, user, draft, field: str, choice: dict) -> None:
@@ -105,5 +118,5 @@ async def commit(ctx, chat_id, user, draft, trace_id: str | None = None) -> None
     await obs.event(trace_id, "draft.commit", payload={"kind": "material",
                                                        "material_id": (created or {}).get("id")})
     await obs.set_trace_status(trace_id, "committed")
-    ctx.resolver.invalidate("materials")
+    ctx.resolver.invalidate()
     await ctx.send(chat_id, material_committed_report(created))

bot/app/handlers/project.py

@@ -72,5 +72,5 @@ async def commit(ctx, chat_id, user, draft, trace_id: str | None = None) -> None
     await obs.event(trace_id, "draft.commit", payload={"kind": "project",
                                                        "project_id": (created or {}).get("id")})
     await obs.set_trace_status(trace_id, "committed")
-    ctx.resolver.invalidate("projects")
+    ctx.resolver.invalidate()
     await ctx.send(chat_id, project_committed_report(created))

bot/app/handlers/quote.py

@@ -2,6 +2,7 @@
 import logging
 
 from app import drafts as draft_store
+from app import matching
 from app import observability as obs
 from app.api.client import ApiError
 from app.drafts import (
@@ -22,6 +23,7 @@ async def handle_entities(
     incoming_msg_id: int | None = None,
     target_draft = None,
     trace_id: str | None = None,
+    jev_map: dict | None = None,
 ) -> None:
     prev_review_msg_id = None
     if target_draft is not None:
@@ -50,7 +52,10 @@ async def handle_entities(
         )
 
     await obs.link_draft(trace_id, draft.id, draft.kind)
-    await resolve_payload(ctx, payload, trace_id)
+    await resolve_payload(ctx, payload, trace_id, jev_map)
+    if jev_map:
+        matching.normalize_brand_spec(payload, jev_map)
+        matching.create_ambiguities(payload, jev_map)
 
     if prev_review_msg_id:
         try:
@@ -71,46 +76,65 @@ async def _log_res(trace_id, field: str, mention, res) -> None:
                              [{"id": c.id, "name": c.name} for c in (res.candidates or [])[:4]])
 
 
-async def resolve_payload(ctx, payload: dict, trace_id: str | None = None) -> None:
+async def resolve_payload(ctx, payload: dict, trace_id: str | None = None,
+                          jev_map: dict | None = None) -> None:
     options = payload.setdefault("options", {})
 
     if not payload.get("supplier_id") and payload.get("supplier_name"):
-        res = await safe_resolve(ctx.resolver, "supplier", payload["supplier_name"])
-        await _log_res(trace_id, "supplier", payload["supplier_name"], res)
-        if res.resolved:
-            c = res.candidates[0]
-            payload["supplier_id"] = c.id
-            payload["supplier_name"] = c.name
-            extra = c.extra or {}
-            for k in ("location", "phone", "contact_person", "category_name"):
-                if extra.get(k):
-                    payload[f"supplier_{k}"] = extra[k]
-            options.pop("sup", None)
-        elif res.candidates:
-            options["sup"] = [{"id": c.id, "name": c.name} for c in res.candidates]
+        hit = matching.lookup(jev_map, payload["supplier_name"])
+        if hit is not None:
+            if hit["status"] == "exact":
+                matching.apply_exact(payload, hit)
+                options.pop("sup", None)
         else:
-            options.pop("sup", None)
+            res = await safe_resolve(ctx.resolver, "supplier", payload["supplier_name"])
+            await _log_res(trace_id, "supplier", payload["supplier_name"], res)
+            if res.resolved:
+                c = res.candidates[0]
+                payload["supplier_id"] = c.id
+                payload["supplier_name"] = c.name
+                extra = c.extra or {}
+                for k in ("location", "phone", "contact_person", "category_name"):
+                    if extra.get(k):
+                        payload[f"supplier_{k}"] = extra[k]
+                options.pop("sup", None)
+            elif res.candidates:
+                options["sup"] = [{"id": c.id, "name": c.name} for c in res.candidates]
+            else:
+                options.pop("sup", None)
 
     resolved_projects = payload.setdefault("resolved_projects", {})
 
     if not payload.get("project_id") and payload.get("project_name"):
-        res = await safe_resolve(ctx.resolver, "project", payload["project_name"])
-        await _log_res(trace_id, "project", payload["project_name"], res)
-        if res.resolved:
-            c = res.candidates[0]
-            payload["project_id"] = c.id
-            payload["project_name"] = c.name
-            resolved_projects[c.name] = c.id
-            resolved_projects[payload["project_name"]] = c.id
-            if payload["project_id"] not in payload.get("project_ids", []):
-                payload.setdefault("project_ids", []).append(payload["project_id"])
-            options.pop("prj", None)
-        elif res.candidates:
-            options["prj"] = [{"id": c.id, "name": c.name} for c in res.candidates]
+        hit = matching.lookup(jev_map, payload["project_name"])
+        if hit is not None:
+            if hit["status"] == "exact":
+                matching.apply_exact(payload, hit)
+                payload["project_name"] = hit["name"]
+                options.pop("prj", None)
         else:
-            options.pop("prj", None)
+            res = await safe_resolve(ctx.resolver, "project", payload["project_name"])
+            await _log_res(trace_id, "project", payload["project_name"], res)
+            if res.resolved:
+                c = res.candidates[0]
+                payload["project_id"] = c.id
+                payload["project_name"] = c.name
+                resolved_projects[c.name] = c.id
+                resolved_projects[payload["project_name"]] = c.id
+                if payload["project_id"] not in payload.get("project_ids", []):
+                    payload.setdefault("project_ids", []).append(payload["project_id"])
+                options.pop("prj", None)
+            elif res.candidates:
+                options["prj"] = [{"id": c.id, "name": c.name} for c in res.candidates]
+            else:
+                options.pop("prj", None)
 
     for p_name in payload.get("project_names") or []:
+        hit = matching.lookup(jev_map, p_name)
+        if hit is not None:
+            if hit["status"] == "exact":
+                matching.apply_exact(payload, hit)
+            continue
         res = await safe_resolve(ctx.resolver, "project", p_name)
         await _log_res(trace_id, "project", p_name, res)
         if res.resolved:
@@ -125,17 +149,23 @@ async def resolve_payload(ctx, payload: dict, trace_id: str | None = None) -> No
 
     for i, item in enumerate(payload.get("items") or [], start=1):
         if not item.get("material_id") and item.get("material_name"):
-            res = await safe_resolve(ctx.resolver, "material", item["material_name"])
-            await _log_res(trace_id, "material", item["material_name"], res)
-            if res.resolved:
-                _set_material(item, res.candidates[0].id, res.candidates[0].name, res.candidates[0].extra)
-                options.pop(f"mat{i}", None)
-            elif res.candidates:
-                options[f"mat{i}"] = [{"id": c.id, "name": c.name} for c in res.candidates]
+            hit = matching.lookup(jev_map, item["material_name"])
+            if hit is not None:
+                if hit["status"] == "exact":
+                    matching.apply_exact(payload, hit)
+                    options.pop(f"mat{i}", None)
             else:
-                options.pop(f"mat{i}", None)
+                res = await safe_resolve(ctx.resolver, "material", item["material_name"])
+                await _log_res(trace_id, "material", item["material_name"], res)
+                if res.resolved:
+                    _set_material(item, res.candidates[0].id, res.candidates[0].name, res.candidates[0].extra)
+                    options.pop(f"mat{i}", None)
+                elif res.candidates:
+                    options[f"mat{i}"] = [{"id": c.id, "name": c.name} for c in res.candidates]
+                else:
+                    options.pop(f"mat{i}", None)
         if not item.get("unit_id"):
-            await _resolve_unit(ctx, item, trace_id)
+            await _resolve_unit(ctx, item, trace_id, jev_map)
 
 
 def _set_material(item: dict, material_id: str, name: str, extra: dict | None = None) -> None:
@@ -143,17 +173,26 @@ def _set_material(item: dict, material_id: str, name: str, extra: dict | None =
     item["material_row"] = {"id": material_id, "canonical_unit_id": (extra or {}).get("canonical_unit_id")}
 
 
-async def _resolve_unit(ctx, item: dict, trace_id: str | None = None) -> None:
+async def _resolve_unit(ctx, item: dict, trace_id: str | None = None,
+                        jev_map: dict | None = None) -> None:
     if item.get("unit_id"):
         return
     if item.get("unit_name"):
-        res = await safe_resolve(ctx.resolver, "unit", item["unit_name"])
-        await _log_res(trace_id, "unit", item["unit_name"], res)
-        if res.resolved:
-            item["unit_id"] = res.candidates[0].id
-            item["unit_explicit"] = True
-            item["unit_name"] = res.candidates[0].name
-            return
+        hit = matching.lookup(jev_map, item["unit_name"])
+        if hit is not None:
+            if hit["status"] == "exact":
+                item["unit_id"] = hit["id"]
+                item["unit_explicit"] = True
+                item["unit_name"] = hit["name"]
+                return
+        else:
+            res = await safe_resolve(ctx.resolver, "unit", item["unit_name"])
+            await _log_res(trace_id, "unit", item["unit_name"], res)
+            if res.resolved:
+                item["unit_id"] = res.candidates[0].id
+                item["unit_explicit"] = True
+                item["unit_name"] = res.candidates[0].name
+                return
     canonical = ((item.get("material_row") or {}).get("canonical_unit_id"))
     if canonical:
         item["unit_id"] = str(canonical)

bot/app/handlers/supplier.py

@@ -110,5 +110,5 @@ async def commit(ctx, chat_id, user, draft, trace_id: str | None = None) -> None
     await obs.event(trace_id, "draft.commit", payload={"kind": "supplier",
                                                        "supplier_id": (created or {}).get("id")})
     await obs.set_trace_status(trace_id, "committed")
-    ctx.resolver.invalidate("suppliers")
+    ctx.resolver.invalidate()
     await ctx.send(chat_id, supplier_committed_report(created))

bot/app/jev.py

@@ -0,0 +1,67 @@
+"""Jev (TypeSafe) client — structured `choice` matching over HTTP (phases 2-5).
+
+Contract (docs.typesafe.ai): POST /v1/systemone with
+  {"state": ..., "model": <pinned>, "questions": {key: {type, instructions, criteria}}}
+→ {"answers": {key: {choice, confidence, probabilities}}, "usage": {...}, "model": "..."}
+
+Design rules (final plan v5):
+- The model version is PINNED (JEV_MODEL=jev-1.13.0 from phase 0); a response
+  whose `model` differs logs a drift warning and is still returned (recorded).
+- Fail-soft: any error/timeout → None. Callers fall back per field to the
+  existing difflib path; with JEV_ENABLED=false nothing here is called at all.
+- One retry only for 429/5xx (after a short backoff), then give up.
+"""
+import asyncio
+import logging
+import time
+
+import httpx
+
+from app.config import settings
+
+logger = logging.getLogger("bot.jev")
+
+_URL = "https://api.typesafe.ai/v1/systemone"
+
+
+async def ask(state: dict, questions: dict, *, timeout: float | None = None) -> dict | None:
+    """One systemone request. Returns {"answers", "usage", "model", "latency_ms"} or None."""
+    if not settings.TYPESAFE_API_KEY:
+        logger.warning("jev: TYPESAFE_API_KEY is not configured")
+        return None
+
+    payload = {"state": state, "model": settings.JEV_MODEL, "questions": questions}
+    headers = {
+        "Authorization": f"Bearer {settings.TYPESAFE_API_KEY}",
+        "Content-Type": "application/json",
+    }
+    t0 = time.perf_counter()
+    for attempt in (1, 2):
+        try:
+            async with httpx.AsyncClient(timeout=timeout or settings.JEV_TIMEOUT) as client:
+                resp = await client.post(_URL, json=payload, headers=headers)
+            if resp.status_code == 200:
+                body = resp.json() or {}
+                actual = str(body.get("model") or "")
+                if actual and actual != settings.JEV_MODEL:
+                    logger.warning("jev: model drift — got %s, pinned %s", actual, settings.JEV_MODEL)
+                return {
+                    "answers": body.get("answers") or {},
+                    "usage": body.get("usage") or {},
+                    "model": actual or settings.JEV_MODEL,
+                    "latency_ms": int((time.perf_counter() - t0) * 1000),
+                }
+            if resp.status_code in (429, 500, 502, 503, 504) and attempt == 1:
+                logger.warning("jev: HTTP %s — retrying once", resp.status_code)
+                await asyncio.sleep(1.0)
+                continue
+            logger.error("jev: HTTP %s: %s", resp.status_code, resp.text[:300])
+            return None
+        except Exception as exc:
+            if attempt == 1:
+                logger.warning("jev: request failed (%s) — retrying once", exc)
+                await asyncio.sleep(1.0)
+                continue
+            logger.error("jev: request failed: %s", exc)
+            return None
+    return None

bot/app/keyboards.py

@@ -81,3 +81,13 @@ def ambiguity_keyboard(prefix: str, amb_id: str) -> dict:
         [_btn("❌ تایید نمی‌شود", f"amb:no:{prefix}:{amb_id}")],
     ])
 
+
+def ambiguity_candidates_keyboard(prefix: str, amb_id: str, candidates: list[dict]) -> dict:
+    """Phase 4: one button per Jev candidate (+ 'new') for an entity-match ambiguity."""
+    rows = [
+        [_btn(f"✅ {str(c.get('name') or '')[:40]}", f"amb:pick:{prefix}:{amb_id}:{i}")]
+        for i, c in enumerate(candidates[:5])
+    ]
+    rows.append([_btn("🆕 هیچ‌کدام — ایجاد مورد جدید", f"amb:new:{prefix}:{amb_id}")])
+    return _kb(rows)
+

bot/app/matching.py

@@ -0,0 +1,321 @@
+"""Entity matching with Jev (TypeSafe) — phases 3-5 of the final plan.
+
+Mechanics (final plan v5 §2):
+- برون‌گروهی: ONE Jev request per FIELD that has mentions (requests run in
+  parallel; latency ≈ one round-trip regardless of field count).
+- درون‌گروهی: inside a field request, ONE `choice` question per mention; every
+  question rides with the SAME full candidate list (+ a `_none` option).
+- درون‌واریانسی: all STT spellings of one entity travel in ONE question, and
+  the instructions ask the model to treat them as one name.
+
+Routing (thresholds from env; calibrated in phase 6):
+  conf >= JEV_CONF_EXACT          → exact  (bind id, no ambiguity sub-message)
+  JEV_CONF_FLOOR..JEV_CONF_EXACT  → ambiguity (candidate buttons sub-message)
+  below floor / _none             → new    (existing auto-create path)
+
+Fail-soft: a failed field simply yields no results for its mentions; callers
+fall back to the difflib path (`safe_resolve`). With JEV_ENABLED=false nothing
+in this module runs (zero regression).
+"""
+import asyncio
+import hashlib
+import json
+import logging
+
+from app import observability as obs
+from app.config import settings
+from app.jev import ask as jev_ask
+from app.nlu import normalize_text
+
+logger = logging.getLogger("bot.matching")
+
+FIELD_ORDER = ["project", "supplier", "material", "brand", "spec", "unit"]
+
+FIELD_FA = {
+    "project": "پروژه",
+    "supplier": "تأمین‌کننده",
+    "material": "متریال",
+    "brand": "برند",
+    "spec": "مشخصه",
+    "unit": "واحد",
+}
+
+
+# ---------------------------------------------------------------------------
+# candidate lists (full lists — no prefilter, per owner decision)
+# ---------------------------------------------------------------------------
+
+async def _candidates(resolver, field: str) -> list:
+    if field == "project":
+        return await resolver.projects()
+    if field == "supplier":
+        return await resolver.all_suppliers()
+    if field == "material":
+        return await resolver.all_materials()
+    if field == "brand":
+        return await resolver.brands()
+    if field == "spec":
+        return await resolver.specs()
+    if field == "unit":
+        return await resolver.units()
+    return []
+
+
+def _hint(field: str, cand) -> str:
+    """Short, token-cheap description shown next to a candidate name."""
+    if field == "supplier":
+        extra = cand.extra or {}
+        bits = [b for b in (extra.get("category_name"), extra.get("location")) if b]
+        return "، ".join(str(b) for b in bits) if bits else "تأمین‌کننده"
+    return ""
+
+
+def _instructions(field: str, variants: list[str]) -> str:
+    fa = FIELD_FA.get(field, field)
+    vlist = "»، «".join(v for v in variants if v)
+    if len(variants) > 1:
+        return (
+            f"کاربر در پیام (صوتی) خود این {fa} را ذکر کرده و مدل‌های مختلف STT آن را به شکل‌های "
+            f"«{vlist}» پیاده کرده‌اند؛ این‌ها چند نگارش از یک نام واحدند. با توجه به بافت جمله و "
+            f"تشابه آوایی میان این واریانت‌ها، مشخص کن به کدام گزینه از فهرست اشاره دارد. "
+            f"اگر هیچ‌کدام منطبق نیست یا تردید جدی داری «_none» را انتخاب کن."
+        )
+    return (
+        f"کاربر «{vlist}» را به‌عنوان {fa} ذکر کرده است. این عبارت به کدام گزینه از فهرست اشاره دارد؟ "
+        f"اگر هیچ‌کدام منطبق نیست یا تردید جدی داری «_none» را انتخاب کن."
+    )
+
+
+# ---------------------------------------------------------------------------
+# the per-field fan-out
+# ---------------------------------------------------------------------------
+
+async def resolve_mentions(ctx, patch: dict, mentions: list[dict], trace_id) -> dict:
+    """Run Jev per field (parallel); return a map keyed by normalized variant.
+
+    Each value: {"field", "status": exact|ambiguity|new", "variants",
+                 "id"?, "name"?, "extra"?, "confidence"?, "candidates": [...]}
+    """
+    by_field: dict[str, list[dict]] = {}
+    for m in mentions or []:
+        field = str((m or {}).get("field") or "").strip()
+        variants = [str(v).strip() for v in ((m or {}).get("variants") or []) if str(v or "").strip()]
+        if field in FIELD_ORDER and variants:
+            by_field.setdefault(field, []).append({"field": field, "variants": variants[:6]})
+
+    results: dict[str, dict] = {}
+    if not by_field:
+        return results
+
+    async def one_field(field: str, ms: list[dict]) -> None:
+        try:
+            cands = await _candidates(ctx.resolver, field)
+        except Exception:
+            logger.exception("matching: candidate fetch failed for %s", field)
+            return
+        if not cands:
+            return
+
+        by_name: dict[str, object] = {}
+        criteria: dict[str, str] = {}
+        for c in cands:
+            if c.name in criteria:
+                continue
+            criteria[c.name] = _hint(field, c)
+            by_name[c.name] = c
+        criteria["_none"] = "هیچ‌کدام از موارد موجود نیست"
+
+        capped = ms[: settings.JEV_MAX_QUESTIONS_PER_FIELD]
+        questions = {
+            f"mention_{i}": {
+                "type": "choice",
+                "instructions": _instructions(field, m["variants"]),
+                "criteria": criteria,
+            }
+            for i, m in enumerate(capped)
+        }
+        state = {
+            "field": field,
+            "patch_context": json.dumps(patch, ensure_ascii=False, default=str)[:1500],
+        }
+        await obs.event(trace_id, "jev.request", field=field, payload={
+            "n_mentions": len(capped), "n_candidates": len(cands),
+        })
+        resp = await jev_ask(state, questions)
+        await obs.event(trace_id, "jev.result", field=field, ok=resp is not None,
+                        model=(resp or {}).get("model"),
+                        latency_ms=(resp or {}).get("latency_ms"),
+                        payload={"answers": (resp or {}).get("answers"),
+                                 "usage": (resp or {}).get("usage")})
+        if resp is None:
+            # Field-wide failure → leave no results so callers fall back to difflib.
+            return
+        answers = resp.get("answers") or {}
+
+        decisions = []
+        for i, m in enumerate(capped):
+            ans = answers.get(f"mention_{i}") or {}
+            if not ans:
+                # No answer for this mention → difflib fallback for it.
+                continue
+            choice = ans.get("choice")
+            conf = float(ans.get("confidence") or 0.0)
+            probs = ans.get("probabilities") or {}
+            cand = by_name.get(choice) if choice else None
+            ranked = sorted(
+                ({"id": by_name[n].id, "name": n, "prob": float(p or 0.0)}
+                 for n, p in probs.items() if n != "_none" and n in by_name),
+                key=lambda x: -x["prob"],
+            )[:6]
+
+            if cand is not None and conf >= settings.JEV_CONF_EXACT:
+                res = {"field": field, "status": "exact", "variants": m["variants"],
+                       "id": cand.id, "name": cand.name, "extra": cand.extra or {},
+                       "confidence": conf}
+            elif cand is not None and conf >= settings.JEV_CONF_FLOOR:
+                res = {"field": field, "status": "ambiguity", "variants": m["variants"],
+                       "id": cand.id, "name": cand.name, "extra": cand.extra or {},
+                       "confidence": conf,
+                       "candidates": ranked or [{"id": cand.id, "name": cand.name, "prob": conf}]}
+            else:
+                res = {"field": field, "status": "new", "variants": m["variants"],
+                       "confidence": conf, "candidates": ranked}
+            for v in m["variants"]:
+                results[normalize_text(v)] = res
+            decisions.append({"variants": m["variants"], "status": res["status"],
+                              "choice": res.get("name"), "confidence": conf,
+                              "probs_top": [{k: round(v, 3)} for k, v in
+                                            sorted(probs.items(), key=lambda kv: -kv[1])[:3]]})
+        await obs.event(trace_id, "routing.decision", field=field,
+                        payload={"source": "jev", "mentions": decisions})
+
+    await asyncio.gather(*(one_field(f, ms) for f, ms in by_field.items()))
+    return results
+
+
+def lookup(jev_map: dict | None, name: str | None) -> dict | None:
+    """Find the Jev result for a raw name (by normalized variant)."""
+    if not jev_map or not name:
+        return None
+    return jev_map.get(normalize_text(name))
+
+
+# ---------------------------------------------------------------------------
+# writing results into drafts
+# ---------------------------------------------------------------------------
+
+def apply_choice(payload: dict, field: str, cand: dict, variants: list[str]) -> None:
+    """Bind one chosen candidate (exact match or a user pick) into the payload."""
+    name, cid = cand.get("name"), cand.get("id")
+    extra = cand.get("extra") or {}
+    variants_norm = {normalize_text(v) for v in (variants or []) if v}
+
+    if field == "supplier":
+        payload["supplier_id"] = cid
+        payload["supplier_name"] = name
+        for k in ("location", "phone", "contact_person", "category_name"):
+            if extra.get(k):
+                payload[f"supplier_{k}"] = extra[k]
+    elif field == "project":
+        resolved = payload.setdefault("resolved_projects", {})
+        for v in (variants or []):
+            resolved[v] = cid
+        resolved[name] = cid
+        if cid not in payload.get("project_ids", []):
+            payload.setdefault("project_ids", []).append(cid)
+        if not payload.get("project_id"):
+            payload["project_id"] = cid
+    elif field == "material":
+        for item in payload.get("items") or []:
+            if normalize_text(item.get("material_name") or "") in variants_norm:
+                item["material_name"] = name
+                item["material_id"] = cid
+                item["material_row"] = {"id": cid,
+                                        "canonical_unit_id": extra.get("canonical_unit_id")}
+    elif field == "unit":
+        for item in payload.get("items") or []:
+            if normalize_text(item.get("unit_name") or "") in variants_norm:
+                item["unit_id"] = cid
+                item["unit_explicit"] = True
+                item["unit_name"] = name
+        if not payload.get("items"):
+            payload["canonical_unit_id"] = cid
+            payload["unit_name"] = name
+    elif field == "brand":
+        for item in payload.get("items") or []:
+            if normalize_text(item.get("brand_name") or "") in variants_norm:
+                item["brand_name"] = name
+    elif field == "spec":
+        for item in payload.get("items") or []:
+            specs = item.get("specs") or []
+            for i, s in enumerate(specs):
+                if normalize_text(s) in variants_norm:
+                    specs[i] = name
+
+
+def apply_exact(payload: dict, res: dict) -> None:
+    """Bind one exact Jev match into a draft payload."""
+    apply_choice(payload, res.get("field"), {"id": res.get("id"), "name": res.get("name"),
+                                             "extra": res.get("extra") or {}},
+                 res.get("variants") or [])
+
+
+def normalize_brand_spec(payload: dict, jev_map: dict | None) -> None:
+    """Canonicalize brand/spec strings via exact matches (spec effect per owner:
+    spelling normalization only)."""
+    if not jev_map:
+        return
+    for item in payload.get("items") or []:
+        brand = item.get("brand_name")
+        if brand:
+            hit = lookup(jev_map, brand)
+            if hit and hit.get("status") == "exact":
+                item["brand_name"] = hit["name"]
+        specs = item.get("specs") or []
+        for i, s in enumerate(specs):
+            hit = lookup(jev_map, s)
+            if hit and hit.get("status") == "exact":
+                specs[i] = hit["name"]
+
+
+# ---------------------------------------------------------------------------
+# phase 4 — ambiguity sub-messages with candidate buttons
+# ---------------------------------------------------------------------------
+
+def _amb_id(field: str, variants: list[str]) -> str:
+    digest = hashlib.sha1("|".join(sorted(variants)).encode()).hexdigest()[:4]
+    return f"em_{field[:4]}_{digest}"
+
+
+def _fa_num(n: int) -> str:
+    return str(n).translate(str.maketrans("0123456789", "۰۱۲۳۴۵۶۷۸۹"))
+
+
+def create_ambiguities(payload: dict, jev_map: dict | None) -> None:
+    """One `entity_match` ambiguity entry per ambiguous mention (deduped)."""
+    if not jev_map:
+        return
+    ambs = payload.setdefault("ambiguities", [])
+    seen = {a.get("id") for a in ambs}
+    unique: dict[int, dict] = {}
+    for res in jev_map.values():
+        unique[id(res)] = res  # dedupe identical result objects
+    for res in unique.values():
+        if res.get("status") != "ambiguity" or not res.get("candidates"):
+            continue
+        amb_id = _amb_id(res["field"], res.get("variants") or [])
+        if amb_id in seen:
+            continue
+        cands = res["candidates"][:6]
+        lines = [f"• {c['name']} — {_fa_num(int(round(c['prob'] * 100)))}٪" for c in cands]
+        ambs.append({
+            "id": amb_id,
+            "kind": "entity_match",
+            "field": res["field"],
+            "variants": res.get("variants") or [],
+            "candidates": [{"id": c["id"], "name": c["name"]} for c in cands],
+            "title": f"تطبیق {FIELD_FA.get(res['field'], res['field'])}: «{' / '.join(res.get('variants') or [])}»",
+            "details": "\n".join(lines),
+            "status": "pending",
+        })
+        seen.add(amb_id)

bot/app/nlu.py

@@ -25,6 +25,23 @@ def normalize_digits(text: str) -> str:
 
 ZWNJ = "\u200c"
 
+MENTION_FIELDS = {"project", "supplier", "material", "brand", "spec", "unit"}
+
+
+def sanitize_mentions(raw) -> list[dict]:
+    """Keep only valid {field, variants} mention objects from the LLM output."""
+    out: list[dict] = []
+    if not isinstance(raw, list):
+        return out
+    for m in raw:
+        if not isinstance(m, dict):
+            continue
+        field = str(m.get("field") or "").strip()
+        variants = [str(v).strip() for v in (m.get("variants") or []) if str(v or "").strip()]
+        if field in MENTION_FIELDS and variants:
+            out.append({"field": field, "variants": variants[:6]})
+    return out
+
 
 def normalize_text(text: str) -> str:
     """Normalize Persian text for fuzzy matching (ZWNJ, arabic ya/ka, spaces)."""
@@ -60,6 +77,7 @@ SYSTEM_PROMPT = """تو دستیار هوشمند تدارکات یک سیستم
   "consensus_text": "بیانیه رسمی، پالایش‌شده، فشرده و بدون تکیه‌کلام از تقاضای تدارکات کاربر",
   "intent": یکی از {intents},
   "patch": {شکل partial هر فعالیتی که کاربر داده},
+  "entity_mentions": [{"field": "project|supplier|material|brand|spec|unit", "variants": ["همه‌ی نگارش‌ها"]}],
   "confidence": 0.0..1.0
 }
 
@@ -95,6 +113,15 @@ SYSTEM_PROMPT = """تو دستیار هوشمند تدارکات یک سیستم
 - اگر درباره پذیرش استعلام است → intent="accept_quote" و در patch.target نام تأمین‌کننده/استعلام.
 - اگر درباره اعلام رسیدن اقلام است → intent="arrive_items" و در patch.target نام تأمین‌کننده/استعلام و در patch.arrivals مقادیر.
 
+۶. استخراج موجودیت‌های قابل تطبیق (entity_mentions):
+   - فیلدهای مجاز: "project" | "supplier" | "material" | "brand" | "spec" | "unit"
+   - هر نام کالا/تأمین‌کننده/پروژه/برند/واحد و هر مشخصه که در پیام آمده، یک عضو جداگانه است.
+   - اگر در نسخه‌های مختلف STT همان نام با املاهای متفاوت پیاده شده (مثل «مصطفوی»، «مستفدی»، «مسطفی»)،
+     همه را در variants همان یک عضو بگذار — نه چند عضو جداگانه.
+   - برای پیام متنی معمولی، variants یک عضو تک‌عنصر است.
+   - مشخصه (spec) و برند (brand) جدا از متریال و جدا از هم استخراج شوند.
+   - وظیفه‌ی تطبیق با موجودیت‌های سیستم بر عهده‌ی تو نیست؛ هیچ نامی را به نزدیک‌ترین نام موجود «حدس» نزن.
+
 شکل patch برای intent="register_quote"/"update_draft":
 {"supplier_name": str, "project_name": str, "project_names": [str], "is_official": bool, "notes": str, "expiry_days": int, "expiry_date": "YYYY-MM-DD",
  "items": [{"material_name": str, "brand_name": str, "specs": [str], "quantity": number, "unit_name": str, "unit_price": number, "project_name": str, "notes": str}],
@@ -148,7 +175,8 @@ def _extract_json(raw: str) -> dict | None:
 async def understand(settings: Settings, text: str | list[dict], draft_summary: str | None = None,
                      trace_id: str | None = None) -> dict:
     """Return {"consensus_text": ..., "intent": ..., "patch": {...}, "confidence": ...}; never raises."""
-    fallback = {"consensus_text": "", "intent": "unknown", "patch": {}, "confidence": 0.0}
+    fallback = {"consensus_text": "", "intent": "unknown", "patch": {},
+                "entity_mentions": [], "confidence": 0.0}
     if not settings.AI_BASE_URL or not settings.AI_API_KEY:
         logger.warning("AI provider not configured; cannot understand message")
         return fallback
@@ -214,6 +242,7 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary:
                             payload={"error": "unparseable", "raw": content[:500]})
             return fallback
         parsed.setdefault("patch", {})
+        parsed["entity_mentions"] = sanitize_mentions(parsed.get("entity_mentions"))
         if not parsed.get("consensus_text"):
             parsed["consensus_text"] = first_text
         await obs.event(trace_id, "nlu.result", ok=True, model=settings.LLM_MODEL, payload={
@@ -221,6 +250,7 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary:
             "confidence": parsed.get("confidence"),
             "patch": parsed.get("patch"),
             "consensus_text": parsed.get("consensus_text"),
+            "entity_mentions": parsed.get("entity_mentions"),
         })
         return parsed
     except (httpx.HTTPError, json.JSONDecodeError, KeyError, TypeError):

bot/app/resolve.py

@@ -153,6 +153,43 @@ async def _db_materials(q: str | None = None) -> list[Candidate]:
         return []
 
 
+async def _db_brands() -> list[Candidate]:
+    try:
+        async with SessionLocal() as s:
+            rows = (await s.execute(text("SELECT id, name FROM brands"))).mappings().all()
+            return [Candidate(id=str(b["id"]), name=b.get("name") or str(b["id"])) for b in rows]
+    except Exception:
+        return []
+
+
+async def _db_specs() -> list[Candidate]:
+    """Distinct spec values: EAV strings from the catalog + specs used in recent drafts."""
+    values: dict[str, None] = {}
+    try:
+        async with SessionLocal() as s:
+            rows = (await s.execute(text(
+                "SELECT DISTINCT value_string FROM material_attribute_values "
+                "WHERE value_string IS NOT NULL AND btrim(value_string) <> '' LIMIT 500"
+            ))).scalars().all()
+            for v in rows:
+                values.setdefault(str(v), None)
+    except Exception:
+        pass
+    try:
+        async with SessionLocal() as s:
+            rows = (await s.execute(text(
+                "SELECT DISTINCT spec FROM bot_drafts d, "
+                "jsonb_array_elements(COALESCE(d.payload->'items', '[]'::jsonb)) it, "
+                "jsonb_array_elements_text(COALESCE(it->'specs', '[]'::jsonb)) spec "
+                "WHERE btrim(spec) <> '' LIMIT 300"
+            ))).scalars().all()
+            for v in rows:
+                values.setdefault(str(v), None)
+    except Exception:
+        pass
+    return [Candidate(id=v, name=v) for v in values]
+
+
 class Resolver:
     """Caches catalog lists (units/categories/projects are stable; suppliers/materials queried on demand)."""
 
@@ -236,6 +273,74 @@ class Resolver:
             items = await _db_materials(q=q)
         return _rank(q, items) if q else items
 
+    # ------------------------------------------------- full lists (Jev, v5)
+
+    async def _fetch_all(self, fetch_page, max_pages: int = 5) -> list[dict]:
+        """Page through a paginated ERP list endpoint (page_size=100)."""
+        out: list[dict] = []
+        for page in range(1, max_pages + 1):
+            rows = await fetch_page(page)
+            out += rows
+            if len(rows) < 100:
+                break
+        return out
+
+    async def all_suppliers(self) -> list[Candidate]:
+        async def fetch():
+            try:
+                rows = await self._fetch_all(
+                    lambda page: self._api.list_suppliers(page=page, page_size=100)
+                )
+                return [
+                    Candidate(
+                        id=str(s["id"]),
+                        name=s.get("name") or str(s["id"]),
+                        extra={
+                            "location": s.get("location"),
+                            "phone": s.get("phone"),
+                            "contact_person": s.get("contact_person"),
+                            "category_name": (s.get("category") or {}).get("name")
+                            if isinstance(s.get("category"), dict) else s.get("category_name"),
+                        },
+                    )
+                    for s in rows
+                ]
+            except Exception:
+                return await _db_suppliers()
+        return await self._cached("all_suppliers", fetch)
+
+    async def all_materials(self) -> list[Candidate]:
+        async def fetch():
+            try:
+                rows = await self._fetch_all(
+                    lambda page: self._api.list_materials(page=page, page_size=100)
+                )
+                return [
+                    Candidate(
+                        id=str(m["id"]),
+                        name=m.get("name") or str(m["id"]),
+                        extra={"canonical_unit_id": m.get("canonical_unit_id")},
+                    )
+                    for m in rows
+                ]
+            except Exception:
+                return await _db_materials()
+        return await self._cached("all_materials", fetch)
+
+    async def brands(self) -> list[Candidate]:
+        async def fetch():
+            try:
+                rows = await self._fetch_all(
+                    lambda page: self._api.list_brands(page=page, page_size=100)
+                )
+                return [Candidate(id=str(b["id"]), name=b.get("name") or str(b["id"])) for b in rows]
+            except Exception:
+                return await _db_brands()
+        return await self._cached("brands", fetch)
+
+    async def specs(self) -> list[Candidate]:
+        return await self._cached("specs", _db_specs)
+
     # ------------------------------------------------------------ resolutions
 
     async def resolve(self, kind: str, name: str | None) -> Resolution: