From 9919c159597e62016214a1d97ae1eb559e547b11 Mon Sep 17 00:00:00 2001 From: elliottwaves-20 Date: Sun, 4 Oct 2026 21:15:49 +0200 Subject: [PATCH] Run TP4 batches concurrently _check_tp4 executed its LLM batches through the serial run_batches_detailed loop, while the semantic analyzers and the meta-review fan out through arun_batches. On skills with many code chunks TP4 became the long tail of the scan and could exhaust the workflow deadline on its own. Use run_async(analyzer.arun_batches_detailed(batches)). The shared limiter still honours SKILLSPECTOR_MAX_LLM_CONCURRENCY (1 keeps the old serial behaviour), failures are recorded per batch as before, and successes keep the batch order, so the TP4_MAX_FINDINGS cap and the ledger projection are unchanged. Co-Authored-By: Claude Opus 5.5 Signed-off-by: elliottwaves-20 --- .../nodes/analyzers/mcp_tool_poisoning.py | 5 +- tests/integration/test_graph.py | 2 +- tests/test_mcp_tool_poisoning.py | 64 ++++++++++++++++--- 3 files changed, 61 insertions(+), 10 deletions(-) diff --git a/src/skillspector/nodes/analyzers/mcp_tool_poisoning.py b/src/skillspector/nodes/analyzers/mcp_tool_poisoning.py index a6dcef182..d669d3a79 100644 --- a/src/skillspector/nodes/analyzers/mcp_tool_poisoning.py +++ b/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 diff --git a/tests/integration/test_graph.py b/tests/integration/test_graph.py index d8c607105..006332535 100644 --- a/tests/integration/test_graph.py +++ b/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 diff --git a/tests/test_mcp_tool_poisoning.py b/tests/test_mcp_tool_poisoning.py index f1ee16740..d33c03e76 100644 --- a/tests/test_mcp_tool_poisoning.py +++ b/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."""