feat(matching): implement two-stage matching with DeepSeek arbiter, fail-stop guard, and zero-probability filter

- Add app/fallback_matcher.py for intelligent crisis entity matching with token and latency observability
- Route Jev failure/timeouts in app/matching.py to DeepSeek fallback engine
- Handle MatchingFailureError defensively in handlers/__init__.py with honest error message
- Enforce strict 'new' entity status in handlers/quote.py and implement defensive get_or_create on commit
- Clean up app/resolve.py to eliminate difflib and character ratio heuristics
- Pass all smoke tests (59/59), fallback matching tests (15/15), and matching unit tests (13/13)

.gitignore

@@ -31,6 +31,7 @@ scratch/
 v1.2 - Copy/
 inv dash/
 voice/
+data/
 
 # ---- Machine-local compose overrides (LAN phone access, etc.) ----
 docker-compose.override.yml

bot/app/audio_enhancer.py

@@ -0,0 +1,197 @@
+"""Audio preprocessing & denoising pipeline for procurement bot.
+
+Workflow:
+  Raw Voice Bytes (Bale)
+    → enhance_voice() [Optional Audio Enhancement / Denoising API]
+        - If disabled or missing URL → zero overhead pass-through
+        - Zero transcoding by default (raw bytes sent as-is)
+        - Transcoding to target format ONLY if explicitly configured
+        - Graceful fail-soft on timeout or error (transparent raw audio fallback)
+        - Enhanced bytes saved to data/voices/{YYYYMMDD}/{trace_id}_{message_id}_enhanced.{ext}
+        - Observability events recorded (audio.enhance.result with ok/size/latency_ms/path)
+    → Multi-STT Providers
+"""
+import asyncio
+import logging
+import subprocess
+import tempfile
+import time
+from pathlib import Path
+
+import httpx
+
+from app import observability as obs
+from app.config import Settings
+
+logger = logging.getLogger("bot.audio_enhancer")
+
+_CONVERT_TIMEOUT = 60
+
+
+def _transcode(raw: bytes, src_ext: str, target_fmt: str) -> bytes:
+    """Convert audio to target format using ffmpeg only when explicitly requested."""
+    target_fmt = target_fmt.lstrip(".")
+    with tempfile.TemporaryDirectory() as tmp:
+        src = Path(tmp) / f"in{src_ext or '.ogg'}"
+        dst = Path(tmp) / f"out.{target_fmt}"
+        src.write_bytes(raw)
+        proc = subprocess.run(
+            ["ffmpeg", "-y", "-i", str(src), "-vn", str(dst)],
+            capture_output=True,
+            timeout=_CONVERT_TIMEOUT,
+        )
+        if proc.returncode != 0 or not dst.exists():
+            raise RuntimeError(f"ffmpeg conversion to {target_fmt} failed: {proc.stderr[-300:]!r}")
+        return dst.read_bytes()
+
+
+def _resolve_extension(filename: str | None, mime_type: str | None) -> str:
+    ext = Path(filename or "").suffix
+    if not ext and mime_type:
+        clean_mime = mime_type.split(";")[0].strip().lower()
+        ext = {
+            "audio/ogg": ".ogg",
+            "audio/opus": ".ogg",
+            "audio/mpeg": ".mp3",
+            "audio/mp4": ".m4a",
+            "audio/wav": ".wav",
+            "audio/x-wav": ".wav",
+            "audio/webm": ".webm",
+            "audio/flac": ".flac",
+        }.get(clean_mime, ".ogg")
+    return ext or ".ogg"
+
+
+async def enhance_voice(
+    settings: Settings,
+    audio_bytes: bytes,
+    mime_type: str | None = None,
+    filename: str | None = None,
+    trace_id: str | None = None,
+    message_id: int | None = None,
+) -> tuple[bytes, dict]:
+    """Enhance and denoise voice audio before STT.
+
+    Returns:
+        (processed_audio_bytes, metadata)
+    Guarantees:
+        - Never raises an unhandled exception.
+        - On any failure or timeout, gracefully returns original audio_bytes (fallback).
+        - Enhanced bytes are saved to daily folder with _enhanced suffix when successful.
+        - Observability events are captured.
+    """
+    if not settings.AUDIO_ENHANCE_ENABLED or not (settings.AUDIO_ENHANCE_BASE_URL or "").strip():
+        return audio_bytes, {"enhanced": False, "skipped": True}
+
+    base_url = settings.AUDIO_ENHANCE_BASE_URL.rstrip("/")
+    api_key = settings.AUDIO_ENHANCE_API_KEY or ""
+    model = settings.AUDIO_ENHANCE_MODEL or ""
+    timeout = float(settings.AUDIO_ENHANCE_TIMEOUT_SECONDS or 60.0)
+    target_format = (settings.AUDIO_ENHANCE_TARGET_FORMAT or "").strip().lower()
+    method = (settings.AUDIO_ENHANCE_METHOD or "auto").strip().lower()
+
+    orig_ext = _resolve_extension(filename, mime_type)
+    t0 = time.perf_counter()
+
+    # Condition 2: Zero transcoding unless explicitly requested by configuration
+    payload_bytes = audio_bytes
+    upload_ext = orig_ext
+    if target_format and not orig_ext.endswith(f".{target_format}"):
+        try:
+            payload_bytes = await asyncio.to_thread(_transcode, audio_bytes, orig_ext, target_format)
+            upload_ext = f".{target_format}"
+        except Exception as exc:
+            logger.warning("Target format conversion failed: %s; using raw bytes", exc)
+            payload_bytes = audio_bytes
+            upload_ext = orig_ext
+
+    headers: dict[str, str] = {}
+    if api_key:
+        headers["Authorization"] = f"Bearer {api_key}"
+
+    try:
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            upload_filename = f"voice{upload_ext}"
+            req_mime = mime_type or "audio/ogg"
+
+            if method == "raw_stream":
+                headers["Content-Type"] = req_mime
+                resp = await client.post(base_url, content=payload_bytes, headers=headers)
+            else:
+                # Default / multipart upload (compatible with standard voice enhancement APIs)
+                data = {"model": model} if model else {}
+                files = {"file": (upload_filename, payload_bytes, req_mime)}
+                resp = await client.post(base_url, files=files, data=data, headers=headers)
+
+        ms = int((time.perf_counter() - t0) * 1000)
+
+        if resp.status_code == 200 and len(resp.content) > 0:
+            enhanced_bytes = resp.content
+
+            # Determine response audio extension
+            resp_ct = (resp.headers.get("content-type") or "").split(";")[0].strip().lower()
+            out_ext = {
+                "audio/ogg": ".ogg",
+                "audio/opus": ".ogg",
+                "audio/mpeg": ".mp3",
+                "audio/mp4": ".m4a",
+                "audio/wav": ".wav",
+                "audio/x-wav": ".wav",
+            }.get(resp_ct, upload_ext)
+
+            saved_path = obs.save_enhanced_voice(
+                trace_id, message_id or 0, enhanced_bytes, ext=out_ext
+            )
+
+            await obs.event(
+                trace_id,
+                "audio.enhance.result",
+                ok=True,
+                latency_ms=ms,
+                model=model,
+                payload={
+                    "path": saved_path,
+                    "size": len(enhanced_bytes),
+                    "original_size": len(audio_bytes),
+                    "provider": model or "audio_enhancer",
+                },
+            )
+            logger.info(
+                "Audio enhancement succeeded in %d ms (size: %d -> %d bytes)",
+                ms,
+                len(audio_bytes),
+                len(enhanced_bytes),
+            )
+            return enhanced_bytes, {
+                "enhanced": True,
+                "path": saved_path,
+                "size": len(enhanced_bytes),
+                "latency_ms": ms,
+            }
+
+        # Non-200 or empty response -> graceful fallback
+        err_msg = f"HTTP {resp.status_code}: {resp.text[:200]}"
+        logger.warning("Audio enhancement failed: %s; falling back to raw audio", err_msg)
+        await obs.event(
+            trace_id,
+            "audio.enhance.result",
+            ok=False,
+            latency_ms=ms,
+            model=model,
+            payload={"error": err_msg, "fallback": True},
+        )
+        return audio_bytes, {"enhanced": False, "fallback": True, "error": err_msg}
+
+    except Exception as exc:
+        ms = int((time.perf_counter() - t0) * 1000)
+        err_msg = str(exc)[:300]
+        logger.warning("Audio enhancement exception: %s; falling back to raw audio", err_msg)
+        await obs.event(
+            trace_id,
+            "audio.enhance.result",
+            ok=False,
+            latency_ms=ms,
+            model=model,
+            payload={"error": err_msg, "fallback": True},
+        )
+        return audio_bytes, {"enhanced": False, "fallback": True, "error": err_msg}

bot/app/config.py

@@ -73,6 +73,15 @@ class Settings(BaseSettings):
     STT_MODEL: str = "whisper-1"
     STT_LANGUAGE: str = "fa"
 
+    # Audio Enhancement / Preprocessing (denoising & speech quality enhancement)
+    AUDIO_ENHANCE_ENABLED: bool = False
+    AUDIO_ENHANCE_BASE_URL: str = ""
+    AUDIO_ENHANCE_API_KEY: str = ""
+    AUDIO_ENHANCE_MODEL: str = ""
+    AUDIO_ENHANCE_TIMEOUT_SECONDS: float = 60.0
+    AUDIO_ENHANCE_TARGET_FORMAT: str = ""  # empty = pass raw bytes directly; e.g. "wav" / "mp3"
+    AUDIO_ENHANCE_METHOD: str = "auto"     # "auto" | "raw_stream" | "multipart"
+
     @property
     def stt_base_url(self) -> str:
         return self.STT_BASE_URL or self.AI_BASE_URL

bot/app/fallback_matcher.py

@@ -0,0 +1,552 @@
+"""Dedicated DeepSeek fallback matching engine for crisis mode.
+
+Invoked when Jev (TypeSafe) fails, times out, returns 5xx/429, or is unavailable.
+Takes:
+  1. raw_input: Raw user text or list of STT transcriptions.
+  2. mentions: Entity mentions extracted by DeepSeek NLU.
+  3. candidates_by_field: Catalog candidate records fetched from the database.
+  4. patch_context: Current draft patch / context.
+  5. trace_id: Observability trace ID.
+
+Outputs:
+  Standard resolution map compatible with jev_map:
+  {
+    "<normalized_variant>": {
+      "field": str,
+      "status": "exact" | "ambiguity" | "new",
+      "variants": list[str],
+      "id": str | None,
+      "name": str | None,
+      "extra": dict,
+      "candidates": list[dict],
+      "inferred_category_slug"?: str,
+    }
+  }
+
+Defensive Guarantee:
+  If DeepSeek call fails (HTTP error, timeout, or unparseable JSON),
+  raises MatchingFailureError so the bot halts gracefully with:
+  "⚠️ سامانه در پردازش هوشمند موجودیت‌ها با اختلال مواجه شد. لطفاً دوباره پیام را ارسال کنید."
+"""
+import hashlib
+import json
+import logging
+import re
+import time
+
+import httpx
+
+from app import observability as obs
+from app.config import settings
+from app.nlu import normalize_text
+
+logger = logging.getLogger("bot.fallback_matcher")
+
+
+class MatchingFailureError(Exception):
+    """Raised when matching cannot proceed due to LLM / network failure."""
+    pass
+
+
+ARBITRATION_SYSTEM_PROMPT = """تو قاضی و داور نهایی هوشمند تطبیق موجودیت‌های تدارکاتی و ساختمانی در یک سیستم ERP هستی.
+یک مدل جستجوی برداری اولیه (به نام Jev) برای هر منشن کاربر، کاتالوگ دیتابیس را بررسی کرده و گزینه‌ای را پیشنهاد داده است.
+وظیفه خطیر تو:
+تحلیل هم‌زمان ۱) پیام اولیه کاربر، ۲) کاندیداهای کاتالوگ دیتابیس، و ۳) پیشنهاد مدل Jev، و صدور «حکم قطعی نهایی».
+
+یک شیء JSON و فقط JSON برگردان با ساختار زیر:
+{
+  "decisions": [
+    {
+      "field": "supplier|material|project|unit|brand|spec",
+      "variant": "نگارش منشن",
+      "status": "exact|ambiguity|new",
+      "id": "شناسه رکورد کاتالوگ یا null",
+      "name": "نام رسمی رکورد کاتالوگ یا null",
+      "candidates": [
+        {"id": "شناسه", "name": "نام", "prob": 0.0..1.0}
+      ],
+      "inferred_category_slug": "یکی از ۸ اسلاگ مجاز فقط برای تامین‌کننده new یا null",
+      "adjudication_action": "endorsed_jev|vetoed_to_new|demoted_to_ambiguity",
+      "reasoning": "توضیح کوتاه فنی دلیل قضاوت"
+    }
+  ]
+}
+
+قوانین حیاتی داوری:
+۱. عدم اعتماد کورکورانه به Jev:
+   - مدل Jev صرفاً یک مدل آماری/برداری است و تفاوت سایزها، رده‌ها، ابعاد فنی و استانداردهای صنعتی (مانند لوله ۶ اینچ در برابر لوله ۴ اینچ، میلگرد ۱۶ در برابر ۲۰، یا تفاوت دو برند متفاوت) را تشخیص نمی‌دهد.
+   - تو باید کلام کاربر را با پیشنهاد Jev مقایسه کنی.
+
+۲. حالات صدور حکم:
+   الف) تأیید انتخاب Jev (status: "exact"):
+      اگر پیشنهاد Jev دقیقاً و با احتساب نام تجاری و مشخصات فنی با خواسته کاربر در پیام مطابقت دارد.
+      - فیلد id و name را با مقادیر دقیق رکورد کاتالوگ پر کن. candidates خالی باشد ([]).
+   
+   ب) وتوی Jev و اعلام قلم جدید (status: "new"):
+      اگر کالا، تأمین‌کننده یا پروژه مورد نظر کاربر در کاتالوگ ارسالی وجود ندارد (حتی اگر Jev به اشتباه به دلیل شباهت حروفی گزینه‌ای را با اطمینان بالا پیشنهاد داده باشد).
+      - پیشنهاد Jev را بلافاصله باطل و وتو کن!
+      - وضعیت را "new" بگذار و id حتماً null باشد. هرگز قلم ناموجود را به زور به کاتالوگ نچسبان.
+      - برای تأمین‌کننده جدید (supplier)، مناسب‌ترین اسلاگ را از بین ۸ دسته مجاز در inferred_category_slug قرار بده.
+
+   ج) تنزل به ابهام (status: "ambiguity"):
+      اگر کاربر عبارت کلی گفته (مثلاً «میلگرد») و در کاتالوگ چند گزینه محتمل وجود دارد، یا بین انتخاب Jev و سایر کاندیداها تردید فنی وجود دارد.
+      - وضعیت را "ambiguity" بگذار و گزینه‌های منتخب را در candidates بگذار.
+
+۳. تحلیل ورودی‌های صوتی (Voice Mode):
+   اگر ورودی صوتی است (input_mode: "voice")، آرایه کامل ترانویسی موتورهای STT (مانند Whisper، Vosk و غیره) به همراه consensus_text در اختیارت است.
+   شواهد صوتی، تلفظ‌ها، خطاهای شنیداری و لهجه‌های کارگاهی را ملاک قضاوت قرار بده.
+
+۴. اسلاگ‌های مجاز دسته‌بندی برای تأمین‌کننده جدید:
+   * metals-steel-sections
+   * mechanical-installations-pipes
+   * electrical-equipment-supplies
+   * construction-materials
+   * construction-chemicals
+   * doors-windows-glass
+   * finishing-interior-decoration
+   * construction-tools-equipment
+"""
+
+
+FALLBACK_SYSTEM_PROMPT = """تو موتور تطبیق و پالایش هوشمند موجودیت‌های تدارکاتی یک سیستم ساختمانی و صنعتی ERP در ایران هستی.
+سرویس جانبی تطبیق کاتالوگ در دسترس نیست و اکنون تو وظیفه داری با درک مهندسی، تجاری و معنایی خود، عبارات ذکرشده توسط کاربر را مستقیماً با رکوردهای کاتالوگ دیتابیس تطبیق دهی.
+
+یک شیء JSON و فقط JSON برگردان (بدون هرگونه متن یا مارک‌داون اضافه) با ساختار زیر:
+{
+  "results": [
+    {
+      "field": "supplier|material|project|unit|brand|spec",
+      "variants": ["تمام نگارش‌های این موجودیت"],
+      "status": "exact|ambiguity|new",
+      "id": "شناسه رکورد کاتالوگ منطبق یا null",
+      "name": "نام رسمی رکورد کاتالوگ منطبق یا null",
+      "candidates": [
+        {"id": "شناسه رکورد کاندیدا", "name": "نام رکورد کاندیدا", "prob": 0.0..1.0}
+      ],
+      "inferred_category_slug": "فقط برای supplier جدید یکی از ۸ اسلاگ مجاز یا null"
+    }
+  ]
+}
+
+قوانین حیاتی تطبیق:
+۱. تطبیق قطعی (status: "exact"):
+   - اگر واریانت کاربر با یکی از رکوردهای کاتالوگ انطباق معنایی و تجاری روشن دارد (با احتساب خطاهای گفتاری STT، لهجه اصفهانی، املای ارقام فارسی/انگلیسی، پیشوند و پسوندهای کارگاهی مانند «آهن»، «بازرگانی»، «لوله»، «میلگرد»)، آن را انتخاب کن.
+   - فیلد id و name را با مقادیر دقیق موجود در کاتالوگ پر کن.
+   - فیلد candidates را خالی بگذار ([]).
+
+۲. ابهام تجاری/ابعادی (status: "ambiguity"):
+   - اگر کاربر عنوان عامی را به کار برده که به چند رکورد مشخص در کاتالوگ قابل تفکیک است (مثلاً کاربر گفته «میلگرد» ولی در کاتالوگ انواع سایزهای ۱۴، ۱۶، ۱۸ داریم؛ یا کاربر نام خانوادگی تأمین‌کننده را گفته و چند تأمین‌کننده مشابه با این نام وجود دارند)، حتماً وضعیت را "ambiguity" بگذار.
+   - رکوردهای کاندیدای محتمل را در لیست candidates با نام، شناسه و تخمین احتمال (prob) قرار بده (حداکثر ۶ مورد).
+
+۳. موجودیت جدید (status: "new"):
+   - قانون طلایی: اگر کالا، تأمین‌کننده یا پروژه در کاتالوگ ارسالی وجود ندارد، وضعیت را حتماً "new" بگذار!
+   - به هیچ عنوان و تحت هیچ شرایطی یک قلم جدید را به زور شباهت کلمات یا کاراکتر به یک قلم نامربوط در کاتالوگ نچسبان (مثلاً لوله ۶ اینچ را به لوله ۴ اینچ یا میلگرد ۲۰ را به میلگرد ۱۶ نچسبان).
+   - در حالت new، فیلد id باید null باشد.
+
+۴. استنتاج خودکار دسته‌بندی برای تأمین‌کننده جدید (inferred_category_slug):
+   - اگر تأمین‌کننده‌ای وضعیتش new است، با توجه به اقلام استعلام‌شده در پیام کاربر، مناسب‌ترین اسلاگ دسته‌بندی را از بین این ۸ مورد انتخاب کن:
+     * metals-steel-sections (فلزات و مقاطع فولادی)
+     * mechanical-installations-pipes (تأسیسات مکانیکی و آبی)
+     * electrical-equipment-supplies (تجهیزات و لوازم برقی)
+     * construction-materials (مصالح ساختمانی)
+     * construction-chemicals (مواد شیمیایی ساختمانی)
+     * doors-windows-glass (درب، پنجره و شیشه)
+     * finishing-interior-decoration (نازک‌کاری و دکوراسیون داخلی)
+     * construction-tools-equipment (ابزار و تجهیزات کارگاهی)
+"""
+
+
+def _extract_json(raw: str) -> dict | None:
+    if not raw or not raw.strip():
+        return None
+    text = raw.strip()
+    match = re.search(r"\{.*\}", text, re.DOTALL)
+    if not match:
+        return None
+    try:
+        return json.loads(match.group(0))
+    except Exception:
+        return None
+
+
+async def match_with_deepseek(
+    raw_input: object,
+    mentions: list[dict],
+    candidates_by_field: dict[str, list],
+    patch_context: dict | None,
+    trace_id: str | None,
+    consensus_text: str | None = None,
+) -> dict[str, dict]:
+    """Execute DeepSeek crisis matching for unmapped mentions.
+
+    Returns a dict keyed by normalized variant, ready to merge into jev_map.
+    Raises MatchingFailureError if DeepSeek fails or returns invalid output.
+    """
+    if not mentions:
+        return {}
+
+    if not settings.AI_BASE_URL or not settings.AI_API_KEY:
+        logger.error("fallback_matcher: AI provider not configured (AI_BASE_URL/AI_API_KEY missing)")
+        raise MatchingFailureError("AI provider not configured")
+
+    # Serialize candidates to clean dicts for prompt efficiency
+    catalog_summary: dict[str, list[dict]] = {}
+    for fld, cands in (candidates_by_field or {}).items():
+        field_cands = []
+        for c in cands:
+            item = {"id": str(getattr(c, "id", "")), "name": str(getattr(c, "name", ""))}
+            extra = getattr(c, "extra", {}) or {}
+            for k in ("category_name", "location", "canonical_unit_id", "code"):
+                if extra.get(k):
+                    item[k] = extra[k]
+            field_cands.append(item)
+        catalog_summary[fld] = field_cands[:80]  # token-safe ceiling per field
+
+    if isinstance(raw_input, list):
+        reference_input = {
+            "input_mode": "voice",
+            "stt_multi_transcripts": raw_input,
+            "consensus_text": consensus_text or "",
+        }
+    else:
+        reference_input = {
+            "input_mode": "text",
+            "raw_text": str(raw_input or ""),
+        }
+
+    payload_data = {
+        "reference_user_input": reference_input,
+        "entity_mentions": mentions,
+        "catalog_candidates": catalog_summary,
+        "patch_context": patch_context or {},
+    }
+    user_content = json.dumps(payload_data, ensure_ascii=False)
+
+    t0 = time.perf_counter()
+    try:
+        async with httpx.AsyncClient(timeout=settings.LLM_TIMEOUT_SECONDS) as client:
+            resp = await client.post(
+                f"{settings.AI_BASE_URL.rstrip('/')}/chat/completions",
+                headers={"Authorization": f"Bearer {settings.AI_API_KEY}"},
+                json={
+                    "model": settings.LLM_MODEL,
+                    "messages": [
+                        {"role": "system", "content": FALLBACK_SYSTEM_PROMPT},
+                        {"role": "user", "content": user_content},
+                    ],
+                    "temperature": 0.1,
+                    "response_format": {"type": "json_object"},
+                },
+            )
+        latency_ms = int((time.perf_counter() - t0) * 1000)
+
+        if resp.status_code != 200:
+            logger.error("fallback_matcher: HTTP %s from DeepSeek: %s", resp.status_code, resp.text[:300])
+            await obs.event(trace_id, "matching.fallback.failure", ok=False, payload={
+                "status_code": resp.status_code,
+                "error": "http_error",
+                "body": resp.text[:300],
+            })
+            raise MatchingFailureError(f"DeepSeek returned HTTP {resp.status_code}")
+
+        content = ((resp.json() or {}).get("choices") or [{}])[0].get("message", {}).get("content", "")
+        usage = (resp.json() or {}).get("usage") or {}
+        parsed = _extract_json(content)
+        if not parsed or not isinstance(parsed.get("results"), list):
+            logger.error("fallback_matcher: unparseable JSON from DeepSeek: %r", content[:300])
+            await obs.event(trace_id, "matching.fallback.failure", ok=False, payload={
+                "error": "unparseable_json",
+                "raw": content[:500],
+            })
+            raise MatchingFailureError("DeepSeek returned unparseable JSON")
+
+    except httpx.HTTPError as exc:
+        latency_ms = int((time.perf_counter() - t0) * 1000)
+        logger.exception("fallback_matcher: HTTP network error calling DeepSeek")
+        await obs.event(trace_id, "matching.fallback.failure", ok=False, payload={
+            "error": "network_exception",
+            "details": str(exc),
+            "latency_ms": latency_ms,
+        })
+        raise MatchingFailureError("Network exception calling DeepSeek") from exc
+    except Exception as exc:
+        if isinstance(exc, MatchingFailureError):
+            raise
+        latency_ms = int((time.perf_counter() - t0) * 1000)
+        logger.exception("fallback_matcher: Unexpected error")
+        await obs.event(trace_id, "matching.fallback.failure", ok=False, payload={
+            "error": "unexpected_exception",
+            "details": str(exc),
+            "latency_ms": latency_ms,
+        })
+        raise MatchingFailureError("Unexpected error calling DeepSeek") from exc
+
+    results_map: dict[str, dict] = {}
+    statuses_count = {"exact": 0, "ambiguity": 0, "new": 0}
+
+    # Build candidate lookup dict for candidate extras
+    cand_by_field_id: dict[tuple[str, str], object] = {}
+    for fld, c_list in (candidates_by_field or {}).items():
+        for c in c_list:
+            cid = str(getattr(c, "id", ""))
+            cand_by_field_id[(fld, cid)] = c
+
+    for item in parsed.get("results", []):
+        if not isinstance(item, dict):
+            continue
+        field = str(item.get("field") or "").strip()
+        variants = [str(v).strip() for v in (item.get("variants") or []) if str(v or "").strip()]
+        status = str(item.get("status") or "").strip().lower()
+        if status not in ("exact", "ambiguity", "new"):
+            status = "new"
+
+        statuses_count[status] = statuses_count.get(status, 0) + 1
+
+        cand_id = str(item.get("id")) if item.get("id") else None
+        cand_name = str(item.get("name")) if item.get("name") else None
+        cand_obj = cand_by_field_id.get((field, cand_id or "")) if cand_id else None
+        extra = getattr(cand_obj, "extra", {}) or {}
+
+        # Format candidates for ambiguity
+        amb_candidates = []
+        for c in item.get("candidates") or []:
+            if isinstance(c, dict) and c.get("id") and c.get("name"):
+                amb_candidates.append({
+                    "id": str(c["id"]),
+                    "name": str(c["name"]),
+                    "prob": float(c.get("prob") or 0.5),
+                })
+
+        entry = {
+            "field": field,
+            "status": status,
+            "variants": variants,
+            "id": cand_id,
+            "name": cand_name,
+            "extra": extra,
+            "candidates": amb_candidates,
+            "source": "deepseek-fallback",
+        }
+        if item.get("inferred_category_slug"):
+            entry["inferred_category_slug"] = str(item["inferred_category_slug"])
+
+        for v in variants:
+            results_map[normalize_text(v)] = entry
+
+    # Observability: matching.fallback.result
+    await obs.event(trace_id, "matching.fallback.result", ok=True, payload={
+        "latency_ms": latency_ms,
+        "usage": usage,
+        "statuses": statuses_count,
+        "n_mentions": len(mentions),
+        "n_resolved": len(results_map),
+    })
+
+    return results_map
+
+
+async def adjudicate_with_deepseek(
+    raw_input: object,
+    consensus_text: str | None,
+    items: list[dict],
+    patch_context: dict | None,
+    trace_id: str | None,
+) -> dict[str, dict]:
+    """Execute DeepSeek final arbitration over Jev recommendations.
+
+    Takes Jev-assessed items and makes final binding decisions.
+    Raises MatchingFailureError if DeepSeek fails or returns unparseable output.
+    """
+    if not items:
+        return {}
+
+    if not settings.AI_BASE_URL or not settings.AI_API_KEY:
+        logger.error("fallback_matcher: AI provider not configured for arbitration")
+        raise MatchingFailureError("AI provider not configured")
+
+    if isinstance(raw_input, list):
+        reference_input = {
+            "input_mode": "voice",
+            "stt_multi_transcripts": raw_input,
+            "consensus_text": consensus_text or "",
+        }
+    else:
+        reference_input = {
+            "input_mode": "text",
+            "raw_text": str(raw_input or ""),
+        }
+
+    arbitration_items_payload = []
+    # Candidate lookup by (field, id) and (field, name) for metadata
+    cand_lookup: dict[tuple[str, str], object] = {}
+
+    for it in items:
+        fld = it["field"]
+        c_list = it.get("candidates") or []
+        serialized_cands = []
+        for c in c_list:
+            cid = str(getattr(c, "id", ""))
+            cname = str(getattr(c, "name", ""))
+            cand_lookup[(fld, cid)] = c
+            cand_lookup[(fld, normalize_text(cname))] = c
+            item_entry = {"id": cid, "name": cname}
+            extra = getattr(c, "extra", {}) or {}
+            for k in ("category_name", "location", "canonical_unit_id", "code"):
+                if extra.get(k):
+                    item_entry[k] = extra[k]
+            serialized_cands.append(item_entry)
+
+        raw_probs = it.get("jev_probabilities") or {}
+        non_zero_probs = {k: v for k, v in raw_probs.items() if float(v or 0.0) > 0.0}
+
+        arbitration_items_payload.append({
+            "field": fld,
+            "variants": it.get("variants") or [],
+            "catalog_candidates": serialized_cands[:60],
+            "jev_assessment": {
+                "recommended_choice": it.get("jev_choice"),
+                "recommended_id": it.get("jev_choice_id"),
+                "confidence": it.get("jev_confidence"),
+                "probabilities": non_zero_probs,
+                "top_ranked": it.get("jev_ranked") or [],
+            },
+        })
+
+    payload_data = {
+        "reference_user_input": reference_input,
+        "items_to_adjudicate": arbitration_items_payload,
+        "patch_context": patch_context or {},
+    }
+    user_content = json.dumps(payload_data, ensure_ascii=False)
+
+    t0 = time.perf_counter()
+    try:
+        async with httpx.AsyncClient(timeout=settings.LLM_TIMEOUT_SECONDS) as client:
+            resp = await client.post(
+                f"{settings.AI_BASE_URL.rstrip('/')}/chat/completions",
+                headers={"Authorization": f"Bearer {settings.AI_API_KEY}"},
+                json={
+                    "model": settings.LLM_MODEL,
+                    "messages": [
+                        {"role": "system", "content": ARBITRATION_SYSTEM_PROMPT},
+                        {"role": "user", "content": user_content},
+                    ],
+                    "temperature": 0.1,
+                    "response_format": {"type": "json_object"},
+                },
+            )
+        latency_ms = int((time.perf_counter() - t0) * 1000)
+
+        if resp.status_code != 200:
+            logger.error("adjudicate_with_deepseek: HTTP %s from DeepSeek: %s", resp.status_code, resp.text[:300])
+            await obs.event(trace_id, "matching.arbitration.failure", ok=False, payload={
+                "status_code": resp.status_code,
+                "error": "http_error",
+                "body": resp.text[:300],
+            })
+            raise MatchingFailureError(f"DeepSeek returned HTTP {resp.status_code}")
+
+        content = ((resp.json() or {}).get("choices") or [{}])[0].get("message", {}).get("content", "")
+        usage = (resp.json() or {}).get("usage") or {}
+        parsed = _extract_json(content)
+        if not parsed or not isinstance(parsed.get("decisions"), list):
+            logger.error("adjudicate_with_deepseek: unparseable JSON from DeepSeek: %r", content[:300])
+            await obs.event(trace_id, "matching.arbitration.failure", ok=False, payload={
+                "error": "unparseable_json",
+                "raw": content[:500],
+            })
+            raise MatchingFailureError("DeepSeek returned unparseable arbitration JSON")
+
+    except httpx.HTTPError as exc:
+        latency_ms = int((time.perf_counter() - t0) * 1000)
+        logger.exception("adjudicate_with_deepseek: HTTP network error calling DeepSeek")
+        await obs.event(trace_id, "matching.arbitration.failure", ok=False, payload={
+            "error": "network_exception",
+            "details": str(exc),
+            "latency_ms": latency_ms,
+        })
+        raise MatchingFailureError("Network exception calling DeepSeek") from exc
+    except Exception as exc:
+        if isinstance(exc, MatchingFailureError):
+            raise
+        latency_ms = int((time.perf_counter() - t0) * 1000)
+        logger.exception("adjudicate_with_deepseek: Unexpected error")
+        await obs.event(trace_id, "matching.arbitration.failure", ok=False, payload={
+            "error": "unexpected_exception",
+            "details": str(exc),
+            "latency_ms": latency_ms,
+        })
+        raise MatchingFailureError("Unexpected error calling DeepSeek") from exc
+
+    results_map: dict[str, dict] = {}
+    actions_count = {"endorsed_jev": 0, "vetoed_to_new": 0, "demoted_to_ambiguity": 0, "other": 0}
+    statuses_count = {"exact": 0, "ambiguity": 0, "new": 0}
+
+    # Map decisions back to results_map
+    for dec in parsed.get("decisions", []):
+        if not isinstance(dec, dict):
+            continue
+        field = str(dec.get("field") or "").strip()
+        variant = str(dec.get("variant") or "").strip()
+        status = str(dec.get("status") or "").strip().lower()
+        if status not in ("exact", "ambiguity", "new"):
+            status = "new"
+
+        action = str(dec.get("adjudication_action") or "other")
+        actions_count[action] = actions_count.get(action, 0) + 1
+        statuses_count[status] = statuses_count.get(status, 0) + 1
+
+        cand_id = str(dec.get("id")) if dec.get("id") else None
+        cand_name = str(dec.get("name")) if dec.get("name") else None
+        cand_obj = None
+        if cand_id:
+            cand_obj = cand_lookup.get((field, cand_id))
+        if not cand_obj and cand_name:
+            cand_obj = cand_lookup.get((field, normalize_text(cand_name)))
+        extra = getattr(cand_obj, "extra", {}) or {}
+
+        amb_candidates = []
+        for c in dec.get("candidates") or []:
+            if isinstance(c, dict) and c.get("id") and c.get("name"):
+                amb_candidates.append({
+                    "id": str(c["id"]),
+                    "name": str(c["name"]),
+                    "prob": float(c.get("prob") or 0.5),
+                })
+
+        # Find matching item from items for variants
+        matched_item = next(
+            (it for it in items if it["field"] == field and any(normalize_text(v) == normalize_text(variant) for v in it.get("variants", []))),
+            None
+        )
+        variants = matched_item.get("variants") if matched_item else [variant]
+
+        entry = {
+            "field": field,
+            "status": status,
+            "variants": variants,
+            "id": cand_id,
+            "name": cand_name,
+            "extra": extra,
+            "candidates": amb_candidates,
+            "source": "deepseek-arbiter",
+            "adjudication_action": action,
+            "reasoning": dec.get("reasoning"),
+        }
+        if dec.get("inferred_category_slug"):
+            entry["inferred_category_slug"] = str(dec["inferred_category_slug"])
+
+        for v in variants:
+            results_map[normalize_text(v)] = entry
+
+    # Observability: matching.arbitration.result
+    await obs.event(trace_id, "matching.arbitration.result", ok=True, payload={
+        "latency_ms": latency_ms,
+        "usage": usage,
+        "actions": actions_count,
+        "statuses": statuses_count,
+        "n_items": len(items),
+        "n_resolved": len(results_map),
+    })
+
+    return results_map

bot/app/handlers/__init__.py

@@ -13,7 +13,8 @@ import logging
 from app import drafts as draft_store
 from app import observability as obs
 from app.config import settings
-from app.nlu import understand
+from app.fallback_matcher import MatchingFailureError
+from app.nlu import NLUFailureError, understand
 from app.runtime import BotContext, LinkedUser
 from app.storage import save_proforma_file
 from app.stt import transcribe, transcribe_multi
@@ -104,6 +105,7 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
                 mime_type=audio_media.get("mime_type"),
                 filename=audio_media.get("file_name"),
                 trace_id=trace_id,
+                message_id=message_id,
             )
         except Exception as exc:
             detail = "هوش مصنوعی تنظیم نشده" if "not configured" in str(exc) else str(exc)[:200]
@@ -221,23 +223,48 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None:
 
     draft_summary = _draft_summary(target_draft) if target_draft else None
     input_payload = voice_transcripts if voice_transcripts else text
-    result = await understand(settings, input_payload, draft_summary, trace_id=trace_id)
+    try:
+        result = await understand(settings, input_payload, draft_summary, trace_id=trace_id)
+    except (MatchingFailureError, NLUFailureError):
+        logger.error("NLU processing failed completely (Universal Fail-Stop Guard)")
+        await ctx.send(
+            chat_id,
+            "⚠️ سامانه در پردازش هوشمند موجودیت‌ها با اختلال مواجه شد. لطفاً دوباره پیام را ارسال کنید.",
+            reply_to=message_id,
+        )
+        return
+
     intent, patch = result.get("intent"), result.get("patch") or {}
-    consensus_text = result.get("consensus_text") or text
+    consensus_text = result.get("consensus_text") or (text if isinstance(text, str) else "")
     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).
+    # Entity matching (Two-Stage: Jev scorer + DeepSeek final arbiter; DeepSeek crisis fallback if Jev down)
     jev_map: dict = {}
-    if settings.JEV_ENABLED and mentions and intent in ("register_quote", "update_draft", "add_material"):
+    if 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)
+            jev_map = await matching.resolve_mentions(
+                ctx, patch, mentions, trace_id=trace_id, raw_input=input_payload,
+                consensus_text=consensus_text,
+            )
+        except matching.MatchingFailureError:
+            logger.error("entity matching failed completely (Universal Fail-Stop Guard)")
+            await ctx.send(
+                chat_id,
+                "⚠️ سامانه در پردازش هوشمند موجودیت‌ها با اختلال مواجه شد. لطفاً دوباره پیام را ارسال کنید.",
+                reply_to=message_id,
+            )
+            return
         except Exception:
-            logger.exception("jev matching failed")
-            jev_map = {}
+            logger.exception("unexpected matching failure")
+            await ctx.send(
+                chat_id,
+                "⚠️ سامانه در پردازش هوشمند موجودیت‌ها با اختلال مواجه شد. لطفاً دوباره پیام را ارسال کنید.",
+                reply_to=message_id,
+            )
+            return
 
     if source == "voice":
         await ctx._log(chat_id, "in", "voice", consensus_text)
@@ -421,7 +448,9 @@ async def _handle_ambiguity_user_reply(
             audio_media = message.get("voice") or message.get("audio")
             audio_bytes = await ctx.bale.download_file(audio_media["file_id"])
             obs.save_voice(trace_id, incoming_msg_id or 0, audio_bytes)
-            text = await transcribe(settings, audio_bytes, trace_id=trace_id)
+            text = await transcribe(
+                settings, audio_bytes, trace_id=trace_id, message_id=incoming_msg_id
+            )
         except Exception:
             text = ""
 

bot/app/handlers/material.py

@@ -9,7 +9,7 @@ from app.drafts import (
     KIND_MATERIAL, merge_material_patch, material_missing, new_material_payload,
 )
 from app.keyboards import pick_buttons, simple_review_keyboard
-from app.resolve import safe_resolve
+from app.nlu import normalize_text
 from app.texts import ERR_GENERIC, esc, material_committed_report, material_review_text
 
 logger = logging.getLogger("bot.material")
@@ -45,16 +45,19 @@ async def resolve_payload(ctx, payload: dict, trace_id: str | None = None,
             elif hit["status"] == "ambiguity" and hit.get("candidates"):
                 options["unit"] = [{"id": c["id"], "name": c["name"]}
                                    for c in hit["candidates"][:6]]
+            elif hit["status"] == "new":
+                options.pop("unit", None)
         else:
-            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
+            try:
+                units = await ctx.resolver.units()
+            except Exception:
+                units = []
+            norm_u = normalize_text(payload["unit_name"])
+            exact = [u for u in units if normalize_text(u.name) == norm_u or (u.extra or {}).get("code") == norm_u]
+            if len(exact) == 1:
+                payload["canonical_unit_id"] = exact[0].id
+                payload["unit_name"] = exact[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)
 

bot/app/handlers/quote.py

@@ -10,6 +10,7 @@ from app.drafts import (
     new_quote_payload, quote_ready,
 )
 from app.keyboards import quote_review_keyboard
+from app.nlu import normalize_text
 from app.resolve import safe_resolve
 from app.texts import ERR_GENERIC, quote_committed_report, quote_review_text, esc
 
@@ -67,15 +68,9 @@ async def handle_entities(
 
 
 # ---------------------------------------------------------------------------
-# resolution
+# resolution (pure intelligence; zero local difflib heuristics)
 # ---------------------------------------------------------------------------
 
-async def _log_res(trace_id, field: str, mention, res) -> None:
-    """Observability: record one difflib resolution attempt (baseline for phase 6)."""
-    await obs.log_resolution(trace_id, field, mention, res.status,
-                             [{"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,
                           jev_map: dict | None = None) -> None:
     options = payload.setdefault("options", {})
@@ -86,9 +81,14 @@ async def resolve_payload(ctx, payload: dict, trace_id: str | None = None,
             if hit["status"] == "exact":
                 matching.apply_exact(payload, hit)
                 options.pop("sup", None)
+            elif hit["status"] == "ambiguity" and hit.get("candidates"):
+                options["sup"] = [{"id": c["id"], "name": c["name"]} for c in hit["candidates"][:6]]
+            elif hit["status"] == "new":
+                options.pop("sup", None)
+                if hit.get("inferred_category_slug"):
+                    payload["supplier_inferred_category_slug"] = hit["inferred_category_slug"]
         else:
             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
@@ -111,18 +111,25 @@ async def resolve_payload(ctx, payload: dict, trace_id: str | None = None,
             if hit["status"] == "exact":
                 matching.apply_exact(payload, hit)
                 payload["project_name"] = hit["name"]
+                resolved_projects[hit["name"]] = hit["id"]
+                resolved_projects[payload["project_name"]] = hit["id"]
+                if hit["id"] not in payload.get("project_ids", []):
+                    payload.setdefault("project_ids", []).append(hit["id"])
+                options.pop("prj", None)
+            elif hit["status"] == "ambiguity" and hit.get("candidates"):
+                options["prj"] = [{"id": c["id"], "name": c["name"]} for c in hit["candidates"][:6]]
+            elif hit["status"] == "new":
                 options.pop("prj", None)
         else:
             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"])
+                if c.id not in payload.get("project_ids", []):
+                    payload.setdefault("project_ids", []).append(c.id)
                 options.pop("prj", None)
             elif res.candidates:
                 options["prj"] = [{"id": c.id, "name": c.name} for c in res.candidates]
@@ -131,21 +138,23 @@ async def resolve_payload(ctx, payload: dict, trace_id: str | None = 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:
-            c = res.candidates[0]
-            p_id = c.id
-            resolved_projects[c.name] = p_id
-            resolved_projects[p_name] = p_id
-            if p_id not in payload.get("project_ids", []):
-                payload.setdefault("project_ids", []).append(p_id)
+        if hit is not None and hit.get("status") == "exact":
+            matching.apply_exact(payload, hit)
+            resolved_projects[hit["name"]] = hit["id"]
+            resolved_projects[p_name] = hit["id"]
+            if hit["id"] not in payload.get("project_ids", []):
+                payload.setdefault("project_ids", []).append(hit["id"])
             if not payload.get("project_id"):
-                payload["project_id"] = p_id
+                payload["project_id"] = hit["id"]
+        elif hit is None:
+            res = await safe_resolve(ctx.resolver, "project", p_name)
+            if res.resolved:
+                resolved_projects[p_name] = res.candidates[0].id
+                resolved_projects[res.candidates[0].name] = res.candidates[0].id
+                if res.candidates[0].id not in payload.get("project_ids", []):
+                    payload.setdefault("project_ids", []).append(res.candidates[0].id)
+                if not payload.get("project_id"):
+                    payload["project_id"] = res.candidates[0].id
 
     for i, item in enumerate(payload.get("items") or [], start=1):
         if not item.get("material_id") and item.get("material_name"):
@@ -154,11 +163,15 @@ async def resolve_payload(ctx, payload: dict, trace_id: str | None = None,
                 if hit["status"] == "exact":
                     matching.apply_exact(payload, hit)
                     options.pop(f"mat{i}", None)
+                elif hit["status"] == "ambiguity" and hit.get("candidates"):
+                    options[f"mat{i}"] = [{"id": c["id"], "name": c["name"]} for c in hit["candidates"][:6]]
+                elif hit["status"] == "new":
+                    options.pop(f"mat{i}", None)
             else:
                 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)
+                    c = res.candidates[0]
+                    _set_material(item, c.id, c.name, c.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]
@@ -179,15 +192,13 @@ async def _resolve_unit(ctx, item: dict, trace_id: str | None = None,
         return
     if item.get("unit_name"):
         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:
+        if hit is not None and hit.get("status") == "exact":
+            item["unit_id"] = hit["id"]
+            item["unit_explicit"] = True
+            item["unit_name"] = hit["name"]
+            return
+        if hit is None:
             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
@@ -369,46 +380,106 @@ async def commit(ctx, chat_id, user, draft, only_resolved: bool = False,
 
     newly_created: dict = {}
 
-    # 1. Sequential pre-creation for new supplier if user wants to commit
+    # 1. Defensive pre-creation for new supplier (get_or_create)
     if not payload.get("supplier_id") and payload.get("supplier_name"):
-        cat_id = payload.get("supplier_category_id")
-        if not cat_id:
-            cats = await ctx.resolver.categories()
-            if cats:
-                cat_id = cats[0].id
-        if cat_id:
-            try:
-                sup_body = {
-                    "name": payload["supplier_name"],
-                    "category_id": cat_id,
-                }
-                if payload.get("supplier_phone"):
-                    sup_body["phone"] = payload["supplier_phone"]
-                created_sup = await ctx.api.create_supplier(sup_body, actor_id=user.actor_id)
-                payload["supplier_id"] = created_sup["id"]
-                newly_created["supplier"] = payload["supplier_name"]
-                ctx.resolver.invalidate("suppliers")
-            except Exception as e:
-                logger.warning("Could not auto-create supplier: %s", e)
+        sup_name = payload["supplier_name"].strip()
+        # 1.a) Check if already in DB
+        try:
+            existing_sups = await ctx.api.list_suppliers(q=sup_name, page_size=10)
+            for s in existing_sups:
+                if normalize_text(s.get("name") or "") == normalize_text(sup_name):
+                    payload["supplier_id"] = str(s["id"])
+                    payload["supplier_name"] = s.get("name")
+                    break
+        except Exception as e:
+            logger.warning("Could not check existing suppliers: %s", e)
 
-    # 2. Sequential pre-creation for new materials
+        # 1.b) If still not in DB, create it
+        if not payload.get("supplier_id"):
+            cat_id = payload.get("supplier_category_id")
+            if not cat_id and payload.get("supplier_inferred_category_slug"):
+                inferred_slug = payload["supplier_inferred_category_slug"]
+                try:
+                    cats = await ctx.resolver.categories()
+                    for c in cats:
+                        if (c.extra or {}).get("slug") == inferred_slug or c.name == inferred_slug:
+                            cat_id = c.id
+                            break
+                except Exception:
+                    pass
+            if not cat_id:
+                try:
+                    cats = await ctx.resolver.categories()
+                    if cats:
+                        cat_id = cats[0].id
+                except Exception:
+                    pass
+
+            if cat_id:
+                try:
+                    sup_body = {
+                        "name": sup_name,
+                        "category_id": cat_id,
+                    }
+                    if payload.get("supplier_phone"):
+                        sup_body["phone"] = payload["supplier_phone"]
+                    created_sup = await ctx.api.create_supplier(sup_body, actor_id=user.actor_id)
+                    payload["supplier_id"] = created_sup["id"]
+                    newly_created["supplier"] = sup_name
+                    ctx.resolver.invalidate("suppliers")
+                except Exception as e:
+                    logger.warning("Could not auto-create supplier: %s — trying fallback fetch", e)
+                    try:
+                        existing_sups = await ctx.api.list_suppliers(q=sup_name, page_size=10)
+                        for s in existing_sups:
+                            if normalize_text(s.get("name") or "") == normalize_text(sup_name):
+                                payload["supplier_id"] = str(s["id"])
+                                payload["supplier_name"] = s.get("name")
+                                break
+                    except Exception:
+                        pass
+
+    # 2. Defensive pre-creation for new materials (get_or_create)
     created_mats = []
     for i, item in enumerate(payload.get("items") or [], start=1):
         if not item.get("material_id") and item.get("material_name") and item.get("unit_id"):
+            mat_name = item["material_name"].strip()
+            # 2.a) Check if already in DB
             try:
-                created_mat = await ctx.api.create_material({
-                    "name": item["material_name"],
-                    "canonical_unit_id": int(item["unit_id"]),
-                }, actor_id=user.actor_id)
-                _set_material(item, created_mat["id"], created_mat.get("name") or item["material_name"])
-                created_mats.append(item["material_name"])
-                ctx.resolver.invalidate("materials")
+                existing_mats = await ctx.api.list_materials(q=mat_name, page_size=10)
+                for m in existing_mats:
+                    if normalize_text(m.get("name") or "") == normalize_text(mat_name):
+                        _set_material(item, str(m["id"]), m.get("name") or mat_name,
+                                      {"canonical_unit_id": m.get("canonical_unit_id")})
+                        break
             except Exception as e:
-                logger.warning("Could not auto-create material: %s", e)
+                logger.warning("Could not check existing materials: %s", e)
+
+            # 2.b) If still not in DB, create it
+            if not item.get("material_id"):
+                try:
+                    created_mat = await ctx.api.create_material({
+                        "name": mat_name,
+                        "canonical_unit_id": int(item["unit_id"]),
+                    }, actor_id=user.actor_id)
+                    _set_material(item, created_mat["id"], created_mat.get("name") or mat_name)
+                    created_mats.append(mat_name)
+                    ctx.resolver.invalidate("materials")
+                except Exception as e:
+                    logger.warning("Could not auto-create material %s: %s — trying fallback fetch", mat_name, e)
+                    try:
+                        existing_mats = await ctx.api.list_materials(q=mat_name, page_size=10)
+                        for m in existing_mats:
+                            if normalize_text(m.get("name") or "") == normalize_text(mat_name):
+                                _set_material(item, str(m["id"]), m.get("name") or mat_name,
+                                              {"canonical_unit_id": m.get("canonical_unit_id")})
+                                break
+                    except Exception:
+                        pass
     if created_mats:
         newly_created["materials"] = created_mats
 
-    # 3. Sequential pre-creation for new projects
+    # 3. Defensive pre-creation for new projects (get_or_create)
     created_prjs = []
     all_p_names = []
     if payload.get("project_names"):
@@ -429,14 +500,26 @@ async def commit(ctx, chat_id, user, draft, only_resolved: bool = False,
                 all_p_names.append(it_p_str)
 
     resolved_map = payload.setdefault("resolved_projects", {})
+    all_known_prjs = None
     for p_name in all_p_names:
         p_id = resolved_map.get(p_name)
         if not p_id:
-            res = await safe_resolve(ctx.resolver, "project", p_name)
-            if res.resolved:
-                p_id = res.candidates[0].id
-                resolved_map[p_name] = p_id
-            else:
+            # 3.a) Check if project already exists in catalog
+            if all_known_prjs is None:
+                try:
+                    all_known_prjs = await ctx.resolver.projects()
+                except Exception:
+                    all_known_prjs = []
+            norm_pn = normalize_text(p_name)
+            for prj in all_known_prjs:
+                if normalize_text(prj.name) == norm_pn:
+                    p_id = prj.id
+                    resolved_map[p_name] = p_id
+                    resolved_map[prj.name] = p_id
+                    break
+
+            # 3.b) If not in catalog, create it
+            if not p_id:
                 try:
                     created_prj = await ctx.api.create_project({"name": p_name}, actor_id=user.actor_id)
                     p_id = created_prj.get("id")
@@ -444,6 +527,7 @@ async def commit(ctx, chat_id, user, draft, only_resolved: bool = False,
                         resolved_map[p_name] = p_id
                         created_prjs.append(p_name)
                         ctx.resolver.invalidate("projects")
+                        all_known_prjs = None
                 except Exception as e:
                     logger.warning("Could not auto-create project %s: %s", p_name, e)
 

bot/app/handlers/supplier.py

@@ -8,7 +8,7 @@ from app.drafts import (
     KIND_SUPPLIER, merge_supplier_patch, new_supplier_payload, supplier_missing,
 )
 from app.keyboards import pick_buttons, simple_review_keyboard
-from app.resolve import safe_resolve
+from app.nlu import normalize_text
 from app.texts import ERR_GENERIC, esc, supplier_committed_report, supplier_review_text
 
 logger = logging.getLogger("bot.supplier")
@@ -33,17 +33,26 @@ async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None,
 async def resolve_payload(ctx, payload: dict, trace_id: str | None = None) -> None:
     options = payload.setdefault("options", {})
     if not payload.get("category_id") and payload.get("category_name"):
-        res = await safe_resolve(ctx.resolver, "category", payload["category_name"])
-        await obs.log_resolution(trace_id, "category", payload["category_name"], res.status,
-                                 [{"id": c.id, "name": c.name} for c in (res.candidates or [])[:4]])
-        if res.resolved:
-            payload["category_id"] = res.candidates[0].id
-            payload["category_name"] = res.candidates[0].name
+        norm_cat = normalize_text(payload["category_name"])
+        try:
+            cats = await ctx.resolver.categories()
+        except Exception:
+            cats = []
+        exact = [c for c in cats if normalize_text(c.name) == norm_cat or (c.extra or {}).get("slug") == norm_cat]
+        if len(exact) == 1:
+            payload["category_id"] = exact[0].id
+            payload["category_name"] = exact[0].name
             options.pop("cat", None)
-        elif res.candidates:
-            options["cat"] = [{"id": c.id, "name": c.name} for c in res.candidates]
         else:
-            options.pop("cat", None)
+            partial = [c for c in cats if norm_cat in normalize_text(c.name) and len(norm_cat) >= 3]
+            if len(partial) == 1:
+                payload["category_id"] = partial[0].id
+                payload["category_name"] = partial[0].name
+                options.pop("cat", None)
+            elif partial:
+                options["cat"] = [{"id": c.id, "name": c.name} for c in partial]
+            else:
+                options.pop("cat", None)
 
 
 async def apply_pick(ctx, chat_id, user, draft, field: str, choice: dict) -> None:

bot/app/matching.py

@@ -86,13 +86,24 @@ def _instructions(field: str, variants: list[str]) -> str:
     )
 
 
+# ---------------------------------------------------------------------------
+from app.fallback_matcher import MatchingFailureError, match_with_deepseek, adjudicate_with_deepseek
+
 # ---------------------------------------------------------------------------
 # 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.
+async def resolve_mentions(
+    ctx,
+    patch: dict,
+    mentions: list[dict],
+    trace_id: str | None = None,
+    raw_input: object = None,
+    consensus_text: str | None = None,
+) -> dict:
+    """Run Two-Stage Matching: Jev scorer + DeepSeek final arbiter; DeepSeek crisis fallback if Jev down.
 
+    Return a map keyed by normalized variant.
     Each value: {"field", "status": exact|ambiguity|new", "variants",
                  "id"?, "name"?, "extra"?, "confidence"?, "candidates": [...]}
     """
@@ -107,15 +118,26 @@ async def resolve_mentions(ctx, patch: dict, mentions: list[dict], trace_id) ->
     if not by_field:
         return results
 
+    unhandled_mentions: list[dict] = []
+    unhandled_fields: set[str] = set()
+    jev_arbitration_items: list[dict] = []
+    cands_by_field: dict[str, list] = {}
+
     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)
+            unhandled_mentions.extend(ms)
+            unhandled_fields.add(field)
             return
         if not cands:
+            unhandled_mentions.extend(ms)
+            unhandled_fields.add(field)
             return
 
+        cands_by_field[field] = cands
+
         by_name: dict[str, object] = {}
         criteria: dict[str, str] = {}
         for c in cands:
@@ -148,7 +170,15 @@ async def resolve_mentions(ctx, patch: dict, mentions: list[dict], trace_id) ->
                         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.
+            # Observability: log jev.failure event
+            await obs.event(trace_id, "jev.failure", field=field, payload={
+                "error": "jev_unavailable_or_error",
+                "reason": "HTTP error or timeout or no response",
+                "model": settings.JEV_MODEL,
+                "n_mentions": len(capped),
+            })
+            unhandled_mentions.extend(capped)
+            unhandled_fields.add(field)
             return
         answers = resp.get("answers") or {}
 
@@ -156,52 +186,80 @@ async def resolve_mentions(ctx, patch: dict, mentions: list[dict], trace_id) ->
         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.
+                unhandled_mentions.append(m)
+                unhandled_fields.add(field)
                 continue
             choice = ans.get("choice")
             conf = float(ans.get("confidence") or 0.0)
             probs = ans.get("probabilities") or {}
+            non_zero_probs = {k: v for k, v in probs.items() if float(v or 0.0) > 0.0}
             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),
+                 for n, p in non_zero_probs.items() if n != "_none" and n in by_name),
                 key=lambda x: -x["prob"],
             )[:6]
 
-            base = {"field": field, "variants": m["variants"], "confidence": conf,
-                    "choice": (cand.name if cand is not None else None),
-                    "candidates": ranked}
+            # Collect for DeepSeek final arbitration (never write directly to results)
+            jev_arbitration_items.append({
+                "field": field,
+                "variants": m["variants"],
+                "candidates": cands,
+                "jev_choice": cand.name if cand else ("_none" if choice == "_none" else None),
+                "jev_choice_id": cand.id if cand else None,
+                "jev_confidence": conf,
+                "jev_probabilities": non_zero_probs,
+                "jev_ranked": ranked,
+            })
+
+            decisions.append({
+                "variants": m["variants"],
+                "jev_choice": cand.name if cand else choice,
+                "confidence": conf,
+                "probs_top": [{k: round(v, 3)} for k, v in
+                              sorted(probs.items(), key=lambda kv: -kv[1])[:3]],
+            })
 
-            if cand is not None and conf >= settings.JEV_CONF_EXACT:
-                res = {**base, "status": "exact", "id": cand.id, "name": cand.name,
-                       "extra": cand.extra or {}}
-            elif cand is not None and conf >= settings.JEV_CONF_FLOOR:
-                res = {**base, "status": "ambiguity", "id": cand.id, "name": cand.name,
-                       "extra": cand.extra or {},
-                       "candidates": ranked or [{"id": cand.id, "name": cand.name, "prob": conf}]}
-            elif cand is None:
-                # Explicit «_none» (or an unusable choice): the model judged the whole
-                # list → treat as new (the auto-create path handles it).
-                res = {**base, "status": "new"}
-            else:
-                # Picked something but below the floor = low trust: leave NO result so
-                # the difflib fallback still gets its chance (suppressing it here would
-                # let a correct-but-under-confident pick create duplicates instead).
-                await obs.event(trace_id, "routing.decision", field=field, payload={
-                    "source": "jev-lowtrust", "variants": m["variants"],
-                    "choice": cand.name, "confidence": conf,
-                })
-                continue
-            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})
+                        payload={"source": "jev_scored", "mentions": decisions})
+
+    if settings.JEV_ENABLED:
+        await asyncio.gather(*(one_field(f, ms) for f, ms in by_field.items()))
+    else:
+        # Jev disabled -> all mentions route to DeepSeek fallback
+        for f, ms in by_field.items():
+            unhandled_mentions.extend(ms)
+            unhandled_fields.add(f)
+
+    # 1. Final Arbitration: Pass Jev recommendations + raw user input + DB catalog to DeepSeek
+    if jev_arbitration_items:
+        adjudicated_res = await adjudicate_with_deepseek(
+            raw_input=raw_input,
+            consensus_text=consensus_text,
+            items=jev_arbitration_items,
+            patch_context=patch,
+            trace_id=trace_id,
+        )
+        results.update(adjudicated_res)
+
+    # 2. Crisis Fallback: If any mentions were unhandled / failed in Jev, run DeepSeek Crisis Fallback
+    if unhandled_mentions:
+        for f in unhandled_fields:
+            if f not in cands_by_field:
+                try:
+                    cands_by_field[f] = await _candidates(ctx.resolver, f)
+                except Exception:
+                    cands_by_field[f] = []
+        fallback_res = await match_with_deepseek(
+            raw_input=raw_input,
+            mentions=unhandled_mentions,
+            candidates_by_field=cands_by_field,
+            patch_context=patch,
+            trace_id=trace_id,
+            consensus_text=consensus_text,
+        )
+        results.update(fallback_res)
 
-    await asyncio.gather(*(one_field(f, ms) for f, ms in by_field.items()))
     return results
 
 

bot/app/nlu.py

@@ -16,6 +16,11 @@ from app.config import Settings
 
 logger = logging.getLogger("bot.nlu")
 
+
+class NLUFailureError(Exception):
+    """Raised when NLU understanding fails due to LLM / network failure."""
+    pass
+
 PERSIAN_DIGITS = str.maketrans("۰۱۲۳۴۵۶۷۸۹٠١٢٣٤٥٦٧٨٩", "01234567890123456789")
 
 
@@ -116,7 +121,7 @@ SYSTEM_PROMPT = """تو دستیار هوشمند تدارکات یک سیستم
 ۶. استخراج موجودیت‌های قابل تطبیق (entity_mentions):
    - فیلدهای مجاز: "project" | "supplier" | "material" | "brand" | "spec" | "unit"
    - هر نام کالا/تأمین‌کننده/پروژه/برند/واحد و هر مشخصه که در پیام آمده، یک عضو جداگانه است.
-   - اگر در نسخه‌های مختلف STT همان نام با املاهای متفاوت پیاده شده (مثل «مصطفوی»، «مستفدی»، «مسطفی»)،
+   - اگر در نسخه‌های مختلف STT همان نام با املاهای متفاوت پیاده شده (مثل «فولاد گستر»، «پولاد گستر»)،
      همه را در variants همان یک عضو بگذار — نه چند عضو جداگانه.
    - برای پیام متنی معمولی، variants یک عضو تک‌عنصر است.
    - مشخصه (spec) و برند (brand) جدا از متریال و جدا از هم استخراج شوند.
@@ -177,10 +182,6 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary:
     """Return {"consensus_text": ..., "intent": ..., "patch": {...}, "confidence": ...}; never raises."""
     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
-
     if isinstance(text, list):
         first_text = text[0].get("text") if text else ""
         lines = ["پیام صوتی کاربر توسط چند مدل تبدیل گفتار به متن (STT) به صورت زیر پیاده‌سازی شده است:"]
@@ -203,6 +204,10 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary:
 
     fallback["consensus_text"] = first_text
 
+    if not settings.AI_BASE_URL or not settings.AI_API_KEY:
+        logger.warning("AI provider not configured; cannot understand message")
+        return fallback
+
     system_prompt = _build_system_prompt(draft_summary)
     prompt_hash = hashlib.sha256(
         f"{settings.OBS_PROMPT_VERSION}|{system_prompt}".encode()
@@ -233,14 +238,14 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary:
             logger.error("LLM HTTP %s: %s", resp.status_code, resp.text[:300])
             await obs.event(trace_id, "nlu.result", ok=False, model=settings.LLM_MODEL,
                             payload={"error": f"HTTP {resp.status_code}", "body": resp.text[:300]})
-            return fallback
+            raise NLUFailureError(f"HTTP {resp.status_code}")
         content = ((resp.json() or {}).get("choices") or [{}])[0].get("message", {}).get("content", "")
         parsed = _extract_json(content)
         if not parsed or parsed.get("intent") not in INTENTS:
             logger.warning("unparseable LLM output: %r", content[:200])
             await obs.event(trace_id, "nlu.result", ok=False, model=settings.LLM_MODEL,
                             payload={"error": "unparseable", "raw": content[:500]})
-            return fallback
+            raise NLUFailureError("unparseable LLM output")
         parsed.setdefault("patch", {})
         parsed["entity_mentions"] = sanitize_mentions(parsed.get("entity_mentions"))
         if not parsed.get("consensus_text"):
@@ -253,8 +258,10 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary:
             "entity_mentions": parsed.get("entity_mentions"),
         })
         return parsed
-    except (httpx.HTTPError, json.JSONDecodeError, KeyError, TypeError):
+    except Exception as exc:
+        if isinstance(exc, NLUFailureError):
+            raise
         logger.exception("NLU failure")
         await obs.event(trace_id, "nlu.result", ok=False, model=settings.LLM_MODEL,
-                        payload={"error": "exception"})
-        return fallback
+                        payload={"error": "exception", "details": str(exc)})
+        raise NLUFailureError(f"NLU exception: {exc}") from exc

bot/app/observability.py

@@ -204,3 +204,20 @@ def save_voice(trace_id: str | None, message_id: int, content: bytes,
     except Exception:
         logger.warning("obs: save_voice failed for trace %s", trace_id, exc_info=True)
         return None
+
+
+def save_enhanced_voice(trace_id: str | None, message_id: int, content: bytes,
+                        ext: str = ".ogg") -> str | None:
+    """Persist an enhanced voice note under OBS_VOICE_DIR/{YYYYMMDD}/{trace_id}_{message_id}_enhanced{ext}."""
+    try:
+        day = datetime.utcnow().strftime("%Y%m%d")
+        base = Path(settings.OBS_VOICE_DIR) / day
+        base.mkdir(parents=True, exist_ok=True)
+        if ext and not ext.startswith("."):
+            ext = f".{ext}"
+        path = base / f"{trace_id}_{message_id}_enhanced{ext or '.ogg'}"
+        path.write_bytes(content)
+        return str(path)
+    except Exception:
+        logger.warning("obs: save_enhanced_voice failed for trace %s", trace_id, exc_info=True)
+        return None

bot/app/resolve.py

@@ -3,7 +3,6 @@
 The ERP search endpoints are ilike-based; we fetch candidates with `q=` and
 rank them with difflib so "میلگرد A3 14" still matches "میلگرد A3 سایز ۱۴".
 """
-import difflib
 import time
 from dataclasses import dataclass, field
 
@@ -44,7 +43,7 @@ def _rank(query: str, items: list[Candidate], limit: int = 6) -> list[Candidate]
         elif name in q and len(name) >= 3:
             ratio = 0.85 + 0.15 * (len(name) / len(q))
         else:
-            ratio = difflib.SequenceMatcher(None, q, name).ratio()
+            ratio = 0.0
         if ratio >= 0.72:
             scored.append((ratio, item))
     scored.sort(key=lambda pair: (-pair[0], pair[1].name))
@@ -365,13 +364,6 @@ class Resolver:
         exact = [c for c in candidates if normalize_text(c.name) == norm_name]
         if len(exact) == 1:
             return Resolution("one", exact)
-        top = candidates[0]
-        top_norm = normalize_text(top.name)
-        ratio = difflib.SequenceMatcher(None, norm_name, top_norm).ratio()
-        if (norm_name in top_norm or top_norm in norm_name) and len(norm_name) >= 3:
-            ratio = max(ratio, 0.90)
-        if ratio >= 0.88 and (len(candidates) == 1 or ratio > 0.95):
-            return Resolution("one", [top])
         return Resolution("many", candidates)
 
     async def unit_for_material(self, unit_name: str | None, material: dict | None) -> tuple[str | None, bool]:

bot/app/stt.py

@@ -76,7 +76,8 @@ async def _call_single_stt(client: httpx.AsyncClient, provider: dict, converted_
 
 
 async def transcribe_multi(settings: Settings, audio: bytes, mime_type: str | None = None,
-                           filename: str | None = None, trace_id: str | None = None) -> list[dict]:
+                           filename: str | None = None, trace_id: str | None = None,
+                           message_id: int | None = None) -> list[dict]:
     """Transcribe audio with all configured STT providers in parallel.
 
     Returns a list of candidate dicts: [{"name": str, "text": str, "model": str}, ...]
@@ -87,6 +88,13 @@ async def transcribe_multi(settings: Settings, audio: bytes, mime_type: str | No
     if not providers:
         raise RuntimeError("No STT providers are configured (PROCUREMENT_BOT_STT_*)")
 
+    # Audio enhancement / denoising step (pluggable, fail-soft, zero-transcode by default)
+    from app.audio_enhancer import enhance_voice
+    audio, _ = await enhance_voice(
+        settings, audio, mime_type=mime_type, filename=filename,
+        trace_id=trace_id, message_id=message_id,
+    )
+
     suffix = Path(filename or "").suffix
     if not suffix and mime_type:
         suffix = {"audio/ogg": ".ogg", "audio/opus": ".ogg", "audio/mpeg": ".mp3",
@@ -117,8 +125,10 @@ async def transcribe_multi(settings: Settings, audio: bytes, mime_type: str | No
 
 
 async def transcribe(settings: Settings, audio: bytes, mime_type: str | None = None,
-                     filename: str | None = None, trace_id: str | None = None) -> str:
+                     filename: str | None = None, trace_id: str | None = None,
+                     message_id: int | None = None) -> str:
     """Return the Persian transcript for a voice note (empty string on silence)."""
     transcripts = await transcribe_multi(settings, audio, mime_type=mime_type,
-                                         filename=filename, trace_id=trace_id)
+                                         filename=filename, trace_id=trace_id,
+                                         message_id=message_id)
     return transcripts[0]["text"] if transcripts else ""

bot/tools/smoke_test.py

@@ -255,6 +255,17 @@ def _install_fake_stores() -> None:
 
     runtime.LinkStore.get, runtime.LinkStore.upsert, runtime.LinkStore.delete = link_get, link_upsert, link_delete
 
+    import app.observability as obs
+    async def fake_obs(*args, **kwargs):
+        pass
+    async def fake_obs_trace(*args, **kwargs):
+        return None
+    obs.event = fake_obs
+    obs.trace_start = fake_obs
+    obs.link_draft = fake_obs
+    obs.set_trace_status = fake_obs
+    obs.latest_trace_for_draft = fake_obs_trace
+
 
 async def fake_log(self, chat_id, direction, kind, body=None):
     pass
@@ -263,16 +274,16 @@ async def fake_log(self, chat_id, direction, kind, body=None):
 NLU_QUEUE: list[dict] = []
 
 
-async def fake_understand(settings, text, draft_summary=None):
+async def fake_understand(settings, text, draft_summary=None, *args, **kwargs):
     return NLU_QUEUE.pop(0) if NLU_QUEUE else {"intent": "unknown", "patch": {}, "confidence": 0}
 
 
-async def fake_transcribe(settings, audio, mime_type=None, filename=None):
+async def fake_transcribe(settings, audio, mime_type=None, filename=None, *args, **kwargs):
     return ("سازان نو برای پروژه برج آسمان ۵۰۰ کیلو میلگرد A3 سایز ۱۴ هر کیلو ۴۵ هزار تومن "
             "پرداخت سه فاز")
 
 
-async def fake_transcribe_multi(settings, audio, mime_type=None, filename=None):
+async def fake_transcribe_multi(settings, audio, mime_type=None, filename=None, *args, **kwargs):
     return [{"name": "fake", "text": "سازان نو برای پروژه برج آسمان ۵۰۰ کیلو میلگرد A3 سایز ۱۴ هر کیلو ۴۵ هزار تومن پرداخت سه فاز"}]
 
 

bot/tools/test_enhancer.py

@@ -0,0 +1,182 @@
+"""Tests for audio preprocessing / enhancement module (audio_enhancer.py).
+
+Validates:
+  1. Disabled enhancer -> zero overhead pass-through of raw bytes.
+  2. Successful enhancement -> receives enhanced audio, saves with _enhanced suffix,
+     emits audio.enhance.result with latency_ms/path/size.
+  3. Failure/timeout -> graceful fallback to raw audio bytes, ok=False event logged.
+  4. Zero transcoding by default (raw bytes sent without ffmpeg).
+  5. Timeout configuration adheres to 60.0s.
+"""
+import asyncio
+import os
+import sys
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import httpx
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from app.audio_enhancer import enhance_voice
+from app.config import Settings
+from app import observability as obs
+
+_passed = _failed = 0
+
+
+def check(name: str, ok: bool, detail: str = "") -> None:
+    global _passed, _failed
+    ok = bool(ok)
+    mark = "PASS" if ok else "FAIL"
+    print(f"[{mark}] {name}" + (f" — {detail}" if detail and not ok else ""))
+    _passed += ok
+    _failed += (not ok)
+
+
+async def test_disabled_enhancer():
+    cfg = Settings(AUDIO_ENHANCE_ENABLED=False, AUDIO_ENHANCE_BASE_URL="http://mock/enhance")
+    raw_audio = b"fake-raw-voice-bytes"
+    out_audio, meta = await enhance_voice(cfg, raw_audio, mime_type="audio/ogg")
+    check("disabled: raw bytes returned directly", out_audio == raw_audio)
+    check("disabled: metadata indicates skipped", meta.get("skipped") is True)
+
+
+async def test_timeout_setting():
+    cfg = Settings(AUDIO_ENHANCE_TIMEOUT_SECONDS=60.0)
+    check("config: timeout defaults to 60.0s", cfg.AUDIO_ENHANCE_TIMEOUT_SECONDS == 60.0)
+
+
+async def test_successful_enhancement():
+    cfg = Settings(
+        AUDIO_ENHANCE_ENABLED=True,
+        AUDIO_ENHANCE_BASE_URL="http://mock-enhancer.internal/v1/enhance",
+        AUDIO_ENHANCE_MODEL="voice-cleaner-v1",
+        AUDIO_ENHANCE_TIMEOUT_SECONDS=60.0,
+    )
+    raw_audio = b"noisy-voice-bytes"
+    enhanced_audio = b"crystal-clear-voice-bytes"
+    trace_id = "test-trace-enh-001"
+    message_id = 12345
+
+    events_captured = []
+
+    async def fake_event(t_id, stage, ok=True, latency_ms=None, model=None, payload=None, **kwargs):
+        events_captured.append({"stage": stage, "ok": ok, "latency_ms": latency_ms, "payload": payload})
+
+    # Mock httpx response
+    mock_resp = httpx.Response(
+        status_code=200,
+        content=enhanced_audio,
+        headers={"content-type": "audio/ogg"},
+        request=httpx.Request("POST", cfg.AUDIO_ENHANCE_BASE_URL),
+    )
+
+    with patch.object(obs, "event", side_effect=fake_event):
+        with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_resp
+            out_bytes, meta = await enhance_voice(
+                cfg,
+                raw_audio,
+                mime_type="audio/ogg",
+                filename="voice.ogg",
+                trace_id=trace_id,
+                message_id=message_id,
+            )
+
+    check("success: enhanced bytes returned", out_bytes == enhanced_audio)
+    check("success: meta indicates enhanced", meta.get("enhanced") is True)
+    check("success: enhanced path exists in meta", bool(meta.get("path")))
+    check("success: path has _enhanced suffix", "_enhanced" in (meta.get("path") or ""))
+
+    # Verify physical file existence
+    if meta.get("path"):
+        saved_file = Path(meta["path"])
+        check("success: enhanced file written to disk", saved_file.exists() and saved_file.read_bytes() == enhanced_audio)
+
+    # Verify observability event
+    enh_events = [e for e in events_captured if e["stage"] == "audio.enhance.result"]
+    check("obs: audio.enhance.result event emitted", len(enh_events) == 1)
+    if enh_events:
+        ev = enh_events[0]
+        check("obs: event ok=True", ev["ok"] is True)
+        check("obs: payload contains size & latency_ms", "size" in ev["payload"] and ev.get("latency_ms") is not None)
+        check("obs: payload recorded correct size", ev["payload"]["size"] == len(enhanced_audio))
+
+
+async def test_graceful_fallback_on_error():
+    cfg = Settings(
+        AUDIO_ENHANCE_ENABLED=True,
+        AUDIO_ENHANCE_BASE_URL="http://mock-enhancer.internal/v1/enhance",
+        AUDIO_ENHANCE_TIMEOUT_SECONDS=60.0,
+    )
+    raw_audio = b"original-unprocessed-audio"
+    trace_id = "test-trace-enh-002"
+    message_id = 999
+
+    events_captured = []
+
+    async def fake_event(t_id, stage, ok=True, latency_ms=None, model=None, payload=None, **kwargs):
+        events_captured.append({"stage": stage, "ok": ok, "latency_ms": latency_ms, "payload": payload})
+
+    # Mock HTTP 500 error
+    mock_resp = httpx.Response(
+        status_code=500,
+        text="Internal Model Error",
+        request=httpx.Request("POST", cfg.AUDIO_ENHANCE_BASE_URL),
+    )
+
+    with patch.object(obs, "event", side_effect=fake_event):
+        with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_resp
+            out_bytes, meta = await enhance_voice(
+                cfg,
+                raw_audio,
+                mime_type="audio/ogg",
+                trace_id=trace_id,
+                message_id=message_id,
+            )
+
+    check("fallback: returns raw original bytes on 500", out_bytes == raw_audio)
+    check("fallback: meta indicates fallback", meta.get("fallback") is True)
+
+    enh_events = [e for e in events_captured if e["stage"] == "audio.enhance.result"]
+    check("obs: fallback event ok=False", len(enh_events) == 1 and enh_events[0]["ok"] is False)
+    if enh_events:
+        check("obs: error detail captured", "HTTP 500" in enh_events[0]["payload"].get("error", ""))
+
+
+async def test_zero_transcoding_by_default():
+    cfg = Settings(
+        AUDIO_ENHANCE_ENABLED=True,
+        AUDIO_ENHANCE_BASE_URL="http://mock-enhancer.internal/v1/enhance",
+        AUDIO_ENHANCE_TARGET_FORMAT="",  # empty -> zero transcoding
+    )
+    raw_audio = b"raw-bale-ogg-bytes"
+    captured_payload = None
+
+    async def fake_post(url, files=None, data=None, headers=None, **kwargs):
+        nonlocal captured_payload
+        captured_payload = files["file"][1]
+        return httpx.Response(status_code=200, content=b"clean", request=httpx.Request("POST", url))
+
+    with patch("httpx.AsyncClient.post", side_effect=fake_post):
+        await enhance_voice(cfg, raw_audio, mime_type="audio/ogg")
+
+    check("transcoding: raw bytes sent directly without modification", captured_payload == raw_audio)
+
+
+async def main():
+    print("Running audio_enhancer unit tests...")
+    await test_disabled_enhancer()
+    await test_timeout_setting()
+    await test_successful_enhancement()
+    await test_graceful_fallback_on_error()
+    await test_zero_transcoding_by_default()
+
+    print(f"\nResult: {_passed} passed, {_failed} failed")
+    return 1 if _failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

bot/tools/test_fallback_matching.py

@@ -0,0 +1,403 @@
+"""Unit and integration tests for DeepSeek crisis fallback matching.
+
+Covers:
+1. Jev failure -> jev.failure observability event + DeepSeek fallback invocation.
+2. DeepSeek fallback results mapping: exact, ambiguity, new (with inferred category).
+3. Defensive failure guard: DeepSeek network/HTTP/JSON failure raises MatchingFailureError.
+4. quote.py resolve_payload: respects new, binds exact, populates ambiguity, zero difflib.
+5. quote.py commit: defensive get_or_create pattern prevents 409 duplicate errors.
+"""
+import asyncio
+import json
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from app import fallback_matcher, matching
+from app.config import settings
+from app.fallback_matcher import MatchingFailureError
+from app.handlers import quote
+from app.nlu import normalize_text
+from app.resolve import Candidate
+
+PASSED: list[str] = []
+FAILED: list[str] = []
+
+
+def check(name: str, ok: bool, extra: str = "") -> None:
+    (PASSED if ok else FAILED).append(name)
+    print(f"{'PASS' if ok else 'FAIL'}  {name}{'  — ' + str(extra) if extra else ''}")
+
+
+class FakeResolver:
+    async def projects(self):
+        return [Candidate(id="p1", name="پروژه پردیس"), Candidate(id="p2", name="برج نگین")]
+
+    async def all_suppliers(self):
+        return [Candidate(
+            id="s1",
+            name="فولاد خوزستان",
+            extra={"location": "اهواز", "category_name": "فلزات و مقاطع فولادی", "slug": "metals-steel-sections"}
+        )]
+
+    async def all_materials(self):
+        return [
+            Candidate(id="m1", name="میلگرد ۱۴ آجدار", extra={"canonical_unit_id": 1}),
+            Candidate(id="m2", name="لوله پی‌وی‌سی ۴ اینچ", extra={"canonical_unit_id": 3}),
+        ]
+
+    async def categories(self):
+        return [
+            Candidate(id="cat1", name="فلزات و مقاطع فولادی", extra={"slug": "metals-steel-sections"}),
+            Candidate(id="cat2", name="تأسیسات مکانیکی و آبی", extra={"slug": "mechanical-installations-pipes"}),
+        ]
+
+    async def units(self):
+        return [Candidate(id="u1", name="شاخه", extra={"code": "branch"}), Candidate(id="u2", name="کیلوگرم", extra={"code": "kg"})]
+
+    def invalidate(self, key=None):
+        pass
+
+
+class FakeApi:
+    def __init__(self):
+        self.created_materials = []
+        self.created_suppliers = []
+        self.created_projects = []
+
+    async def list_materials(self, q=None, page_size=20, page=1):
+        if q and "میلگرد ۱۴" in q:
+            return [{"id": "m1", "name": "میلگرد ۱۴ آجدار", "canonical_unit_id": 1}]
+        return []
+
+    async def create_material(self, body, actor_id=None):
+        if body["name"] == "میلگرد ۱۴ آجدار":
+            raise Exception("Conflict 409")
+        item = {"id": f"m_new_{len(self.created_materials)+1}", **body}
+        self.created_materials.append(item)
+        return item
+
+    async def list_suppliers(self, q=None, page_size=20, page=1):
+        if q and "فولاد خوزستان" in q:
+            return [{"id": "s1", "name": "فولاد خوزستان"}]
+        return []
+
+    async def create_supplier(self, body, actor_id=None):
+        item = {"id": f"s_new_{len(self.created_suppliers)+1}", **body}
+        self.created_suppliers.append(item)
+        return item
+
+    async def list_projects(self, page_size=100):
+        return [{"id": "p1", "name": "پروژه پردیس"}]
+
+    async def create_project(self, body, actor_id=None):
+        item = {"id": f"p_new_{len(self.created_projects)+1}", **body}
+        self.created_projects.append(item)
+        return item
+
+
+async def test_jev_failure_triggers_fallback() -> None:
+    """1. When Jev returns None (HTTP 500/timeout), DeepSeek fallback is triggered."""
+    ctx = SimpleNamespace(resolver=FakeResolver())
+    events_logged = []
+
+    async def fake_event(trace_id, event_type, ok=True, **kwargs):
+        events_logged.append((event_type, kwargs))
+
+    orig_event = matching.obs.event
+    matching.obs.event = fake_event
+
+    saved_key = settings.TYPESAFE_API_KEY
+    settings.TYPESAFE_API_KEY = "test-typesafe-key"
+
+    orig_jev_ask = matching.jev_ask
+    matching.jev_ask = lambda state, qs: asyncio.sleep(0.001, result=None)
+
+    # Mock DeepSeek fallback matcher
+    async def fake_fallback(*args, **kwargs):
+        return {
+            normalize_text("میلگرد ۱۴"): {
+                "field": "material",
+                "status": "exact",
+                "id": "m1",
+                "name": "میلگرد ۱۴ آجدار",
+                "variants": ["میلگرد ۱۴"],
+                "candidates": [],
+            },
+            normalize_text("لوله ۶ اینچ"): {
+                "field": "material",
+                "status": "new",
+                "id": None,
+                "name": None,
+                "variants": ["لوله ۶ اینچ"],
+                "candidates": [],
+            },
+            normalize_text("آهن"): {
+                "field": "material",
+                "status": "ambiguity",
+                "id": None,
+                "name": None,
+                "variants": ["آهن"],
+                "candidates": [{"id": "m1", "name": "میلگرد ۱۴ آجدار", "prob": 0.8}],
+            },
+            normalize_text("تأمین‌کننده البرز"): {
+                "field": "supplier",
+                "status": "new",
+                "id": None,
+                "name": None,
+                "variants": ["تأمین‌کننده البرز"],
+                "candidates": [],
+                "inferred_category_slug": "metals-steel-sections",
+            },
+        }
+
+    orig_deepseek = matching.match_with_deepseek
+    matching.match_with_deepseek = fake_fallback
+
+    try:
+        mentions = [
+            {"field": "material", "variants": ["میلگرد ۱۴"]},
+            {"field": "material", "variants": ["لوله ۶ اینچ"]},
+            {"field": "material", "variants": ["آهن"]},
+            {"field": "supplier", "variants": ["تأمین‌کننده البرز"]},
+        ]
+        res = await matching.resolve_mentions(ctx, {"items": []}, mentions, trace_id="t1", raw_input="تست")
+
+        check("Jev failure recorded in observability", any(e[0] == "jev.failure" for e in events_logged))
+        check("Exact match resolved via fallback", res[normalize_text("میلگرد ۱۴")]["status"] == "exact")
+        check("New item strictly preserved as new (no difflib forced match)",
+              res[normalize_text("لوله ۶ اینچ")]["status"] == "new" and res[normalize_text("لوله ۶ اینچ")]["id"] is None)
+        check("Ambiguity item has candidates",
+              res[normalize_text("آهن")]["status"] == "ambiguity" and len(res[normalize_text("آهن")]["candidates"]) == 1)
+        check("New supplier has inferred_category_slug",
+              res[normalize_text("تأمین‌کننده البرز")]["inferred_category_slug"] == "metals-steel-sections")
+
+    finally:
+        settings.TYPESAFE_API_KEY = saved_key
+        matching.obs.event = orig_event
+        matching.jev_ask = orig_jev_ask
+        matching.match_with_deepseek = orig_deepseek
+
+
+async def test_two_stage_adjudication_flow() -> None:
+    """2. When Jev succeeds, outputs are NOT directly committed; they are sent to DeepSeek Arbiter."""
+    ctx = SimpleNamespace(resolver=FakeResolver())
+    saved_key = settings.TYPESAFE_API_KEY
+    settings.TYPESAFE_API_KEY = "test-typesafe-key"
+
+    orig_jev_ask = matching.jev_ask
+    matching.jev_ask = lambda state, qs: asyncio.sleep(0.001, result={
+        "answers": {
+            "mention_0": {
+                "choice": "میلگرد ۱۴ آجدار",
+                "confidence": 0.95,
+                "probabilities": {"میلگرد ۱۴ آجدار": 0.95, "تسمه فلزی": 0.0, "_none": 0.05},
+            },
+            "mention_1": {
+                "choice": "لوله پی‌وی‌سی ۴ اینچ",
+                "confidence": 0.82,
+                "probabilities": {"لوله پی‌وی‌سی ۴ اینچ": 0.82, "لوله پلی‌اتیلن": 0.0, "_none": 0.18},
+            },
+        }
+    })
+
+    arbiter_called_with = []
+
+    async def fake_arbiter(*args, **kwargs):
+        arbiter_called_with.append((args, kwargs))
+        items = kwargs.get("items") or (args[2] if len(args) > 2 else [])
+        res = {}
+        for it in items:
+            v0 = it["variants"][0]
+            if v0 == "میلگرد ۱۴":
+                # Endorsed
+                res[normalize_text(v0)] = {
+                    "field": it["field"], "status": "exact", "id": "m1", "name": "میلگرد ۱۴ آجدار",
+                    "variants": it["variants"], "source": "deepseek-arbiter", "adjudication_action": "endorsed_jev",
+                }
+            elif v0 == "لوله ۶ اینچ":
+                # Vetoed Jev (Jev wrongly picked 4-inch pipe)
+                res[normalize_text(v0)] = {
+                    "field": it["field"], "status": "new", "id": None, "name": None,
+                    "variants": it["variants"], "source": "deepseek-arbiter", "adjudication_action": "vetoed_to_new",
+                }
+        return res
+
+    orig_arbiter = matching.adjudicate_with_deepseek
+    matching.adjudicate_with_deepseek = fake_arbiter
+
+    try:
+        mentions = [
+            {"field": "material", "variants": ["میلگرد ۱۴"]},
+            {"field": "material", "variants": ["لوله ۶ اینچ"]},
+        ]
+        res = await matching.resolve_mentions(
+            ctx, {"items": []}, mentions, trace_id="t_arb",
+            raw_input="استعلام میلگرد ۱۴ و لوله ۶ اینچ",
+        )
+
+        check("Two-stage arbitration called when Jev succeeds", len(arbiter_called_with) == 1)
+        passed_items = arbiter_called_with[0][1].get("items") or arbiter_called_with[0][0][2]
+        m0_probs = passed_items[0]["jev_probabilities"]
+        check("Zero probability items filtered out", "تسمه فلزی" not in m0_probs)
+        check("Non-zero probability preserved exactly", m0_probs.get("میلگرد ۱۴ آجدار") == 0.95)
+        check("_none with non-zero probability preserved", m0_probs.get("_none") == 0.05)
+        check("Catalog candidates list complete and untouched", len(passed_items[0]["candidates"]) > 0)
+        check("Arbiter endorsed valid Jev choice as exact", res[normalize_text("میلگرد ۱۴")]["status"] == "exact")
+        check("Arbiter vetoed wrong Jev choice to new", res[normalize_text("لوله ۶ اینچ")]["status"] == "new" and res[normalize_text("لوله ۶ اینچ")]["id"] is None)
+    finally:
+        settings.TYPESAFE_API_KEY = saved_key
+        matching.jev_ask = orig_jev_ask
+        matching.adjudicate_with_deepseek = orig_arbiter
+
+
+async def test_defensive_failure_guard() -> None:
+    """3. Universal Fail-Stop Guard: If DeepSeek (fallback or arbiter or NLU) fails, MatchingFailureError is raised."""
+    ctx = SimpleNamespace(resolver=FakeResolver())
+    matching.jev_ask = lambda state, qs: asyncio.sleep(0.001, result=None)
+
+    async def boom_fallback(*args, **kwargs):
+        raise MatchingFailureError("DeepSeek API 500 error")
+
+    orig_deepseek = matching.match_with_deepseek
+    matching.match_with_deepseek = boom_fallback
+
+    try:
+        mentions = [{"field": "material", "variants": ["سیمان"]}]
+        failed = False
+        try:
+            await matching.resolve_mentions(ctx, {"items": []}, mentions, trace_id="t2")
+        except MatchingFailureError:
+            failed = True
+        check("Crisis fallback failure raises MatchingFailureError (stops cleanly)", failed)
+    finally:
+        matching.match_with_deepseek = orig_deepseek
+
+    # Also test Arbiter failure raises MatchingFailureError
+    matching.jev_ask = lambda state, qs: asyncio.sleep(0.001, result={"answers": {"mention_0": {"choice": "m1", "confidence": 0.9}}})
+    async def boom_arbiter(*args, **kwargs):
+        raise MatchingFailureError("Arbiter DeepSeek timeout")
+
+    orig_arbiter = matching.adjudicate_with_deepseek
+    matching.adjudicate_with_deepseek = boom_arbiter
+    try:
+        failed_arb = False
+        try:
+            await matching.resolve_mentions(ctx, {"items": []}, mentions, trace_id="t2_arb")
+        except MatchingFailureError:
+            failed_arb = True
+        check("Arbiter failure raises MatchingFailureError (stops cleanly)", failed_arb)
+    finally:
+        matching.adjudicate_with_deepseek = orig_arbiter
+
+    # Test NLUFailureError is subclass of Exception
+    from app.nlu import NLUFailureError
+    check("NLUFailureError is subclass of Exception", issubclass(NLUFailureError, Exception))
+
+
+async def test_quote_resolve_payload_zero_difflib() -> None:
+    """3. resolve_payload in quote.py respects Jev/fallback map and never runs difflib."""
+    ctx = SimpleNamespace(resolver=FakeResolver())
+    payload = {
+        "supplier_name": "تأمین‌کننده البرز",
+        "items": [
+            {"material_name": "لوله ۶ اینچ", "unit_name": "شاخه"},
+            {"material_name": "میلگرد ۱۴", "unit_name": "شاخه"},
+        ],
+    }
+    jev_map = {
+        normalize_text("تأمین‌کننده البرز"): {
+            "field": "supplier", "status": "new", "variants": ["تأمین‌کننده البرز"],
+            "inferred_category_slug": "metals-steel-sections",
+        },
+        normalize_text("لوله ۶ اینچ"): {
+            "field": "material", "status": "new", "variants": ["لوله ۶ اینچ"],
+        },
+        normalize_text("میلگرد ۱۴"): {
+            "field": "material", "status": "exact", "id": "m1", "name": "میلگرد ۱۴ آجدار",
+            "variants": ["میلگرد ۱۴"],
+        },
+        normalize_text("شاخه"): {
+            "field": "unit", "status": "exact", "id": "u1", "name": "شاخه", "variants": ["شاخه"],
+        },
+    }
+
+    await quote.resolve_payload(ctx, payload, trace_id="t3", jev_map=jev_map)
+
+    check("New supplier kept id=None (no difflib forced match)", payload.get("supplier_id") is None)
+    check("New supplier inferred category saved", payload.get("supplier_inferred_category_slug") == "metals-steel-sections")
+    check("New material kept id=None (no difflib forced match to 4-inch pipe)", payload["items"][0].get("material_id") is None)
+    check("Exact material bound to catalog ID", payload["items"][1].get("material_id") == "m1")
+    check("Exact unit bound to catalog ID", payload["items"][0].get("unit_id") == "u1")
+
+
+async def test_quote_commit_defensive_get_or_create() -> None:
+    """4. commit in quote.py uses get_or_create to prevent 409 conflict errors."""
+    api = FakeApi()
+    resolver = FakeResolver()
+    ctx = SimpleNamespace(api=api, resolver=resolver, pending={},
+                          send=lambda *a, **k: asyncio.sleep(0.0001))
+    user = SimpleNamespace(actor_id="usr_1")
+
+    # Mock obs calls so offline test doesn't require live DB connection
+    orig_latest = quote.obs.latest_trace_for_draft
+    orig_link = quote.obs.link_draft
+    orig_event = quote.obs.event
+    quote.obs.latest_trace_for_draft = lambda did: asyncio.sleep(0.0001, result="mock_trace")
+    quote.obs.link_draft = lambda *a, **k: asyncio.sleep(0.0001)
+    quote.obs.event = lambda *a, **k: asyncio.sleep(0.0001)
+
+    # Item with name "میلگرد ۱۴ آجدار" that already exists in DB
+    payload = {
+        "supplier_name": "فولاد خوزستان",  # Already in DB
+        "supplier_inferred_category_slug": "metals-steel-sections",
+        "items": [
+            {"material_name": "میلگرد ۱۴ آجدار", "unit_id": 1, "quantity": 10, "unit_price": 25000},
+            {"material_name": "تیرآهن ۱۸", "unit_id": 1, "quantity": 5, "unit_price": 50000},  # Brand new
+        ],
+        "project_name": "پروژه پردیس",  # Already in DB
+    }
+    draft = SimpleNamespace(id="d1", payload=payload, review_message_id=None)
+
+    # Mock create_quote
+    api.create_quote = lambda body, actor_id: asyncio.sleep(0.001, result={"id": "q1"})
+    # Mock draft_store.set_status
+    quote.draft_store.set_status = lambda did, st: asyncio.sleep(0.001)
+
+    try:
+        await quote.commit(ctx, chat_id=123, user=user, draft=draft, only_resolved=False)
+
+        check("Existing supplier reused ID from DB (not recreated)", payload["supplier_id"] == "s1")
+        check("Existing material reused ID from DB on conflict (not crashing)", payload["items"][0]["material_id"] == "m1")
+        check("New material created cleanly", payload["items"][1]["material_id"] == "m_new_1")
+        check("Existing project reused ID from DB", payload["project_id"] == "p1")
+    finally:
+        quote.obs.latest_trace_for_draft = orig_latest
+        quote.obs.link_draft = orig_link
+        quote.obs.event = orig_event
+
+
+async def main() -> None:
+    print("Running DeepSeek Crisis Fallback & Zero-Difflib Tests...\n")
+    orig_fb_event = fallback_matcher.obs.event
+    fallback_matcher.obs.event = lambda *a, **k: asyncio.sleep(0.0001)
+    try:
+        await test_jev_failure_triggers_fallback()
+        await test_two_stage_adjudication_flow()
+        await test_defensive_failure_guard()
+        await test_quote_resolve_payload_zero_difflib()
+        await test_quote_commit_defensive_get_or_create()
+    finally:
+        fallback_matcher.obs.event = orig_fb_event
+
+    total = len(PASSED) + len(FAILED)
+    print(f"\n{len(PASSED)}/{total} passed")
+    if FAILED:
+        print("FAILED:", FAILED)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

bot/tools/test_matching.py

@@ -74,13 +74,44 @@ async def mock_tests() -> None:
         "mention_3": {"choice": "دفتر", "confidence": 0.5,
                       "probabilities": {"دفتر": 0.52, "_none": 0.48}},
     })
+    orig_fallback = matching.match_with_deepseek
+    async def mock_fallback_below_floor(*args, **kwargs):
+        ms = kwargs.get("mentions") or (args[1] if len(args) > 1 else [])
+        return {normalize_text(v): {"status": "new", "field": m.get("field"), "variants": m.get("variants"), "id": None}
+                for m in ms for v in m.get("variants", [])}
+    matching.match_with_deepseek = mock_fallback_below_floor
+
+    orig_arbiter = matching.adjudicate_with_deepseek
+    async def mock_arbiter(*args, **kwargs):
+        items = kwargs.get("items") or (args[2] if len(args) > 2 else [])
+        res = {}
+        for it in items:
+            choice = it.get("jev_choice")
+            conf = it.get("jev_confidence", 0.0)
+            if choice == "منشانه" and conf >= 0.8:
+                entry = {"status": "exact", "id": "p2", "name": "منشانه", "field": it["field"], "variants": it["variants"]}
+            elif choice == "دفتر" and conf >= 0.6:
+                entry = {"status": "ambiguity", "id": "p1", "name": "دفتر", "field": it["field"], "variants": it["variants"],
+                         "candidates": [{"id": "p1", "name": "دفتر", "prob": conf}]}
+            else:
+                entry = {"status": "new", "id": None, "name": None, "field": it["field"], "variants": it["variants"], "candidates": []}
+            for v in it["variants"]:
+                res[normalize_text(v)] = entry
+        return res
+    matching.adjudicate_with_deepseek = mock_arbiter
+
     mentions = [
         {"field": "project", "variants": ["منشیانه"]},
         {"field": "project", "variants": ["دفتر", "دفر"]},
         {"field": "project", "variants": ["چیز عجیب"]},
         {"field": "project", "variants": ["دفر دو"]},
     ]
-    res = await matching.resolve_mentions(ctx, {"items": []}, mentions, None)
+    try:
+        res = await matching.resolve_mentions(ctx, {"items": []}, mentions, None)
+    finally:
+        matching.match_with_deepseek = orig_fallback
+        matching.adjudicate_with_deepseek = orig_arbiter
+
     r0 = matching.lookup(res, "منشیانه")
     r1 = matching.lookup(res, "دفر")          # variant key resolves too
     r2 = matching.lookup(res, "چیز عجیب")
@@ -90,15 +121,23 @@ async def mock_tests() -> None:
     check("variant of same mention shares one result", r1 is r0 if False else
           (r1 and matching.lookup(res, "دفتر") is r1))
     check("_none / low conf -> new", r2 and r2["status"] == "new")
-    check("below-floor pick -> NO result (difflib fallback keeps its chance)",
-          matching.lookup(res, "دفر دو") is None)
+    check("below-floor pick -> routed to crisis fallback",
+          matching.lookup(res, "دفر دو") and matching.lookup(res, "دفر دو")["status"] == "new")
     check("below-floor pick did not pollute other mentions", r1 and r1["status"] == "ambiguity")
 
-    # 2) Jev failure -> empty map (callers fall back to difflib)
+    # 2) Jev failure -> DeepSeek crisis fallback
     matching.jev_ask = make_fake_ask({})
     matching.jev_ask.fail = True
-    res_fail = await matching.resolve_mentions(ctx, {"items": []}, mentions, None)
-    check("jev failure -> empty results (difflib fallback path)", res_fail == {})
+    async def mock_deepseek(*args, **kwargs):
+        return {normalize_text("منشیانه"): {"status": "exact", "id": "p2", "name": "منشانه",
+                                            "field": "project", "variants": ["منشیانه"]}}
+    matching.match_with_deepseek = mock_deepseek
+    try:
+        res_fail = await matching.resolve_mentions(ctx, {"items": []}, mentions, None)
+        check("jev failure -> triggers crisis fallback",
+              normalize_text("منشیانه") in res_fail and res_fail[normalize_text("منشیانه")]["status"] == "exact")
+    finally:
+        matching.match_with_deepseek = orig_fallback
 
     # 3) apply_exact writes supplier/project ids into a payload
     payload = {"items": []}