fix(scanner): include adjudicator token usage

docs/architecture/analyzers/adjudicator.md

@@ -9,6 +9,7 @@ When enabled via the `--adjudicate` CLI flag (or `ScanPolicy.adjudicator.enabled
 - **False positive demotion**: Uses an LLM to reason about whether a deterministic HIGH/CRITICAL match represents a real instance of the threat, or a coincidental regex hit on benign prose.
 - **Cascade prevention**: Because it runs before the LLM analyzer, demoted findings never enter the LLM analyzer's static-finding enrichment context — so a wrong deterministic HIGH cannot be amplified into LLM findings citing the same pattern hit.
 - **Audit trail**: Every finding it considers is recorded in `scan_metadata.adjudicator.audit` with the LLM's verdict, confidence, and reason.
+- **Usage accounting**: Adjudication input, output, and total tokens are included in the scan result's aggregate `llm_usage` values.
 
 The adjudicator only touches deterministic findings (static, pipeline, behavioral, bytecode, yara analyzers) at HIGH or CRITICAL severity. LLM and other advisory findings are outside its scope.
 

docs/reference/output-formats.md

@@ -111,7 +111,7 @@ skill-scanner scan evals/skills/data-exfiltration/environment-secrets --format j
 }
 ```
 
-`llm_usage` is only present when at least one LLM call was made (`--use-llm` and/or `--enable-meta`), aggregated across every LLM analyzer and meta-analysis call for the scan. It's omitted entirely on static-only scans.
+`llm_usage` is only present when at least one LLM call was made (`--use-llm`, `--enable-meta`, and/or `--adjudicate`). It is aggregated across every LLM analyzer, meta-analysis, and adjudication call for the scan, and omitted entirely when no LLM call runs.
 
 Use `--compact` to remove pretty-printing for machine pipelines.
 

skill_scanner/core/analyzers/adjudicator.py

@@ -51,6 +51,12 @@
 
 from ..models import Finding, Severity, Skill
 from ..rule_registry import PackLoader
+from .llm_request_handler import (
+    LLMTokenUsage,
+    _add_token_usage,
+    _empty_token_usage,
+    _extract_token_usage,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -277,6 +283,15 @@ def __init__(
         # traceability.
         self.audit: list[AdjudicationResult] = []
 
+        # Cumulative token usage across all LLM calls in the most recent
+        # adjudicate() run, including calls whose response cannot be parsed.
+        self._llm_usage: LLMTokenUsage = _empty_token_usage()
+
+    @property
+    def llm_usage(self) -> LLMTokenUsage:
+        """Token usage from the most recent :meth:`adjudicate` run."""
+        return dict(self._llm_usage)  # type: ignore[return-value]
+
     def is_available(self) -> bool:
         """Whether the adjudicator has enough config to run.
 
@@ -346,6 +361,9 @@ def _call_llm(self, prompt: str) -> dict[str, Any] | None:
             for attempt in range(self.max_retries + 1):
                 try:
                     response = litellm.completion(**request, drop_params=True)
+                    # The provider charged for a successful completion even if
+                    # its content is malformed and cannot yield a verdict.
+                    _add_token_usage(self._llm_usage, _extract_token_usage(response))
                     content = response["choices"][0]["message"]["content"] or ""
                     content = content.strip()
                     break
@@ -498,6 +516,7 @@ def adjudicate(self, findings: list[Finding], skill: Skill) -> list[Finding]:
 
         Returns the same list (mutated) so callers can chain.
         """
+        self._llm_usage = _empty_token_usage()
         if not self.is_available():
             logger.debug("adjudicator not configured (no model env var); skipping all findings")
             return findings

skill_scanner/core/scanner.py

@@ -257,33 +257,39 @@ def _scan_single_skill(self, skill: Skill, skill_directory: Path) -> ScanResult:
             # raised. LLM errors leave findings at their original severity,
             # so enabling this pass cannot introduce false negatives.
             adjudicator_audit: list[dict[str, Any]] = []
+            adjudicator_usage = _empty_token_usage()
             if self.policy.adjudicator.enabled and all_findings:
                 try:
                     from .analyzers.adjudicator import Adjudicator
 
                     adj = Adjudicator(
                         min_fp_confidence=self.policy.adjudicator.min_fp_confidence,
                     )
-                    if adj.is_available():
-                        adj.adjudicate(all_findings, skill)
-                        analyzer_names.append("adjudicator")
-                        adjudicator_audit = [
-                            {
-                                "rule_id": r.rule_id,
-                                "verdict": r.verdict,
-                                "confidence": r.confidence,
-                                "reason": r.reason,
-                                "demoted_to": r.demoted_to,
-                                "model_id": r.model_id,
-                            }
-                            for r in adj.audit
-                        ]
-                    else:
-                        logger.debug(
-                            "adjudicator enabled but no LLM model configured; "
-                            "set SKILL_SCANNER_LLM_MODEL or "
-                            "SKILL_SCANNER_ADJUDICATOR_LLM_MODEL to activate"
-                        )
+                    try:
+                        if adj.is_available():
+                            adj.adjudicate(all_findings, skill)
+                            analyzer_names.append("adjudicator")
+                            adjudicator_audit = [
+                                {
+                                    "rule_id": r.rule_id,
+                                    "verdict": r.verdict,
+                                    "confidence": r.confidence,
+                                    "reason": r.reason,
+                                    "demoted_to": r.demoted_to,
+                                    "model_id": r.model_id,
+                                }
+                                for r in adj.audit
+                            ]
+                        else:
+                            logger.debug(
+                                "adjudicator enabled but no LLM model configured; "
+                                "set SKILL_SCANNER_LLM_MODEL or "
+                                "SKILL_SCANNER_ADJUDICATOR_LLM_MODEL to activate"
+                            )
+                    finally:
+                        # Preserve billed usage even if a later adjudication
+                        # step raises and findings remain fail-closed.
+                        _add_token_usage(adjudicator_usage, adj.llm_usage)
                 except Exception as exc:
                     logger.warning("Adjudication failed: %s", exc)
 
@@ -328,6 +334,7 @@ def _scan_single_skill(self, skill: Skill, skill_directory: Path) -> ScanResult:
 
             # Aggregate token usage across all LLM analyzers that ran.
             aggregated_usage = _empty_token_usage()
+            _add_token_usage(aggregated_usage, adjudicator_usage)
             for analyzer in llm_analyzers:
                 if hasattr(analyzer, "llm_usage"):
                     _add_token_usage(aggregated_usage, analyzer.llm_usage)

tests/test_adjudicator.py

@@ -45,6 +45,8 @@
     Adjudicator,
 )
 from skill_scanner.core.models import Finding, Severity, Skill, SkillFile, SkillManifest
+from skill_scanner.core.scan_policy import ScanPolicy
+from skill_scanner.core.scanner import SkillScanner
 
 # ----- Fixtures -------------------------------------------------------------
 
@@ -96,10 +98,23 @@ def _finding(
     )
 
 
-def _mock_litellm_response(verdict: str, confidence: int, reason: str = "test") -> Any:
+def _mock_litellm_response(
+    verdict: str,
+    confidence: int,
+    reason: str = "test",
+    *,
+    prompt_tokens: int = 0,
+    completion_tokens: int = 0,
+) -> Any:
     """Build a mock LiteLLM response whose choices[0].message.content is JSON."""
     payload = json.dumps({"verdict": verdict, "confidence": confidence, "reason": reason})
-    resp: Any = {"choices": [{"message": {"content": payload}}]}
+    response_type = type("MockLiteLLMResponse", (dict,), {})
+    resp: Any = response_type({"choices": [{"message": {"content": payload}}]})
+    resp.usage = MagicMock(
+        prompt_tokens=prompt_tokens,
+        completion_tokens=completion_tokens,
+        total_tokens=prompt_tokens + completion_tokens,
+    )
     return resp
 
 
@@ -194,6 +209,91 @@ def test_llm_exception_keeps_original_severity(self, tmp_path: Path, with_model_
         assert finding.severity == Severity.HIGH
         assert "adjudication" not in (finding.metadata or {})
 
+
+class TestAdjudicatorTokenUsage:
+    """Adjudicator reports every billed LiteLLM completion."""
+
+    def test_accumulates_usage_across_findings(self, tmp_path: Path, with_model_env: None) -> None:
+        skill = _make_skill(tmp_path, "---\nname: test\n---\n\nSome content.\n")
+        findings = [
+            _finding("PROMPT_INJECTION_CONCEALMENT", Severity.HIGH, line_number=4),
+            _finding("PIPELINE_TAINT_FLOW", Severity.HIGH, analyzer="pipeline", line_number=4),
+        ]
+
+        with patch(
+            "litellm.completion",
+            side_effect=[
+                _mock_litellm_response("real", 5, prompt_tokens=100, completion_tokens=20),
+                _mock_litellm_response("real", 5, prompt_tokens=40, completion_tokens=10),
+            ],
+        ):
+            adj = Adjudicator()
+            adj.adjudicate(findings, skill)
+
+        assert adj.llm_usage == {
+            "input_tokens": 140,
+            "output_tokens": 30,
+            "total_tokens": 170,
+        }
+
+    def test_counts_usage_when_response_content_is_malformed(self, tmp_path: Path, with_model_env: None) -> None:
+        skill = _make_skill(tmp_path, "---\nname: test\n---\n\nSome content.\n")
+        finding = _finding("PROMPT_INJECTION_CONCEALMENT", Severity.HIGH, line_number=4)
+        response = _mock_litellm_response("real", 5, prompt_tokens=75, completion_tokens=8)
+        response["choices"][0]["message"]["content"] = "not json"
+
+        with patch("litellm.completion", return_value=response):
+            adj = Adjudicator()
+            adj.adjudicate([finding], skill)
+
+        assert finding.severity == Severity.HIGH
+        assert adj.llm_usage == {
+            "input_tokens": 75,
+            "output_tokens": 8,
+            "total_tokens": 83,
+        }
+
+    def test_scanner_combines_adjudicator_and_analyzer_usage(self, tmp_path: Path, with_model_env: None) -> None:
+        skill = _make_skill(
+            tmp_path,
+            "---\nname: test-skill\ndescription: test skill.\n---\n\nSome content.\n",
+        )
+        finding = _finding("PROMPT_INJECTION_CONCEALMENT", Severity.HIGH, line_number=5)
+
+        class DeterministicAnalyzer:
+            def get_name(self) -> str:
+                return "static_analyzer"
+
+            def analyze(self, _skill: Skill) -> list[Finding]:
+                return [finding]
+
+        class UsageLLMAnalyzer:
+            llm_usage = {"input_tokens": 300, "output_tokens": 50, "total_tokens": 350}
+
+            def get_name(self) -> str:
+                return "llm_analyzer"
+
+            def analyze(self, _skill: Skill) -> list[Finding]:
+                return []
+
+        policy = ScanPolicy.default()
+        policy.adjudicator.enabled = True
+        scanner = SkillScanner(analyzers=[DeterministicAnalyzer(), UsageLLMAnalyzer()], policy=policy)  # type: ignore[list-item]
+        response = _mock_litellm_response("real", 5, prompt_tokens=80, completion_tokens=20)
+
+        with patch("litellm.completion", return_value=response):
+            result = scanner._scan_single_skill(skill, Path(skill.directory))
+
+        assert result.llm_usage == {
+            "input_tokens": 380,
+            "output_tokens": 70,
+            "total_tokens": 450,
+        }
+
+
+class TestAdjudicatorMalformedResponses:
+    """Malformed adjudicator responses keep findings fail-closed."""
+
     def test_malformed_json_keeps_original_severity(self, tmp_path: Path, with_model_env: None) -> None:
         skill = _make_skill(tmp_path, "---\nname: test\n---\n\nSome content.\n")
         finding = _finding("PROMPT_INJECTION_CONCEALMENT", Severity.HIGH, line_number=4)