Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 65 additions & 3 deletions src/skillspector/nodes/meta_analyzer.py
Original file line number Diff line number Diff line change
Expand Up @@ -215,8 +215,70 @@ def _format_findings_for_prompt(findings: list[Finding]) -> str:
return "\n".join(lines)


_NO_LLM_CONFIDENCE_THRESHOLD = 0.4
_HIGH_SEVERITY_PASS_THROUGH = frozenset({"CRITICAL", "HIGH"})
_CODE_EXAMPLE_DOWNWEIGHT = 0.5


def _fallback_filtered(findings: list[Finding]) -> list[Finding]:
"""Apply default remediation to all findings (pass-through with defaults)."""
"""Heuristic fallback filter for --no-llm mode.

Applies rule-based filtering when LLM analysis is unavailable:
1. Drop findings with confidence below threshold (0.4), UNLESS severity
is CRITICAL or HIGH (high-severity findings are never dropped on
confidence alone)
2. Downweight findings whose context matches code-example indicators
(0.5x confidence reduction) — never hard-drop, as there is no LLM
safety net in this mode
3. Apply default remediations from pattern_defaults
"""
from skillspector.nodes.analyzers.common import is_code_example

result: list[Finding] = []
for f in findings:
severity_upper = f.severity.upper()
confidence = f.confidence
if f.context and is_code_example(f.context):
confidence *= _CODE_EXAMPLE_DOWNWEIGHT
if confidence < _NO_LLM_CONFIDENCE_THRESHOLD:
if severity_upper not in _HIGH_SEVERITY_PASS_THROUGH:
continue
result.append(
Finding(
rule_id=f.rule_id,
message=f.message,
severity=f.severity,
confidence=confidence,
file=f.file,
start_line=f.start_line,
end_line=f.end_line,
remediation=f.remediation or get_remediation(f.rule_id),
tags=f.tags,
context=f.context,
matched_text=f.matched_text,
category=getattr(f, "category", None),
pattern=getattr(f, "pattern", None),
finding=getattr(f, "finding", None),
explanation=getattr(f, "explanation", None),
code_snippet=getattr(f, "code_snippet", None) or f.context,
intent=None,
)
)
logger.info(
"Heuristic fallback filter (--no-llm): %d → %d findings",
len(findings),
len(result),
)
return result


def _passthrough_with_defaults(findings: list[Finding]) -> list[Finding]:
"""Pass all findings through with default remediations (fail-closed).

Used on LLM failure path: when the LLM call fails, we pass ALL findings
through unchanged (except adding default remediations). A security tool
should fail-closed — showing more findings is safer than silently dropping.
"""
return [
Finding(
rule_id=f.rule_id,
Expand Down Expand Up @@ -431,5 +493,5 @@ def meta_analyzer(state: SkillspectorState) -> MetaAnalyzerResponse:
except ValueError:
raise
except Exception as e:
logger.warning("LLM call failed, using fallback: %s", e)
return {"filtered_findings": _fallback_filtered(findings)}
logger.warning("LLM call failed, passing all findings through (fail-closed): %s", e)
return {"filtered_findings": _passthrough_with_defaults(findings)}
4 changes: 4 additions & 0 deletions src/skillspector/nodes/report.py
Original file line number Diff line number Diff line change
Expand Up @@ -256,12 +256,16 @@ def _format_terminal(
def _build_metadata(has_executable_scripts: bool, use_llm: bool) -> dict[str, object]:
"""Build the metadata section shared by all output formats."""
llm_available, llm_error = is_llm_available()
meta_analysis_applied = use_llm and llm_available
meta: dict[str, object] = {
"has_executable_scripts": has_executable_scripts,
"skillspector_version": skillspector_version,
"llm_requested": use_llm,
"llm_available": llm_available,
"meta_analysis_applied": meta_analysis_applied,
}
if not meta_analysis_applied:
meta["filtering_mode"] = "heuristic"
if use_llm and not llm_available:
meta["llm_error"] = llm_error
return meta
Expand Down
239 changes: 239 additions & 0 deletions tests/nodes/test_meta_analyzer_fallback.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,239 @@
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for meta_analyzer heuristic fallback filter (--no-llm mode)."""

from __future__ import annotations

from unittest.mock import patch

from skillspector.models import Finding
from skillspector.nodes.meta_analyzer import (
_fallback_filtered,
_passthrough_with_defaults,
meta_analyzer,
)


def _finding(
rule_id: str = "TM1",
confidence: float = 0.8,
severity: str = "HIGH",
context: str | None = "import subprocess\nsubprocess.run(cmd, shell=True)",
matched_text: str = "subprocess.run(cmd, shell=True)",
file: str = "tool.py",
) -> Finding:
return Finding(
rule_id=rule_id,
message=f"Test {rule_id}",
severity=severity,
confidence=confidence,
file=file,
start_line=1,
context=context,
matched_text=matched_text,
)


class TestConfidenceThreshold:
"""Findings below confidence threshold are dropped (unless high severity)."""

def test_low_confidence_low_severity_dropped(self) -> None:
"""LOW severity finding with confidence 0.3 is below threshold and dropped."""
findings = [_finding(confidence=0.3, severity="LOW")]
result = _fallback_filtered(findings)
assert len(result) == 0

def test_low_confidence_medium_severity_dropped(self) -> None:
"""MEDIUM severity finding with confidence 0.3 is dropped."""
findings = [_finding(confidence=0.3, severity="MEDIUM")]
result = _fallback_filtered(findings)
assert len(result) == 0

def test_at_threshold_kept(self) -> None:
"""Finding with confidence exactly 0.4 is kept (>= 0.4)."""
findings = [_finding(confidence=0.4)]
result = _fallback_filtered(findings)
assert len(result) == 1

def test_high_confidence_kept(self) -> None:
"""Finding with high confidence passes through."""
findings = [_finding(confidence=0.9)]
result = _fallback_filtered(findings)
assert len(result) == 1


class TestSeverityFloor:
"""HIGH and CRITICAL findings are never dropped on confidence alone."""

def test_critical_below_threshold_retained(self) -> None:
"""CRITICAL finding at 0.35 confidence is retained (severity floor)."""
findings = [_finding(confidence=0.35, severity="CRITICAL")]
result = _fallback_filtered(findings)
assert len(result) == 1
assert result[0].severity == "CRITICAL"

def test_high_below_threshold_retained(self) -> None:
"""HIGH finding at 0.2 confidence is retained (severity floor)."""
findings = [_finding(confidence=0.2, severity="HIGH")]
result = _fallback_filtered(findings)
assert len(result) == 1
assert result[0].severity == "HIGH"

def test_low_severity_below_threshold_still_dropped(self) -> None:
"""LOW finding at 0.2 confidence is still dropped (no severity protection)."""
findings = [_finding(confidence=0.2, severity="LOW")]
result = _fallback_filtered(findings)
assert len(result) == 0


class TestCodeExampleFiltering:
"""Findings in code example context are downweighted, not hard-dropped."""

def test_fenced_code_block_context_downweighted(self) -> None:
"""Finding whose context contains ``` gets confidence halved."""
findings = [
_finding(
context="```bash\ncurl -k https://api.example.com\n```",
confidence=0.8,
)
]
result = _fallback_filtered(findings)
assert len(result) == 1
assert result[0].confidence == 0.4

def test_example_keyword_context_downweighted(self) -> None:
"""Finding whose context contains 'example:' gets downweighted."""
findings = [
_finding(
context="Example: how to use subprocess\nsubprocess.run(cmd)",
confidence=0.8,
)
]
result = _fallback_filtered(findings)
assert len(result) == 1
assert result[0].confidence == 0.4

def test_code_example_low_confidence_low_severity_dropped(self) -> None:
"""LOW severity finding at 0.6 conf in code-example context: 0.6*0.5=0.3 < 0.4, dropped."""
findings = [
_finding(
context="```\ncurl -k https://api.example.com\n```",
confidence=0.6,
severity="LOW",
)
]
result = _fallback_filtered(findings)
assert len(result) == 0

def test_code_example_high_severity_retained(self) -> None:
"""HIGH severity finding in code-example context at low conf: retained by severity floor."""
findings = [
_finding(
context="```\ncurl -k https://api.example.com\n```",
confidence=0.6,
severity="HIGH",
)
]
result = _fallback_filtered(findings)
assert len(result) == 1

def test_normal_code_context_kept(self) -> None:
"""Finding with regular code context (no example indicators) passes."""
findings = [
_finding(
context="import subprocess\nresult = subprocess.run(cmd, shell=True)",
confidence=0.8,
)
]
result = _fallback_filtered(findings)
assert len(result) == 1

def test_no_context_kept(self) -> None:
"""Finding with no context (None) passes through."""
findings = [_finding(context=None, confidence=0.8)]
result = _fallback_filtered(findings)
assert len(result) == 1


class TestCombinedFiltering:
"""Both filters work together."""

def test_mixed_findings_filtered(self) -> None:
"""Mix of low-confidence, code-example, and genuine findings."""
findings = [
_finding(confidence=0.2, severity="LOW"), # dropped: low conf + low sev
_finding(
confidence=0.8,
context="```\ncurl -k https://example.com\n```",
), # kept but downweighted (HIGH severity protects)
_finding(confidence=0.8), # kept: genuine finding
_finding(confidence=0.6), # kept: above threshold, normal context
]
result = _fallback_filtered(findings)
assert len(result) == 3

def test_remediation_applied(self) -> None:
"""Kept findings get default remediation if none set."""
findings = [_finding(confidence=0.8)]
result = _fallback_filtered(findings)
assert len(result) == 1
assert result[0].remediation is not None
assert len(result[0].remediation) > 0

def test_empty_input(self) -> None:
"""Empty findings list returns empty."""
assert _fallback_filtered([]) == []


class TestLLMFailurePassthrough:
"""On LLM failure, all findings pass through (fail-closed)."""

def test_passthrough_preserves_all_findings(self) -> None:
"""_passthrough_with_defaults keeps all findings regardless of confidence."""
findings = [
_finding(confidence=0.1, severity="LOW"),
_finding(confidence=0.3, severity="MEDIUM"),
_finding(confidence=0.9, severity="CRITICAL"),
]
result = _passthrough_with_defaults(findings)
assert len(result) == 3

def test_passthrough_adds_default_remediation(self) -> None:
"""Passthrough adds default remediation to findings without one."""
findings = [_finding(confidence=0.8)]
result = _passthrough_with_defaults(findings)
assert len(result) == 1
assert result[0].remediation is not None

def test_meta_analyzer_llm_failure_uses_passthrough(self) -> None:
"""When LLM call raises, meta_analyzer passes all findings through."""
findings = [
_finding(confidence=0.2, severity="LOW"),
_finding(confidence=0.8, severity="HIGH"),
]
state = {
"findings": findings,
"use_llm": True,
"file_cache": {"tool.py": "import subprocess"},
"manifest": {},
"model_config": {},
}
with patch(
"skillspector.nodes.meta_analyzer.LLMMetaAnalyzer"
) as mock_cls:
mock_cls.return_value.get_batches.side_effect = RuntimeError("API timeout")
result = meta_analyzer(state)
assert len(result["filtered_findings"]) == 2