From 1d1ed29e763f438742d40b76cdb6d8d6032f3a87 Mon Sep 17 00:00:00 2001
From: Madhu Goutham Reddy Ambati <mambati@redhat.com>
Date: Tue, 4 Aug 2026 17:45:46 -0400
Subject: [PATCH 1/3] feat(preflight): add plugin, API version, and image
 checks before deploy

Signed-off-by: Madhu Goutham Reddy Ambati <mambati@redhat.com>
---
 src/conformance/preflight.py | 313 +++++++++++++++++++++++++++++++++++
 tests/test_conformance.py    |  43 +++++
 tests/test_preflight.py      | 226 +++++++++++++++++++++++++
 3 files changed, 582 insertions(+)
 create mode 100644 src/conformance/preflight.py
 create mode 100644 tests/test_preflight.py

diff --git a/src/conformance/preflight.py b/src/conformance/preflight.py
new file mode 100644
index 0000000..7f26b30
--- /dev/null
+++ b/src/conformance/preflight.py
@@ -0,0 +1,313 @@
+"""Pre-flight compatibility checks for LLMInferenceService manifests.
+
+Three checks, all dynamic — no hardcoded versions:
+
+  1. **Plugin compatibility** — manifest plugins vs installed EPP
+     (live probe from running EPP, or source extract from router repo)
+  2. **API version** — manifest apiVersion vs CRD served versions
+  3. **Image existence** — EPP image pullable? (skopeo inspect)
+
+Each check degrades safely — skip with warning, never false-fail.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+import subprocess
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import yaml
+
+log = logging.getLogger(__name__)
+
+
+@dataclass
+class PreflightResult:
+    compatible: bool
+    missing_plugins: set[str] = field(default_factory=set)
+    required_plugins: set[str] = field(default_factory=set)
+    available_count: int = 0
+    source: str = ""
+    diagnosis: str = ""
+    skipped_reason: str = ""
+
+
+def extract_required_plugins(manifest_path: Path) -> set[str]:
+    """Extract plugin type names from a manifest's scheduler config."""
+    with open(manifest_path) as f:
+        docs = list(yaml.safe_load_all(f))
+    manifest = docs[0] if docs else {}
+
+    plugins: set[str] = set()
+    inline = (
+        manifest.get("spec", {})
+        .get("router", {})
+        .get("scheduler", {})
+        .get("config", {})
+        .get("inline", {})
+    )
+    for plugin in (inline.get("plugins") or []):
+        ptype = plugin.get("type", "")
+        if ptype and not plugin.get("optional", False):
+            plugins.add(ptype)
+
+    sat_ref = (
+        inline.get("flowControl", {})
+        .get("saturationDetector", {})
+        .get("pluginRef", "")
+    )
+    if sat_ref:
+        plugins.add(sat_ref)
+
+    return plugins
+
+
+def probe_epp_plugins(kubectl_fn, namespace: str) -> frozenset[str] | None:
+    """Layer 1: query a running EPP pod for registered plugins."""
+    try:
+        pods_raw = kubectl_fn(
+            "get", "pods",
+            "-l", "app.kubernetes.io/component=llminferenceservice-router-scheduler",
+            "--field-selector", "status.phase=Running",
+            "-o", "jsonpath={.items[0].metadata.name}",
+            "-n", namespace, check=False,
+        )
+        if not pods_raw or not pods_raw.strip():
+            return None
+
+        pod = pods_raw.strip().split()[0]
+        logs = kubectl_fn("logs", pod, "--tail=200", "-n", namespace, check=False)
+        if not logs:
+            return None
+
+        plugins = set()
+        for line in logs.splitlines():
+            if "registered plugin" in line.lower() or "plugin type" in line.lower():
+                plugins.update(re.findall(r'"([a-z][a-z0-9-]+)"', line))
+
+        if plugins:
+            return frozenset(plugins)
+    except Exception:
+        pass
+    return None
+
+
+def extract_plugins_from_source(router_repo: str | Path, tag: str) -> frozenset[str]:
+    """Layer 2: extract plugin type strings from router Go source at a git tag."""
+    result = subprocess.run(
+        ["git", "grep", "-h", r'Type\s*\(PluginType\)\?=\s*"', tag, "--", "*.go"],
+        capture_output=True, text=True, cwd=str(router_repo),
+    )
+    if not result.stdout.strip():
+        result = subprocess.run(
+            ["git", "grep", "-h", r'Type\s*=\s*"', tag, "--", "*.go"],
+            capture_output=True, text=True, cwd=str(router_repo),
+        )
+
+    plugins = set(re.findall(r'"([a-z][a-z0-9-]+)"', result.stdout))
+    non_plugins = {
+        "content-type", "custom", "default", "colliding-source-type",
+        "decode-only", "encode-decode", "encode-prefill-decode",
+        "header-based-testing-filter", "destination-endpoint-served-verifier",
+    }
+    return frozenset(plugins - non_plugins)
+
+
+def _detect_epp_version_tag(kubectl_fn) -> str:
+    """Try to extract a version tag from the installed EPP image reference."""
+    for ns in ("redhat-ods-applications", "rhai-gitops", "rhaii"):
+        try:
+            raw = kubectl_fn(
+                "get", "llminferenceserviceconfig",
+                "-o", "jsonpath={.items[0].spec.router.scheduler.template.containers[0].image}",
+                "-n", ns, check=False,
+            )
+            if not raw or not raw.strip():
+                continue
+            image = raw.strip().split()[0]
+            if ":" in image and "@" not in image:
+                tag = image.rsplit(":", 1)[-1]
+                if tag.startswith("v"):
+                    return tag
+        except Exception:
+            continue
+    return ""
+
+
+def resolve_available_plugins(
+    kubectl_fn,
+    namespace: str,
+    router_repo: str | Path | None = None,
+) -> tuple[frozenset[str] | None, str]:
+    """Try both layers to determine available plugins."""
+    live = probe_epp_plugins(kubectl_fn, namespace)
+    if live:
+        return live, f"live probe ({len(live)} plugins)"
+
+    if router_repo:
+        router_path = Path(router_repo)
+        if router_path.exists():
+            try:
+                epp_tag = _detect_epp_version_tag(kubectl_fn)
+                if epp_tag:
+                    plugins = extract_plugins_from_source(router_path, epp_tag)
+                    if plugins:
+                        return plugins, f"source extract at {epp_tag} ({len(plugins)} plugins)"
+                    log.info("Tag %s found but no plugins extracted — tag may not exist in router repo", epp_tag)
+
+                if not epp_tag:
+                    result = subprocess.run(
+                        ["git", "tag", "--list", "v*.*.*", "--sort=v:refname"],
+                        capture_output=True, text=True, cwd=str(router_path),
+                    )
+                    tags = [t.strip() for t in result.stdout.splitlines()
+                            if t.strip() and "rc" not in t and "alpha" not in t]
+                    if tags:
+                        plugins = extract_plugins_from_source(router_path, tags[-1])
+                        if plugins:
+                            log.warning(
+                                "Using latest tag %s — could not determine cluster EPP version",
+                                tags[-1],
+                            )
+                            return plugins, f"source extract at {tags[-1]} (latest — cluster version unknown)"
+            except Exception:
+                pass
+
+    return None, "no detection method available"
+
+
+def check_manifest_compatibility(
+    manifest_path: Path,
+    available_plugins: frozenset[str] | None,
+    source: str = "",
+) -> PreflightResult:
+    """Check if a manifest's plugins are compatible with available plugins."""
+    required = extract_required_plugins(manifest_path)
+
+    if not required:
+        return PreflightResult(
+            compatible=True, source=source,
+            diagnosis="No custom plugins — uses defaults",
+        )
+
+    if available_plugins is None:
+        return PreflightResult(
+            compatible=True, required_plugins=required,
+            skipped_reason="could not detect available plugins",
+            diagnosis="Pre-flight skipped: no detection method available",
+        )
+
+    missing = required - available_plugins
+
+    if not missing:
+        return PreflightResult(
+            compatible=True, required_plugins=required,
+            available_count=len(available_plugins), source=source,
+            diagnosis=f"All {len(required)} required plugins available ({source})",
+        )
+
+    return PreflightResult(
+        compatible=False, missing_plugins=missing,
+        required_plugins=required, available_count=len(available_plugins),
+        source=source,
+        diagnosis=(
+            f"Manifest requires {len(missing)} plugin(s) not available "
+            f"({len(available_plugins)} plugins detected via {source}):\n"
+            + "\n".join(f"  - '{p}'" for p in sorted(missing))
+            + "\nUse manifests compatible with the installed EPP version."
+        ),
+    )
+
+
+# --- Check 2: API version compatibility ---
+
+def check_api_version(manifest_path: Path, kubectl_fn) -> PreflightResult:
+    """Check if the manifest's apiVersion is served by the cluster's CRD."""
+    with open(manifest_path) as f:
+        docs = list(yaml.safe_load_all(f))
+    doc = docs[0] if docs else {}
+    api_version = doc.get("apiVersion", "")
+    version = api_version.rsplit("/", 1)[-1] if "/" in api_version else ""
+
+    if not version:
+        return PreflightResult(compatible=True, skipped_reason="no apiVersion in manifest")
+
+    try:
+        raw = kubectl_fn(
+            "get", "crd", "llminferenceservices.serving.kserve.io",
+            "-o", "jsonpath={.spec.versions[?(@.served==true)].name}",
+            check=False,
+        )
+        if not raw or not raw.strip():
+            return PreflightResult(compatible=True, skipped_reason="CRD not found")
+
+        served = set(raw.strip().split())
+    except Exception:
+        return PreflightResult(compatible=True, skipped_reason="could not query CRD versions")
+
+    if version not in served:
+        return PreflightResult(
+            compatible=False,
+            diagnosis=(
+                f"Manifest uses apiVersion '{api_version}' but the cluster CRD "
+                f"only serves {sorted(served)}."
+            ),
+        )
+
+    return PreflightResult(
+        compatible=True,
+        diagnosis=f"API version '{version}' is served by the CRD",
+    )
+
+
+# --- Check 3: Image existence ---
+
+def check_image_existence(kubectl_fn) -> PreflightResult:
+    """Check if the EPP image referenced in LLMInferenceServiceConfig is pullable."""
+    epp_image = ""
+    for ns in ("redhat-ods-applications", "rhai-gitops", "rhaii"):
+        try:
+            raw = kubectl_fn(
+                "get", "llminferenceserviceconfig",
+                "-o", "jsonpath={.items[0].spec.router.scheduler.template.containers[0].image}",
+                "-n", ns, check=False,
+            )
+            if raw and raw.strip():
+                epp_image = raw.strip().split()[0]
+                break
+        except Exception:
+            continue
+
+    if not epp_image:
+        return PreflightResult(compatible=True, skipped_reason="no EPP image found in config")
+
+    try:
+        result = subprocess.run(
+            ["skopeo", "inspect", "--raw", f"docker://{epp_image}"],
+            capture_output=True, text=True, timeout=30,
+        )
+        if result.returncode != 0:
+            stderr = result.stderr.strip()[:200].lower()
+            if "manifest unknown" in stderr or "not found" in stderr:
+                return PreflightResult(
+                    compatible=False,
+                    diagnosis=(
+                        f"EPP image '{epp_image}' does not exist: {result.stderr.strip()[:200]}\n"
+                        f"Verify the image was pushed to the registry."
+                    ),
+                )
+            return PreflightResult(
+                compatible=True,
+                skipped_reason=f"image check inconclusive: {result.stderr.strip()[:100]}",
+            )
+    except FileNotFoundError:
+        return PreflightResult(compatible=True, skipped_reason="skopeo not installed")
+    except Exception as e:
+        return PreflightResult(compatible=True, skipped_reason=f"image check failed: {e}")
+
+    return PreflightResult(
+        compatible=True,
+        diagnosis=f"EPP image '{epp_image.split('/')[-1][:50]}' exists and is pullable",
+    )
diff --git a/tests/test_conformance.py b/tests/test_conformance.py
index c6a3e1f..e1ea83f 100644
--- a/tests/test_conformance.py
+++ b/tests/test_conformance.py
@@ -34,6 +34,12 @@
 from conformance.config import TestCase, chat_prompt_to_messages
 from conformance.client import LLMClient
 from conformance.deployer import Deployer
+from conformance.preflight import (
+    check_api_version,
+    check_image_existence,
+    check_manifest_compatibility,
+    resolve_available_plugins,
+)
 from conformance.metrics import (
     Scraper,
     dump_raw_metrics,
@@ -81,6 +87,43 @@ def _check_threshold(name: str, value: float, min_value: float | None = None, ma
 class TestConformance:
     """Ordered conformance phases for each test case."""
 
+    def test_00_preflight(self, deployer: Deployer, tc: TestCase):
+        """Pre-flight: verify manifest is compatible with the cluster.
+
+        Three checks: API version, image existence, plugin compatibility.
+        Each skips safely if detection is unavailable.
+        """
+        _require_manifest(tc)
+        manifest_path = _MANIFEST_DIR / tc.deployment.manifest_path
+
+        checks = [
+            ("api_version", lambda: check_api_version(manifest_path, deployer.kubectl)),
+            ("image_existence", lambda: check_image_existence(deployer.kubectl)),
+        ]
+
+        for name, check_fn in checks:
+            result = check_fn()
+            if result.skipped_reason:
+                _log(f"Pre-flight {name}: skipped ({result.skipped_reason})")
+            elif not result.compatible:
+                _log(f"Pre-flight {name} FAIL: {result.diagnosis}")
+                pytest.fail(f"pre-flight [{name}]: {result.diagnosis}")
+            else:
+                _log(f"Pre-flight {name}: {result.diagnosis}")
+
+        available, source = resolve_available_plugins(
+            deployer.kubectl, deployer.namespace,
+            router_repo=getattr(deployer, "router_repo", None),
+        )
+        result = check_manifest_compatibility(manifest_path, available, source)
+        if result.skipped_reason:
+            _log(f"Pre-flight plugins: skipped ({result.skipped_reason})")
+        elif not result.compatible:
+            _log(f"Pre-flight plugins FAIL: {result.diagnosis}")
+            pytest.fail(f"pre-flight [plugins]: {result.diagnosis}")
+        else:
+            _log(f"Pre-flight plugins: {result.diagnosis}")
+
     def test_01_prereq(self, deployer: Deployer, tc: TestCase):
         """LLMInferenceService CRD must be installed and manifest must exist."""
         _require_manifest(tc)
diff --git a/tests/test_preflight.py b/tests/test_preflight.py
new file mode 100644
index 0000000..c21a727
--- /dev/null
+++ b/tests/test_preflight.py
@@ -0,0 +1,226 @@
+"""Tests for pre-flight plugin compatibility — no static registry, no hardcoded versions."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import yaml
+
+from conformance.preflight import (
+    check_api_version,
+    check_image_existence,
+    check_manifest_compatibility,
+    extract_plugins_from_source,
+    extract_required_plugins,
+)
+
+
+def _write_manifest(tmp_path: Path, plugins: list[dict], extras: dict | None = None) -> Path:
+    inline = {"plugins": plugins}
+    if extras:
+        inline.update(extras)
+    manifest = {
+        "apiVersion": "serving.kserve.io/v1alpha2",
+        "kind": "LLMInferenceService",
+        "metadata": {"name": "test"},
+        "spec": {"model": {"name": "m"}, "router": {"scheduler": {"config": {"inline": inline}}}},
+    }
+    path = tmp_path / "test.yaml"
+    with open(path, "w") as f:
+        yaml.dump(manifest, f)
+    return path
+
+
+class TestExtractRequiredPlugins:
+
+    def test_extracts_types(self, tmp_path):
+        path = _write_manifest(tmp_path, [{"type": "a"}, {"type": "b"}])
+        assert extract_required_plugins(path) == {"a", "b"}
+
+    def test_empty_when_no_plugins(self, tmp_path):
+        path = tmp_path / "s.yaml"
+        with open(path, "w") as f:
+            yaml.dump({"apiVersion": "v1", "spec": {"model": {"name": "m"}}}, f)
+        assert extract_required_plugins(path) == set()
+
+    def test_optional_excluded(self, tmp_path):
+        path = _write_manifest(tmp_path, [{"type": "req"}, {"type": "opt", "optional": True}])
+        assert extract_required_plugins(path) == {"req"}
+
+    def test_plugins_null_returns_empty(self, tmp_path):
+        path = tmp_path / "null.yaml"
+        with open(path, "w") as f:
+            yaml.dump({
+                "apiVersion": "v1",
+                "spec": {"router": {"scheduler": {"config": {"inline": {"plugins": None}}}}},
+            }, f)
+        assert extract_required_plugins(path) == set()
+
+    def test_saturation_detector_ref(self, tmp_path):
+        path = _write_manifest(tmp_path, [{"type": "a"}], extras={
+            "flowControl": {"saturationDetector": {"pluginRef": "det"}},
+        })
+        assert "det" in extract_required_plugins(path)
+
+
+class TestCheckManifestCompatibility:
+
+    def test_passes_when_all_available(self, tmp_path):
+        path = _write_manifest(tmp_path, [{"type": "a"}])
+        r = check_manifest_compatibility(path, frozenset({"a", "b"}), "test")
+        assert r.compatible is True
+
+    def test_fails_when_missing(self, tmp_path):
+        path = _write_manifest(tmp_path, [{"type": "a"}, {"type": "new"}])
+        r = check_manifest_compatibility(path, frozenset({"a"}), "test")
+        assert r.compatible is False
+        assert "new" in r.missing_plugins
+
+    def test_no_plugins_always_passes(self, tmp_path):
+        path = tmp_path / "s.yaml"
+        with open(path, "w") as f:
+            yaml.dump({"apiVersion": "v1", "spec": {"model": {"name": "m"}}}, f)
+        assert check_manifest_compatibility(path, frozenset(), "test").compatible is True
+
+    def test_none_available_skips(self, tmp_path):
+        path = _write_manifest(tmp_path, [{"type": "a"}])
+        r = check_manifest_compatibility(path, None)
+        assert r.compatible is True
+        assert "could not detect" in r.skipped_reason
+
+
+class TestApiVersionCheck:
+
+    def _make_manifest(self, tmp_path, api_version):
+        path = tmp_path / "test.yaml"
+        with open(path, "w") as f:
+            yaml.dump({
+                "apiVersion": api_version,
+                "kind": "LLMInferenceService",
+                "metadata": {"name": "test"},
+                "spec": {"model": {"name": "m"}},
+            }, f)
+        return path
+
+    def test_served_version_passes(self, tmp_path):
+        path = self._make_manifest(tmp_path, "serving.kserve.io/v1alpha2")
+        def kubectl(*a, **k):
+            return "v1alpha1 v1alpha2"
+        r = check_api_version(path, kubectl)
+        assert r.compatible is True
+
+    def test_unserved_version_fails(self, tmp_path):
+        path = self._make_manifest(tmp_path, "serving.kserve.io/v1alpha2")
+        def kubectl(*a, **k):
+            return "v1alpha1"
+        r = check_api_version(path, kubectl)
+        assert r.compatible is False
+        assert "v1alpha2" in r.diagnosis
+
+    def test_crd_not_found_skips(self, tmp_path):
+        path = self._make_manifest(tmp_path, "serving.kserve.io/v1alpha2")
+        def kubectl(*a, **k):
+            return ""
+        r = check_api_version(path, kubectl)
+        assert r.compatible is True
+        assert r.skipped_reason
+
+
+class TestImageExistence:
+
+    def test_skopeo_not_installed_skips(self):
+        from unittest.mock import patch
+        def kubectl(*a, **k):
+            return "quay.io/rhoai/some-image@sha256:abc"
+        with patch("subprocess.run", side_effect=FileNotFoundError):
+            r = check_image_existence(kubectl)
+        assert r.compatible is True
+        assert "skopeo" in r.skipped_reason
+
+    def test_no_epp_image_skips(self):
+        def kubectl(*a, **k):
+            return ""
+        r = check_image_existence(kubectl)
+        assert r.compatible is True
+        assert "no EPP image" in r.skipped_reason
+
+    def test_image_not_found_fails(self):
+        from unittest.mock import patch
+        import subprocess as sp
+        def kubectl(*a, **k):
+            return "quay.io/rhoai/fake-image:latest"
+        with patch("subprocess.run", return_value=sp.CompletedProcess(
+            args=[], returncode=1, stdout="", stderr="manifest unknown",
+        )):
+            r = check_image_existence(kubectl)
+        assert r.compatible is False
+        assert "does not exist" in r.diagnosis
+
+    def test_auth_error_skips_not_fails(self):
+        from unittest.mock import patch
+        import subprocess as sp
+        def kubectl(*a, **k):
+            return "quay.io/rhoai/private-image:latest"
+        with patch("subprocess.run", return_value=sp.CompletedProcess(
+            args=[], returncode=1, stdout="", stderr="unauthorized: access denied",
+        )):
+            r = check_image_existence(kubectl)
+        assert r.compatible is True
+        assert "inconclusive" in r.skipped_reason
+
+    def test_network_error_skips_not_fails(self):
+        from unittest.mock import patch
+        import subprocess as sp
+        def kubectl(*a, **k):
+            return "quay.io/rhoai/some-image:latest"
+        with patch("subprocess.run", return_value=sp.CompletedProcess(
+            args=[], returncode=1, stdout="", stderr="connection refused",
+        )):
+            r = check_image_existence(kubectl)
+        assert r.compatible is True
+        assert "inconclusive" in r.skipped_reason
+
+    def test_image_exists_passes(self):
+        from unittest.mock import patch
+        import subprocess as sp
+        def kubectl(*a, **k):
+            return "quay.io/rhoai/real-image@sha256:abc"
+        with patch("subprocess.run", return_value=sp.CompletedProcess(
+            args=[], returncode=0, stdout="{}", stderr="",
+        )):
+            r = check_image_existence(kubectl)
+        assert r.compatible is True
+
+
+class TestSourceExtract:
+
+    def test_extracts_from_router(self):
+        router = Path.home() / "redhat" / "llm-d-router"
+        if not router.exists():
+            import pytest
+            pytest.skip("router repo not cloned")
+        import subprocess
+        tags = [t.strip() for t in subprocess.run(
+            ["git", "tag", "--list", "v*.*.*", "--sort=v:refname"],
+            capture_output=True, text=True, cwd=str(router),
+        ).stdout.splitlines() if t.strip() and "rc" not in t]
+        if not tags:
+            import pytest
+            pytest.skip("no tags")
+        assert len(extract_plugins_from_source(router, tags[-1])) > 0
+
+    def test_older_has_fewer(self):
+        router = Path.home() / "redhat" / "llm-d-router"
+        if not router.exists():
+            import pytest
+            pytest.skip("router repo not cloned")
+        import subprocess
+        tags = [t.strip() for t in subprocess.run(
+            ["git", "tag", "--list", "v*.*.*", "--sort=v:refname"],
+            capture_output=True, text=True, cwd=str(router),
+        ).stdout.splitlines() if t.strip() and "rc" not in t]
+        if len(tags) < 2:
+            import pytest
+            pytest.skip("need 2+ tags")
+        assert len(extract_plugins_from_source(router, tags[0])) < len(
+            extract_plugins_from_source(router, tags[-1]))

From 36c1888fdd8f986b5861712bc334392925a31736 Mon Sep 17 00:00:00 2001
From: Madhu Goutham Reddy Ambati <mambati@redhat.com>
Date: Tue, 11 Aug 2026 13:29:52 -0400
Subject: [PATCH 2/3] fix(preflight): address review findings with fail-closed
 Layer-2

Null-safe YAML walks, wire --router-repo, latest-tag only when cluster
tag is absent locally, and skip image checks only in discover mode.

Signed-off-by: Madhu Goutham Reddy Ambati <mambati@redhat.com>
---
 src/conformance/cli.py       |   9 +++
 src/conformance/deployer.py  |   2 +
 src/conformance/preflight.py | 104 ++++++++++++++++++++++++-----------
 tests/conftest.py            |   6 ++
 tests/test_conformance.py    |  12 +++-
 tests/test_preflight.py      |  76 +++++++++++++++++++++++++
 6 files changed, 173 insertions(+), 36 deletions(-)

diff --git a/src/conformance/cli.py b/src/conformance/cli.py
index 21ee1e8..3f40756 100644
--- a/src/conformance/cli.py
+++ b/src/conformance/cli.py
@@ -78,6 +78,14 @@ def main():
     parser.add_argument("--decode-node-selector", default="", help="Node selector for decode pods (key=value)")
     parser.add_argument("--prefill-node-selector", default="", help="Node selector for prefill pods (key=value)")
 
+    # Preflight
+    parser.add_argument(
+        "--router-repo",
+        default="",
+        metavar="PATH",
+        help="Local llm-d-router checkout for preflight Layer-2 plugin source extract",
+    )
+
     # Behavior
     parser.add_argument("--nocleanup", action="store_true", help="Keep resources after test")
     parser.add_argument(
@@ -157,6 +165,7 @@ def main():
         "guidellm_image": "--guidellm-image",
         "decode_node_selector": "--decode-node-selector",
         "prefill_node_selector": "--prefill-node-selector",
+        "router_repo": "--router-repo",
     }
 
     for attr, flag in flag_map.items():
diff --git a/src/conformance/deployer.py b/src/conformance/deployer.py
index 5100fcd..a4eb12f 100644
--- a/src/conformance/deployer.py
+++ b/src/conformance/deployer.py
@@ -81,6 +81,7 @@ def __init__(
         manifest_dir: str = "deploy/manifests",
         decode_node_selector: str = "",
         prefill_node_selector: str = "",
+        router_repo: str = "",
     ):
         self.kubeconfig = kubeconfig
         self.platform = platform
@@ -93,6 +94,7 @@ def __init__(
         self.manifest_dir = Path(manifest_dir)
         self.decode_node_selector = _parse_node_selector(decode_node_selector)
         self.prefill_node_selector = _parse_node_selector(prefill_node_selector)
+        self.router_repo = router_repo
         self._port_forward_proc: subprocess.Popen | None = None
         self._port_forward_port: int = 0
         self._pod_pf_proc: subprocess.Popen | None = None
diff --git a/src/conformance/preflight.py b/src/conformance/preflight.py
index 01d6acd..573e89e 100644
--- a/src/conformance/preflight.py
+++ b/src/conformance/preflight.py
@@ -41,13 +41,22 @@ def extract_required_plugins(manifest_path: Path) -> set[str]:
     manifest = docs[0] if docs else {}
 
     plugins: set[str] = set()
-    inline = manifest.get("spec", {}).get("router", {}).get("scheduler", {}).get("config", {}).get("inline", {})
+    # Explicit YAML nulls (e.g. scheduler: null) must not crash — treat like missing.
+    spec = manifest.get("spec") or {}
+    router = spec.get("router") or {}
+    scheduler = router.get("scheduler") or {}
+    config = scheduler.get("config") or {}
+    inline = config.get("inline") or {}
     for plugin in inline.get("plugins") or []:
+        if not isinstance(plugin, dict):
+            continue
         ptype = plugin.get("type", "")
         if ptype and not plugin.get("optional", False):
             plugins.add(ptype)
 
-    sat_ref = inline.get("flowControl", {}).get("saturationDetector", {}).get("pluginRef", "")
+    flow = inline.get("flowControl") or {}
+    sat = flow.get("saturationDetector") or {}
+    sat_ref = sat.get("pluginRef") or ""
     if sat_ref:
         plugins.add(sat_ref)
 
@@ -146,6 +155,17 @@ def _detect_epp_version_tag(kubectl_fn) -> str:
     return ""
 
 
+def _list_release_tags(router_path: Path) -> list[str]:
+    """Stable release tags in the router checkout (no rc/alpha), sorted ascending."""
+    result = subprocess.run(
+        ["git", "tag", "--list", "v*.*.*", "--sort=v:refname"],
+        capture_output=True,
+        text=True,
+        cwd=str(router_path),
+    )
+    return [t.strip() for t in result.stdout.splitlines() if t.strip() and "rc" not in t and "alpha" not in t]
+
+
 def resolve_available_plugins(
     kubectl_fn,
     namespace: str,
@@ -156,39 +176,57 @@ def resolve_available_plugins(
     if live:
         return live, f"live probe ({len(live)} plugins)"
 
-    if router_repo:
-        router_path = Path(router_repo)
-        if router_path.exists():
-            try:
-                epp_tag = _detect_epp_version_tag(kubectl_fn)
+    if not router_repo:
+        return None, "no detection method available"
+
+    router_path = Path(router_repo)
+    if not router_path.exists():
+        return None, "no detection method available"
+
+    try:
+        epp_tag = _detect_epp_version_tag(kubectl_fn)
+        tags = _list_release_tags(router_path)
+
+        if epp_tag:
+            plugins = extract_plugins_from_source(router_path, epp_tag)
+            if plugins:
+                return plugins, f"source extract at {epp_tag} ({len(plugins)} plugins)"
+            # Tag present in local repo but extract empty → fail closed (skip).
+            # Using "latest" here would false-pass: newer source ≠ cluster EPP.
+            if epp_tag in tags:
+                log.info(
+                    "Tag %s exists locally but yielded no plugins — skipping Layer-2 "
+                    "(not falling back to latest; that could false-pass)",
+                    epp_tag,
+                )
+                return None, "no detection method available"
+            log.info(
+                "Cluster tag %s not in local router tags — falling back to latest",
+                epp_tag,
+            )
+
+        # Latest-tag fallback: no cluster tag, OR cluster tag absent from local clone.
+        if tags:
+            latest = tags[-1]
+            plugins = extract_plugins_from_source(router_path, latest)
+            if plugins:
                 if epp_tag:
-                    plugins = extract_plugins_from_source(router_path, epp_tag)
-                    if plugins:
-                        return plugins, f"source extract at {epp_tag} ({len(plugins)} plugins)"
-                    log.info("Tag %s found but no plugins extracted — tag may not exist in router repo", epp_tag)
-
-                if not epp_tag:
-                    result = subprocess.run(
-                        ["git", "tag", "--list", "v*.*.*", "--sort=v:refname"],
-                        capture_output=True,
-                        text=True,
-                        cwd=str(router_path),
+                    log.warning(
+                        "Using latest tag %s — cluster tag %s missing from local router repo",
+                        latest,
+                        epp_tag,
                     )
-                    tags = [
-                        t.strip()
-                        for t in result.stdout.splitlines()
-                        if t.strip() and "rc" not in t and "alpha" not in t
-                    ]
-                    if tags:
-                        plugins = extract_plugins_from_source(router_path, tags[-1])
-                        if plugins:
-                            log.warning(
-                                "Using latest tag %s — could not determine cluster EPP version",
-                                tags[-1],
-                            )
-                            return plugins, f"source extract at {tags[-1]} (latest — cluster version unknown)"
-            except Exception:
-                pass
+                    return (
+                        plugins,
+                        f"source extract at {latest} (latest — cluster tag {epp_tag} absent locally)",
+                    )
+                log.warning(
+                    "Using latest tag %s — could not determine cluster EPP version",
+                    latest,
+                )
+                return plugins, f"source extract at {latest} (latest — cluster version unknown)"
+    except Exception:
+        pass
 
     return None, "no detection method available"
 
diff --git a/tests/conftest.py b/tests/conftest.py
index 68f241d..ad79359 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -81,6 +81,11 @@ def pytest_addoption(parser):
     parser.addoption("--guidellm-image", default="", help="GuideLLM benchmark image override")
     parser.addoption("--decode-node-selector", default="", help="Node selector for decode pods (key=value)")
     parser.addoption("--prefill-node-selector", default="", help="Node selector for prefill pods (key=value)")
+    parser.addoption(
+        "--router-repo",
+        default="",
+        help="Local llm-d-router checkout for preflight Layer-2 plugin source extract",
+    )
 
 
 def _resolve_test_cases(config) -> list[TestCase]:
@@ -119,6 +124,7 @@ def deployer(request) -> Deployer:
         disable_auth=request.config.getoption("--disable-auth"),
         decode_node_selector=request.config.getoption("--decode-node-selector"),
         prefill_node_selector=request.config.getoption("--prefill-node-selector"),
+        router_repo=request.config.getoption("--router-repo"),
     )
     yield d
     d.stop_port_forward()
diff --git a/tests/test_conformance.py b/tests/test_conformance.py
index 6e8becd..2b127b7 100644
--- a/tests/test_conformance.py
+++ b/tests/test_conformance.py
@@ -97,19 +97,25 @@ def _check_threshold(name: str, value: float, min_value: float | None = None, ma
 class TestConformance:
     """Ordered conformance phases for each test case."""
 
-    def test_00_preflight(self, deployer: Deployer, tc: TestCase):
+    def test_00_preflight(self, deployer: Deployer, tc: TestCase, test_mode: str):
         """Pre-flight: verify manifest is compatible with the cluster.
 
         Three checks: API version, image existence, plugin compatibility.
         Each skips safely if detection is unavailable.
+
+        In discover mode, skip image existence (apply/pull oriented) but keep
+        API version and plugin checks — live EPP probe is most useful there.
         """
         _require_manifest(tc)
         manifest_path = _MANIFEST_DIR / tc.deployment.manifest_path
 
         checks = [
             ("api_version", lambda: check_api_version(manifest_path, deployer.kubectl)),
-            ("image_existence", lambda: check_image_existence(deployer.kubectl)),
         ]
+        if test_mode != "discover":
+            checks.append(("image_existence", lambda: check_image_existence(deployer.kubectl)))
+        else:
+            _log("Pre-flight image_existence: skipped (discover mode)")
 
         for name, check_fn in checks:
             result = check_fn()
@@ -124,7 +130,7 @@ def test_00_preflight(self, deployer: Deployer, tc: TestCase):
         available, source = resolve_available_plugins(
             deployer.kubectl,
             deployer.namespace,
-            router_repo=getattr(deployer, "router_repo", None),
+            router_repo=deployer.router_repo or None,
         )
         result = check_manifest_compatibility(manifest_path, available, source)
         if result.skipped_reason:
diff --git a/tests/test_preflight.py b/tests/test_preflight.py
index be2b5b5..bf8a715 100644
--- a/tests/test_preflight.py
+++ b/tests/test_preflight.py
@@ -12,6 +12,7 @@
     check_manifest_compatibility,
     extract_plugins_from_source,
     extract_required_plugins,
+    resolve_available_plugins,
 )
 
 
@@ -58,6 +59,18 @@ def test_plugins_null_returns_empty(self, tmp_path):
             )
         assert extract_required_plugins(path) == set()
 
+    def test_scheduler_null_does_not_raise(self, tmp_path):
+        path = tmp_path / "sched-null.yaml"
+        with open(path, "w") as f:
+            yaml.dump(
+                {
+                    "apiVersion": "v1",
+                    "spec": {"router": {"scheduler": None}},
+                },
+                f,
+            )
+        assert extract_required_plugins(path) == set()
+
     def test_saturation_detector_ref(self, tmp_path):
         path = _write_manifest(
             tmp_path,
@@ -94,6 +107,69 @@ def test_none_available_skips(self, tmp_path):
         assert "could not detect" in r.skipped_reason
 
 
+class TestResolveAvailablePlugins:
+    def test_falls_back_to_latest_when_cluster_tag_absent_locally(self, tmp_path, monkeypatch):
+        """Cluster tag not in local repo tags → use latest (clone is incomplete)."""
+        router = tmp_path / "router"
+        router.mkdir()
+
+        def kubectl(*_a, **_k):
+            return ""
+
+        monkeypatch.setattr(
+            "conformance.preflight.probe_epp_plugins",
+            lambda *_a, **_k: None,
+        )
+        monkeypatch.setattr(
+            "conformance.preflight._detect_epp_version_tag",
+            lambda *_a, **_k: "v9.9.9",
+        )
+
+        def fake_extract(_repo, tag: str):
+            if tag == "v9.9.9":
+                return frozenset()
+            if tag == "v1.2.3":
+                return frozenset({"plugin-a"})
+            return frozenset()
+
+        monkeypatch.setattr("conformance.preflight.extract_plugins_from_source", fake_extract)
+        monkeypatch.setattr(
+            "conformance.preflight._list_release_tags",
+            lambda _repo: ["v1.0.0", "v1.2.3"],
+        )
+
+        plugins, source = resolve_available_plugins(kubectl, "ns", router_repo=router)
+        assert plugins == frozenset({"plugin-a"})
+        assert "v1.2.3" in source
+        assert "v9.9.9" in source
+
+    def test_skips_when_exact_tag_exists_but_extract_empty(self, tmp_path, monkeypatch):
+        """Tag exists locally but extract empty → fail closed (do not use latest)."""
+        router = tmp_path / "router"
+        router.mkdir()
+
+        monkeypatch.setattr(
+            "conformance.preflight.probe_epp_plugins",
+            lambda *_a, **_k: None,
+        )
+        monkeypatch.setattr(
+            "conformance.preflight._detect_epp_version_tag",
+            lambda *_a, **_k: "v1.2.3",
+        )
+        monkeypatch.setattr(
+            "conformance.preflight.extract_plugins_from_source",
+            lambda _repo, _tag: frozenset(),
+        )
+        monkeypatch.setattr(
+            "conformance.preflight._list_release_tags",
+            lambda _repo: ["v1.0.0", "v1.2.3"],
+        )
+
+        plugins, source = resolve_available_plugins(lambda *_a, **_k: "", "ns", router_repo=router)
+        assert plugins is None
+        assert source == "no detection method available"
+
+
 class TestApiVersionCheck:
     def _make_manifest(self, tmp_path, api_version):
         path = tmp_path / "test.yaml"

From 45f5f9fe723d2867cfb2f83097b309224eda1d69 Mon Sep 17 00:00:00 2001
From: Madhu Goutham Reddy Ambati <mambati@redhat.com>
Date: Tue, 11 Aug 2026 15:15:40 -0400
Subject: [PATCH 3/3] fix(preflight): address all review findings

Null-safe YAML, wire --router-repo, fail-closed Layer-2 fallthrough,
narrow discover image skip, tighten PluginType/log extractors, soften
brittle tag-count test.

Signed-off-by: Madhu Goutham Reddy Ambati <mambati@redhat.com>
---
 src/conformance/preflight.py | 32 +++++++++++++------
 tests/test_preflight.py      | 59 ++++++++++++++++++++++++++++++++++--
 2 files changed, 78 insertions(+), 13 deletions(-)

diff --git a/src/conformance/preflight.py b/src/conformance/preflight.py
index 573e89e..f0e620f 100644
--- a/src/conformance/preflight.py
+++ b/src/conformance/preflight.py
@@ -88,9 +88,14 @@ def probe_epp_plugins(kubectl_fn, namespace: str) -> frozenset[str] | None:
             return None
 
         plugins = set()
+        # Only the name after "registered plugin" / "plugin type" — not every
+        # quoted token on the line (handlers, modes, etc.).
+        name_re = re.compile(
+            r'(?:registered plugin|plugin type)\s+"([a-z][a-z0-9-]+)"',
+            re.IGNORECASE,
+        )
         for line in logs.splitlines():
-            if "registered plugin" in line.lower() or "plugin type" in line.lower():
-                plugins.update(re.findall(r'"([a-z][a-z0-9-]+)"', line))
+            plugins.update(name_re.findall(line))
 
         if plugins:
             return frozenset(plugins)
@@ -99,23 +104,30 @@ def probe_epp_plugins(kubectl_fn, namespace: str) -> frozenset[str] | None:
     return None
 
 
+_PLUGIN_TYPE_ASSIGN = re.compile(r'(?:Type\s*\(PluginType\)|PluginType|Type)\s*=\s*"([a-z][a-z0-9-]+)"')
+
+
 def extract_plugins_from_source(router_repo: str | Path, tag: str) -> frozenset[str]:
     """Layer 2: extract plugin type strings from router Go source at a git tag."""
-    result = subprocess.run(
-        ["git", "grep", "-h", r'Type\s*\(PluginType\)\?=\s*"', tag, "--", "*.go"],
-        capture_output=True,
-        text=True,
-        cwd=str(router_repo),
+    # Prefer PluginType assignments; avoid bare Type = "..." which matches
+    # unrelated Go fields (content-type, json, grpc, etc.).
+    patterns = (
+        r'Type\s*\(PluginType\)\s*=\s*"',
+        r'PluginType\s*=\s*"',
     )
-    if not result.stdout.strip():
+    stdout_parts: list[str] = []
+    for pattern in patterns:
         result = subprocess.run(
-            ["git", "grep", "-h", r'Type\s*=\s*"', tag, "--", "*.go"],
+            ["git", "grep", "-h", pattern, tag, "--", "*.go"],
             capture_output=True,
             text=True,
             cwd=str(router_repo),
         )
+        if result.stdout.strip():
+            stdout_parts.append(result.stdout)
 
-    plugins = set(re.findall(r'"([a-z][a-z0-9-]+)"', result.stdout))
+    combined = "\n".join(stdout_parts)
+    plugins = set(_PLUGIN_TYPE_ASSIGN.findall(combined))
     non_plugins = {
         "content-type",
         "custom",
diff --git a/tests/test_preflight.py b/tests/test_preflight.py
index bf8a715..8da157a 100644
--- a/tests/test_preflight.py
+++ b/tests/test_preflight.py
@@ -12,6 +12,7 @@
     check_manifest_compatibility,
     extract_plugins_from_source,
     extract_required_plugins,
+    probe_epp_plugins,
     resolve_available_plugins,
 )
 
@@ -315,6 +316,46 @@ def kubectl(*a, **k):
         assert r.compatible is True
 
 
+class TestProbeEppPlugins:
+    def test_ignores_handler_names_on_registered_line(self):
+        logs = (
+            'Registered plugin "flow-control-dispatcher" using handler "round-robin"\n'
+            'plugin type "prefix-cache-scorer" ready\n'
+        )
+
+        def kubectl(*args, **_k):
+            if args and args[0] == "get":
+                return "epp-0"
+            if args and args[0] == "logs":
+                return logs
+            return ""
+
+        plugins = probe_epp_plugins(kubectl, "ns")
+        assert plugins == frozenset({"flow-control-dispatcher", "prefix-cache-scorer"})
+        assert "round-robin" not in plugins
+
+
+class TestSourceExtractParsing:
+    def test_parse_ignores_unrelated_type_assignments(self, tmp_path, monkeypatch):
+        """Bare Type = \"json\" style lines must not become available plugins."""
+        router = tmp_path / "router"
+        router.mkdir()
+
+        import subprocess as sp
+
+        def fake_run(cmd, **kwargs):
+            # Only PluginType greps return plugin lines; no broad Type= dump.
+            pattern = cmd[3] if len(cmd) > 3 else ""
+            if "PluginType" in pattern:
+                stdout = 'const PluginType = "prefix-cache-scorer"\nPluginType = "token-producer"\n'
+                return sp.CompletedProcess(args=cmd, returncode=0, stdout=stdout, stderr="")
+            return sp.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="")
+
+        monkeypatch.setattr("conformance.preflight.subprocess.run", fake_run)
+        plugins = extract_plugins_from_source(router, "v1.0.0")
+        assert plugins == frozenset({"prefix-cache-scorer", "token-producer"})
+
+
 class TestSourceExtract:
     def test_extracts_from_router(self):
         router = Path.home() / "redhat" / "llm-d-router"
@@ -338,9 +379,17 @@ def test_extracts_from_router(self):
             import pytest
 
             pytest.skip("no tags")
-        assert len(extract_plugins_from_source(router, tags[-1])) > 0
+        plugins = extract_plugins_from_source(router, tags[-1])
+        assert len(plugins) > 0
+        # Broad Type= junk from CRD/condition enums must not leak in.
+        assert not {"exact", "accepted", "console"} & plugins
+
+    def test_two_tags_both_extractable(self):
+        """Extract on two tags without asserting older_count < newer_count.
 
-    def test_older_has_fewer(self):
+        Plugin removals/renames are legitimate; only require newest to work and
+        that an older tag either yields plugins or is skippable (pre-PluginType era).
+        """
         router = Path.home() / "redhat" / "llm-d-router"
         if not router.exists():
             import pytest
@@ -362,4 +411,8 @@ def test_older_has_fewer(self):
             import pytest
 
             pytest.skip("need 2+ tags")
-        assert len(extract_plugins_from_source(router, tags[0])) < len(extract_plugins_from_source(router, tags[-1]))
+        newest = extract_plugins_from_source(router, tags[-1])
+        assert len(newest) > 0
+        oldest = extract_plugins_from_source(router, tags[0])
+        # Older tags may predate PluginType constants — empty is OK, crash is not.
+        assert isinstance(oldest, frozenset)