isolate malformed async LLM batches

Signed-off-by: kigland <shuaizhicheng336@gmail.com>

Author: koriyoshi2041

src/skillspector/llm_analyzer_base.py

@@ -374,7 +374,9 @@ 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.
+        return dicts or other types. Malformed structured responses are isolated
+        per batch and omitted from the result so callers can detect partial
+        analysis by comparing submitted and returned batches.
         """
         results: list[tuple[Batch, list]] = []
         for batch in batches:
@@ -399,6 +401,9 @@ def run_batches(
             else:
                 try:
                     response = _message_text(self._llm.invoke(prompt))
+                except ValidationError as exc:
+                    logger.warning("LLM batch failed for %s: %s", batch.file_label, exc)
+                    continue
                 except (ValueError, NotImplementedError):
                     raise
                 except Exception as exc:
@@ -432,12 +437,12 @@ async def arun_batches(
         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.
+        oversized-chunk 400, malformed structured response, ...) 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`.
         """
@@ -462,6 +467,9 @@ async def _process(batch: Batch) -> tuple[Batch, list]:
         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, ValidationError):
+                logger.warning("LLM batch failed for %s: %s", batch.file_label, result)
+                continue
             if isinstance(result, (ValueError, NotImplementedError)):
                 raise result
             if isinstance(result, BaseException):

src/skillspector/nodes/analyzers/semantic_developer_intent.py

@@ -179,6 +179,12 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse:
         results = asyncio.run(analyzer.arun_batches(batches))
         findings = analyzer.collect_findings(results)
         logger.info("%s: %d findings", ANALYZER_ID, len(findings))
+        if len(results) < len(batches):
+            error = f"{len(batches) - len(results)}/{len(batches)} LLM batches failed"
+            return {
+                "findings": findings,
+                "llm_call_log": [llm_call_record(ANALYZER_ID, ok=False, error=error)],
+            }
         return {"findings": findings, "llm_call_log": [llm_call_record(ANALYZER_ID, ok=True)]}
     except ValueError:
         raise

src/skillspector/nodes/analyzers/semantic_quality_policy.py

@@ -148,6 +148,12 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse:
         results = asyncio.run(analyzer.arun_batches(batches))
         findings = analyzer.collect_findings(results)
         logger.info("%s: %d findings", ANALYZER_ID, len(findings))
+        if len(results) < len(batches):
+            error = f"{len(batches) - len(results)}/{len(batches)} LLM batches failed"
+            return {
+                "findings": findings,
+                "llm_call_log": [llm_call_record(ANALYZER_ID, ok=False, error=error)],
+            }
         return {"findings": findings, "llm_call_log": [llm_call_record(ANALYZER_ID, ok=True)]}
     except ValueError:
         raise

src/skillspector/nodes/analyzers/semantic_security_discovery.py

@@ -90,6 +90,12 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse:
         results = analyzer.run_batches(batches)
         findings = analyzer.collect_findings(results)
         logger.info("%s: %d findings", ANALYZER_ID, len(findings))
+        if len(results) < len(batches):
+            error = f"{len(batches) - len(results)}/{len(batches)} LLM batches failed"
+            return {
+                "findings": findings,
+                "llm_call_log": [llm_call_record(ANALYZER_ID, ok=False, error=error)],
+            }
         return {"findings": findings, "llm_call_log": [llm_call_record(ANALYZER_ID, ok=True)]}
     except ValidationError as exc:
         # Malformed LLM response — degrade gracefully rather than crashing the graph

src/skillspector/nodes/meta_analyzer.py

@@ -563,6 +563,12 @@ def meta_analyzer(state: SkillspectorState) -> MetaAnalyzerResponse:
             len(findings),
             len(filtered),
         )
+        if unanalysed:
+            error = f"{len(batches) - len(batch_results)}/{len(batches)} LLM batches failed"
+            return {
+                "filtered_findings": filtered,
+                "llm_call_log": [llm_call_record("meta_analyzer", ok=False, error=error)],
+            }
         return {
             "filtered_findings": filtered,
             "llm_call_log": [llm_call_record("meta_analyzer", ok=True)],

tests/nodes/analyzers/test_semantic_developer_intent.py

@@ -22,7 +22,7 @@
 
 import pytest
 
-from skillspector.llm_analyzer_base import LLMAnalysisResult, LLMFinding
+from skillspector.llm_analyzer_base import Batch, LLMAnalysisResult, LLMFinding
 from skillspector.models import Finding
 from skillspector.nodes.analyzers.semantic_developer_intent import (
     ANALYZER_ID,
@@ -240,7 +240,12 @@ class TestLLMCallTelemetry:
     def test_success_records_ok_true(self) -> None:
         from skillspector.llm_analyzer_base import LLMAnalyzerBase
 
-        with patch.object(LLMAnalyzerBase, "arun_batches", new_callable=AsyncMock, return_value=[]):
+        with patch.object(
+            LLMAnalyzerBase,
+            "arun_batches",
+            new_callable=AsyncMock,
+            return_value=[(Batch(file_path="main.py", content="import os"), [])],
+        ):
             result = node({"file_cache": {"main.py": "import os"}})
         assert result["llm_call_log"] == [{"node": ANALYZER_ID, "ok": True, "error": None}]
 

tests/nodes/analyzers/test_semantic_security_discovery.py

@@ -23,7 +23,7 @@
 import pytest
 from pydantic import ValidationError
 
-from skillspector.llm_analyzer_base import LLMAnalysisResult, LLMFinding
+from skillspector.llm_analyzer_base import Batch, LLMAnalysisResult, LLMFinding
 from skillspector.models import Finding
 from skillspector.nodes.analyzers.semantic_security_discovery import (
     ANALYZER_ID,
@@ -316,7 +316,11 @@ class TestLLMCallTelemetry:
     def test_success_records_ok_true(self, base_state) -> None:
         from skillspector.llm_analyzer_base import LLMAnalyzerBase
 
-        with patch.object(LLMAnalyzerBase, "run_batches", return_value=[]):
+        with patch.object(
+            LLMAnalyzerBase,
+            "run_batches",
+            return_value=[(Batch(file_path="SKILL.md", content="# Skill"), [])],
+        ):
             result = node(base_state)
         assert result["llm_call_log"] == [{"node": ANALYZER_ID, "ok": True, "error": None}]
 
@@ -374,7 +378,6 @@ def _build_file_cache(skill_dir: Path) -> dict[str, str]:
 
 def _make_file_aware_run_batches(responses: dict[str, LLMAnalysisResult]):
     """Return a mock run_batches that dispatches based on file_path in each batch."""
-    from skillspector.llm_analyzer_base import Batch
 
     def _run_batches(self_inner, batches: list[Batch], **_kwargs):
         results = []

tests/nodes/test_llm_analyzer_base.py

@@ -621,6 +621,32 @@ async def _flaky_ainvoke(prompt: str) -> LLMAnalysisResult:
         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_malformed_structured_batch_does_not_abort_the_others(self) -> None:
+        """A malformed structured response is isolated even though it is a ValueError."""
+
+        async def _ainvoke(prompt: str) -> LLMAnalysisResult:
+            if "b.py" in prompt:
+                return LLMAnalysisResult.model_validate({"findings": 'We{"findings":[]}'})
+            return LLMAnalysisResult(
+                findings=[
+                    LLMFinding(rule_id="T-1", message="hit", severity="LOW", start_line=1),
+                ]
+            )
+
+        analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL)
+        analyzer._structured_llm.ainvoke = _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"}
+        assert [items[0].rule_id for _, items in results] == ["T-1", "T-1"]
+
     @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)

tests/nodes/test_meta_analyzer.py

@@ -261,15 +261,18 @@ def _degr_state(**overrides: object) -> SkillspectorState:
 
 
 def test_records_ok_true_on_success() -> None:
+    state = _degr_state()
     with (
         patch("skillspector.llm_analyzer_base.get_chat_model", return_value=MagicMock()),
         patch(
             "skillspector.nodes.meta_analyzer.LLMMetaAnalyzer.arun_batches",
             new_callable=AsyncMock,
-            return_value=[],
+            return_value=[
+                (Batch(file_path="SKILL.md", content="# Skill", findings=state["findings"]), [])
+            ],
         ),
     ):
-        result = meta_analyzer(_degr_state())
+        result = meta_analyzer(state)
     assert result["llm_call_log"] == [{"node": "meta_analyzer", "ok": True, "error": None}]
 
 

tests/nodes/test_semantic_quality_policy.py

@@ -22,7 +22,7 @@
 
 import pytest
 
-from skillspector.llm_analyzer_base import LLMAnalysisResult, LLMFinding
+from skillspector.llm_analyzer_base import Batch, LLMAnalysisResult, LLMFinding
 from skillspector.models import Finding
 from skillspector.nodes.analyzers.semantic_quality_policy import (
     ANALYZER_ID,
@@ -269,7 +269,12 @@ class TestLLMCallTelemetry:
     def test_success_records_ok_true(self) -> None:
         from skillspector.llm_analyzer_base import LLMAnalyzerBase
 
-        with patch.object(LLMAnalyzerBase, "arun_batches", new_callable=AsyncMock, return_value=[]):
+        with patch.object(
+            LLMAnalyzerBase,
+            "arun_batches",
+            new_callable=AsyncMock,
+            return_value=[(Batch(file_path="SKILL.md", content="# Skill"), [])],
+        ):
             result = node({"file_cache": {"SKILL.md": "# Skill"}})
         assert result["llm_call_log"] == [{"node": ANALYZER_ID, "ok": True, "error": None}]