diff --git a/src/skillspector/cli.py b/src/skillspector/cli.py index 4bb3e34d..5fb880cb 100644 --- a/src/skillspector/cli.py +++ b/src/skillspector/cli.py @@ -294,6 +294,7 @@ def _scan_state( "input_path": input_path, "output_format": format.value, "use_llm": not no_llm, + "llm_requested": not no_llm, } if yara_rules_dir is not None: state["yara_rules_dir"] = yara_rules_dir diff --git a/src/skillspector/graph.py b/src/skillspector/graph.py index 805769d4..54701da3 100644 --- a/src/skillspector/graph.py +++ b/src/skillspector/graph.py @@ -24,7 +24,6 @@ from langgraph.graph import END, START, StateGraph from skillspector.inspection_ledger import guard_analyzer_node -from skillspector.llm_utils import is_llm_available from skillspector.logging_config import get_logger from skillspector.nodes.analyzers import ANALYZER_MODULES, ANALYZER_NODE_IDS, ANALYZER_NODES from skillspector.nodes.build_context import build_context @@ -60,13 +59,6 @@ def create_graph(): logger.warning("Skipping analyzer %s: is_available() returned False", analyzer_id) continue - requires_api_key = getattr(mod, "requires_api_key", False) - if requires_api_key: - has_llm, _ = is_llm_available() - if not has_llm: - logger.warning("Skipping analyzer %s: required API key is missing", analyzer_id) - continue - workflow.add_node( analyzer_id, guard_analyzer_node(analyzer_id, ANALYZER_NODES[analyzer_id]) ) diff --git a/src/skillspector/mcp_server.py b/src/skillspector/mcp_server.py index ef2dd351..dce2a757 100644 --- a/src/skillspector/mcp_server.py +++ b/src/skillspector/mcp_server.py @@ -27,6 +27,7 @@ from __future__ import annotations +from collections.abc import Mapping from pathlib import Path from typing import TYPE_CHECKING, Any @@ -36,6 +37,8 @@ from skillspector.graph import graph from skillspector.llm_utils import is_llm_available from skillspector.logging_config import get_logger +from skillspector.nodes.analyzers import ANALYZER_MODULES +from skillspector.semantic_runtime import llm_runtime_available, semantic_runtime_accounting from skillspector.suppression import effective_findings if TYPE_CHECKING: @@ -46,6 +49,23 @@ VALID_FORMATS = ("json", "markdown", "sarif", "terminal") +def _llm_runtime_accounting(*, enabled: bool, result: Mapping[str, object]) -> tuple[bool, bool]: + """Apply shared semantic runtime accounting to the discovered registry.""" + return semantic_runtime_accounting( + enabled=enabled, + result=result, + discovered_modules=ANALYZER_MODULES, + ) + + +def _llm_runtime_available(*, preflight_available: bool, result: Mapping[str, object]) -> bool: + """Apply shared provider and meta-analysis runtime availability.""" + return llm_runtime_available( + preflight_available=preflight_available, + result=result, + ) + + def _is_local_target(target: str) -> bool: """Return True when ``target`` names local filesystem content.""" stripped = target.strip() @@ -105,22 +125,23 @@ async def run_scan( if local_target or local_yara_rules: raise ValueError("local targets are disabled for this MCP transport") - llm_available, _ = is_llm_available() - llm_used = use_llm and llm_available + llm_preflight_available, _ = is_llm_available() + llm_enabled = use_llm and llm_preflight_available state: dict[str, Any] = { "input_path": target, "output_format": output_format, - "use_llm": llm_used, + "use_llm": llm_enabled, + "llm_requested": use_llm, } if yara_rules_dir: state["yara_rules_dir"] = yara_rules_dir logger.debug( - "MCP scan started: target=%s, format=%s, llm_used=%s", + "MCP scan started: target=%s, format=%s, llm_enabled=%s", target, output_format, - llm_used, + llm_enabled, ) result: dict[str, Any] | None = None @@ -132,7 +153,8 @@ async def run_scan( "tags": ["skillspector", "mcp"], "metadata": { "input_path": target, - "use_llm": llm_used, + "use_llm": llm_enabled, + "llm_requested": use_llm, "output_format": output_format, "version": __version__, }, @@ -140,20 +162,30 @@ async def run_scan( ) findings = effective_findings(result) risk_score = int(result.get("risk_score") or 0) + llm_used, llm_runtime_complete = _llm_runtime_accounting(enabled=llm_enabled, result=result) + llm_available = _llm_runtime_available( + preflight_available=llm_preflight_available, + result=result, + ) execution_successful = bool(result.get("execution_successful", True)) analysis_completeness = result.get("analysis_completeness") or {} entirely_uninspected = int(analysis_completeness.get("entirely_uninspected_files", 0)) + analysis_requirement_met = not use_llm or llm_runtime_complete safe_to_install = ( risk_score <= RISK_THRESHOLD and execution_successful and entirely_uninspected == 0 and bool(analysis_completeness.get("is_complete", True)) + and analysis_requirement_met ) + recommendation = result.get("risk_recommendation") + if not analysis_requirement_met and recommendation == "SAFE": + recommendation = "CAUTION" return { "target": target, "risk_score": risk_score, "severity": result.get("risk_severity"), - "recommendation": result.get("risk_recommendation"), + "recommendation": recommendation, "safe_to_install": safe_to_install, "execution_successful": execution_successful, "analysis_completeness": analysis_completeness, diff --git a/src/skillspector/multi_skill.py b/src/skillspector/multi_skill.py index fea6d869..cc232eb2 100644 --- a/src/skillspector/multi_skill.py +++ b/src/skillspector/multi_skill.py @@ -283,7 +283,7 @@ def detect_skills(directory: Path) -> MultiSkillDetectionResult: continue except OSError as exc: raise _read_error("multi_skill_directory_entry") from exc - if entry.name in _SKIP_DIRS or entry.name.startswith("."): + if entry.name in _SKIP_DIRS: continue has_manifest = _has_skill_md(child, budget=budget) diff --git a/src/skillspector/nodes/report.py b/src/skillspector/nodes/report.py index ab4b814d..63daf92d 100644 --- a/src/skillspector/nodes/report.py +++ b/src/skillspector/nodes/report.py @@ -41,6 +41,7 @@ from skillspector.llm_utils import is_llm_available from skillspector.logging_config import get_logger from skillspector.models import Finding +from skillspector.nodes.analyzers import ANALYZER_MODULES from skillspector.nodes.deduplicate import deduplicate from skillspector.python_ast import clear_python_ast_cache from skillspector.sarif_models import ( @@ -61,6 +62,11 @@ SarifTool, validate_sarif_report, ) +from skillspector.semantic_runtime import ( + llm_runtime_available, + semantic_runtime_accounting, + successful_llm_record, +) from skillspector.state import SkillspectorState from skillspector.suppression import Baseline, SuppressedFinding, partition_findings @@ -879,6 +885,7 @@ def _format_terminal( has_executable_scripts: bool, use_llm: bool = True, llm_call_log: Sequence[Mapping[str, object]] | None = None, + degraded_notice: str | None = None, suppressed: list[SuppressedFinding] | None = None, structured_summaries: list[dict[str, object]] | None = None, show_suppressed: bool = False, @@ -937,12 +944,14 @@ def _format_terminal( comp_table.add_row(f"... and {len(component_metadata) - 15} more", "", "", "") console.print(comp_table) - degraded_notice = _llm_degradation_notice(use_llm, llm_call_log or []) - if degraded_notice: + effective_degraded_notice = degraded_notice or _llm_degradation_notice( + use_llm, llm_call_log or [] + ) + if effective_degraded_notice: console.print() console.print( Panel( - f"[bold]Degraded scan[/bold]\n{degraded_notice}", + f"[bold]Degraded scan[/bold]\n{effective_degraded_notice}", title="[bold red]WARNING[/bold red]", border_style="red", ) @@ -1032,7 +1041,7 @@ def _llm_runtime_status( pass is degraded too, not just a total one. """ attempted = len(llm_call_log) - succeeded = sum(1 for r in llm_call_log if r.get("ok")) + succeeded = sum(1 for record in llm_call_log if successful_llm_record(record)) degraded = bool(use_llm and attempted > 0 and succeeded < attempted) return attempted, succeeded, degraded @@ -1059,12 +1068,14 @@ def _build_metadata( transitive_targets_scanned: int | None = None, transitive_bytes_scanned: int | None = None, transitive_truncation_reasons: Sequence[str] | None = None, + llm_execution_enabled: bool | None = None, + semantic_runtime_incomplete: bool = False, + runtime_available: bool | None = None, ) -> dict[str, object]: """Build the metadata section shared by all output formats.""" llm_call_log = llm_call_log or [] provider_available, llm_error = is_llm_available() - attempted, succeeded, degraded = _llm_runtime_status(use_llm, llm_call_log) - + attempted, succeeded, call_log_degraded = _llm_runtime_status(use_llm, llm_call_log) # meta_analyzer's own record, independent of whether a DIFFERENT # LLM-backed node (a semantic_* analyzer) lost coverage to a dropped # batch. A missing record means meta_analyzer never ran (e.g. there were @@ -1072,13 +1083,16 @@ def _build_metadata( # vacuously ok for llm_available (provider/runtime truth). When it does # run it always emits exactly one record. meta_analyzer_records = [r for r in llm_call_log if r.get("node") == "meta_analyzer"] - meta_analyzer_ok = all(bool(r.get("ok")) for r in meta_analyzer_records) + meta_analyzer_ok = all(successful_llm_record(record) for record in meta_analyzer_records) # meta_analysis_applied is stricter: "did meta-analysis actually run" # cannot be satisfied vacuously. all([]) is True on an empty list, so # meta_analyzer_ok alone is also True when meta_analyzer made no call at # all (the no-findings path) - require at least one record, and that # record must have succeeded. meta_analyzer_succeeded = bool(meta_analyzer_records) and meta_analyzer_ok + effective_runtime_available = ( + provider_available and meta_analyzer_ok if runtime_available is None else runtime_available + ) # meta_analysis_applied / llm_available answer different questions. # llm_available is provider availability: the binary/credentials were @@ -1092,7 +1106,11 @@ def _build_metadata( # llm_calls_succeeded, and must not flip these two fields on its own - # that would conflate two independent contracts (meta-analysis ran vs. # some coverage was lost) into one boolean. - meta_analysis_applied = use_llm and provider_available and meta_analyzer_succeeded + execution_enabled = use_llm if llm_execution_enabled is None else llm_execution_enabled + unavailable_before_execution = bool(use_llm and not execution_enabled) + meta_analysis_applied = ( + use_llm and execution_enabled and provider_available and meta_analyzer_succeeded + ) meta: dict[str, object] = { "has_executable_scripts": has_executable_scripts, @@ -1100,7 +1118,7 @@ def _build_metadata( "llm_requested": use_llm, # llm_available reflects runtime truth: the binary/credentials were # available AND meta_analyzer's own call (if it ran) succeeded. - "llm_available": provider_available and meta_analyzer_ok, + "llm_available": (effective_runtime_available and not unavailable_before_execution), "meta_analysis_applied": meta_analysis_applied, # A list (including an empty list) makes observability explicit. Empty # means the provider/transport supplied no counters; it is never an @@ -1112,10 +1130,19 @@ def _build_metadata( if use_llm and attempted: meta["llm_calls_attempted"] = attempted meta["llm_calls_succeeded"] = succeeded - if degraded: + if unavailable_before_execution: + meta["llm_error"] = ( + "LLM analysis was requested but unavailable during preflight; " + "results reflect static analysis only." + ) + elif call_log_degraded: meta["llm_degraded"] = True reasons = sorted( - {str(r.get("error")) for r in llm_call_log if not r.get("ok") and r.get("error")} + { + str(record.get("error")) + for record in llm_call_log + if not successful_llm_record(record) and record.get("error") + } ) detail = f" Reasons: {'; '.join(reasons)}" if reasons else "" failed = attempted - succeeded @@ -1123,6 +1150,12 @@ def _build_metadata( f"LLM analysis was requested but {failed} of {attempted} LLM call(s) failed; " f"results reflect static analysis only for the affected batch(es).{detail}" ) + elif semantic_runtime_incomplete: + meta["llm_degraded"] = True + meta["llm_error"] = ( + "LLM analysis was requested but semantic runtime telemetry was incomplete; " + "results may reflect static analysis only." + ) elif use_llm and not provider_available: meta["llm_error"] = llm_error if transitive_targets_scanned is not None: @@ -1154,6 +1187,9 @@ def _format_json( transitive_bytes_scanned: int | None = None, transitive_truncation_reasons: Sequence[str] | None = None, structured_summaries: list[dict[str, object]] | None = None, + llm_execution_enabled: bool | None = None, + semantic_runtime_incomplete: bool = False, + runtime_available: bool | None = None, ) -> str: """Generate JSON report string.""" suppressed = suppressed or [] @@ -1195,6 +1231,9 @@ def _format_json( transitive_targets_scanned, transitive_bytes_scanned, transitive_truncation_reasons, + llm_execution_enabled, + semantic_runtime_incomplete, + runtime_available, ), "execution_successful": execution_successful, } @@ -1273,6 +1312,7 @@ def _format_markdown( has_executable_scripts: bool, use_llm: bool = True, llm_call_log: Sequence[Mapping[str, object]] | None = None, + degraded_notice: str | None = None, suppressed: list[SuppressedFinding] | None = None, structured_summaries: list[dict[str, object]] | None = None, show_suppressed: bool = False, @@ -1291,9 +1331,11 @@ def _format_markdown( lines.append(f"**Scanned:** {datetime.now(UTC).strftime('%Y-%m-%d %H:%M:%S UTC')} ") lines.append("") - degraded_notice = _llm_degradation_notice(use_llm, llm_call_log or []) - if degraded_notice: - lines.append(f"> ⚠️ **Degraded scan:** {degraded_notice}") + effective_degraded_notice = degraded_notice or _llm_degradation_notice( + use_llm, llm_call_log or [] + ) + if effective_degraded_notice: + lines.append(f"> ⚠️ **Degraded scan:** {effective_degraded_notice}") lines.append("") lines.append("## Risk Assessment\n") @@ -1442,8 +1484,15 @@ def report(state: SkillspectorState) -> dict[str, object]: manifest = state.get("manifest") or {} skill_path = state.get("skill_path") output_format = state.get("output_format") or "sarif" - use_llm = state.get("use_llm", True) - llm_call_log = state.get("llm_call_log") or [] + use_llm = state.get("use_llm") is not False + raw_llm_requested = state.get("llm_requested") + llm_requested = raw_llm_requested if isinstance(raw_llm_requested, bool) else use_llm + raw_llm_call_log = state.get("llm_call_log") + llm_call_log: list[Mapping[str, object]] = ( + [record for record in raw_llm_call_log if isinstance(record, Mapping)] + if isinstance(raw_llm_call_log, list) + else [] + ) inference_usage = state.get("inference_usage") or [] transitive_targets_scanned = state.get("transitive_targets_scanned") transitive_bytes_scanned = state.get("transitive_bytes_scanned") @@ -1462,18 +1511,57 @@ def report(state: SkillspectorState) -> dict[str, object]: analysis_completeness["limitations"] = limitations analysis_completeness["is_complete"] = False - _attempted, _succeeded, degraded = _llm_runtime_status(use_llm, llm_call_log) + _llm_used, semantic_runtime_complete = semantic_runtime_accounting( + enabled=bool(llm_requested and use_llm), + result=state, + discovered_modules=ANALYZER_MODULES, + ) + semantic_runtime_incomplete = bool(llm_requested and use_llm and not semantic_runtime_complete) + _attempted, _succeeded, degraded = _llm_runtime_status(llm_requested, llm_call_log) provider_available, provider_error = is_llm_available() - has_recorded_failure = any(not r.get("ok") for r in llm_call_log) - provider_unavailable = bool(use_llm and not provider_available and has_recorded_failure) - degraded = degraded or provider_unavailable - degraded_notice = _llm_degradation_notice(use_llm, llm_call_log) - if provider_unavailable and degraded_notice is None: + runtime_available = llm_runtime_available( + preflight_available=provider_available, + result=state, + ) + has_recorded_failure = any(not successful_llm_record(record) for record in llm_call_log) + unavailable_before_execution = bool(llm_requested and not use_llm) + provider_unavailable = bool( + llm_requested + and not provider_available + and (has_recorded_failure or unavailable_before_execution) + ) + degraded = ( + degraded + or provider_unavailable + or unavailable_before_execution + or semantic_runtime_incomplete + ) + degraded_notice = _llm_degradation_notice(llm_requested, llm_call_log) + if unavailable_before_execution: + degraded_notice = ( + "LLM analysis was requested but unavailable during preflight; " + "results reflect STATIC analysis only." + ) + elif provider_unavailable and degraded_notice is None: degraded_notice = ( "LLM analysis was requested but the configured provider was unavailable" f" ({provider_error or 'unknown reason'}); results may reflect static analysis only." ) - if degraded: + elif semantic_runtime_incomplete and degraded_notice is None: + degraded_notice = ( + "LLM analysis was requested but semantic runtime telemetry was incomplete; " + "results may reflect static analysis only." + ) + if unavailable_before_execution: + logger.warning( + "LLM stage unavailable during preflight; report reflects static analysis only" + ) + elif semantic_runtime_incomplete: + logger.warning( + "LLM stage degraded: semantic runtime telemetry was incomplete; " + "report may reflect static analysis only" + ) + elif degraded: logger.warning( "LLM stage degraded: %d/%d LLM call(s) failed; report reflects static analysis only", _attempted - _succeeded, @@ -1545,8 +1633,9 @@ def report(state: SkillspectorState) -> dict[str, object]: risk_severity, risk_recommendation, has_executable_scripts, - use_llm=use_llm, + use_llm=llm_requested, llm_call_log=llm_call_log, + degraded_notice=degraded_notice, suppressed=suppressed, structured_summaries=structured_summaries, show_suppressed=show_suppressed, @@ -1563,7 +1652,7 @@ def report(state: SkillspectorState) -> dict[str, object]: risk_severity, risk_recommendation, has_executable_scripts, - use_llm=use_llm, + use_llm=llm_requested, llm_call_log=llm_call_log, inference_usage=inference_usage, analysis_completeness=analysis_completeness, @@ -1579,6 +1668,9 @@ def report(state: SkillspectorState) -> dict[str, object]: ), transitive_truncation_reasons=transitive_truncation_reasons, structured_summaries=structured_summaries, + llm_execution_enabled=use_llm, + semantic_runtime_incomplete=semantic_runtime_incomplete, + runtime_available=runtime_available, ) elif output_format == "markdown": report_body = _format_markdown( @@ -1590,8 +1682,9 @@ def report(state: SkillspectorState) -> dict[str, object]: risk_severity, risk_recommendation, has_executable_scripts, - use_llm=use_llm, + use_llm=llm_requested, llm_call_log=llm_call_log, + degraded_notice=degraded_notice, suppressed=suppressed, structured_summaries=structured_summaries, show_suppressed=show_suppressed, diff --git a/src/skillspector/semantic_runtime.py b/src/skillspector/semantic_runtime.py new file mode 100644 index 00000000..96bd6c26 --- /dev/null +++ b/src/skillspector/semantic_runtime.py @@ -0,0 +1,137 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Provider-independent semantic analyzer runtime accounting.""" + +from __future__ import annotations + +from collections.abc import Mapping + +CANONICAL_SEMANTIC_ANALYZER_IDS = frozenset( + { + "semantic_developer_intent", + "semantic_quality_policy", + "semantic_security_discovery", + } +) + + +def required_semantic_analyzer_ids( + discovered_modules: Mapping[str, object], +) -> frozenset[str]: + """Return stable requirements plus newly discovered credential-gated analyzers.""" + discovered = frozenset( + analyzer_id + for analyzer_id, module in discovered_modules.items() + if getattr(module, "requires_api_key", False) + ) + return CANONICAL_SEMANTIC_ANALYZER_IDS | discovered + + +def successful_llm_record(record: object) -> bool: + """Return whether ``record`` is a well-formed successful LLM call.""" + return ( + isinstance(record, Mapping) + and isinstance(record.get("node"), str) + and bool(record.get("node")) + and record.get("ok") is True + and record.get("error") is None + ) + + +def _has_effective_findings(result: Mapping[str, object]) -> bool: + """Return whether meta-analysis had effective findings to process.""" + effective_ids = result.get("effective_finding_ids") + if isinstance(effective_ids, list): + return bool(effective_ids) + filtered_findings = result.get("filtered_findings") + return isinstance(filtered_findings, list) and bool(filtered_findings) + + +def llm_runtime_available( + *, + preflight_available: bool, + result: Mapping[str, object], +) -> bool: + """Return provider availability after applying meta-analysis runtime evidence.""" + if not preflight_available: + return False + call_log = result.get("llm_call_log") + if not isinstance(call_log, list): + return True + meta_analyzer_records = [ + record + for record in call_log + if isinstance(record, Mapping) and record.get("node") == "meta_analyzer" + ] + return all(successful_llm_record(record) for record in meta_analyzer_records) + + +def semantic_runtime_accounting( + *, + enabled: bool, + result: Mapping[str, object], + discovered_modules: Mapping[str, object], +) -> tuple[bool, bool]: + """Return ``(used, complete)`` for an enabled semantic LLM pass. + + A requested pass is complete only when every required semantic analyzer + explicitly reports either ``completed`` with successful telemetry or + ``not_applicable``. Empty telemetry never proves use. Meta-analysis also + needs a successful record when effective findings exist. + """ + if not enabled: + return False, False + + raw_call_log = result.get("llm_call_log", []) + if not isinstance(raw_call_log, list): + return False, False + used = any(successful_llm_record(record) for record in raw_call_log) + if not all(successful_llm_record(record) for record in raw_call_log): + return used, False + + raw_statuses = result.get("analyzer_status_events") + if not isinstance(raw_statuses, list): + return used, False + required_analyzer_ids = required_semantic_analyzer_ids(discovered_modules) + statuses_by_analyzer: dict[str, list[str]] = {} + for status in raw_statuses: + if not isinstance(status, Mapping): + return used, False + analyzer_id = status.get("analyzer_id") + analyzer_status = status.get("status") + if ( + not isinstance(analyzer_id, str) + or not analyzer_id + or not isinstance(analyzer_status, str) + or not analyzer_status + ): + return used, False + if analyzer_id in required_analyzer_ids: + statuses_by_analyzer.setdefault(analyzer_id, []).append(analyzer_status) + + for analyzer_id in required_analyzer_ids: + statuses = statuses_by_analyzer.get(analyzer_id) + if statuses is None or len(statuses) != 1: + return used, False + status = statuses[0] + has_successful_call = any( + successful_llm_record(record) and record.get("node") == analyzer_id + for record in raw_call_log + ) + if status == "completed": + if not has_successful_call: + return used, False + elif status == "not_applicable": + if has_successful_call: + return used, False + else: + return used, False + + if _has_effective_findings(result) and not any( + successful_llm_record(record) and record.get("node") == "meta_analyzer" + for record in raw_call_log + ): + return used, False + + return used, True diff --git a/src/skillspector/state.py b/src/skillspector/state.py index c5f80b6e..4fffdb3e 100644 --- a/src/skillspector/state.py +++ b/src/skillspector/state.py @@ -288,6 +288,10 @@ class SkillspectorState(TypedDict, total=False): # and the semantic_* analyzers) return immediately without calling the LLM. # Each such node checks use_llm itself; there is no graph-level routing. use_llm: bool + # Optional caller intent when preflight disables execution before the graph. + # Report generation uses this to distinguish unavailable requested analysis from an + # explicit static-only scan while analyzers continue to honor use_llm. + llm_requested: bool # Risk: report node sets these from risk_score risk_severity: str diff --git a/tests/nodes/test_report.py b/tests/nodes/test_report.py index 58f30bae..5026e347 100644 --- a/tests/nodes/test_report.py +++ b/tests/nodes/test_report.py @@ -18,6 +18,7 @@ from __future__ import annotations import json +import logging import pytest @@ -395,6 +396,8 @@ def test_report_empty_findings_zero_risk(self) -> None: "manifest": {}, "skill_path": "/tmp/skill", "output_format": "sarif", + "use_llm": False, + "llm_requested": False, } result = report(state) assert result["risk_score"] == 0 @@ -625,6 +628,11 @@ def test_report_json_structured_summary_survives_llm_mode(self) -> None: "output_format": "json", "use_llm": True, "llm_call_log": [], + "analyzer_status_events": [ + {"analyzer_id": "semantic_developer_intent", "status": "not_applicable"}, + {"analyzer_id": "semantic_quality_policy", "status": "not_applicable"}, + {"analyzer_id": "semantic_security_discovery", "status": "not_applicable"}, + ], } result = report(state) assert result["risk_score"] == 0 @@ -868,6 +876,8 @@ def test_report_baseline_suppresses_finding_and_lowers_score() -> None: "skill_path": None, "output_format": "json", "baseline": baseline, + "use_llm": False, + "llm_requested": False, } result = report(state) assert result["risk_score"] == 0 @@ -1275,8 +1285,8 @@ def test_report_meta_analysis_not_applied_when_no_meta_analyzer_record( assert meta["filtering_mode"] == "heuristic" -def test_report_not_degraded_when_no_llm_calls(monkeypatch: pytest.MonkeyPatch) -> None: - """use_llm True but no LLM calls attempted (e.g. empty skill) -> not degraded.""" +def test_report_static_only_without_calls_is_not_degraded(monkeypatch: pytest.MonkeyPatch) -> None: + """Explicit static-only intent needs no LLM telemetry and is not degraded.""" monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) state: SkillspectorState = { "filtered_findings": [], @@ -1284,7 +1294,8 @@ def test_report_not_degraded_when_no_llm_calls(monkeypatch: pytest.MonkeyPatch) "has_executable_scripts": False, "manifest": {}, "output_format": "json", - "use_llm": True, + "use_llm": False, + "llm_requested": False, "llm_call_log": [], } meta = _meta_from_json_report(state) @@ -1439,6 +1450,8 @@ def test_report_sarif_projects_complete_analysis_completeness() -> None: "has_executable_scripts": False, "manifest": {}, "output_format": "sarif", + "use_llm": False, + "llm_requested": False, "analysis_completeness": { # type: ignore[typeddict-item] "total_components": 2, "coverage_percent": 100.0, @@ -1550,6 +1563,8 @@ def test_report_sarif_bounds_completeness_notifications( "has_executable_scripts": False, "manifest": {}, "output_format": "sarif", + "use_llm": False, + "llm_requested": False, "analysis_completeness": { # type: ignore[typeddict-item] "total_components": 4, "coverage_percent": 0.0, @@ -1597,6 +1612,215 @@ def test_degraded_scan_floors_recommendation_at_caution() -> None: assert result["risk_recommendation"] == "CAUTION" # but never SAFE when degraded +def test_explicit_requested_llm_with_missing_telemetry_degrades_json( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A requested semantic pass needs verified runtime evidence before JSON can say SAFE.""" + monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) + state: SkillspectorState = { + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {}, + "output_format": "json", + "use_llm": True, + "llm_requested": True, + "llm_call_log": [], + "analyzer_status_events": [], + } + + result = report(state) + payload = json.loads(result["report_body"]) + + assert result["risk_recommendation"] == "CAUTION" + assert payload["risk_assessment"]["recommendation"] == "CAUTION" + assert payload["metadata"]["llm_degraded"] is True + assert "runtime telemetry was incomplete" in payload["metadata"]["llm_error"] + + +def test_use_llm_fallback_with_missing_telemetry_degrades_json( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Omitted request metadata inherits enabled LLM intent and cannot report SAFE.""" + monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) + state: SkillspectorState = { + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {}, + "output_format": "json", + "use_llm": True, + "llm_call_log": [], + "analyzer_status_events": [], + } + + result = report(state) + payload = json.loads(result["report_body"]) + + assert result["risk_recommendation"] == "CAUTION" + assert payload["risk_assessment"]["recommendation"] == "CAUTION" + assert payload["metadata"]["llm_degraded"] is True + assert "runtime telemetry was incomplete" in payload["metadata"]["llm_error"] + + +@pytest.mark.parametrize("malformed_request", [None, "false"]) +def test_malformed_llm_request_intent_falls_back_to_enabled_llm( + malformed_request: object, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Non-boolean request metadata cannot override enabled semantic analysis.""" + monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) + state: SkillspectorState = { + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {}, + "output_format": "json", + "use_llm": True, + "llm_requested": malformed_request, # type: ignore[typeddict-item] + "llm_call_log": [], + "analyzer_status_events": [], + } + + result = report(state) + payload = json.loads(result["report_body"]) + + assert result["risk_recommendation"] == "CAUTION" + assert payload["metadata"]["llm_requested"] is True + assert payload["metadata"]["llm_degraded"] is True + + +def test_truthy_malformed_request_intent_cannot_override_static_only( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A non-boolean request value inherits an explicit static-only execution mode.""" + monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) + state: SkillspectorState = { + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {}, + "output_format": "json", + "use_llm": False, + "llm_requested": "true", # type: ignore[typeddict-item] + "llm_call_log": [], + } + + result = report(state) + payload = json.loads(result["report_body"]) + + assert result["risk_recommendation"] == "SAFE" + assert payload["metadata"]["llm_requested"] is False + assert "llm_degraded" not in payload["metadata"] + + +def test_malformed_use_llm_value_cannot_opt_out_of_semantic_accounting( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Only the literal boolean False selects static-only execution.""" + monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) + state: SkillspectorState = { + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {}, + "output_format": "json", + "use_llm": None, # type: ignore[typeddict-item] + "llm_call_log": [], + "analyzer_status_events": [], + } + + result = report(state) + payload = json.loads(result["report_body"]) + + assert result["risk_recommendation"] == "CAUTION" + assert payload["metadata"]["llm_requested"] is True + assert payload["metadata"]["llm_degraded"] is True + + +def test_explicit_requested_llm_with_invalid_call_telemetry_degrades_without_crashing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Malformed runtime evidence cannot bypass the floor or break report rendering.""" + monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) + state: SkillspectorState = { + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {}, + "output_format": "json", + "use_llm": True, + "llm_requested": True, + "llm_call_log": ["invalid"], # type: ignore[list-item] + "analyzer_status_events": [], + } + + result = report(state) + payload = json.loads(result["report_body"]) + + assert result["risk_recommendation"] == "CAUTION" + assert payload["metadata"]["llm_degraded"] is True + assert "runtime telemetry was incomplete" in payload["metadata"]["llm_error"] + + +@pytest.mark.parametrize("output_format", ["terminal", "markdown", "sarif"]) +def test_explicit_requested_llm_with_missing_telemetry_warns_every_report_surface( + output_format: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Human-readable and SARIF reports expose the shared semantic coverage gap.""" + monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) + state: SkillspectorState = { + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {}, + "output_format": output_format, + "use_llm": True, + "llm_requested": True, + "llm_call_log": [], + "analyzer_status_events": [], + } + + result = report(state) + + assert result["risk_recommendation"] == "CAUTION" + assert "runtime telemetry was incomplete" in result["report_body"] + if output_format == "sarif": + notification = result["sarif_report"]["runs"][0]["invocations"][0][ + "toolExecutionNotifications" + ][0] + assert notification["level"] == "warning" + + +def test_explicit_all_not_applicable_semantic_pass_stays_safe( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Verified no-work statuses are complete without claiming any LLM calls.""" + monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, None)) + state: SkillspectorState = { + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {}, + "output_format": "json", + "use_llm": True, + "llm_requested": True, + "llm_call_log": [], + "analyzer_status_events": [ + {"analyzer_id": "semantic_developer_intent", "status": "not_applicable"}, + {"analyzer_id": "semantic_quality_policy", "status": "not_applicable"}, + {"analyzer_id": "semantic_security_discovery", "status": "not_applicable"}, + ], + } + + result = report(state) + metadata = json.loads(result["report_body"])["metadata"] + + assert result["risk_recommendation"] == "SAFE" + assert "llm_degraded" not in metadata + + def test_unavailable_provider_floors_recommendation_even_with_success_records( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -1720,6 +1944,31 @@ async def partially_succeeds(self: LLMAnalyzerBase, batches: list, **_kwargs: ob assert result["risk_recommendation"] == "CAUTION" +def test_preflight_unavailable_log_does_not_claim_runtime_calls_failed( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Preflight failure is logged without inventing zero attempted runtime calls.""" + monkeypatch.setattr( + "skillspector.nodes.report.is_llm_available", + lambda: (False, "not configured"), + ) + state: SkillspectorState = { + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {}, + "output_format": "json", + "use_llm": False, + "llm_requested": True, + } + + with caplog.at_level(logging.WARNING, logger="skillspector.nodes.report"): + report(state) + + assert "unavailable during preflight" in caplog.text + assert "0/0" not in caplog.text + + def test_non_degraded_clean_scan_stays_safe() -> None: """Without degradation, a clean scan still reports SAFE (no over-flooring).""" state: SkillspectorState = { @@ -1729,7 +1978,16 @@ def test_non_degraded_clean_scan_stays_safe() -> None: "manifest": {}, "output_format": "json", "use_llm": True, - "llm_call_log": [llm_call_record("semantic_security_discovery", ok=True)], + "llm_call_log": [ + llm_call_record("semantic_developer_intent", ok=True), + llm_call_record("semantic_quality_policy", ok=True), + llm_call_record("semantic_security_discovery", ok=True), + ], + "analyzer_status_events": [ + {"analyzer_id": "semantic_developer_intent", "status": "completed"}, + {"analyzer_id": "semantic_quality_policy", "status": "completed"}, + {"analyzer_id": "semantic_security_discovery", "status": "completed"}, + ], } result = report(state) assert result["risk_recommendation"] == "SAFE" diff --git a/tests/test_multi_skill.py b/tests/test_multi_skill.py index e13dfc61..de6c1b18 100644 --- a/tests/test_multi_skill.py +++ b/tests/test_multi_skill.py @@ -279,20 +279,41 @@ def test_single_sub_skill_not_multi(self, tmp_path: Path) -> None: assert result.is_multi_skill is False assert len(result.skills) == 1 - def test_hidden_directories_skipped(self, tmp_path: Path) -> None: - """Directories starting with '.' are not scanned for skills.""" + def test_dot_prefixed_child_skill_is_discovered_with_explicit_skips( + self, tmp_path: Path + ) -> None: + """A dot-prefixed child skill is scanned without traversing explicit skips or links.""" for name in ("skill-a", "skill-b"): sub = tmp_path / name sub.mkdir() (sub / "SKILL.md").write_text(f"---\nname: {name}\n---\n", encoding="utf-8") - hidden = tmp_path / ".hidden-skill" - hidden.mkdir() - (hidden / "SKILL.md").write_text("---\nname: hidden\n---\n", encoding="utf-8") + dot_prefixed = tmp_path / ".review-helper" + dot_prefixed.mkdir() + (dot_prefixed / "SKILL.md").write_text("---\nname: review-helper\n---\n", encoding="utf-8") + skipped = tmp_path / ".git" + skipped.mkdir() + (skipped / "SKILL.md").write_text("---\nname: skipped\n---\n", encoding="utf-8") + linked_target = tmp_path.parent / f"{tmp_path.name}-linked-target" + linked_target.mkdir() + (linked_target / "SKILL.md").write_text("---\nname: linked\n---\n", encoding="utf-8") + try: + (tmp_path / "linked-skill").symlink_to(linked_target, target_is_directory=True) + except OSError: + pytest.skip("symlinks are not supported on this filesystem") + result = detect_skills(tmp_path) + assert result.is_multi_skill is True - assert len(result.skills) == 2 - names = {s.name for s in result.skills} - assert "hidden" not in names + assert [skill.relative_path for skill in result.skills] == [ + ".review-helper", + "skill-a", + "skill-b", + ] + assert {skill.name for skill in result.skills} == { + "review-helper", + "skill-a", + "skill-b", + } def test_symlinked_skill_directory_is_skipped(self, tmp_path: Path) -> None: """Detection must not read a skill manifest through a directory symlink.""" diff --git a/tests/test_semantic_runtime.py b/tests/test_semantic_runtime.py new file mode 100644 index 00000000..945926f0 --- /dev/null +++ b/tests/test_semantic_runtime.py @@ -0,0 +1,74 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for provider-independent semantic runtime accounting.""" + +from __future__ import annotations + +from skillspector.semantic_runtime import ( + required_semantic_analyzer_ids, + semantic_runtime_accounting, + successful_llm_record, +) + + +def test_empty_discovery_registry_cannot_shrink_canonical_semantic_requirements() -> None: + """An import failure cannot erase a required semantic completion check.""" + assert required_semantic_analyzer_ids({}) == frozenset( + { + "semantic_developer_intent", + "semantic_quality_policy", + "semantic_security_discovery", + } + ) + + +def test_successful_llm_record_requires_strict_well_formed_evidence() -> None: + """Truthy substitutes and errored records cannot prove a successful call.""" + assert successful_llm_record({"node": "meta_analyzer", "ok": True, "error": None}) + assert not successful_llm_record({"node": "meta_analyzer", "ok": "false", "error": None}) + assert not successful_llm_record({"node": "", "ok": True, "error": None}) + assert not successful_llm_record( + {"node": "meta_analyzer", "ok": True, "error": "runtime failure"} + ) + + +def test_discovered_api_key_analyzers_extend_canonical_semantic_requirements() -> None: + """Future credential-gated analyzers automatically join the required set.""" + + class _FutureSemanticAnalyzer: + requires_api_key = True + + class _StaticAnalyzer: + requires_api_key = False + + discovered = { + "semantic_future_policy": _FutureSemanticAnalyzer(), + "static_example": _StaticAnalyzer(), + } + + assert required_semantic_analyzer_ids(discovered) == frozenset( + { + "semantic_developer_intent", + "semantic_future_policy", + "semantic_quality_policy", + "semantic_security_discovery", + } + ) + + +def test_incomplete_registry_cannot_make_incomplete_canonical_telemetry_complete() -> None: + """Runtime accounting still requires canonical analyzers absent from discovery.""" + result = { + "llm_call_log": [], + "analyzer_status_events": [ + {"analyzer_id": "semantic_developer_intent", "status": "not_applicable"}, + {"analyzer_id": "semantic_quality_policy", "status": "not_applicable"}, + ], + } + + assert semantic_runtime_accounting( + enabled=True, + result=result, + discovered_modules={}, + ) == (False, False) diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index bbb62c6e..89b9f436 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -88,6 +88,14 @@ def test_cli_version() -> None: assert "v" in result.output +@pytest.mark.parametrize(("no_llm", "expected"), [(False, True), (True, False)]) +def test_scan_state_records_explicit_llm_request_intent(no_llm: bool, expected: bool) -> None: + """Report finalization can distinguish real CLI intent from legacy direct calls.""" + state = cli._scan_state("skill", FormatChoice.json, no_llm) + + assert state["llm_requested"] is expected + + def test_cli_scan_local_directory(tmp_path: Path) -> None: """scan with local directory runs graph and prints report.""" (tmp_path / "SKILL.md").write_text("---\nname: scan-test\n---\n# Safe", encoding="utf-8") @@ -264,6 +272,49 @@ def test_recursive_scan_exception_marks_combined_execution_as_failed(tmp_path: P assert payload["skills"][1] == {"name": "two", "error": "child scan crashed"} +def test_recursive_scan_dispatches_dot_prefixed_child_skill( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Recursive dispatch scans bounded dot-prefixed children and excludes links and skips.""" + for name in ("skill-a", "skill-b", ".review-helper"): + child = tmp_path / name + child.mkdir() + (child / "SKILL.md").write_text(f"---\nname: {name.lstrip('.')}\n---\n", encoding="utf-8") + skipped = tmp_path / ".git" + skipped.mkdir() + (skipped / "SKILL.md").write_text("---\nname: skipped\n---\n", encoding="utf-8") + linked_target = tmp_path.parent / f"{tmp_path.name}-linked-target" + linked_target.mkdir() + (linked_target / "SKILL.md").write_text("---\nname: linked\n---\n", encoding="utf-8") + try: + (tmp_path / "linked-skill").symlink_to(linked_target, target_is_directory=True) + except OSError: + pytest.skip("symlinks are not supported on this filesystem") + + scanned_paths: list[str] = [] + + def fake_scan_skill(*, input_path: str, **_kwargs: object) -> dict[str, object]: + scanned_paths.append(input_path) + return { + "filtered_findings": [], + "risk_score": 0, + "risk_severity": "LOW", + "report_body": "{}", + "execution_successful": True, + "analysis_completeness": {"is_complete": True}, + } + + monkeypatch.setattr(cli_module, "_scan_skill", fake_scan_skill) + result = runner.invoke(app, ["scan", str(tmp_path), "--recursive", "--no-llm"]) + + assert result.exit_code == 0 + assert scanned_paths == [ + str(tmp_path / ".review-helper"), + str(tmp_path / "skill-a"), + str(tmp_path / "skill-b"), + ] + + def test_cli_scan_slack_p6_pe3_regression(tmp_path: Path) -> None: """Benign context stays distinguishable without deleting deterministic CLI evidence.""" (tmp_path / "references").mkdir() diff --git a/tests/unit/test_mcp_server.py b/tests/unit/test_mcp_server.py index 1d2430c9..f4ccd872 100644 --- a/tests/unit/test_mcp_server.py +++ b/tests/unit/test_mcp_server.py @@ -16,6 +16,8 @@ """Tests for the MCP server wrapper (run_scan core + scan_skill tool).""" import asyncio +import importlib +import json import os import sys from pathlib import Path @@ -25,8 +27,10 @@ import pytest from skillspector import mcp_server -from skillspector.mcp_server import run_scan +from skillspector.graph import graph as workflow_graph +from skillspector.mcp_server import _llm_runtime_accounting, run_scan from skillspector.models import Finding +from skillspector.nodes.report import report from skillspector.providers import reset_provider, use_provider from skillspector.suppression import SuppressedFinding @@ -43,7 +47,7 @@ async def test_run_scan_returns_structured_verdict( monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (False, "no llm")) _write_skill(tmp_path) - result = await run_scan(str(tmp_path), use_llm=True, output_format="json") + result = await run_scan(str(tmp_path), use_llm=False, output_format="json") assert result["target"] == str(tmp_path) assert isinstance(result["risk_score"], int) @@ -69,6 +73,657 @@ async def test_run_scan_llm_accounting_is_honest_without_credentials( assert result["scan_mode"] == "static-only" +def _complete_zero_risk_graph_result() -> dict[str, object]: + """Return a complete graph verdict suitable for MCP safety-predicate tests.""" + return { + "filtered_findings": [], + "risk_score": 0, + "risk_severity": "LOW", + "risk_recommendation": "SAFE", + "execution_successful": True, + "analysis_completeness": { + "is_complete": True, + "status": "complete", + "entirely_uninspected_files": 0, + }, + "report_body": "{}", + } + + +def test_empty_llm_telemetry_is_not_a_complete_requested_pass() -> None: + """Empty telemetry cannot prove that a requested semantic pass ran.""" + assert _llm_runtime_accounting(enabled=True, result={"llm_call_log": []}) == (False, False) + + +def test_all_not_applicable_semantic_statuses_are_complete_without_llm_use() -> None: + """A real no-work pass is complete only when every semantic node says so.""" + result = { + "llm_call_log": [], + "analyzer_status_events": [ + {"analyzer_id": "semantic_security_discovery", "status": "not_applicable"}, + {"analyzer_id": "semantic_developer_intent", "status": "not_applicable"}, + {"analyzer_id": "semantic_quality_policy", "status": "not_applicable"}, + ], + } + + assert _llm_runtime_accounting(enabled=True, result=result) == (False, True) + + +def test_effective_findings_require_successful_meta_analyzer_telemetry() -> None: + """Meta-analysis is required when effective findings gave it work to do.""" + result = { + "effective_finding_ids": ["finding-1"], + "analyzer_status_events": [ + {"analyzer_id": "semantic_security_discovery", "status": "completed"}, + {"analyzer_id": "semantic_developer_intent", "status": "completed"}, + {"analyzer_id": "semantic_quality_policy", "status": "completed"}, + ], + "llm_call_log": [ + {"node": "semantic_security_discovery", "ok": True, "error": None}, + {"node": "semantic_developer_intent", "ok": True, "error": None}, + {"node": "semantic_quality_policy", "ok": True, "error": None}, + ], + } + + assert _llm_runtime_accounting(enabled=True, result=result) == (True, False) + + +def test_completed_semantic_analyzer_requires_its_own_successful_telemetry() -> None: + """A completed semantic status cannot stand in for its missing call record.""" + result = { + "effective_finding_ids": ["finding-1"], + "analyzer_status_events": [ + {"analyzer_id": "semantic_security_discovery", "status": "completed"}, + {"analyzer_id": "semantic_developer_intent", "status": "completed"}, + {"analyzer_id": "semantic_quality_policy", "status": "completed"}, + ], + "llm_call_log": [ + {"node": "semantic_security_discovery", "ok": True, "error": None}, + {"node": "semantic_quality_policy", "ok": True, "error": None}, + {"node": "meta_analyzer", "ok": True, "error": None}, + ], + } + + assert _llm_runtime_accounting(enabled=True, result=result) == (True, False) + + +def _fully_accounted_semantic_result() -> dict[str, object]: + """Return hand-authored complete semantic telemetry for validation tests.""" + return { + "effective_finding_ids": ["finding-1"], + "analyzer_status_events": [ + {"analyzer_id": "semantic_security_discovery", "status": "completed"}, + {"analyzer_id": "semantic_developer_intent", "status": "completed"}, + {"analyzer_id": "semantic_quality_policy", "status": "completed"}, + ], + "llm_call_log": [ + {"node": "semantic_security_discovery", "ok": True, "error": None}, + {"node": "semantic_developer_intent", "ok": True, "error": None}, + {"node": "semantic_quality_policy", "ok": True, "error": None}, + {"node": "meta_analyzer", "ok": True, "error": None}, + ], + } + + +@pytest.mark.parametrize( + "malformed_status", + [ + {"analyzer_id": [], "status": "completed"}, + {"analyzer_id": {}, "status": "completed"}, + {"status": "completed"}, + {"analyzer_id": "semantic_security_discovery"}, + {"analyzer_id": "semantic_security_discovery", "status": None}, + {"analyzer_id": "semantic_security_discovery", "status": []}, + [], + ], + ids=[ + "list-analyzer-id", + "mapping-analyzer-id", + "missing-analyzer-id", + "missing-status", + "none-status", + "list-status", + "non-mapping-event", + ], +) +def test_malformed_analyzer_status_event_is_incomplete_without_crashing( + malformed_status: object, +) -> None: + """Malformed global or semantic status evidence cannot be silently discarded.""" + result = _fully_accounted_semantic_result() + result["analyzer_status_events"].append(malformed_status) # type: ignore[index] + + assert _llm_runtime_accounting(enabled=True, result=result) == (True, False) + + +def test_duplicate_semantic_status_evidence_is_incomplete() -> None: + """Each semantic analyzer must produce exactly one terminal status.""" + result = _fully_accounted_semantic_result() + result["analyzer_status_events"].append( # type: ignore[index] + {"analyzer_id": "semantic_security_discovery", "status": "completed"} + ) + + assert _llm_runtime_accounting(enabled=True, result=result) == (True, False) + + +def test_not_applicable_status_cannot_have_successful_semantic_telemetry() -> None: + """A no-work terminal status conflicts with a successful call for that node.""" + result = _fully_accounted_semantic_result() + result["analyzer_status_events"][0]["status"] = "not_applicable" # type: ignore[index] + + assert _llm_runtime_accounting(enabled=True, result=result) == (True, False) + + +async def test_all_not_applicable_semantic_pass_remains_install_eligible( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Explicit no-work statuses are complete without claiming an LLM call.""" + graph_result = _complete_zero_risk_graph_result() + graph_result["llm_call_log"] = [] + graph_result["analyzer_status_events"] = [ + {"analyzer_id": "semantic_security_discovery", "status": "not_applicable"}, + {"analyzer_id": "semantic_developer_intent", "status": "not_applicable"}, + {"analyzer_id": "semantic_quality_policy", "status": "not_applicable"}, + ] + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (True, None)) + monkeypatch.setattr(mcp_server.graph, "ainvoke", AsyncMock(return_value=graph_result)) + + verdict = await run_scan("fixture", use_llm=True, output_format="json") + + assert verdict["llm_used"] is False + assert verdict["scan_mode"] == "static-only" + assert verdict["safe_to_install"] is True + assert verdict["recommendation"] == "SAFE" + + +def test_static_only_graph_keeps_semantic_nodes_without_constructing_a_model( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Static graph execution skips semantic model construction but keeps nodes wired.""" + capability_probes = 0 + + class _CapabilityProvider: + DEFAULT_MODEL = "test-model" + SLOT_DEFAULTS = {"meta_analyzer": "test-model"} + + def get_context_length(self, model: str) -> int | None: + return 4096 if model == "test-model" else None + + def get_max_output_tokens(self, model: str) -> int | None: + return 128 if model == "test-model" else None + + def resolve_model(self, slot: str = "default") -> str: + del slot + return "test-model" + + def resolve_credentials(self) -> tuple[str, str | None] | None: + return None + + def create_chat_model( + self, + model: str, + *, + max_tokens: int, + timeout: float | None = 120, + ) -> object: + nonlocal capability_probes + del model, max_tokens, timeout + capability_probes += 1 + return object() + + semantic_model_factory = MagicMock( + side_effect=AssertionError("static-only semantic nodes must not construct a chat model") + ) + monkeypatch.setattr("skillspector.llm_analyzer_base.get_chat_model", semantic_model_factory) + token = use_provider(_CapabilityProvider()) + _write_skill(tmp_path) + try: + result = workflow_graph.invoke( + {"skill_path": str(tmp_path), "use_llm": False, "output_format": "json"} + ) + finally: + reset_provider(token) + + assert capability_probes > 0 + semantic_model_factory.assert_not_called() + assert json.loads(result["report_body"])["metadata"]["llm_available"] is True + semantic_statuses = { + status["analyzer_id"]: status["status"] + for status in result["analyzer_status_events"] + if status["analyzer_id"].startswith("semantic_") # type: ignore[index] + } + assert semantic_statuses == { + "semantic_security_discovery": "disabled", + "semantic_developer_intent": "disabled", + "semantic_quality_policy": "disabled", + } + + +async def test_late_provider_binding_cannot_claim_a_complete_semantic_scan( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A graph built without credentials keeps semantic nodes for a later provider.""" + _write_skill(tmp_path) + graph_module = importlib.import_module("skillspector.graph") + monkeypatch.setattr( + graph_module, + "is_llm_available", + lambda: (False, "not configured"), + raising=False, + ) + late_bound_graph = graph_module.create_graph() + + def transport_failure(*_args: object, **_kwargs: object) -> object: + raise RuntimeError("simulated late-bound provider failure") + + captured: dict[str, object] = {} + + async def invoke_real_graph( + state: dict[str, object], config: dict[str, object] + ) -> dict[str, object]: + captured["result"] = await late_bound_graph.ainvoke(state, config=config) + return captured["result"] # type: ignore[return-value] + + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (True, None)) + monkeypatch.setattr("skillspector.llm_analyzer_base.get_chat_model", transport_failure) + monkeypatch.setattr(mcp_server, "graph", SimpleNamespace(ainvoke=invoke_real_graph)) + + verdict = await run_scan(str(tmp_path), use_llm=True, output_format="json") + graph_result = captured["result"] + + statuses = { + status["analyzer_id"]: status["status"] + for status in graph_result["analyzer_status_events"] # type: ignore[index] + if status["analyzer_id"].startswith("semantic_") # type: ignore[index] + } + assert statuses == { + "semantic_security_discovery": "unavailable", + "semantic_developer_intent": "unavailable", + "semantic_quality_policy": "unavailable", + } + assert verdict["scan_mode"] == "static-only" + assert verdict["safe_to_install"] is False + assert verdict["recommendation"] != "SAFE" + + +async def test_requested_unavailable_llm_blocks_safe_to_install( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An unmet requested analysis pass cannot produce an install-safe verdict.""" + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (False, "not configured")) + monkeypatch.setattr( + mcp_server.graph, + "ainvoke", + AsyncMock(return_value=_complete_zero_risk_graph_result()), + ) + + verdict = await run_scan("fixture", use_llm=True, output_format="json") + + assert verdict["llm_requested"] is True + assert verdict["llm_available"] is False + assert verdict["llm_used"] is False + assert verdict["scan_mode"] == "static-only" + assert verdict["risk_score"] == 0 + assert verdict["safe_to_install"] is False + assert verdict["recommendation"] == "CAUTION" + + +async def test_unrequested_llm_keeps_static_scan_eligible_for_safe_to_install( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Explicit static-only scans retain the normal complete low-risk safety predicate.""" + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (False, "not configured")) + monkeypatch.setattr( + mcp_server.graph, + "ainvoke", + AsyncMock(return_value=_complete_zero_risk_graph_result()), + ) + + verdict = await run_scan("fixture", use_llm=False, output_format="json") + + assert verdict["llm_requested"] is False + assert verdict["llm_available"] is False + assert verdict["llm_used"] is False + assert verdict["scan_mode"] == "static-only" + assert verdict["risk_score"] == 0 + assert verdict["safe_to_install"] is True + assert verdict["recommendation"] == "SAFE" + + +async def test_all_failed_runtime_llm_calls_block_safe_to_install( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Successful preflight cannot hide a requested LLM pass that failed at runtime.""" + graph_result = _complete_zero_risk_graph_result() + graph_result["llm_call_log"] = [ + {"node": "semantic_security_discovery", "ok": False, "error": "transport error"}, + {"node": "semantic_quality_policy", "ok": False, "error": "invalid response"}, + ] + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (True, None)) + monkeypatch.setattr( + mcp_server.graph, + "ainvoke", + AsyncMock(return_value=graph_result), + ) + + verdict = await run_scan("fixture", use_llm=True, output_format="json") + + assert verdict["llm_requested"] is True + assert verdict["llm_available"] is True + assert verdict["llm_used"] is False + assert verdict["scan_mode"] == "static-only" + assert verdict["risk_score"] == 0 + assert verdict["safe_to_install"] is False + assert verdict["recommendation"] == "CAUTION" + + +async def test_partial_runtime_llm_failure_blocks_safe_to_install( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A partially completed requested pass remains degraded and cannot be install-safe.""" + graph_result = _complete_zero_risk_graph_result() + graph_result["llm_call_log"] = [ + {"node": "semantic_security_discovery", "ok": True, "error": None}, + {"node": "semantic_quality_policy", "ok": False, "error": "rate limited"}, + ] + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (True, None)) + monkeypatch.setattr( + mcp_server.graph, + "ainvoke", + AsyncMock(return_value=graph_result), + ) + + verdict = await run_scan("fixture", use_llm=True, output_format="json") + + assert verdict["llm_requested"] is True + assert verdict["llm_available"] is True + assert verdict["llm_used"] is True + assert verdict["scan_mode"] == "static+llm" + assert verdict["risk_score"] == 0 + assert verdict["safe_to_install"] is False + assert verdict["recommendation"] == "CAUTION" + + +async def test_successful_runtime_llm_calls_keep_complete_scan_install_eligible( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A complete low-risk pass remains eligible for an install-safe verdict.""" + graph_result = _complete_zero_risk_graph_result() + graph_result["effective_finding_ids"] = ["finding-1"] + graph_result["analyzer_status_events"] = [ + {"analyzer_id": "semantic_security_discovery", "status": "completed"}, + {"analyzer_id": "semantic_developer_intent", "status": "completed"}, + {"analyzer_id": "semantic_quality_policy", "status": "completed"}, + ] + graph_result["llm_call_log"] = [ + {"node": "semantic_security_discovery", "ok": True, "error": None}, + {"node": "semantic_developer_intent", "ok": True, "error": None}, + {"node": "semantic_quality_policy", "ok": True, "error": None}, + {"node": "meta_analyzer", "ok": True, "error": None}, + ] + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (True, None)) + monkeypatch.setattr( + mcp_server.graph, + "ainvoke", + AsyncMock(return_value=graph_result), + ) + + verdict = await run_scan("fixture", use_llm=True, output_format="json") + + assert verdict["llm_available"] is True + assert verdict["llm_used"] is True + assert verdict["scan_mode"] == "static+llm" + assert verdict["safe_to_install"] is True + assert verdict["recommendation"] == "SAFE" + + +async def _render_complete_zero_risk_result( + state: dict[str, object], config: dict[str, object] +) -> dict[str, object]: + """Render a canonical complete report from the wrapper-provided graph state.""" + del config + return report( + { + **state, + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {"name": "mcp-test"}, + "analysis_completeness": { + "is_complete": True, + "status": "complete", + "execution_successful": True, + "entirely_uninspected_files": 0, + }, + "execution_successful": True, + } + ) + + +async def test_unavailable_requested_llm_aligns_embedded_json_report( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The embedded JSON report reflects requested-but-unavailable LLM analysis.""" + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (False, "not configured")) + monkeypatch.setattr( + "skillspector.nodes.report.is_llm_available", + lambda: (False, "not configured"), + ) + monkeypatch.setattr(mcp_server.graph, "ainvoke", _render_complete_zero_risk_result) + + verdict = await run_scan("fixture", use_llm=True, output_format="json") + payload = json.loads(verdict["report"]) + + assert verdict["risk_score"] == payload["risk_assessment"]["score"] == 0 + assert verdict["recommendation"] == payload["risk_assessment"]["recommendation"] == "CAUTION" + assert payload["metadata"]["llm_requested"] is True + assert payload["metadata"]["llm_available"] is False + assert payload["metadata"]["meta_analysis_applied"] is False + assert payload["metadata"]["filtering_mode"] == "heuristic" + + +async def test_empty_runtime_telemetry_aligns_mcp_and_embedded_json_caution( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A successful preflight cannot leave the embedded report fail-open.""" + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (True, None)) + monkeypatch.setattr( + "skillspector.nodes.report.is_llm_available", + lambda: (True, None), + ) + monkeypatch.setattr(mcp_server.graph, "ainvoke", _render_complete_zero_risk_result) + + verdict = await run_scan("fixture", use_llm=True, output_format="json") + payload = json.loads(verdict["report"]) + + assert verdict["recommendation"] == "CAUTION" + assert verdict["safe_to_install"] is False + assert payload["risk_assessment"]["recommendation"] == "CAUTION" + assert payload["metadata"]["llm_degraded"] is True + assert "runtime telemetry was incomplete" in payload["metadata"]["llm_error"] + + +async def test_malformed_runtime_telemetry_preserves_failed_meta_availability( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Malformed siblings cannot erase valid failed meta-analysis evidence.""" + llm_call_log: list[object] = [ + {"node": "meta_analyzer", "ok": False, "error": "runtime failure"}, + "malformed-record", + ] + + async def render_mixed_telemetry( + state: dict[str, object], config: dict[str, object] + ) -> dict[str, object]: + del config + completeness = { + "is_complete": True, + "status": "complete", + "execution_successful": True, + "entirely_uninspected_files": 0, + } + return { + **report( + { + **state, + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {"name": "mcp-test"}, + "llm_call_log": llm_call_log, # type: ignore[typeddict-item] + "analysis_completeness": completeness, + "execution_successful": True, + } + ), + "llm_call_log": llm_call_log, + "analysis_completeness": completeness, + } + + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (True, None)) + monkeypatch.setattr( + "skillspector.nodes.report.is_llm_available", + lambda: (True, None), + ) + monkeypatch.setattr(mcp_server.graph, "ainvoke", render_mixed_telemetry) + + verdict = await run_scan("fixture", use_llm=True, output_format="json") + payload = json.loads(verdict["report"]) + + assert verdict["llm_available"] is False + assert payload["metadata"]["llm_available"] is False + assert verdict["recommendation"] == "CAUTION" + assert payload["risk_assessment"]["recommendation"] == "CAUTION" + assert verdict["safe_to_install"] is False + + +async def test_truthy_malformed_ok_is_not_counted_as_runtime_success( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Only the literal boolean True can prove a successful LLM call.""" + llm_call_log = [ + {"node": "meta_analyzer", "ok": "false", "error": None}, + ] + + async def render_truthy_malformed_telemetry( + state: dict[str, object], config: dict[str, object] + ) -> dict[str, object]: + del config + completeness = { + "is_complete": True, + "status": "complete", + "execution_successful": True, + "entirely_uninspected_files": 0, + } + return { + **report( + { + **state, + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {"name": "mcp-test"}, + "llm_call_log": llm_call_log, # type: ignore[typeddict-item] + "analysis_completeness": completeness, + "execution_successful": True, + } + ), + "llm_call_log": llm_call_log, + "analysis_completeness": completeness, + } + + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (True, None)) + monkeypatch.setattr( + "skillspector.nodes.report.is_llm_available", + lambda: (True, None), + ) + monkeypatch.setattr(mcp_server.graph, "ainvoke", render_truthy_malformed_telemetry) + + verdict = await run_scan("fixture", use_llm=True, output_format="json") + payload = json.loads(verdict["report"]) + metadata = payload["metadata"] + + assert verdict["recommendation"] == "CAUTION" + assert verdict["safe_to_install"] is False + assert verdict["llm_available"] is False + assert payload["risk_assessment"]["recommendation"] == "CAUTION" + assert metadata["llm_available"] is False + assert metadata["meta_analysis_applied"] is False + assert metadata["llm_calls_succeeded"] == 0 + assert metadata["llm_degraded"] is True + + +async def test_failed_meta_analysis_aligns_mcp_and_embedded_json_availability( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """MCP and its embedded report expose the same runtime availability.""" + llm_call_log = [ + {"node": "meta_analyzer", "ok": False, "error": "runtime failure"}, + ] + + async def render_failed_meta_analysis( + state: dict[str, object], config: dict[str, object] + ) -> dict[str, object]: + del config + completeness = { + "is_complete": True, + "status": "complete", + "execution_successful": True, + "entirely_uninspected_files": 0, + } + return { + **report( + { + **state, + "filtered_findings": [], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {"name": "mcp-test"}, + "llm_call_log": llm_call_log, + "analysis_completeness": completeness, + "execution_successful": True, + } + ), + "llm_call_log": llm_call_log, + "analysis_completeness": completeness, + } + + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (True, None)) + monkeypatch.setattr( + "skillspector.nodes.report.is_llm_available", + lambda: (True, None), + ) + monkeypatch.setattr(mcp_server.graph, "ainvoke", render_failed_meta_analysis) + + verdict = await run_scan("fixture", use_llm=True, output_format="json") + payload = json.loads(verdict["report"]) + + assert verdict["llm_available"] is False + assert verdict["llm_available"] == payload["metadata"]["llm_available"] + assert verdict["llm_used"] is False + assert verdict["safe_to_install"] is False + assert verdict["recommendation"] == "CAUTION" + + +async def test_explicit_static_only_keeps_embedded_json_report_unchanged( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Explicit static-only mode keeps its existing SAFE report and request metadata.""" + monkeypatch.setattr(mcp_server, "is_llm_available", lambda: (False, "not configured")) + monkeypatch.setattr( + "skillspector.nodes.report.is_llm_available", + lambda: (False, "not configured"), + ) + monkeypatch.setattr(mcp_server.graph, "ainvoke", _render_complete_zero_risk_result) + + verdict = await run_scan("fixture", use_llm=False, output_format="json") + payload = json.loads(verdict["report"]) + + assert verdict["risk_score"] == payload["risk_assessment"]["score"] == 0 + assert verdict["recommendation"] == payload["risk_assessment"]["recommendation"] == "SAFE" + assert payload["metadata"]["llm_requested"] is False + assert payload["metadata"]["meta_analysis_applied"] is False + + async def test_run_scan_reports_llm_available_with_credentials( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -84,10 +739,10 @@ async def test_run_scan_reports_llm_available_with_credentials( assert result["scan_mode"] == "static-only" -async def test_run_scan_uses_bound_provider_without_credentials( +async def test_run_scan_reports_missing_telemetry_for_bound_provider_without_credentials( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - """An injected provider can own the LLM client without exposing raw credentials.""" + """A bound provider alone cannot make an empty semantic pass look used.""" class _InjectedProvider: DEFAULT_MODEL = "injected-default" @@ -123,6 +778,7 @@ async def ainvoke(self, state, config): "risk_severity": "LOW", "risk_recommendation": "OK", "report_body": "report", + "llm_call_log": [], } token = use_provider(_InjectedProvider()) @@ -136,8 +792,9 @@ async def ainvoke(self, state, config): assert result["llm_available"] is True assert result["llm_requested"] is True - assert result["llm_used"] is True - assert result["scan_mode"] == "static+llm" + assert result["llm_used"] is False + assert result["scan_mode"] == "static-only" + assert result["safe_to_install"] is False async def test_run_scan_disables_llm_for_unavailable_bound_provider(