diff --git a/src/skillspector/llm_analyzer_base.py b/src/skillspector/llm_analyzer_base.py index 8e592f23..397ee0d8 100644 --- a/src/skillspector/llm_analyzer_base.py +++ b/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 -------------------------------------------------------- diff --git a/src/skillspector/nodes/meta_analyzer.py b/src/skillspector/nodes/meta_analyzer.py index acb6ca2a..616bb621 100644 --- a/src/skillspector/nodes/meta_analyzer.py +++ b/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", diff --git a/tests/nodes/test_llm_analyzer_base.py b/tests/nodes/test_llm_analyzer_base.py index 954dfda9..cd521119 100644 --- a/tests/nodes/test_llm_analyzer_base.py +++ b/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) diff --git a/tests/nodes/test_meta_analyzer.py b/tests/nodes/test_meta_analyzer.py index 0087f463..5cecb7b1 100644 --- a/tests/nodes/test_meta_analyzer.py +++ b/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")}