diff --git a/src/skillspector/llm_analyzer_base.py b/src/skillspector/llm_analyzer_base.py index c5ab9dce..38268245 100644 --- a/src/skillspector/llm_analyzer_base.py +++ b/src/skillspector/llm_analyzer_base.py @@ -32,8 +32,10 @@ from dataclasses import dataclass, field from typing import Literal +import openai +from langchain_core.exceptions import OutputParserException from langchain_core.messages import BaseMessage -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel, Field, ValidationError, field_validator from skillspector.llm_utils import get_chat_model from skillspector.logging_config import get_logger @@ -244,11 +246,56 @@ def _message_text(response: object) -> str: - Be precise: report only genuine issues, not speculative ones.""" +# Structured-output method. langchain-openai's default is "json_schema", which +# has the model emit the result as ordinary message *content* under a strict +# JSON grammar. On some OpenAI-compatible serving stacks, content-mode JSON +# generation degenerates into a whitespace-repetition loop — the model emits one +# valid finding, then thousands of whitespace tokens until the output cap +# (finish_reason="length"), which the strict parser then discards. This was +# observed across unrelated models (NVIDIA Nemotron, gpt-oss), i.e. it is a +# property of the serving stack's content generation, not any one model. Emitting +# the same data as a *tool call* ("function_calling") avoids the content path and +# eliminates the runaway empirically (0/6 vs 4/6 on a 90-finding meta call). Free +# (non-guided) generation runs away just as badly, so this is specifically about +# tool-call vs content output — not guided-decoding-vs-not. +_STRUCTURED_OUTPUT_METHOD = "function_calling" + +# function_calling still occasionally returns tool-call arguments that don't +# satisfy the schema (a fast, retryable ValidationError — not a slow runaway), so +# retry a failed structured call a few times; each attempt is a fresh sample. +_LLM_MAX_RETRIES = 2 + +# Structured-output failures that a fresh sample usually fixes. All are caught +# per attempt and retried; on the final attempt they are wrapped in +# StructuredOutputError (see its docstring for why wrapping matters). +_RETRYABLE_STRUCTURED_ERRORS = ( + openai.LengthFinishReasonError, + OutputParserException, + ValidationError, +) + + # --------------------------------------------------------------------------- # Base LLM Analyzer # --------------------------------------------------------------------------- +class StructuredOutputError(RuntimeError): + """A structured LLM call failed after all retries. + + Deliberately a ``RuntimeError`` (NOT a ``ValueError``): the analyzer nodes + re-raise ``ValueError`` (to surface credential errors) but fall back on any + other ``Exception``. Two of the underlying failures — ``OutputParserException`` + and ``ValidationError`` — *are* ``ValueError`` subclasses, so raising them + directly would be re-raised by the nodes and crash the scan instead of + degrading gracefully. (``openai.LengthFinishReasonError`` is *not* a + ``ValueError`` and would already fall through to the nodes' generic + ``except Exception``, but it is wrapped here too for uniform handling.) + Wrapping breaks the ``ValueError`` re-raise path so a failed structured + call degrades instead of crashing. + """ + + class LLMAnalyzerBase: """Per-file / per-chunk LLM analyzer. @@ -275,9 +322,63 @@ def __init__(self, base_prompt: str, model: str): self._input_budget = get_max_input_tokens(model) self._llm = get_chat_model(model=model) self._structured_llm = ( - self._llm.with_structured_output(self.response_schema) if self.response_schema else None + self._llm.with_structured_output(self.response_schema, method=_STRUCTURED_OUTPUT_METHOD) + if self.response_schema + else None ) + # -- Structured-output invocation (retry) ------------------------------- + + @staticmethod + def _check_structured_result(result: object) -> object: + """Treat a ``None`` structured result as a retryable failure. + + The model emitted no tool call at all — common on oversized outputs. + ``None`` isn't an exception, so surface it as one so the retry loop + re-samples instead of returning ``None`` to ``parse_response``. + """ + if result is None: + raise OutputParserException("structured call returned no tool call") + return result + + @staticmethod + def _handle_structured_failure(attempt: int, exc: Exception) -> None: + """On the final attempt wrap-and-raise; otherwise log and let the caller retry.""" + if attempt == _LLM_MAX_RETRIES: + raise StructuredOutputError(str(exc)) from exc + logger.warning( + "structured LLM call failed (attempt %d/%d), retrying: %s", + attempt + 1, + _LLM_MAX_RETRIES + 1, + exc, + ) + + def _invoke_structured(self, prompt: str): + """Invoke the structured LLM, retrying on a failed structured response. + + Failures are sampling-dependent, so a fresh attempt usually succeeds: + ``LengthFinishReasonError`` (content-mode whitespace runaway hitting the + token cap), ``OutputParserException`` (unparseable tool call or no tool + call at all), or ``ValidationError`` (tool-call args that miss the schema). + """ + for attempt in range(_LLM_MAX_RETRIES + 1): + try: + return self._check_structured_result(self._structured_llm.invoke(prompt)) + except _RETRYABLE_STRUCTURED_ERRORS as exc: + self._handle_structured_failure(attempt, exc) + # _handle_structured_failure always raises on the final attempt; this is + # an explicit guard so the method never silently returns None. + raise StructuredOutputError("structured call exhausted retries") + + async def _ainvoke_structured(self, prompt: str): + """Async counterpart of :meth:`_invoke_structured`.""" + for attempt in range(_LLM_MAX_RETRIES + 1): + try: + return self._check_structured_result(await self._structured_llm.ainvoke(prompt)) + except _RETRYABLE_STRUCTURED_ERRORS as exc: + self._handle_structured_failure(attempt, exc) + raise StructuredOutputError("structured call exhausted retries") + # -- Batching ----------------------------------------------------------- def _estimate_extra_overhead(self, findings: list[Finding]) -> int: @@ -375,6 +476,12 @@ def run_batches( The element type of the inner list depends on the subclass: the default :meth:`parse_response` returns :class:`Finding` objects; subclasses may return dicts or other types. + + A batch whose structured call fails after all retries + (:class:`StructuredOutputError`) is logged and skipped rather than + aborting the whole run, so the batches that did succeed still + contribute their results. Other exceptions (e.g. credential + ``ValueError``) propagate. """ results: list[tuple[Batch, list]] = [] for batch in batches: @@ -385,10 +492,14 @@ def run_batches( estimate_tokens(prompt), len(batch.findings), ) - if self._structured_llm: - response = self._structured_llm.invoke(prompt) - else: - response = _message_text(self._llm.invoke(prompt)) + try: + if self._structured_llm: + response = self._invoke_structured(prompt) + else: + response = _message_text(self._llm.invoke(prompt)) + except StructuredOutputError as exc: + logger.warning("Skipping %s: %s", batch.file_label, exc) + continue logger.debug("LLM response for %s", batch.file_label) parsed = self.parse_response(response, batch) results.append((batch, parsed)) @@ -413,7 +524,10 @@ async def arun_batches( of the fan-out. Callers can detect partial results by comparing the returned batches against the submitted ones. ``ValueError`` and ``NotImplementedError`` signal misconfiguration rather than infra - trouble and keep propagating. + trouble and keep propagating. A batch whose structured call fails after + all retries (:class:`StructuredOutputError`) is one such isolated failure: + it is a ``RuntimeError``, so it is logged and dropped like any other infra + error rather than aborting the gather. The return type mirrors :meth:`run_batches`. """ @@ -429,7 +543,7 @@ async def _process(batch: Batch) -> tuple[Batch, list]: len(batch.findings), ) if self._structured_llm: - response = await self._structured_llm.ainvoke(prompt) + response = await self._ainvoke_structured(prompt) else: response = _message_text(await self._llm.ainvoke(prompt)) logger.debug("LLM response for %s", batch.file_label) diff --git a/src/skillspector/nodes/meta_analyzer.py b/src/skillspector/nodes/meta_analyzer.py index 58c5b634..d27e021f 100644 --- a/src/skillspector/nodes/meta_analyzer.py +++ b/src/skillspector/nodes/meta_analyzer.py @@ -24,6 +24,7 @@ import asyncio import json +from dataclasses import replace from typing import Literal from pydantic import BaseModel, Field, field_validator @@ -87,18 +88,16 @@ def _normalize_confidence(cls, v: object) -> float: remediation: str = Field(default="", description="How to fix the issue (actionable steps)") -class OverallAssessment(BaseModel): - """Overall risk assessment for the analyzed file.""" - - risk_level: str = Field(description="Overall risk level: LOW, MEDIUM, HIGH, or CRITICAL") - summary: str = Field(description="Brief summary of findings") - - class MetaAnalyzerResult(BaseModel): - """Top-level structured response from the meta-analyzer LLM.""" + """Top-level structured response from the meta-analyzer LLM. + + Only ``findings`` is modelled: it is the sole field the node consumes. A + smaller required output is also less likely to truncate, which matters for + the served-model reliability the structured-output retry/batching machinery + targets. + """ findings: list[MetaAnalyzerFinding] = Field(default_factory=list) - overall_assessment: OverallAssessment | None = None @field_validator("findings", mode="before") @classmethod @@ -112,17 +111,6 @@ def _parse_stringified_findings(cls, v: object) -> object: return parsed if isinstance(parsed, list) else [] return v - @field_validator("overall_assessment", mode="before") - @classmethod - def _parse_stringified_assessment(cls, v: object) -> object: - """LLMs sometimes return nested objects as JSON strings.""" - if isinstance(v, str): - try: - return json.loads(v) - except (json.JSONDecodeError, TypeError): - return None - return v - # --------------------------------------------------------------------------- # Prompt (no JSON format instructions — schema handles the structure) @@ -218,6 +206,33 @@ def _format_findings_for_prompt(findings: list[Finding]) -> str: return "\n".join(lines) +def _fallback_finding(f: Finding) -> Finding: + """Pass *f* through unchanged but with a default remediation, no LLM enrichment. + + Used on the fail-closed paths (LLM unavailable or failed) to preserve a + finding rather than silently dropping an unreviewed one. + """ + return Finding( + rule_id=f.rule_id, + message=f.message, + severity=f.severity, + confidence=f.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, + ) + + _NO_LLM_CONFIDENCE_THRESHOLD = 0.4 _HIGH_SEVERITY_PASS_THROUGH = frozenset({"CRITICAL", "HIGH"}) _CODE_EXAMPLE_DOWNWEIGHT = 0.5 @@ -282,28 +297,7 @@ def _passthrough_with_defaults(findings: list[Finding]) -> list[Finding]: 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, - message=f.message, - severity=f.severity, - confidence=f.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, - ) - for f in findings - ] + return [_fallback_finding(f) for f in findings] # --------------------------------------------------------------------------- @@ -311,6 +305,19 @@ def _passthrough_with_defaults(findings: list[Finding]) -> list[Finding]: # --------------------------------------------------------------------------- +# Max findings enriched per LLM call. Served models are unreliable producing a +# large structured result in one tool call: on big outputs they either emit no +# tool call at all, or stringify the findings array and truncate that string +# (unparseable). Smaller batches keep the result short enough to complete +# reliably (observed: 90→none, 10→occasional truncation, small→reliable). +# A batch that still fails after retries is skipped individually (run loops +# don't abort on one failure) and its findings are preserved un-enriched by the +# node's fallback path, so more batches no longer means a higher chance of +# losing the whole pass — 10 balances per-call reliability against re-sent input +# cost (each batch re-sends the file content; see the cost note in get_batches). +_MAX_FINDINGS_PER_BATCH = 10 + + class LLMMetaAnalyzer(LLMAnalyzerBase): """Per-file LLM filter/enrichment of static findings. @@ -323,6 +330,40 @@ class LLMMetaAnalyzer(LLMAnalyzerBase): def __init__(self, model: str): super().__init__(base_prompt=PER_FILE_ANALYSIS_PROMPT, model=model) + def get_batches( + self, + file_paths: list[str], + file_cache: dict[str, str], + findings: list[Finding] | None = None, + ) -> list[Batch]: + """Split each file's findings into groups of ``_MAX_FINDINGS_PER_BATCH``. + + The base batcher splits by *input* size, so a file with many static + findings becomes one call that must emit a large tool-call result. Served + models can't reliably produce a big structured result in one shot — they + return no tool call at all on very large outputs. Bounding findings per + call keeps each tool call small and reliable; ``apply_filter`` re-merges + across batches by (file, rule_id, line). + + Cost note: each sub-batch re-sends the full file content and prompt, so + a file with N findings costs ~ceil(N / _MAX_FINDINGS_PER_BATCH)x the + input tokens of a single call. _MAX_FINDINGS_PER_BATCH trades that + re-sent input against per-call output reliability. + """ + batches = super().get_batches(file_paths, file_cache, findings) + bounded: list[Batch] = [] + for batch in batches: + if len(batch.findings) <= _MAX_FINDINGS_PER_BATCH: + bounded.append(batch) + continue + for i in range(0, len(batch.findings), _MAX_FINDINGS_PER_BATCH): + # replace() carries every other Batch field forward, so a new + # field added to Batch isn't silently dropped for split batches. + bounded.append( + replace(batch, findings=batch.findings[i : i + _MAX_FINDINGS_PER_BATCH]) + ) + return bounded + def _estimate_extra_overhead(self, findings: list[Finding]) -> int: if not findings: return 0