diff --git a/contrib/batch_scan/api_pool.py b/contrib/batch_scan/api_pool.py index f694954c6..b068f8b06 100644 --- a/contrib/batch_scan/api_pool.py +++ b/contrib/batch_scan/api_pool.py @@ -475,99 +475,87 @@ async def ainvoke_with_usage(self, prompt: str, collector: object) -> object: # -- Internal ------------------------------------------------------------- + @staticmethod + def _remaining(deadline: float) -> float: + from skillspector.llm_analyzer_base import LLMRuntimeLimitError + + remaining = deadline - time.monotonic() + if remaining <= 0: + raise LLMRuntimeLimitError("pooled request runtime limit reached") + return remaining + def _invoke_with_retry( - self, - prompt: str, - *, - callbacks: list[object] | None = None, + self, prompt: str, *, callbacks: list[object] | None = None, ) -> object: - """Sync retry loop — acquire slot, call LLM, release, retry on 429.""" - last_exception: Exception | None = None - + """Share one deadline across slot waiting, requests, and key retries.""" + deadline = time.monotonic() + self._timeout for attempt in range(self._max_retries + 1): - key = self._pool.acquire() - llm = self._build_llm(key) try: - if callbacks is None: - result = llm.invoke(prompt) - else: - result = llm.invoke(prompt, config={"callbacks": callbacks}) - self._pool.release(key, success=True) - if attempt > 0: - self._pool.record_retry_success() - return result + key = self._pool.acquire(timeout=self._remaining(deadline)) + except Exception: + self._remaining(deadline) + raise + rate_limited = False + try: + llm = self._build_llm(key, timeout=self._remaining(deadline)) + result = llm.invoke(prompt) if callbacks is None else llm.invoke( + prompt, config={"callbacks": callbacks} + ) except Exception as exc: - if self._is_rate_limit(exc) and attempt < self._max_retries: - self._pool.release(key, success=False) - logger.debug( - "PooledChatModel: rate-limited, retrying " - "(attempt %d/%d)", - attempt + 1, - self._max_retries, - ) + rate_limited = self._is_rate_limit(exc) + if rate_limited and attempt < self._max_retries: continue - self._pool.release(key, success=True) - last_exception = exc raise - - raise RuntimeError( - f"PooledChatModel: exhausted {self._max_retries} retries " - "due to rate-limit errors" - ) from last_exception + finally: + self._pool.release(key, success=not rate_limited) + if attempt > 0: + self._pool.record_retry_success() + return result + raise RuntimeError("PooledChatModel exhausted key retries") async def _ainvoke_with_retry( - self, - prompt: str, - *, - callbacks: list[object] | None = None, + self, prompt: str, *, callbacks: list[object] | None = None, ) -> object: - """Async retry loop — non-blocking acquire first, block only if full.""" + """Bound asynchronous slot waiting without leaving an acquiring thread.""" import asyncio - last_exception: Exception | None = None + deadline = time.monotonic() + self._timeout for attempt in range(self._max_retries + 1): + self._remaining(deadline) key = self._pool.try_acquire() - if key is None: - key = await asyncio.to_thread(self._pool.acquire) - llm = self._build_llm(key) + while key is None: + await asyncio.sleep(min(0.05, self._remaining(deadline))) + self._remaining(deadline) + key = self._pool.try_acquire() + rate_limited = False try: - if callbacks is None: - result = await llm.ainvoke(prompt) - else: - result = await llm.ainvoke(prompt, config={"callbacks": callbacks}) - self._pool.release(key, success=True) - if attempt > 0: - self._pool.record_retry_success() - return result + llm = self._build_llm(key, timeout=self._remaining(deadline)) + result = await llm.ainvoke(prompt) if callbacks is None else await llm.ainvoke( + prompt, config={"callbacks": callbacks} + ) except Exception as exc: - if self._is_rate_limit(exc) and attempt < self._max_retries: - self._pool.release(key, success=False) - logger.debug( - "PooledChatModel: rate-limited, retrying " - "(attempt %d/%d)", - attempt + 1, - self._max_retries, - ) + rate_limited = self._is_rate_limit(exc) + if rate_limited and attempt < self._max_retries: continue - self._pool.release(key, success=True) - last_exception = exc raise - - raise RuntimeError( - f"PooledChatModel: exhausted {self._max_retries} retries " - "due to rate-limit errors" - ) from last_exception - - def _build_llm(self, key: ApiKey): + finally: + self._pool.release(key, success=not rate_limited) + if attempt > 0: + self._pool.record_retry_success() + return result + raise RuntimeError("PooledChatModel exhausted key retries") + + def _build_llm(self, key: ApiKey, *, timeout: float | None = None): """Build a fresh :class:`~langchain_openai.ChatOpenAI` for *key*.""" from langchain_openai import ChatOpenAI from pydantic import SecretStr + timeout = self._timeout if timeout is None else timeout try: import httpx - _timeout = httpx.Timeout(self._timeout, connect=8.0) + _timeout = httpx.Timeout(timeout, connect=min(8.0, timeout)) except ImportError: - _timeout = self._timeout + _timeout = timeout return ChatOpenAI( model=key.model, @@ -575,6 +563,7 @@ def _build_llm(self, key: ApiKey): api_key=SecretStr(key.key), max_completion_tokens=self._max_tokens, timeout=_timeout, + max_retries=0, ) @staticmethod diff --git a/contrib/batch_scan/docs/DESIGN.md b/contrib/batch_scan/docs/DESIGN.md index a60c20aa7..2cdd8d464 100644 --- a/contrib/batch_scan/docs/DESIGN.md +++ b/contrib/batch_scan/docs/DESIGN.md @@ -284,22 +284,22 @@ raw LLM string → _strip_markdown_fences() → json.loads() → model_validate( The two-step parse (stdlib `json.loads` then Pydantic `model_validate`) exists because: -1. `json.loads` is fast, deterministic, and raises clear `JSONDecodeError` on - malformed output — we catch this and return `[]` (empty findings). -2. `model_validate` enforces the schema: required fields, literal enums, - confidence range, string length. Schema violations are caught and returned - as `[]` with a warning log. - -**Error propagation:** If the LLM returns invalid JSON or schema-mismatched -output, the analyzer returns `[]` (no findings for that file). The scan -continues — a single malformed LLM response never blocks the pipeline. -The warning is logged at `WARNING` level so operators can monitor parse-failure -rates without sifting through debug logs. - -Patch 3 adds a `_sanitize_meta_finding()` pass after validation to handle -known LLM quirks: `null` string fields → `""`, unrecognized enum values -(e.g., `"none"`) → `"low"`. These are applied post-validation because they -represent recoverable soft errors, not hard schema violations. +1. `json.loads` parses the response. Invalid JSON raises the core structured-response + validation error rather than producing an empty findings list. +2. `model_validate` checks the required fields and value types. Schema failures + use the same retry and reporting path. + +**Error propagation:** Invalid JSON and schema failures are retried up to four +attempts. If they persist, the ledger records `skipped` with reason +`llm_structured_response_invalid`; the analyzer is degraded and the skill counts +as incomplete. The scan continues and keeps findings from completed work. +Core logs `LLM structured response validation failed for ... retrying` and, +on exhaustion, `... after 4 attempts`, without logging the raw response. + +Patch 3 sanitizes known soft quirks before validation: null explanation and +remediation become empty strings, impact labels are case-folded, and unknown +impact labels use `low`. Invalid findings still fail validation. Optional prose +in `overall_assessment` is ignored so valid finding verdicts are retained. ## Gap-Fill Rule Selection Criteria diff --git a/contrib/batch_scan/docs/README.md b/contrib/batch_scan/docs/README.md index 3e0c58208..ad012f42e 100644 --- a/contrib/batch_scan/docs/README.md +++ b/contrib/batch_scan/docs/README.md @@ -341,10 +341,11 @@ fi classified as English and lose gap-fill coverage. 3. **No SARIF output.** Upstream supports it; this contrib adds terminal/JSON/Markdown. 4. **Gap-fill quality not benchmarked for non-English.** No ground-truth comparison exists. -5. **`parse_response` JSON recovery is best-effort.** When the LLM returns - malformed JSON, the analyzer returns empty findings (no crash). This is a - graceful-degradation choice: a single malformed response won't block the - pipeline, but the user won't know which findings were lost. +5. **Malformed provider responses reduce coverage.** Compatibility-mode discovery + and meta responses are retried up to four attempts, then recorded as + `llm_structured_response_invalid` and counted as incomplete. Valid findings + from completed batches are kept. Gap-fill parsing still drops malformed + responses until the separate gap-fill failure fix is included. See `DESIGN.md` for architecture details and `docs/archive/FUTURE_WORK.md` for suggested directions. diff --git a/contrib/batch_scan/runner.py b/contrib/batch_scan/runner.py index e6e3217ed..42f4945b0 100644 --- a/contrib/batch_scan/runner.py +++ b/contrib/batch_scan/runner.py @@ -37,8 +37,14 @@ from datetime import UTC, datetime from pathlib import Path +from pydantic import ValidationError + from skillspector.graph import graph -from skillspector.llm_analyzer_base import LLMAnalyzerBase, LLMAnalysisResult +from skillspector.llm_analyzer_base import ( + LLMAnalyzerBase, + LLMAnalysisResult, + _StructuredResponseValidationError, +) from skillspector.logging_config import get_logger from skillspector.nodes.meta_analyzer import LLMMetaAnalyzer, MetaAnalyzerResult @@ -85,13 +91,16 @@ def set_api_pool(pool: "ApiKeyPool | None") -> None: if _original_get_chat_model is None: _original_get_chat_model = _llm_utils.get_chat_model - def _pooled_get_chat_model(model=None): + def _pooled_get_chat_model(model=None, *, timeout=None): if _api_pool: from .api_pool import PooledChatModel - pooled_model = PooledChatModel(_api_pool) + pooled_model = PooledChatModel( + _api_pool, + timeout=_DEFAULT_REQUEST_TIMEOUT if timeout is None else timeout, + ) _llm_utils.register_chat_model_provider(pooled_model, "openai") return pooled_model - return _original_get_chat_model(model) + return _original_get_chat_model(model=model, timeout=timeout) _llm_utils.get_chat_model = _pooled_get_chat_model _llm_analyzer_base.get_chat_model = _pooled_get_chat_model @@ -124,7 +133,7 @@ def _pooled_get_chat_model(model=None): _original_base_init = LLMAnalyzerBase.__init__ -def _patched_base_init(self, base_prompt, model, *, node="llm_analyzer"): +def _patched_base_init(self, base_prompt, model, *, node="llm_analyzer", timeout=None): """Set response_schema=None on the instance dict BEFORE original init. Relies on Python MRO guarantee: instance.__dict__ is always checked @@ -132,7 +141,7 @@ def _patched_base_init(self, base_prompt, model, *, node="llm_analyzer"): a library internal. """ self.response_schema = None - _original_base_init(self, base_prompt, model, node=node) + _original_base_init(self, base_prompt, model, node=node, timeout=timeout) # -- Patch 2: LLMAnalyzerBase.parse_response handles raw JSON -------------- @@ -146,23 +155,11 @@ def _patched_base_parse(self, response, batch): text = _strip_markdown_fences(str(response)) try: data = json.loads(text) - except json.JSONDecodeError as exc: - logger.warning( - "LLMAnalyzerBase.parse_response: invalid JSON for %s: %s", - batch.file_label, - exc, - ) - return [] - try: result = LLMAnalysisResult.model_validate(data) - return [f.to_finding(batch.file_path) for f in result.findings] - except Exception as exc: - logger.warning( - "LLMAnalyzerBase.parse_response: schema validation failed for %s: %s", - batch.file_label, - exc, - ) - return [] + except (json.JSONDecodeError, ValidationError) as exc: + # Keep raw-response providers on the core retry and failure-ledger path. + raise _StructuredResponseValidationError from exc + return [f.to_finding(batch.file_path) for f in result.findings] # -- Patch 3: LLMMetaAnalyzer.parse_response handles raw JSON --------------- @@ -174,7 +171,9 @@ def _sanitize_meta_finding(d: dict) -> dict: for key in ("remediation", "explanation"): if d.get(key) is None: d[key] = "" - if d.get("impact") not in ("critical", "high", "medium", "low"): + impact = d.get("impact") + d["impact"] = impact.casefold() if isinstance(impact, str) else "low" + if d["impact"] not in ("critical", "high", "medium", "low"): d["impact"] = "low" return d @@ -186,28 +185,20 @@ def _patched_meta_parse(self, response, batch): text = _strip_markdown_fences(str(response)) try: data = json.loads(text) - except json.JSONDecodeError as exc: - logger.warning( - "LLMMetaAnalyzer.parse_response: invalid JSON for %s: %s", - batch.file_label, - exc, - ) - return [] - try: + if isinstance(data, dict) and isinstance(data.get("findings"), list): + data["findings"] = [ + _sanitize_meta_finding(dict(item)) if isinstance(item, dict) else item + for item in data["findings"] + ] result = MetaAnalyzerResult.model_validate(data) - items = [] - for f in result.findings: - d = _sanitize_meta_finding(f.model_dump()) - d["_file"] = batch.file_path - items.append(d) - return items - except Exception as exc: - logger.warning( - "LLMMetaAnalyzer.parse_response: schema validation failed for %s: %s", - batch.file_label, - exc, - ) - return [] + except (json.JSONDecodeError, ValidationError) as exc: + raise _StructuredResponseValidationError from exc + items = [] + for f in result.findings: + d = f.model_dump() + d["_file"] = batch.file_path + items.append(d) + return items # -- Patch 4: append JSON output format to base prompt --------------------- @@ -264,10 +255,14 @@ def _patched_meta_build_prompt(self, batch, **kwargs): def _patched_chatopenai_init(self, **kwargs): import httpx - _to = httpx.Timeout( - _DEFAULT_REQUEST_TIMEOUT, - connect=_DEFAULT_CONNECT_TIMEOUT, - ) + requested = kwargs.get("request_timeout", kwargs.get("timeout")) + _to = httpx.Timeout(_DEFAULT_REQUEST_TIMEOUT, connect=_DEFAULT_CONNECT_TIMEOUT) + if requested is not None: + requested = httpx.Timeout(requested).as_dict() + _to = httpx.Timeout(**{ + key: min(cap, requested[key]) if requested[key] is not None else cap + for key, cap in _to.as_dict().items() + }) # Set both the Pydantic alias AND the canonical field name so we don't # depend on alias-precedence behaviour (which is a Pydantic v2 internal). kwargs["timeout"] = _to @@ -327,12 +322,13 @@ def _verify_patch_targets() -> None: "LLMAnalyzerBase.__init__", 1, ) - _node_param = inspect.signature(LLMAnalyzerBase.__init__).parameters.get("node") - if _node_param is None or _node_param.kind != inspect.Parameter.KEYWORD_ONLY: - raise RuntimeError( - "Patch 1 target changed: LLMAnalyzerBase.__init__ must retain its " - "keyword-only 'node' parameter." - ) + for name in ("node", "timeout"): + parameter = inspect.signature(LLMAnalyzerBase.__init__).parameters.get(name) + if parameter is None or parameter.kind != inspect.Parameter.KEYWORD_ONLY: + raise RuntimeError( + "Patch 1 target changed: LLMAnalyzerBase.__init__ must retain its " + f"keyword-only '{name}' parameter." + ) if not hasattr(LLMAnalyzerBase, "response_schema"): raise RuntimeError( "Patch 1 target lost: LLMAnalyzerBase no longer has " @@ -347,7 +343,7 @@ def _verify_patch_targets() -> None: "LLMAnalyzerBase.parse_response", 2, ) - # Deep deps (called inside try/except — silent degradation if broken): + # Validate dependencies before patching rather than failing during a scan: if not hasattr(LLMAnalysisResult, "model_validate"): raise RuntimeError( "Patch 2 deep dependency lost: LLMAnalysisResult.model_validate " diff --git a/contrib/batch_scan/tests/docs/TEST_DESIGN.md b/contrib/batch_scan/tests/docs/TEST_DESIGN.md index 372c0b06b..e01d31170 100644 --- a/contrib/batch_scan/tests/docs/TEST_DESIGN.md +++ b/contrib/batch_scan/tests/docs/TEST_DESIGN.md @@ -95,7 +95,7 @@ It checks every assumption our patches depend on. If anything changed, it raises |------|-----------------| | **Guard passes current upstream** | Verifies no false positive. Tested against NVIDIA/SkillSpector@ab0431f (130+ commits, 89 files) — guard must not raise on the currently-installed upstream. Also tested after apply+restore cycle (state corruption check). | | **Each of 7 patches individually verified** | For each patch, we temporarily break its specific target and verify the guard catches it with the correct patch number in the error message. This proves every guard check is unique and distinguishable — an operator seeing "Patch 3" in the error knows exactly what broke. | -| **Deep dependency detection** | Beyond function signatures, our patches call `model_validate()`, `to_finding()`, `Batch.file_path`, `MetaAnalyzerResult.findings`, `asyncio.new_event_loop`. These are inside `try/except` blocks — if they silently disappear, the patch catches the exception and returns `[]`, masking the problem. The guard checks these BEFORE patching. | +| **Deep dependency detection** | Beyond function signatures, our patches call `model_validate()`, `to_finding()`, `Batch.file_path`, `MetaAnalyzerResult.findings`, `asyncio.new_event_loop`. Missing dependencies would fail during scanning; the guard checks them BEFORE patching so installed-core API drift is reported immediately. Malformed provider data follows the retry and incomplete-ledger path. | | **Keyword-only migration** | Python 3.x can change positional params to keyword-only. `_check_signature` detects `Parameter.KEYWORD_ONLY` kind and raises — our call sites pass these positionally. | | **Atomicity** | Guard failure must leave the process in its original state. We break a target, call `_apply_patches()`, and verify all 5 methods are still originals — the guard raised before any assignment happened. | diff --git a/contrib/batch_scan/tests/test_monkeypatch_fragility.py b/contrib/batch_scan/tests/test_monkeypatch_fragility.py index 950cd0778..77cc33744 100644 --- a/contrib/batch_scan/tests/test_monkeypatch_fragility.py +++ b/contrib/batch_scan/tests/test_monkeypatch_fragility.py @@ -263,22 +263,28 @@ def _broken_init(self, base_prompt, model): LLMAnalyzerBase.__init__ = original def test_patched_init_forwards_keyword_only_node(self) -> None: - instance = SimpleNamespace() - with patch("contrib.batch_scan.runner._original_base_init") as original_init: - _patched_base_init( - instance, - "prompt", - "model", - node="semantic_security_discovery", - ) - - original_init.assert_called_once_with( - instance, - "prompt", - "model", - node="semantic_security_discovery", - ) - self.assertIsNone(instance.response_schema) + for timeout in (None, 7, lambda: 7): + with self.subTest(timeout=timeout): + instance = SimpleNamespace() + with patch("contrib.batch_scan.runner._original_base_init") as original_init: + _patched_base_init( + instance, "prompt", "model", + node="semantic_security_discovery", timeout=timeout, + ) + original_init.assert_called_once_with( + instance, "prompt", "model", + node="semantic_security_discovery", timeout=timeout, + ) + self.assertIs(original_init.call_args.kwargs["timeout"], timeout) + self.assertIsNone(instance.response_schema) + + def test_guard_catches_missing_timeout_param(self) -> None: + def without_timeout(self, base_prompt, model, *, node="llm_analyzer"): + pass + + with patch.object(LLMAnalyzerBase, "__init__", without_timeout): + with self.assertRaisesRegex(RuntimeError, "timeout"): + _verify_patch_targets() def test_guard_catches_missing_response_schema_attr(self) -> None: """If upstream removes response_schema class attr, guard must raise.""" diff --git a/src/skillspector/llm_analyzer_base.py b/src/skillspector/llm_analyzer_base.py index 4543ab0d3..cc661f307 100644 --- a/src/skillspector/llm_analyzer_base.py +++ b/src/skillspector/llm_analyzer_base.py @@ -610,7 +610,10 @@ def _clamp_start_line(cls, v: int) -> int: @classmethod def _normalize_confidence(cls, v: object) -> float: # Accept 0-100 scale values from some models, then clamp into [0, 1]. - value = float(cast(Any, v)) + try: + value = float(cast(Any, v)) + except (TypeError, ValueError) as exc: + raise ValueError("confidence must be numeric") from exc if value > 2.0: value = value / 100.0 return min(1.0, max(0.0, value)) diff --git a/src/skillspector/nodes/meta_analyzer.py b/src/skillspector/nodes/meta_analyzer.py index b9909f173..252134d9d 100644 --- a/src/skillspector/nodes/meta_analyzer.py +++ b/src/skillspector/nodes/meta_analyzer.py @@ -97,7 +97,10 @@ class MetaAnalyzerFinding(BaseModel): @classmethod def _normalize_confidence(cls, v: object) -> float: # Accept 0-100 scale values from some models, then clamp into [0, 1]. - value = float(v) # type: ignore[arg-type] + try: + value = float(v) # type: ignore[arg-type] + except (TypeError, ValueError) as exc: + raise ValueError("confidence must be numeric") from exc if value > 2.0: value = value / 100.0 return min(1.0, max(0.0, value)) @@ -130,21 +133,17 @@ class MetaAnalyzerResult(BaseModel): def _parse_stringified_findings(cls, v: object) -> object: """LLMs sometimes return the findings array as a JSON string.""" if isinstance(v, str): - try: - parsed = json.loads(v) - except (json.JSONDecodeError, TypeError): - return [] - return parsed if isinstance(parsed, list) else [] + return json.loads(v) return v @field_validator("overall_assessment", mode="before") @classmethod def _parse_stringified_assessment(cls, v: object) -> object: - """LLMs sometimes return nested objects as JSON strings.""" + """Ignore optional prose summaries without losing valid finding verdicts.""" if isinstance(v, str): try: return json.loads(v) - except (json.JSONDecodeError, TypeError): + except json.JSONDecodeError: return None return v diff --git a/tests/nodes/test_llm_analyzer_base.py b/tests/nodes/test_llm_analyzer_base.py index 8a53db58f..f73418166 100644 --- a/tests/nodes/test_llm_analyzer_base.py +++ b/tests/nodes/test_llm_analyzer_base.py @@ -992,13 +992,13 @@ def test_findings_as_native_list(self) -> None: result = MetaAnalyzerResult.model_validate({"findings": [self._FINDING]}) assert len(result.findings) == 1 - def test_findings_invalid_string_yields_empty(self) -> None: - result = MetaAnalyzerResult.model_validate({"findings": "not json"}) - assert result.findings == [] + def test_findings_invalid_string_rejected(self) -> None: + with pytest.raises(ValidationError): + MetaAnalyzerResult.model_validate({"findings": "not json"}) - def test_findings_non_list_json_yields_empty(self) -> None: - result = MetaAnalyzerResult.model_validate({"findings": json.dumps({"a": 1})}) - assert result.findings == [] + def test_findings_non_list_json_rejected(self) -> None: + with pytest.raises(ValidationError): + MetaAnalyzerResult.model_validate({"findings": json.dumps({"a": 1})}) # --------------------------------------------------------------------------- diff --git a/tests/nodes/test_meta_analyzer.py b/tests/nodes/test_meta_analyzer.py index 4df095d04..c2896f0db 100644 --- a/tests/nodes/test_meta_analyzer.py +++ b/tests/nodes/test_meta_analyzer.py @@ -24,6 +24,8 @@ from unittest.mock import AsyncMock, MagicMock, patch +import pytest + from skillspector.inspection_ledger import LedgerOutcome, LedgerReason, finalize_ledger from skillspector.llm_analyzer_base import Batch, BatchExecutionResult, BatchFailure from skillspector.models import Finding @@ -31,6 +33,7 @@ from skillspector.nodes.analyzers.static_runner import analyzer_finding_to_finding from skillspector.nodes.meta_analyzer import ( LLMMetaAnalyzer, + MetaAnalyzerResult, _meta_ledger_response, meta_analyzer, ) @@ -1092,7 +1095,46 @@ def test_use_llm_false_records_nothing() -> None: assert "filtered_findings" not in result +@pytest.mark.parametrize("value", ["invalid JSON", "", "{not json", "null", "42", '"wrong type"']) +def test_invalid_stringified_meta_findings_fail_validation(value) -> None: + from pydantic import ValidationError + + with pytest.raises(ValidationError): + MetaAnalyzerResult.model_validate({"findings": value}) + + +def test_valid_stringified_meta_fields_are_supported() -> None: + result = MetaAnalyzerResult.model_validate( + { + "findings": "[]", + "overall_assessment": '{"risk_level": "LOW", "summary": "No issues found"}', + } + ) + assert result.findings == [] + assert result.overall_assessment.summary == "No issues found" + + def test_no_findings_records_nothing() -> None: result = meta_analyzer(_degr_state(findings=[])) assert "llm_call_log" not in result assert "filtered_findings" not in result + + +@pytest.mark.parametrize("assessment", ["HIGH risk: exfiltrates credentials", "LOW", ""]) +def test_optional_prose_assessment_keeps_finding_verdict(assessment): + verdict = { + "pattern_id": "P1", + "is_vulnerability": True, + "confidence": 0.9, + "intent": "malicious", + "impact": "high", + "explanation": "Unsafe instruction", + "remediation": "Remove instruction", + } + result = MetaAnalyzerResult.model_validate( + {"findings": [verdict], "overall_assessment": assessment} + ) + assert len(result.findings) == 1 + assert result.findings[0].pattern_id == "P1" + assert result.findings[0].is_vulnerability is True + assert result.overall_assessment is None diff --git a/tests/test_batch_scan_security.py b/tests/test_batch_scan_security.py index b1c7577f0..2bfa498a1 100644 --- a/tests/test_batch_scan_security.py +++ b/tests/test_batch_scan_security.py @@ -29,6 +29,91 @@ _SAFE_TEXT = "# 安全助手\n这是一个帮助用户整理资料的安全技能。\n" +@pytest.mark.parametrize("analyzer_kind", ["discovery", "meta"]) +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize( + "invalid_response", + ["not JSON", '{"findings": "invalid"}', "null-confidence", "list-confidence"], +) +@pytest.mark.parametrize("recovers", [False, True]) +async def test_compat_parse_failures_retry_and_remain_visible( + monkeypatch: pytest.MonkeyPatch, + analyzer_kind: str, + asynchronous: bool, + invalid_response: str, + recovers: bool, +) -> None: + if invalid_response in {"null-confidence", "list-confidence"}: + confidence = None if invalid_response == "null-confidence" else [1] + item = ( + { + "rule_id": "P1", + "message": "Unsafe instruction", + "severity": "HIGH", + "start_line": 1, + "explanation": "Unsafe", + "remediation": "Remove", + } + if analyzer_kind == "discovery" + else { + "pattern_id": "P1", + "is_vulnerability": True, + "intent": "malicious", + "impact": "high", + "explanation": "Unsafe", + "remediation": "Remove", + } + ) + invalid_response = json.dumps({"findings": [dict(item, confidence=confidence)]}) + calls = [] + + def invoke(prompt): + calls.append(prompt) + response = '{"findings": []}' if recovers and len(calls) > 1 else invalid_response + return AIMessage(content=response) + + async def ainvoke(prompt): + return invoke(prompt) + + monkeypatch.setattr( + llm_analyzer_base, + "get_chat_model", + lambda **kwargs: SimpleNamespace(invoke=invoke, ainvoke=ainvoke), + ) + monkeypatch.setattr(llm_analyzer_base, "get_max_input_tokens", lambda model: 100_000) + monkeypatch.setattr(llm_analyzer_base, "STRUCTURED_RESPONSE_RETRY_DELAYS_SECONDS", (0, 0, 0)) + batch = llm_analyzer_base.Batch(file_path="SKILL.md", content=_SAFE_TEXT) + + with runner.deepseek_compat(): + analyzer = ( + runner.LLMAnalyzerBase(base_prompt="Review the supplied skill", model="test", timeout=7) + if analyzer_kind == "discovery" + else runner.LLMMetaAnalyzer(model="test", timeout=7) + ) + assert analyzer._timeout == 7 + outcome = ( + await analyzer.arun_batches_detailed([batch]) + if asynchronous + else analyzer.run_batches_detailed([batch]) + ) + + if recovers: + assert len(calls) == 2 + assert outcome.successful == [(batch, [])] + assert outcome.failures == [] + else: + assert len(calls) == llm_analyzer_base.STRUCTURED_RESPONSE_MAX_ATTEMPTS + assert outcome.successful == [] + assert len(outcome.failures) == 1 + assert outcome.failures[0].reason.value == "llm_structured_response_invalid" + events, _ = llm_analyzer_base.ledger_events_for_batches("compat-test", outcome) + assert len(events) == 1 + # Core policy records malformed output as skipped, so coverage is partial. + assert events[0]["outcome"].value == "skipped" + assert events[0]["reason_code"] == "llm_structured_response_invalid" + assert invalid_response not in json.dumps(events) + + @pytest.fixture def batch_skill(tmp_path: Path) -> tuple[Path, Path]: skill = tmp_path / "safe-skill" @@ -541,3 +626,148 @@ def test_scan_forwards_verbose_logging(batch_skill, monkeypatch): skill, skill.parent, use_llm=False, lang="en", require_llm=False, verbose=True ) assert levels == ["DEBUG"] + + +@pytest.mark.parametrize("impact", ["High", "none", None, "catastrophic"]) +def test_compat_meta_repairs_soft_fields_before_validation(impact): + batch = llm_analyzer_base.Batch(file_path="SKILL.md", content=_SAFE_TEXT) + result = runner._patched_meta_parse( + None, + json.dumps( + { + "findings": [ + { + "pattern_id": "P1", + "is_vulnerability": True, + "confidence": 0.9, + "intent": "malicious", + "impact": impact, + "explanation": None, + "remediation": None, + } + ] + } + ), + batch, + ) + assert len(result) == 1 + assert result[0]["impact"] == ("high" if impact == "High" else "low") + assert result[0]["explanation"] == result[0]["remediation"] == "" + assert result[0]["_file"] == "SKILL.md" + + +@pytest.mark.parametrize("timeout", [None, 7, lambda: 7]) +def test_compat_pooled_constructor_forwards_timeout(monkeypatch, timeout): + from contrib.batch_scan.api_pool import PooledChatModel + + pool = object() + monkeypatch.setattr(llm_analyzer_base, "get_max_input_tokens", lambda model: 100_000) + # Register and construct the real adapter; invoking it would require a real pool. + try: + runner.set_api_pool(pool) + with runner.deepseek_compat(): + analyzer = runner.LLMAnalyzerBase("Review", "test", timeout=timeout) + assert isinstance(analyzer._llm, PooledChatModel) + assert analyzer._llm._pool is pool + assert analyzer._llm._timeout == (30 if timeout is None else 7) + finally: + runner.set_api_pool(None) + + +@pytest.mark.parametrize("requested", [2, 60, None]) +def test_compat_http_timeout_preserves_shorter_deadline(monkeypatch, requested): + observed = {} + monkeypatch.setattr(runner, "_original_chatopenai_init", lambda self, **kw: observed.update(kw)) + runner._patched_chatopenai_init(object(), timeout=requested) + assert observed["timeout"] is observed["request_timeout"] + assert observed["timeout"].read == (2 if requested == 2 else 30) + assert observed["timeout"].connect == (2 if requested == 2 else 8) + + +@pytest.mark.parametrize("asynchronous", [False, True]) +async def test_pooled_request_slot_wait_obeys_deadline(asynchronous): + from contrib.batch_scan.api_pool import ApiKey, ApiKeyPool, PooledChatModel + + pool = ApiKeyPool([ApiKey("synthetic", None, "test", max_concurrent=1)]) + occupied = pool.acquire() + model = PooledChatModel(pool, timeout=0.02) + started = time.monotonic() + try: + with pytest.raises(llm_analyzer_base.LLMRuntimeLimitError): + if asynchronous: + await model.ainvoke("unused") + else: + model.invoke("unused") + assert time.monotonic() - started < 1 + assert occupied.active_requests == 1 + finally: + pool.release(occupied) + assert occupied.active_requests == 0 + + +@pytest.mark.parametrize("asynchronous", [False, True]) +async def test_pooled_request_releases_slot_when_client_build_fails(monkeypatch, asynchronous): + from contrib.batch_scan.api_pool import ApiKey, ApiKeyPool, PooledChatModel + + key = ApiKey("synthetic", None, "test", max_concurrent=1) + model = PooledChatModel(ApiKeyPool([key]), timeout=1) + + def fail(*args, **kwargs): + raise ValueError("synthetic constructor failure") + + monkeypatch.setattr(model, "_build_llm", fail) + with pytest.raises(ValueError, match="synthetic constructor failure"): + if asynchronous: + await model.ainvoke("unused") + else: + model.invoke("unused") + assert key.active_requests == 0 + + +@pytest.mark.parametrize("asynchronous", [False, True]) +async def test_pooled_key_retries_share_one_deadline(monkeypatch, asynchronous): + import contrib.batch_scan.api_pool as api_pool + + clock = [0.0] + monkeypatch.setattr(api_pool, "time", SimpleNamespace(monotonic=lambda: clock[0])) + limits, releases = [], [] + key = object() + pool = SimpleNamespace( + acquire=lambda **kwargs: key, + try_acquire=lambda: key, + release=lambda key, **kwargs: releases.append(kwargs), + ) + model = api_pool.PooledChatModel(pool, timeout=2, max_retries=5) + + def invoke(prompt): + clock[0] += 1 + raise RuntimeError("429 rate limit") + + async def ainvoke(prompt): + return invoke(prompt) + + def build(key, *, timeout): + limits.append(timeout) + return SimpleNamespace(invoke=invoke, ainvoke=ainvoke) + + monkeypatch.setattr(model, "_build_llm", build) + with pytest.raises(llm_analyzer_base.LLMRuntimeLimitError): + if asynchronous: + await model.ainvoke("unused") + else: + model.invoke("unused") + assert limits == [2, 1] + assert releases == [{"success": False}, {"success": False}] + + +def test_pooled_client_bounds_connect_and_disables_sdk_retries(monkeypatch): + import langchain_openai + + from contrib.batch_scan.api_pool import ApiKey, ApiKeyPool, PooledChatModel + + key = ApiKey("synthetic", None, "test") + observed = {} + monkeypatch.setattr(langchain_openai, "ChatOpenAI", lambda **kw: observed.update(kw)) + PooledChatModel(ApiKeyPool([key]))._build_llm(key, timeout=0.1) + assert observed["timeout"].connect == observed["timeout"].read == 0.1 + assert observed["max_retries"] == 0