fix(sec-core): degrade prompt scan to L1 when ML model not downloaded

src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/prompt_scanner/detectors/ml_classifier.py

@@ -51,6 +51,27 @@ def __init__(
     def name(self) -> str:
         return "ml_classifier"
 
+    def is_available(self) -> bool:
+        """Return True only if torch/transformers are importable AND the
+        model is downloaded locally.
+
+        Returns False (instead of raising) when dependencies are missing or
+        the model has not been fetched yet (e.g. ``scan-prompt warmup`` never
+        ran), so the scanner can skip L2 and degrade to L1 rather than
+        failing every scan.
+        """
+        import importlib.util
+
+        if (
+            importlib.util.find_spec("torch") is None
+            or importlib.util.find_spec("transformers") is None
+        ):
+            return False
+
+        return self._classifier._manager.is_model_downloaded(
+            self._classifier._model_name
+        )
+
     def warmup(self) -> None:
         """Eagerly download and load the ML model.
 

src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/prompt_scanner/models/model_manager.py

@@ -170,18 +170,28 @@ def detect_device() -> str:
 
     # ------------------------------------------------------------------
 
+    def is_model_downloaded(self, model_name: str) -> bool:
+        """Return True if *model_name* is present in the local cache.
+
+        Checks the same on-disk condition as ``_resolve_local_model_path``
+        (cache dir exists and contains ``config.json``) but returns a bool
+        instead of raising, so callers like ``MLClassifier.is_available()``
+        can probe availability without exception handling.
+        """
+        cache_dir = Path(self._cache_dir).expanduser()
+        candidate = cache_dir / Path(model_name)
+        return candidate.is_dir() and (candidate / "config.json").exists()
+
     def _resolve_local_model_path(self, model_name: str) -> str:
         """Return the local cache path for *model_name* without triggering a download.
 
         Raises:
             ModelLoadError: if the model has not been downloaded yet, with a
                 user-friendly message pointing to ``scan-prompt warmup``.
         """
-        cache_dir = Path(self._cache_dir).expanduser()
-        candidate = cache_dir / Path(model_name)
-
-        if candidate.is_dir() and (candidate / "config.json").exists():
-            return str(candidate)
+        if self.is_model_downloaded(model_name):
+            cache_dir = Path(self._cache_dir).expanduser()
+            return str(cache_dir / Path(model_name))
 
         raise ModelLoadError(
             f"Model '{model_name}' is not available locally.\n"

src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/prompt_scanner/scanner.py

@@ -33,9 +33,12 @@
 }
 
 # Detectors that can be skipped silently when unavailable.
-# L1 (rule_engine) and L2 (ml_classifier) are mandatory — their deps ship
-# with the package.  Only future optional layers (e.g. L3 semantic) go here.
-_OPTIONAL_DETECTORS = frozenset({"semantic"})
+# L1 (rule_engine) is mandatory — its deps ship with the package.
+# L2 (ml_classifier) is skippable: torch/transformers ship with the package
+# but the model must be downloaded via `scan-prompt warmup`. When the model
+# is absent, skip L2 and degrade to L1 rather than failing every scan.
+# L3 (semantic) is not yet implemented.
+_OPTIONAL_DETECTORS = frozenset({"ml_classifier", "semantic"})
 
 
 class PromptScanner:
@@ -77,8 +80,20 @@ def warmup(self) -> None:
 
             agent-sec-cli scan-prompt warmup
         """
-        log.info("Warming up %d detector(s)...", len(self._detectors))
-        for detector in self._detectors:
+        # Build detectors directly from config, bypassing the is_available()
+        # gate in _init_detectors: warmup's job is to MAKE detectors available
+        # (download the model), so it must not skip a detector merely because
+        # the model is not downloaded yet.
+        log.info("Warming up %d configured detector(s)...", len(self._config.layers))
+        for name in self._config.layers:
+            cls = _DETECTOR_REGISTRY.get(name)
+            if cls is None:
+                continue
+            detector = (
+                cls(model_name=self._config.model_name)
+                if name == "ml_classifier"
+                else cls()
+            )
             if hasattr(detector, "warmup"):
                 detector.warmup()
         log.info("Warmup complete.")

src/agent-sec-core/tests/unit-test/prompt_scanner/test_ml_classifier.py

@@ -255,6 +255,37 @@ def test_device_property(self) -> None:
         mgr = ModelManager(device="cpu")
         self.assertEqual(mgr.device, "cpu")
 
+    def test_is_model_downloaded_false_when_missing(self) -> None:
+        mgr = self._make_manager()
+        self.assertFalse(mgr.is_model_downloaded("nonexistent/model"))
+
+    def test_is_model_downloaded_true_when_present(self) -> None:
+        import os
+        import tempfile
+
+        mgr = self._make_manager()
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mgr._cache_dir = tmpdir
+            model_dir = os.path.join(tmpdir, "test-model")
+            os.makedirs(model_dir)
+            with open(os.path.join(model_dir, "config.json"), "w") as f:
+                f.write("{}")
+            self.assertTrue(mgr.is_model_downloaded("test-model"))
+
+    def test_resolve_local_model_path_returns_path_when_downloaded(self) -> None:
+        import os
+        import tempfile
+
+        mgr = self._make_manager()
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mgr._cache_dir = tmpdir
+            model_dir = os.path.join(tmpdir, "test-model")
+            os.makedirs(model_dir)
+            with open(os.path.join(model_dir, "config.json"), "w") as f:
+                f.write("{}")
+            path = mgr._resolve_local_model_path("test-model")
+            self.assertEqual(path, model_dir)
+
     def test_load_model_raises_without_deps(self) -> None:
         from agent_sec_cli.prompt_scanner.exceptions import ModelLoadError
         from agent_sec_cli.prompt_scanner.models.model_manager import (
@@ -447,17 +478,31 @@ def test_name_property(self) -> None:
         layer = self.MLClassifier.__new__(self.MLClassifier)
         self.assertEqual(layer.name, "ml_classifier")
 
-    def test_is_available_false_without_deps(self) -> None:
-        # MLClassifier.is_available() is inherited from DetectionLayer and
-        # always returns True (deps are mandatory); this verifies it returns True.
+    def _make_layer_with_model(self, downloaded: bool):
         layer = self.MLClassifier.__new__(self.MLClassifier)
-        self.assertTrue(layer.is_available())
+        mock_clf = MagicMock()
+        mock_clf._model_name = "test-model"
+        mock_clf._manager.is_model_downloaded.return_value = downloaded
+        layer._classifier = mock_clf
+        return layer
 
-    def test_is_available_true_with_deps(self) -> None:
-        fake_torch = _make_fake_torch()
-        fake_tf = types.ModuleType("transformers")
-        layer = self.MLClassifier.__new__(self.MLClassifier)
-        with patch.dict(sys.modules, {"torch": fake_torch, "transformers": fake_tf}):
+    def test_is_available_false_when_model_missing(self) -> None:
+        # Deps present but model NOT downloaded -> False, so the scanner skips
+        # L2 and degrades to L1 instead of failing every scan.
+        layer = self._make_layer_with_model(downloaded=False)
+        with patch("importlib.util.find_spec", return_value=object()):
+            self.assertFalse(layer.is_available())
+
+    def test_is_available_false_without_deps(self) -> None:
+        # torch/transformers missing -> False even if the model is present
+        # (deps are probed first, before the on-disk model check).
+        layer = self._make_layer_with_model(downloaded=True)
+        with patch("importlib.util.find_spec", return_value=None):
+            self.assertFalse(layer.is_available())
+
+    def test_is_available_true_with_deps_and_model(self) -> None:
+        layer = self._make_layer_with_model(downloaded=True)
+        with patch("importlib.util.find_spec", return_value=object()):
             self.assertTrue(layer.is_available())
 
     def test_detect_raises_when_deps_missing(self) -> None:

src/agent-sec-core/tests/unit-test/prompt_scanner/test_scanner.py

@@ -52,6 +52,34 @@ def test_standard_mode_skips_ml_when_unavailable(self) -> None:
         # present depending on the test environment; just check no exception raised
         self.assertGreaterEqual(len(scanner._detectors), 1)
 
+    def test_warmup_bypasses_availability_gate(self) -> None:
+        from agent_sec_cli.prompt_scanner.detectors.ml_classifier import (
+            MLClassifier,
+        )
+
+        with patch.object(MLClassifier, "is_available", return_value=False):
+            scanner = PromptScanner(mode=ScanMode.STANDARD)
+            self.assertEqual(len(scanner._detectors), 1)
+            with patch.object(MLClassifier, "warmup", return_value=None) as mock_warmup:
+                scanner.warmup()
+                mock_warmup.assert_called_once()
+
+    def test_standard_mode_degrades_to_l1_when_ml_model_missing(self) -> None:
+        # When ml_classifier.is_available() is False (e.g. model not downloaded),
+        # STANDARD mode must degrade to L1 (rule_engine) only — NOT raise.
+        # This is the discriminating test for the fix: before adding
+        # ml_classifier to _OPTIONAL_DETECTORS, this raised LayerNotAvailableError.
+        from agent_sec_cli.prompt_scanner.detectors.ml_classifier import (
+            MLClassifier,
+        )
+
+        with patch.object(MLClassifier, "is_available", return_value=False):
+            scanner = PromptScanner(mode=ScanMode.STANDARD)
+        names = [d.name for d in scanner._detectors]
+        self.assertIn("rule_engine", names)
+        self.assertNotIn("ml_classifier", names)
+        self.assertEqual(len(scanner._detectors), 1)
+
     def test_custom_config_unknown_detector_raises(self) -> None:
         config = ScanConfig(layers=["nonexistent_layer"])
         with self.assertRaises(ValueError):