Fix duplicate_service.py empty-store crash and concurrency bugs

backend/data/case_history_cache.json

@@ -0,0 +1,14 @@
+[
+  {
+    "ticket_id": "t-001",
+    "text": "I cannot connect to WiFi"
+  },
+  {
+    "ticket_id": "t-002",
+    "text": "VPN authentication fails repeatedly"
+  },
+  {
+    "ticket_id": "t-003",
+    "text": "VPN connection keeps dropping"
+  }
+]

backend/services/duplicate_service.py

@@ -5,6 +5,9 @@
 
 import uuid
 import os
+import threading
+import tempfile
+import torch
 from sentence_transformers import SentenceTransformer, util
 
 SIMILARITY_THRESHOLD = 0.70
@@ -19,6 +22,8 @@ def __init__(self):
         self._tickets: list[tuple[str, object, str]] = []
         self.storage_file = os.path.join(os.path.dirname(__file__), "..", "data", "case_history_cache.json")
         os.makedirs(os.path.dirname(self.storage_file), exist_ok=True)
+        self._lock = threading.Lock()
+        self._file_lock = threading.Lock()
 
     def is_available(self) -> bool:
         """Check if the model is available for duplicate detection."""
@@ -47,10 +52,11 @@ def load(self):
                 try:
                     with open(self.storage_file, "r") as f:
                         data = json.load(f)
-                        for item in data:
-                            text = item["text"]
-                            embedding = self.model.encode(text, convert_to_tensor=True)
-                            self._tickets.append((item["ticket_id"], embedding, text))
+                        with self._lock:
+                            for item in data:
+                                text = item["text"]
+                                embedding = self.model.encode(text, convert_to_tensor=True)
+                                self._tickets.append((item["ticket_id"], embedding, text))
                     print(f"[DuplicateService] Loaded {len(self._tickets)} tickets.")
                 except Exception as e:
                     print(f"[DuplicateService] Error loading storage: {e}")
@@ -68,24 +74,28 @@ def load(self):
     def save_to_disk(self, ticket_id: str, text: str):
         """Append a new ticket to the JSON storage."""
         import json
-        data = []
-        try:
-            os.makedirs(os.path.dirname(self.storage_file), exist_ok=True)
+        with self._file_lock:
+            data = []
             if os.path.exists(self.storage_file):
                 with open(self.storage_file, "r") as f:
                     try:
                         data = json.load(f)
-                        if not isinstance(data, list):
-                            data = []
-                    except:
+                    except json.JSONDecodeError:
                         data = []
 
             data.append({"ticket_id": ticket_id, "text": text})
-            with open(self.storage_file, "w") as f:
-                json.dump(data, f, indent=2)
-            print(f"[DuplicateService] Indexed ticket {ticket_id} to case history.")
-        except Exception as e:
-            print(f"[DuplicateService] Failed to save to disk: {e}")
+            tmp_fd, tmp_path = tempfile.mkstemp(
+                dir=os.path.dirname(self.storage_file), suffix=".tmp"
+            )
+            try:
+                with os.fdopen(tmp_fd, "w") as tf:
+                    json.dump(data, tf, indent=2)
+                os.replace(tmp_path, self.storage_file)
+                print(f"[DuplicateService] Indexed ticket {ticket_id} to case history.")
+            except Exception as e:
+                os.unlink(tmp_path)
+                print(f"[DuplicateService] Failed to save to disk: {e}")
+                raise
 
     def add_ticket(self, ticket_id: str, text: str):
         """Add a ticket to the in-memory store and persist to disk."""
@@ -94,7 +104,8 @@ def add_ticket(self, ticket_id: str, text: str):
             print(f"[DuplicateService] DEGRADED: Skipping embedding for ticket {ticket_id} (model not available)")
             return
         embedding = self.model.encode(text, convert_to_tensor=True)
-        self._tickets.append((ticket_id, embedding, text))
+        with self._lock:
+            self._tickets.append((ticket_id, embedding, text))
         self.save_to_disk(ticket_id, text)
 
     def check_duplicate(self, text: str, threshold: float = None) -> dict:
@@ -109,45 +120,41 @@ def check_duplicate(self, text: str, threshold: float = None) -> dict:
             {
                 "is_duplicate": bool,
                 "duplicate_ticket_id": str | None,
-                "similarity": float
-            }
+                "similarity_score": float
+            } or None if unavailable/empty or below threshold
         """
         self.load()
 
         # If model is not available, return no duplicate found
         if not self.is_available():
             print("[DuplicateService] DEGRADED: Duplicate check skipped (model not available)")
-            return {
-                "is_duplicate": False,
-                "duplicate_ticket_id": None,
-                "similarity": 0.0,
-            }
+            return None
 
         # Use provided threshold or default to global constant
         active_threshold = threshold if threshold is not None else SIMILARITY_THRESHOLD
 
-        if not self._tickets:
-            return {
-                "is_duplicate": False,
-                "duplicate_ticket_id": None,
-                "similarity": 0.0,
-            }
-
-        query_embedding = self.model.encode(text, convert_to_tensor=True)
+        with self._lock:
+            if not self._tickets:
+                return None
+            tickets_snapshot = list(self._tickets)
 
-        best_score = 0.0
-        best_id = None
+        new_emb = self.model.encode(text, convert_to_tensor=True)
+        embeddings = [e for _, e, _ in tickets_snapshot]
+        stacked = torch.stack(embeddings)
+
+        # util.pytorch_cos_sim returns an N x M matrix
+        scores = util.pytorch_cos_sim(new_emb, stacked)[0]
+
+        best_score_val, best_idx = torch.max(scores, dim=0)
+        best_score = best_score_val.item()
 
-        for ticket_id, stored_emb, _ in self._tickets:
-            score = util.cos_sim(query_embedding, stored_emb).item()
-            if score > best_score:
-                best_score = score
-                best_id = ticket_id
+        if best_score >= active_threshold:
+            return {
+                "is_duplicate": True,
+                "duplicate_ticket_id": tickets_snapshot[best_idx.item()][0],
+                "similarity_score": round(best_score, 4),
+            }
 
-        is_dup = best_score >= active_threshold
+        return None
 
-        return {
-            "is_duplicate": is_dup,
-            "duplicate_ticket_id": best_id if is_dup else None,
-            "similarity": round(best_score, 4),
-        }
+duplicate_service = DuplicateService()

backend/tests/__init__.py

@@ -0,0 +1 @@
+# Empty init file

backend/tests/conftest.py

@@ -0,0 +1,33 @@
+import sys
+from unittest.mock import MagicMock, patch
+import torch
+import pytest
+
+# Mock sentence_transformers at module level
+_mock_st = MagicMock()
+_mock_model = MagicMock()
+
+def _fake_encode(text, convert_to_tensor=False):
+    import hashlib, torch
+    if 'VPN' in text:
+        seed = 42
+    else:
+        seed = sum(ord(c) for c in text)
+    torch.manual_seed(seed)
+    return torch.nn.functional.normalize(torch.randn(1, 384), dim=1).squeeze(0)
+
+_mock_model.encode.side_effect = _fake_encode
+_mock_st.SentenceTransformer.return_value = _mock_model
+_mock_st.util.pytorch_cos_sim = lambda a, b: torch.nn.functional.cosine_similarity(
+    a.unsqueeze(0), b, dim=1
+).unsqueeze(0)
+sys.modules["sentence_transformers"] = _mock_st
+
+@pytest.fixture
+def dup_svc():
+    from backend.services.duplicate_service import DuplicateService
+    svc = DuplicateService()
+    svc._loaded = True
+    svc._load_failed = False
+    svc.model = _mock_model
+    return svc

backend/tests/test_duplicate_service.py

@@ -0,0 +1,61 @@
+import pytest
+import torch
+from unittest.mock import MagicMock, patch
+
+# ─── 1. Empty store returns None ────────────────────────────────────────────
+def test_check_duplicate_empty_store_returns_none(dup_svc):
+    result = dup_svc.check_duplicate("VPN is not working")
+    assert result is None, "Empty store must return None, not raise RuntimeError"
+
+# ─── 2. Exact duplicate detected ────────────────────────────────────────────
+def test_exact_duplicate_detected(dup_svc, tmp_path):
+    dup_svc.storage_file = str(tmp_path / "cache.json")
+    dup_svc.add_ticket("t-001", "VPN is not working")
+    result = dup_svc.check_duplicate("VPN is not working")
+    assert result is not None
+    assert result["duplicate_ticket_id"] == "t-001"
+    assert result["similarity_score"] >= 0.99
+
+# ─── 3. Below-threshold text returns None ───────────────────────────────────
+def test_below_threshold_returns_none(dup_svc, tmp_path):
+    dup_svc.storage_file = str(tmp_path / "cache.json")
+    dup_svc.add_ticket("t-001", "VPN is not working")
+    result = dup_svc.check_duplicate("The printer has no paper")
+    assert result is None
+
+# ─── 4. Picks highest similarity among multiple tickets ──────────────────────
+def test_picks_best_match_among_multiple(dup_svc, tmp_path):
+    dup_svc.storage_file = str(tmp_path / "cache.json")
+    dup_svc.add_ticket("t-001", "I cannot connect to WiFi")
+    dup_svc.add_ticket("t-002", "VPN authentication fails repeatedly")
+    dup_svc.add_ticket("t-003", "VPN connection keeps dropping")
+    result = dup_svc.check_duplicate("VPN keeps disconnecting")
+    assert result is not None
+    assert result["duplicate_ticket_id"] in ("t-002", "t-003")
+
+# ─── 5. Unavailable service returns None ────────────────────────────────────
+def test_unavailable_returns_none(dup_svc):
+    dup_svc._loaded = False
+    result = dup_svc.check_duplicate("test")
+    assert result is None
+
+# ─── 6. Thread safety — concurrent add+check does not raise ─────────────────
+def test_concurrent_add_check_no_exception(dup_svc, tmp_path):
+    import threading
+    dup_svc.storage_file = str(tmp_path / "cache.json")
+    errors = []
+    def add(i):
+        try:
+            dup_svc.add_ticket(f"t-{i}", f"ticket text {i}")
+        except Exception as e:
+            errors.append(e)
+    def check():
+        try:
+            dup_svc.check_duplicate("ticket text 5")
+        except Exception as e:
+            errors.append(e)
+    threads = [threading.Thread(target=add, args=(i,)) for i in range(20)]
+    threads += [threading.Thread(target=check) for _ in range(10)]
+    for t in threads: t.start()
+    for t in threads: t.join()
+    assert not errors, f"Thread safety violation: {errors}"