Upload folder using huggingface_hub

app/validator/engine.py

@@ -226,6 +226,23 @@ class EmailValidationEngine:
         # ---- Layer 4: Disposable Detection ---------------------------
         disposable = check_disposable(normalized_email, domain)
 
+        # ML advisory (single path): unknown domains only, fail-open. Mirrors
+        # the batch pipeline so both surfaces agree on the same address.
+        ml_model = get_model()
+        if (
+            ml_model is not None
+            and not domain_is_known(domain)
+            and deep
+        ):
+            try:
+                preds = ml_model.predict([f"john.smith [SEP] {domain}"])
+            except Exception:
+                preds = None
+            if preds:
+                disposable = disposable.model_copy(
+                    update={"ml_label": preds[0][0], "ml_confidence": preds[0][1]}
+                )
+
         if not deep:
             # Fast path: skip SMTP for quick validation
             elapsed = (time.monotonic() - start_time) * 1000

tests/test_engine_ml.py

@@ -46,7 +46,10 @@ class FakeProvider:
 
 def _engine_with_ml(flag=True, repo="user/model", predict=None):
     engine = EmailValidationEngine(mailbox=FakeProvider(), cache=NullCache())
-    if predict is not None:
+    # The get_model patcher must honor the flag: with the flag off the REAL
+    # get_model returns None (that IS the production contract), so the double
+    # must not force a model into a disabled engine.
+    if predict is not None and flag:
         fake_model = MagicMock()
         fake_model.predict = MagicMock(side_effect=predict)
         fake_model.load = MagicMock(return_value=True)
@@ -79,6 +82,13 @@ def _stop(engine):
 
 
 class EngineMLTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self._engines = []
+
+    def tearDown(self):
+        # Stop settings/model patchers so they never leak into other tests.
+        for e in self._engines:
+            _stop(e)
     def tearDown(self):
         for e in getattr(self, "_engines", []):
             _stop(e)
@@ -165,5 +175,40 @@ class EngineMLTests(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(results[1].disposable.ml_label, "NORMAL")
 
 
+    async def test_single_validate_also_applies_the_ml_signal(self):
+        # The single endpoint runs engine.validate() - it must get the same
+        # advisory signal as bulk, or the two surfaces disagree on one address.
+        engine = _engine_with_ml(
+            flag=True, predict=[[("DISPOSABLE", 0.93)]]
+        )
+        self._engines.append(engine)
+        with patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())):
+            result = await engine.validate("a@brand-new-disposable-xyz.tk", deep=True)
+
+        self.assertEqual(result.disposable.ml_label, "DISPOSABLE")
+        self.assertEqual(
+            result.scoring.breakdown.get("ML_DISPOSABLE_SUSPECT"), -5
+        )
+
+    async def test_single_validate_flag_off_unchanged(self):
+        engine = _engine_with_ml(flag=False, predict=[[("DISPOSABLE", 0.93)]])
+        self._engines.append(engine)
+        with patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())):
+            result = await engine.validate("a@brand-new-disposable-xyz.tk", deep=True)
+
+        self.assertIsNone(result.disposable.ml_label)
+
+    async def test_single_validate_known_domain_skips_ml(self):
+        engine = _engine_with_ml(
+            flag=True, predict=[AssertionError("must not run")]
+        )
+        self._engines.append(engine)
+        with patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())):
+            result = await engine.validate("a@mailinator.com", deep=True)
+
+        self.assertEqual(result.status, ValidationStatus.INVALID)
+        self.assertIsNone(result.disposable.ml_label)
+
+
 if __name__ == "__main__":
     unittest.main()