fix(exaforce): let LLM verdict override static severity floor

docs/superpowers/EXPECTED_TEST_FAILURES.md

@@ -1,4 +1,4 @@
-# Expected test failures (fork: exaforce schema pruning)
+# Expected test failures (fork: exaforce runtime patches)
 
 These upstream tests are kept at upstream parity on purpose and therefore
 assert the *un-pruned* schema, which the exaforce runtime patch removes. They
@@ -15,8 +15,26 @@ Captured from:
 
 All four fail with an `AssertionError` (or `KeyError`) about a pruned key
 (`explanation`, `intent`) being absent — not an import/collection error.
+
+## Severity floor (added 2026-09-02, `exaforce/_filter_patches.py`)
+
+Upstream asserts that CRITICAL/HIGH *static* findings survive LLM filtering,
+tagged `llm-unconfirmed`. The fork keeps that floor only for LLM-backed
+findings (`SQP-*`, `SDI-*`, `SSD-*`, `TP4`) and lets the meta-analyzer overrule
+static rules, so these fail under the default
+`SKILLSPECTOR_META_SEVERITY_FLOOR=semantic` (and pass with `=upstream`):
+
+- tests/nodes/test_llm_analyzer_base.py::TestApplyFilterSeverityFloor::test_critical_unconfirmed_kept_with_llm_unconfirmed_tag
+- tests/nodes/test_llm_analyzer_base.py::TestApplyFilterSeverityFloor::test_high_unconfirmed_kept_with_llm_unconfirmed_tag
+- tests/nodes/test_llm_analyzer_base.py::TestApplyFilterSeverityFloor::test_llm_unconfirmed_tag_not_duplicated
+- tests/nodes/test_llm_analyzer_base.py::TestApplyFilterSeverityFloor::test_llm_unconfirmed_tag_surfaced_in_to_dict
+
+Each fails on `assert len(result) == 1` / a missing `llm-unconfirmed` tag —
+not an import/collection error. Note `.github/workflows/ci.yml` runs the full
+suite via `make test-ci`, so fork CI is red by design (8 failures). Do not deselect, xfail, or
+edit these upstream tests — that creates conflicts on every upstream sync.
 Confirmed bounded to these two files via `uv run pytest -q -rf`:
 
 ```
-4 failed, 1261 passed, 13 skipped, 34 deselected, 6 xfailed
+8 failed, 1917 passed, 13 skipped, 38 deselected, 4 xfailed   (2026-09-02)
 ```

src/skillspector/exaforce/__init__.py

@@ -2,14 +2,15 @@
 """ExaForce fork-local runtime patches.
 
 Keeps fork behavior — pruning unused LLM structured-output keys and prompt text
-to shrink requests and reduce LLM timeouts — out of upstream-tracked source
-files. All mutations are guarded: an upstream rename/rewrite raises
-``PatchDriftError`` at import time rather than silently going stale.
+to shrink requests and reduce LLM timeouts, and trusting the meta-analyzer LLM
+verdict over static severity — out of upstream-tracked source files. All
+mutations are guarded: an upstream rename/rewrite raises ``PatchDriftError``
+at import time rather than silently going stale.
 """
 
 from __future__ import annotations
 
-from . import _prompt_patches, _sampling_patches, _schema_patches
+from . import _filter_patches, _prompt_patches, _sampling_patches, _schema_patches
 
 _PATCHED = False
 
@@ -22,4 +23,5 @@ def apply_patches() -> None:
     _schema_patches.apply()
     _prompt_patches.apply()
     _sampling_patches.apply()
+    _filter_patches.apply()
     _PATCHED = True

src/skillspector/exaforce/_filter_patches.py

@@ -0,0 +1,170 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Control how far the meta-analyzer LLM verdict overrides a finding's severity (fork behavior).
+
+Upstream ``LLMMetaAnalyzer.apply_filter`` keeps CRITICAL/HIGH findings even when
+the meta-analyzer LLM denies or omits them, tagging them ``llm-unconfirmed``.
+That floor is a prompt-injection defense: a skill's content could talk the LLM
+into dropping a real finding. It also applies to *every* upstream finding,
+including the LLM-backed analyzers (semantic ``SQP-*``/``SDI-*``/``SSD-*`` and
+the ``TP4`` description-behavior check), so in practice it shields first-pass
+LLM findings from a second LLM's re-verification as much as it shields static
+regex hits.
+
+``SKILLSPECTOR_META_SEVERITY_FLOOR`` selects the policy, read on every
+``apply_filter`` call so it can be flipped at runtime:
+
+``none``
+    Empty the floor. Every finding, any severity, is dropped unless the
+    meta-analyzer confirms it. Fewest false positives, clearly worse recall.
+``semantic``  (default)
+    Keep the upstream floor for LLM-backed findings only; static findings of
+    any severity follow the meta-analyzer verdict. This is the literal "trust
+    the LLM over static": the meta-analyzer may overrule a regex, but one LLM
+    does not silently overrule another.
+``upstream``
+    Leave the floor untouched.
+
+Batches that raise or never return are unaffected in every mode: upstream
+routes those findings through its no-verdict fallback before ``apply_filter``
+sees them. A batch that *returns* an empty verdict list is treated by upstream
+as a successful "nothing confirmed" response, and under ``none``/``semantic``
+that now drops the batch's static findings where upstream kept CRITICAL/HIGH;
+the wrapper logs a warning when that happens so it is observable.
+
+Measured 2026-09-02 on nvidia.nemotron-super-3-120b, same-day, two replicates
+each, re-scanning the 94 borderline units (87 malicious / 7 benign) that a
+900-unit ``none`` run had got wrong (188 unit-scans per mode):
+
+    mode       TP   FP   correct
+    upstream   68   10   72
+    semantic   65    7   72      <- default: upstream accuracy, 30 % fewer FPs
+    none       54    3   65
+
+The ``none`` losses are not "correct": spot-checked drops included obfuscated
+PowerShell download-and-execute and Fernet-decrypted ``exec()`` in ``setup.py``
+that the semantic analyzers had flagged CRITICAL at confidence >= 0.9 and the
+meta-analyzer then rejected.
+"""
+
+from __future__ import annotations
+
+import functools
+import inspect
+import os
+from typing import Any
+
+import skillspector.nodes.meta_analyzer as meta
+from skillspector.logging_config import get_logger
+from skillspector.models import Finding
+
+from ._patchlib import PatchDriftError
+
+logger = get_logger(__name__)
+
+ENV_VAR = "SKILLSPECTOR_META_SEVERITY_FLOOR"
+MODES = ("none", "semantic", "upstream")
+DEFAULT_MODE = "semantic"
+
+_UPSTREAM_FLOOR = frozenset({"CRITICAL", "HIGH"})
+
+# ``Finding`` carries no source-analyzer field, so the rule id is the only
+# stable discriminator for LLM-backed findings. Prefixes cover the three
+# semantic analyzers; ``TP4`` is emitted by mcp_tool_poisoning from a
+# ``chat_completion`` reply. Matching is case-insensitive on the stripped id
+# because the semantic analyzers' rule ids are free-form LLM output and the
+# benchmark corpus shows rare variants such as ``ssd-2`` or ``SQP-2 L160``.
+LLM_RULE_PREFIXES = ("SQP-", "SDI-", "SSD-")
+LLM_RULE_IDS = frozenset({"TP4"})
+
+
+def resolve_mode() -> str:
+    raw = os.environ.get(ENV_VAR, "").strip().lower()
+    if not raw:
+        return DEFAULT_MODE
+    if raw not in MODES:
+        logger.warning("%s=%r is not one of %s — using %r.", ENV_VAR, raw, MODES, DEFAULT_MODE)
+        return DEFAULT_MODE
+    return raw
+
+
+def is_llm_finding(finding: Finding) -> bool:
+    rule_id = (finding.rule_id or "").strip().upper()
+    return rule_id in LLM_RULE_IDS or rule_id.startswith(LLM_RULE_PREFIXES)
+
+
+def _warn_on_empty_verdicts(batch_results: Any) -> None:
+    for batch, llm_items in batch_results:
+        if batch.findings and not llm_items:
+            logger.warning(
+                "Meta-analyzer returned no verdicts for %s (%d findings); under "
+                "%s=%s its unconfirmed static findings will be dropped.",
+                batch.file_path,
+                len(batch.findings),
+                ENV_VAR,
+                resolve_mode(),
+            )
+
+
+def _mode_dispatching_apply_filter(original: Any) -> Any:
+    """Wrap upstream ``apply_filter`` to apply the env-selected floor policy per call.
+
+    Upstream reads the floor via ``self._HIGH_SEVERITY_FLOOR``; an instance
+    attribute shadows the class-level frozenset for the duration of one call
+    and is removed in ``finally`` so a raise inside upstream code cannot leave
+    the analyzer mis-configured.
+    """
+
+    @functools.wraps(original)
+    def apply_filter(self: Any, findings: list[Finding], batch_results: Any) -> list[Finding]:
+        mode = resolve_mode()
+        if mode == "upstream":
+            return list(original(self, findings, batch_results))
+        _warn_on_empty_verdicts(batch_results)
+        if mode == "none":
+            floored: list[Finding] = []
+            unfloored = list(findings)
+        else:  # semantic
+            floored = [f for f in findings if is_llm_finding(f)]
+            unfloored = [f for f in findings if not is_llm_finding(f)]
+        kept: list[Finding] = []
+        try:
+            if floored:
+                self._HIGH_SEVERITY_FLOOR = _UPSTREAM_FLOOR
+                kept.extend(original(self, floored, batch_results))
+            self._HIGH_SEVERITY_FLOOR = frozenset()
+            kept.extend(original(self, unfloored, batch_results))
+        finally:
+            self.__dict__.pop("_HIGH_SEVERITY_FLOOR", None)
+        # Upstream forwards ``finding_id`` unchanged, so restore the caller's
+        # ordering by it — keeps the contract identical to upstream's single pass.
+        order = {f.finding_id: i for i, f in enumerate(findings)}
+        kept.sort(key=lambda f: order.get(f.finding_id, len(order)))
+        return kept
+
+    apply_filter._exaforce_wrapped = True  # type: ignore[attr-defined]
+    return apply_filter
+
+
+def apply() -> None:
+    cls = meta.LLMMetaAnalyzer
+    qual = f"{cls.__module__}.{cls.__qualname__}"
+    if cls.__dict__.get("_HIGH_SEVERITY_FLOOR") != _UPSTREAM_FLOOR:
+        raise PatchDriftError(
+            f"{qual}._HIGH_SEVERITY_FLOOR is not {sorted(_UPSTREAM_FLOOR)}; "
+            "upstream changed — update the exaforce patch."
+        )
+    current = cls.__dict__.get("apply_filter")
+    if current is None:
+        raise PatchDriftError(
+            f"{qual}.apply_filter is missing; upstream changed — update the exaforce patch."
+        )
+    if getattr(current, "_exaforce_wrapped", False):
+        return  # already applied
+    # The per-call shadowing above only works if upstream reads the floor
+    # through the instance. Fail at import time if that access path changes.
+    if "self._HIGH_SEVERITY_FLOOR" not in inspect.getsource(current):
+        raise PatchDriftError(
+            f"{qual}.apply_filter no longer reads self._HIGH_SEVERITY_FLOOR; "
+            "upstream changed — update the exaforce patch."
+        )
+    setattr(cls, "apply_filter", _mode_dispatching_apply_filter(current))  # noqa: B010

tests/exaforce/test_patches.py

@@ -78,3 +78,115 @@ def test_apply_patches_is_idempotent(run_in_subprocess):
         """
     )
     assert "OK" in out
+
+
+_FLOOR_HARNESS = """
+    import os
+    os.environ["SKILLSPECTOR_META_SEVERITY_FLOOR"] = {mode!r}
+    from unittest.mock import MagicMock, patch
+    import skillspector
+    from skillspector.exaforce import apply_patches
+    apply_patches()
+    from skillspector.llm_analyzer_base import Batch
+    from skillspector.models import Finding
+    from skillspector.nodes.meta_analyzer import LLMMetaAnalyzer
+
+    def _finding(rule_id, severity, line):
+        return Finding(
+            rule_id=rule_id, message="msg", severity=severity,
+            confidence=0.8, file="skill.md", start_line=line,
+        )
+
+    with patch("skillspector.llm_analyzer_base.get_chat_model", lambda **kw: MagicMock()):
+        analyzer = LLMMetaAnalyzer(model="test-model")
+
+    # Static rules (regex/YARA style ids) and semantic rules (SQP/SDI/SSD), each
+    # with a CRITICAL the LLM explicitly denies, a HIGH the LLM omits, and a
+    # MEDIUM the LLM confirms.
+    findings = [
+        _finding("E2", "CRITICAL", 10), _finding("PE3", "HIGH", 11), _finding("P1", "MEDIUM", 12),
+        _finding("SDI-1", "CRITICAL", 20), _finding("SSD-1", "HIGH", 21), _finding("SQP-2", "MEDIUM", 22),
+        _finding("TP4", "HIGH", 30),  # LLM-backed but not prefix-named; LLM omits it
+    ]
+    batch = Batch(file_path="skill.md", content="code", findings=findings)
+    llm_items = [
+        {{"pattern_id": "E2", "start_line": 10, "is_vulnerability": False, "confidence": 0.2, "_file": "skill.md"}},
+        {{"pattern_id": "P1", "start_line": 12, "is_vulnerability": True, "confidence": 0.9, "_file": "skill.md"}},
+        {{"pattern_id": "SDI-1", "start_line": 20, "is_vulnerability": False, "confidence": 0.2, "_file": "skill.md"}},
+        {{"pattern_id": "SQP-2", "start_line": 22, "is_vulnerability": True, "confidence": 0.9, "_file": "skill.md"}},
+    ]
+    result = analyzer.apply_filter(findings, [(batch, llm_items)])
+    kept = [f.rule_id for f in result]
+    unconfirmed = sorted(f.rule_id for f in result if "llm-unconfirmed" in f.tags)
+    assert kept == {expected_kept!r}, kept
+    assert unconfirmed == {expected_unconfirmed!r}, unconfirmed
+    print("OK")
+"""
+
+
+def test_floor_mode_none_drops_every_unconfirmed_finding(run_in_subprocess):
+    out = run_in_subprocess(
+        _FLOOR_HARNESS.format(
+            mode="none",
+            expected_kept=["P1", "SQP-2"],
+            expected_unconfirmed=[],
+        )
+    )
+    assert "OK" in out
+
+
+def test_floor_mode_semantic_keeps_only_semantic_high_severity(run_in_subprocess):
+    """Static CRITICAL/HIGH follow the LLM verdict; semantic CRITICAL/HIGH keep the
+    upstream floor. Output order matches input order despite the two-pass filter."""
+    out = run_in_subprocess(
+        _FLOOR_HARNESS.format(
+            mode="semantic",
+            expected_kept=["P1", "SDI-1", "SSD-1", "SQP-2", "TP4"],
+            expected_unconfirmed=["SDI-1", "SSD-1", "TP4"],
+        )
+    )
+    assert "OK" in out
+
+
+def test_floor_mode_upstream_is_untouched(run_in_subprocess):
+    out = run_in_subprocess(
+        _FLOOR_HARNESS.format(
+            mode="upstream",
+            expected_kept=["E2", "PE3", "P1", "SDI-1", "SSD-1", "SQP-2", "TP4"],
+            expected_unconfirmed=["E2", "PE3", "SDI-1", "SSD-1", "TP4"],
+        )
+    )
+    assert "OK" in out
+
+
+def test_floor_mode_default_is_semantic_and_switchable_at_runtime(run_in_subprocess):
+    """Mode is read per apply_filter call, so an in-process A/B can flip it after import."""
+    out = run_in_subprocess(
+        """
+        import os
+        os.environ.pop("SKILLSPECTOR_META_SEVERITY_FLOOR", None)
+        from unittest.mock import MagicMock, patch
+        import skillspector
+        from skillspector.exaforce import _filter_patches
+        from skillspector.llm_analyzer_base import Batch
+        from skillspector.models import Finding
+        from skillspector.nodes.meta_analyzer import LLMMetaAnalyzer
+        assert _filter_patches.resolve_mode() == "semantic"
+        assert getattr(LLMMetaAnalyzer.apply_filter, "_exaforce_wrapped", False)
+        with patch("skillspector.llm_analyzer_base.get_chat_model", lambda **kw: MagicMock()):
+            analyzer = LLMMetaAnalyzer(model="test-model")
+        f = Finding(rule_id="E2", message="m", severity="CRITICAL", confidence=0.8,
+                    file="skill.md", start_line=1)
+        batch = Batch(file_path="skill.md", content="c", findings=[f])
+        denied = [{"pattern_id": "E2", "start_line": 1, "is_vulnerability": False,
+                   "confidence": 0.1, "_file": "skill.md"}]
+        assert analyzer.apply_filter([f], [(batch, denied)]) == []           # semantic: static dropped
+        os.environ["SKILLSPECTOR_META_SEVERITY_FLOOR"] = "upstream"
+        assert len(analyzer.apply_filter([f], [(batch, denied)])) == 1       # upstream: floor kept
+        os.environ["SKILLSPECTOR_META_SEVERITY_FLOOR"] = "none"
+        assert analyzer.apply_filter([f], [(batch, denied)]) == []
+        assert "_HIGH_SEVERITY_FLOOR" not in analyzer.__dict__               # instance state restored
+        print("OK")
+        """
+    )
+    assert "OK" in out