Fix runaway whitespace issue

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)

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,35 +297,27 @@ 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]
 
 
 # ---------------------------------------------------------------------------
 # LLMMetaAnalyzer (filter / enrich mode)
 # ---------------------------------------------------------------------------
 
 
+# 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