Run TP4 batches concurrently

src/skillspector/nodes/analyzers/mcp_tool_poisoning.py

@@ -45,6 +45,7 @@
     append_output_language_instruction,
     estimate_tokens,
 )
+from skillspector.llm_utils import run_async
 from skillspector.model_info import get_max_input_tokens
 from skillspector.models import Finding, compute_match_fingerprint
 from skillspector.nodes.analyzers.static_runner import MAX_FINDINGS_PER_ANALYZER
@@ -1482,7 +1483,9 @@ def iter_candidates() -> Iterator[_TP4Candidate]:
         )
         analyzer = _TP4Analyzer(model, timeout=timeout)
         attempted = True
-        batch_outcome = analyzer.run_batches_detailed(batches)
+        # Fan out like the semantic analyzers do. The shared limiter still honours
+        # SKILLSPECTOR_MAX_LLM_CONCURRENCY, and results keep the batch order.
+        batch_outcome = run_async(analyzer.arun_batches_detailed(batches))
         result.inference_usage = cast(list[InferenceUsageRecord], analyzer.inference_usage)
         seen_finding_ids: set[str] = set()
         unexpected_response = False

tests/integration/test_graph.py

@@ -290,7 +290,7 @@ def inference_usage(self) -> list[object]:
         def __init__(self, _model: str, **_kwargs: object) -> None:
             pass
 
-        def run_batches_detailed(self, _batches: object) -> object:
+        async def arun_batches_detailed(self, _batches: object) -> object:
             raise RuntimeError("simulated LLM transport failure")
 
     # Semantic analyzers and meta_analyzer fail while constructing their shared

tests/test_mcp_tool_poisoning.py

@@ -17,17 +17,19 @@
 
 from __future__ import annotations
 
+import asyncio
 import base64
 import json
 import re
 from pathlib import Path
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 import yaml
 from pydantic import BaseModel
 
 from skillspector.inspection_ledger import LedgerOutcome, LedgerReason
+from skillspector.llm_analyzer_base import LLMAnalyzerBase
 from skillspector.llm_utils import AgentCLIChatModel
 from skillspector.nodes.analyzers import mcp_tool_poisoning
 from skillspector.nodes.deduplicate import deduplicate
@@ -214,6 +216,9 @@ def invoke_with_usage(self, prompt: str, collector: object) -> object:
             return self.response_schema.model_validate(response)
         return response
 
+    async def ainvoke_with_usage(self, prompt: str, collector: object) -> object:
+        return self.invoke_with_usage(prompt, collector)
+
 
 class _FakeChatModel:
     def __init__(self, structured_llm: _FakeStructuredLLM) -> None:
@@ -1641,7 +1646,7 @@ def test_llm_call_failure_returns_empty(self, monkeypatch: pytest.MonkeyPatch):
 
     def test_persistently_malformed_response_returns_empty(self, monkeypatch: pytest.MonkeyPatch):
         state = _make_state("mcp_mismatched_skill", use_llm=True)
-        monkeypatch.setattr("skillspector.llm_analyzer_base.time.sleep", lambda _delay: None)
+        monkeypatch.setattr(LLMAnalyzerBase, "_asleep_before_retry", AsyncMock())
         structured_llm = _mock_tp4_structured_llm(
             monkeypatch,
             [{}, {}, {}, {}],
@@ -1659,8 +1664,8 @@ def test_persistently_malformed_response_returns_empty(self, monkeypatch: pytest
 
     def test_malformed_response_is_retried(self, monkeypatch: pytest.MonkeyPatch):
         state = _make_state("mcp_mismatched_skill", use_llm=True)
-        sleep = MagicMock()
-        monkeypatch.setattr("skillspector.llm_analyzer_base.time.sleep", sleep)
+        sleep = AsyncMock()
+        monkeypatch.setattr(LLMAnalyzerBase, "_asleep_before_retry", sleep)
         structured_llm = _mock_tp4_structured_llm(
             monkeypatch,
             [{}, {"is_mismatch": False}],
@@ -1675,8 +1680,8 @@ def test_malformed_response_is_retried(self, monkeypatch: pytest.MonkeyPatch):
 
     def test_cli_parse_error_is_retried(self, monkeypatch: pytest.MonkeyPatch):
         state = _make_state("mcp_mismatched_skill", use_llm=True)
-        sleep = MagicMock()
-        monkeypatch.setattr("skillspector.llm_analyzer_base.time.sleep", sleep)
+        sleep = AsyncMock()
+        monkeypatch.setattr(LLMAnalyzerBase, "_asleep_before_retry", sleep)
         provider = MagicMock()
         provider.complete.side_effect = ["not JSON", '{"is_mismatch": false}']
         monkeypatch.setattr(
@@ -1692,8 +1697,8 @@ def test_cli_parse_error_is_retried(self, monkeypatch: pytest.MonkeyPatch):
 
     def test_out_of_range_confidence_is_retried(self, monkeypatch: pytest.MonkeyPatch):
         state = _make_state("mcp_mismatched_skill", use_llm=True)
-        sleep = MagicMock()
-        monkeypatch.setattr("skillspector.llm_analyzer_base.time.sleep", sleep)
+        sleep = AsyncMock()
+        monkeypatch.setattr(LLMAnalyzerBase, "_asleep_before_retry", sleep)
         structured_llm = _mock_tp4_structured_llm(
             monkeypatch,
             [{"is_mismatch": True, "confidence": 1.7}, {"is_mismatch": False}],
@@ -1707,6 +1712,49 @@ def test_out_of_range_confidence_is_retried(self, monkeypatch: pytest.MonkeyPatc
         assert result["llm_call_log"] == [{"node": "mcp_tool_poisoning", "ok": True, "error": None}]
 
 
+class TestTP4Concurrency:
+    def test_batches_run_concurrently_within_the_shared_limit(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """TP4 fans out its batches like the semantic analyzers instead of one by one."""
+        state = {
+            "manifest": {"name": "concurrency", "description": "Runs local calculations."},
+            "file_cache": {
+                "tool.py": "".join(f"value_{index} = {index}\n" for index in range(300))
+            },
+            "component_metadata": [{"path": "tool.py", "type": "python"}],
+            "use_llm": True,
+            "model_config": {"default": "test-model"},
+        }
+        monkeypatch.setenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", "2")
+        monkeypatch.setattr(mcp_tool_poisoning, "TP4_MAX_BATCH_INPUT_TOKENS", 256)
+        monkeypatch.setattr(mcp_tool_poisoning, "TP4_MAX_BATCHES", 4)
+        monkeypatch.setattr(mcp_tool_poisoning, "TP4_MIN_CODE_TOKENS", 1)
+        monkeypatch.setattr(mcp_tool_poisoning, "get_max_input_tokens", lambda _model: 2048)
+        structured = _mock_tp4_structured_llm(
+            monkeypatch, [{"is_mismatch": False} for _ in range(4)]
+        )
+        in_flight = 0
+        peak = 0
+
+        async def slow_ainvoke(llm: _FakeStructuredLLM, prompt: str, collector: object) -> object:
+            nonlocal in_flight, peak
+            in_flight += 1
+            peak = max(peak, in_flight)
+            await asyncio.sleep(0.05)
+            in_flight -= 1
+            return llm.invoke_with_usage(prompt, collector)
+
+        monkeypatch.setattr(_FakeStructuredLLM, "ainvoke_with_usage", slow_ainvoke)
+
+        result = node(state)
+
+        assert structured.calls == 4
+        assert peak == 2
+        semantic = [event for event in result["inspection_ledger"] if event["phase"] == "semantic"]
+        assert sum(event["outcome"] is LedgerOutcome.COMPLETED for event in semantic) == 4
+
+
 class TestTP4Telemetry:
     """TP4 records llm_call_log so the report's degradation detector counts it
     consistently with the semantic analyzers and the meta-analyzer."""