fix(llm): isolate Stage 2 batch failures and keep unanalysed findings

src/skillspector/llm_analyzer_base.py

@@ -407,6 +407,14 @@ async def arun_batches(
         *max_concurrency* LLM requests in parallel.  Both cross-file and
         cross-chunk batches are parallelized in a single gather call.
 
+        Failures are isolated per batch: a transient error (timeout, 429,
+        oversized-chunk 400, ...) costs only its own batch, which is logged
+        and omitted from the result, so one bad call cannot cancel the rest
+        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.
+
         The return type mirrors :meth:`run_batches`.
         """
         sem = asyncio.Semaphore(max_concurrency)
@@ -427,7 +435,18 @@ async def _process(batch: Batch) -> tuple[Batch, list]:
                 logger.debug("LLM response for %s", batch.file_label)
                 return (batch, self.parse_response(response, batch))
 
-        return list(await asyncio.gather(*[_process(b) for b in batches]))
+        results = await asyncio.gather(*[_process(b) for b in batches], return_exceptions=True)
+        successful: list[tuple[Batch, list]] = []
+        for batch, result in zip(batches, results, strict=True):
+            if isinstance(result, (ValueError, NotImplementedError)):
+                raise result
+            if isinstance(result, asyncio.CancelledError):
+                raise result
+            if isinstance(result, BaseException):
+                logger.warning("LLM batch failed for %s: %s", batch.file_label, result)
+                continue
+            successful.append(result)
+        return successful
 
     # -- Convenience --------------------------------------------------------
 

src/skillspector/nodes/meta_analyzer.py

@@ -484,7 +484,31 @@ def meta_analyzer(state: SkillspectorState) -> MetaAnalyzerResponse:
         )
 
         batch_results = asyncio.run(analyzer.arun_batches(batches, metadata_text=metadata_text))
-        filtered = analyzer.apply_filter(findings, batch_results)
+
+        if len(batch_results) < len(batches):
+            # Some batches never returned. A finding the LLM never saw has no
+            # verdict — keep it via the fallback path instead of letting
+            # apply_filter treat the missing confirmation as a rejection.
+            # get_batches passes through the same Finding objects from
+            # `findings`; if that ever changes, id-based partitioning fails
+            # closed by keeping copied findings as unanalysed.
+            analysed_ids = {id(f) for batch, _ in batch_results for f in batch.findings}
+            analysed = [f for f in findings if id(f) in analysed_ids]
+            unanalysed = [f for f in findings if id(f) not in analysed_ids]
+        else:
+            analysed, unanalysed = findings, []
+
+        filtered = analyzer.apply_filter(analysed, batch_results)
+        if unanalysed:
+            logger.warning(
+                "Meta-analyzer: %d/%d batches failed; keeping %d findings in %d "
+                "files unfiltered (no LLM verdict)",
+                len(batches) - len(batch_results),
+                len(batches),
+                len(unanalysed),
+                len({f.file for f in unanalysed}),
+            )
+            filtered.extend(_fallback_filtered(unanalysed))
 
         logger.debug(
             "LLM filtering done: %d findings -> %d after filter",

tests/nodes/test_llm_analyzer_base.py

@@ -557,6 +557,53 @@ async def _delayed_ainvoke(prompt: str) -> LLMAnalysisResult:
 
         assert seen_files == {f"file_{i}.py" for i in range(num_batches)}
 
+    @patch(MOCK_PATCH_TARGET, _mock_get_chat_model)
+    async def test_failed_batch_does_not_abort_the_others(self) -> None:
+        """A transient failure costs only its own batch, not the whole fan-out."""
+
+        async def _flaky_ainvoke(prompt: str) -> LLMAnalysisResult:
+            if "b.py" in prompt:
+                raise RuntimeError("429 Too Many Requests")
+            return LLMAnalysisResult(findings=[])
+
+        analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL)
+        analyzer._structured_llm.ainvoke = _flaky_ainvoke
+
+        batches = [
+            Batch(file_path="a.py", content="code a"),
+            Batch(file_path="b.py", content="code b"),
+            Batch(file_path="c.py", content="code c"),
+        ]
+        results = await analyzer.arun_batches(batches)
+        assert {batch.file_path for batch, _ in results} == {"a.py", "c.py"}
+
+    @patch(MOCK_PATCH_TARGET, _mock_get_chat_model)
+    async def test_all_batches_failed_returns_empty(self) -> None:
+        analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL)
+        analyzer._structured_llm.ainvoke = AsyncMock(side_effect=RuntimeError("boom"))
+        batches = [Batch(file_path="a.py", content="code")]
+        assert await analyzer.arun_batches(batches) == []
+
+    @patch(MOCK_PATCH_TARGET, _mock_get_chat_model)
+    async def test_value_error_still_propagates(self) -> None:
+        """ValueError signals misconfiguration, not infra trouble — never swallowed."""
+        analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL)
+        analyzer._structured_llm.ainvoke = AsyncMock(side_effect=ValueError("no API key"))
+        batches = [Batch(file_path="a.py", content="code")]
+        with pytest.raises(ValueError, match="no API key"):
+            await analyzer.arun_batches(batches)
+
+    @patch(MOCK_PATCH_TARGET, _mock_get_chat_model)
+    async def test_cancelled_error_still_propagates(self) -> None:
+        """Cooperative cancellation must not be treated as a transient batch failure."""
+        import asyncio
+
+        analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL)
+        analyzer._structured_llm.ainvoke = AsyncMock(side_effect=asyncio.CancelledError())
+        batches = [Batch(file_path="a.py", content="code")]
+        with pytest.raises(asyncio.CancelledError):
+            await analyzer.arun_batches(batches)
+
 
 # ---------------------------------------------------------------------------
 # _format_findings_for_prompt (per-file, no truncation)

tests/nodes/test_meta_analyzer.py

@@ -13,11 +13,25 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Unit tests for LLMMetaAnalyzer.apply_filter (no LLM / no network)."""
+"""Tests for LLMMetaAnalyzer filtering and partial batch failure handling."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, patch
 
 from skillspector.llm_analyzer_base import Batch
 from skillspector.models import Finding
-from skillspector.nodes.meta_analyzer import LLMMetaAnalyzer
+from skillspector.nodes.meta_analyzer import LLMMetaAnalyzer, meta_analyzer
+
+MOCK_PATCH_TARGET = "skillspector.llm_analyzer_base.get_chat_model"
+
+
+def _mock_get_chat_model(*_args, **_kwargs):
+    from unittest.mock import MagicMock
+
+    mock_llm = MagicMock()
+    mock_llm.with_structured_output.return_value = MagicMock()
+    return mock_llm
 
 
 def _analyzer() -> LLMMetaAnalyzer:
@@ -49,11 +63,22 @@ def _llm_item(rule_id: str, start_line: int, **kw: object) -> dict[str, object]:
     return item
 
 
+def _confirm(pattern_id: str, file: str, start_line: int) -> dict[str, object]:
+    """LLM item confirming a finding, as parse_response would emit it."""
+    return {
+        "pattern_id": pattern_id,
+        "is_vulnerability": True,
+        "confidence": 0.9,
+        "explanation": "confirmed by llm",
+        "remediation": "fix it",
+        "_file": file,
+        "start_line": start_line,
+        "end_line": None,
+    }
+
+
 def test_confirmed_finding_kept_when_model_returns_end_line() -> None:
-    """Regression: a static finding with end_line=None must still match a
-    confirmation whose end_line is populated (e.g. end_line == start_line, as
-    some models return). Previously these confirmed findings were silently
-    dropped. See issue #67."""
+    """A finding with end_line=None matches confirmation with end_line=start_line."""
     findings = [_finding("SC4", 4), _finding("SC4", 5)]
     items = [_llm_item("SC4", 4, end_line=4), _llm_item("SC4", 5, end_line=5)]
     batch = Batch(file_path="requirements.txt", content="", findings=findings)
@@ -65,8 +90,7 @@ def test_confirmed_finding_kept_when_model_returns_end_line() -> None:
 
 
 def test_rejected_finding_still_dropped() -> None:
-    """The end_line-agnostic fallback must not resurrect findings the LLM
-    rejected (is_vulnerability=False)."""
+    """The end_line-agnostic fallback must not resurrect rejected findings."""
     findings = [_finding("SC4", 4)]
     items = [_llm_item("SC4", 4, end_line=4, is_vulnerability=False)]
     batch = Batch(file_path="requirements.txt", content="", findings=findings)
@@ -88,8 +112,7 @@ def test_low_confidence_finding_dropped() -> None:
 
 
 def test_exact_end_line_match_still_works() -> None:
-    """Existing behaviour: when both sides carry the same concrete end_line,
-    the finding is kept (no regression from the new fallback)."""
+    """Existing behavior: matching concrete end_line keeps the finding."""
     findings = [_finding("AST1", 21, end_line=21)]
     items = [_llm_item("AST1", 21, end_line=21)]
     batch = Batch(file_path="requirements.txt", content="", findings=findings)
@@ -98,3 +121,95 @@ def test_exact_end_line_match_still_works() -> None:
 
     assert len(kept) == 1
     assert kept[0].rule_id == "AST1"
+
+
+@patch(MOCK_PATCH_TARGET, _mock_get_chat_model)
+class TestMetaAnalyzerPartialBatchFailure:
+    def _state(self, findings: list[Finding]) -> dict[str, object]:
+        return {
+            "findings": findings,
+            "use_llm": True,
+            "file_cache": {"a.py": "code a", "b.py": "code b"},
+            "manifest": {},
+            "model_config": {},
+        }
+
+    def test_unanalysed_findings_survive_a_failed_batch(self) -> None:
+        """Findings whose batch failed are kept (no verdict != rejection)."""
+        f_confirmed = Finding(rule_id="R1", message="m", file="a.py", start_line=1)
+        f_rejected = Finding(rule_id="R2", message="m", file="a.py", start_line=5)
+        f_unseen = Finding(rule_id="R1", message="m", file="b.py", start_line=3)
+
+        batch_a = Batch(file_path="a.py", content="code a", findings=[f_confirmed, f_rejected])
+        batch_b = Batch(file_path="b.py", content="code b", findings=[f_unseen])
+
+        # batch_b never returned (timeout/429): only batch_a's verdicts exist,
+        # and the LLM confirmed R1 but stayed silent on R2 (= rejection).
+        partial_results = [(batch_a, [_confirm("R1", "a.py", 1)])]
+
+        with (
+            patch.object(LLMMetaAnalyzer, "get_batches", return_value=[batch_a, batch_b]),
+            patch.object(
+                LLMMetaAnalyzer,
+                "arun_batches",
+                new_callable=AsyncMock,
+                return_value=partial_results,
+            ),
+        ):
+            result = meta_analyzer(self._state([f_confirmed, f_rejected, f_unseen]))
+
+        filtered = result["filtered_findings"]
+        kept = {(f.file, f.rule_id) for f in filtered}
+
+        assert ("a.py", "R1") in kept
+        assert ("a.py", "R2") not in kept
+        assert ("b.py", "R1") in kept
+
+        confirmed = next(f for f in filtered if f.file == "a.py")
+        assert confirmed.explanation == "confirmed by llm"
+
+    def test_all_batches_failed_keeps_everything_via_fallback(self) -> None:
+        f1 = Finding(rule_id="R1", message="m", file="a.py", start_line=1)
+        f2 = Finding(rule_id="R2", message="m", file="b.py", start_line=2)
+        batch_a = Batch(file_path="a.py", content="code a", findings=[f1])
+        batch_b = Batch(file_path="b.py", content="code b", findings=[f2])
+
+        with (
+            patch.object(LLMMetaAnalyzer, "get_batches", return_value=[batch_a, batch_b]),
+            patch.object(
+                LLMMetaAnalyzer,
+                "arun_batches",
+                new_callable=AsyncMock,
+                return_value=[],
+            ),
+        ):
+            result = meta_analyzer(self._state([f1, f2]))
+
+        kept = {(f.file, f.rule_id) for f in result["filtered_findings"]}
+        assert kept == {("a.py", "R1"), ("b.py", "R2")}
+
+    def test_no_failures_keeps_strict_confirm_or_drop(self) -> None:
+        """When every batch returns, unconfirmed findings are dropped as before."""
+        f_confirmed = Finding(rule_id="R1", message="m", file="a.py", start_line=1)
+        f_rejected = Finding(rule_id="R2", message="m", file="b.py", start_line=2)
+        batch_a = Batch(file_path="a.py", content="code a", findings=[f_confirmed])
+        batch_b = Batch(file_path="b.py", content="code b", findings=[f_rejected])
+
+        full_results = [
+            (batch_a, [_confirm("R1", "a.py", 1)]),
+            (batch_b, []),
+        ]
+
+        with (
+            patch.object(LLMMetaAnalyzer, "get_batches", return_value=[batch_a, batch_b]),
+            patch.object(
+                LLMMetaAnalyzer,
+                "arun_batches",
+                new_callable=AsyncMock,
+                return_value=full_results,
+            ),
+        ):
+            result = meta_analyzer(self._state([f_confirmed, f_rejected]))
+
+        kept = {(f.file, f.rule_id) for f in result["filtered_findings"]}
+        assert kept == {("a.py", "R1")}