From 28d34e3e4041a5d576df72ad2873b36a61bf7e98 Mon Sep 17 00:00:00 2001 From: Gadi Evron Date: Thu, 28 May 2026 12:22:17 -0700 Subject: [PATCH] fix(scanner): isolate optional-stage failures and disambiguate skip causes Three bugs in core/scanner.py, fixed together as they share the optional-stage skip/error machinery. Error-handling: the optional enhance, verify and dynamic-test stages ran inside step_context but had no inner try/except, unlike app-context and llm-reachability. step_context re-raises (core/step_report.py:57), so an optional-stage error escaped scan_repository to cli.py's blanket except and discarded the completed parse/analyze work. Each optional stage now warns and continues on failure, matching the existing app-context pattern. Required stages (parse, analyze, build-output) still propagate their errors. Skip-cause conflation: skipped_steps recorded one bare string for distinct causes (verify auto-skip vs opt-out both -> 'verify'; dynamic-test collapsed several causes -> 'dynamic-test'). Fixed additively: a new skipped_step_reasons dict on ScanResult (emitted as steps_skipped_reasons in scan.report.json) records the disambiguated cause per skipped step. The bare skipped_steps / steps_skipped strings are unchanged so existing telemetry consumers keep working. New tests/test_scanner.py (7 tests; LLM/Docker stages monkeypatched so the orchestration runs offline). Full suite 183 passed, 63 skipped, 0 failed. ruff clean. Co-Authored-By: Claude Opus 4.7 (1M context) --- libs/openant-core/core/scanner.py | 238 ++++++++++++-------- libs/openant-core/core/schemas.py | 6 + libs/openant-core/tests/test_scanner.py | 287 ++++++++++++++++++++++++ 3 files changed, 434 insertions(+), 97 deletions(-) create mode 100644 libs/openant-core/tests/test_scanner.py diff --git a/libs/openant-core/core/scanner.py b/libs/openant-core/core/scanner.py index 04246725..b7368c3b 100644 --- a/libs/openant-core/core/scanner.py +++ b/libs/openant-core/core/scanner.py @@ -213,11 +213,11 @@ def _step_label(name: str) -> str: elif generate_context: print(_step_label("Skipping application context (module not available)."), file=sys.stderr) - result.skipped_steps.append("app-context") + _record_skip(result, "app-context", "module_unavailable") else: print(_step_label("Skipping application context (--no-context)."), file=sys.stderr) - result.skipped_steps.append("app-context") + _record_skip(result, "app-context", "not_requested") print(file=sys.stderr) # --------------------------------------------------------------- @@ -344,7 +344,7 @@ def _step_label(name: str) -> str: _load_step_report(output_dir, "llm-reachability") ) else: - result.skipped_steps.append("llm-reachability") + _record_skip(result, "llm-reachability", "not_requested") print(file=sys.stderr) # --------------------------------------------------------------- @@ -363,40 +363,50 @@ def _step_label(name: str) -> str: "repo_path": repo_path, "mode": enhance_mode, }) as ctx: - enhance_result = enhance_dataset( - dataset_path=active_dataset_path, - output_path=enhanced_path, - analyzer_output_path=parse_result.analyzer_output_path, - repo_path=repo_path, - mode=enhance_mode, - workers=workers, - backoff_seconds=backoff_seconds, - # checkpoint_path auto-derived from output_path - ) + # Enhance is OPTIONAL: a failure here must not discard the completed + # parse work. Catch-and-continue (matching app-context / + # llm-reachability), since step_context re-raises otherwise. + try: + enhance_result = enhance_dataset( + dataset_path=active_dataset_path, + output_path=enhanced_path, + analyzer_output_path=parse_result.analyzer_output_path, + repo_path=repo_path, + mode=enhance_mode, + workers=workers, + backoff_seconds=backoff_seconds, + # checkpoint_path auto-derived from output_path + ) - ctx.summary = { - "units_enhanced": enhance_result.units_enhanced, - "error_count": enhance_result.error_count, - "classifications": enhance_result.classifications, - "mode": enhance_mode, - } - if enhance_result.error_summary: - ctx.summary["error_summary"] = enhance_result.error_summary - ctx.outputs = { - "enhanced_dataset_path": enhance_result.enhanced_dataset_path, - } - - result.enhanced_dataset_path = enhance_result.enhanced_dataset_path - active_dataset_path = enhance_result.enhanced_dataset_path - collected_step_reports.append(_load_step_report(output_dir, "enhance")) + ctx.summary = { + "units_enhanced": enhance_result.units_enhanced, + "error_count": enhance_result.error_count, + "classifications": enhance_result.classifications, + "mode": enhance_mode, + } + if enhance_result.error_summary: + ctx.summary["error_summary"] = enhance_result.error_summary + ctx.outputs = { + "enhanced_dataset_path": enhance_result.enhanced_dataset_path, + } + + result.enhanced_dataset_path = enhance_result.enhanced_dataset_path + active_dataset_path = enhance_result.enhanced_dataset_path - print(f" Enhanced: {enhance_result.units_enhanced} units", file=sys.stderr) - print(f" Classifications: {enhance_result.classifications}", file=sys.stderr) - if enhance_result.error_summary: - print(f" Errors: {enhance_result.error_count} ({enhance_result.error_summary})", file=sys.stderr) + print(f" Enhanced: {enhance_result.units_enhanced} units", file=sys.stderr) + print(f" Classifications: {enhance_result.classifications}", file=sys.stderr) + if enhance_result.error_summary: + print(f" Errors: {enhance_result.error_count} ({enhance_result.error_summary})", file=sys.stderr) + except Exception as e: + print(f" WARNING: Enhancement failed: {e}", file=sys.stderr) + print(" Continuing with the un-enhanced dataset.", file=sys.stderr) + ctx.summary = {"skipped": True, "reason": str(e)} + _record_skip(result, "enhance", "failed") + + collected_step_reports.append(_load_step_report(output_dir, "enhance")) else: print(_step_label("Skipping enhancement (--no-enhance)."), file=sys.stderr) - result.skipped_steps.append("enhance") + _record_skip(result, "enhance", "not_requested") print(file=sys.stderr) # --------------------------------------------------------------- @@ -462,55 +472,64 @@ def _step_label(name: str) -> str: "results_path": analyze_result.results_path, "analyzer_output_path": parse_result.analyzer_output_path, }) as ctx: - verify_result = run_verification( - results_path=analyze_result.results_path, - output_dir=output_dir, - analyzer_output_path=parse_result.analyzer_output_path, - app_context_path=app_context_path, - repo_path=repo_path, - workers=workers, - backoff_seconds=backoff_seconds, - ) + # Verify is OPTIONAL: a failure here must not discard completed + # parse/analyze work (step_context re-raises otherwise). + try: + verify_result = run_verification( + results_path=analyze_result.results_path, + output_dir=output_dir, + analyzer_output_path=parse_result.analyzer_output_path, + app_context_path=app_context_path, + repo_path=repo_path, + workers=workers, + backoff_seconds=backoff_seconds, + ) - ctx.summary = { - "findings_input": verify_result.findings_input, - "findings_verified": verify_result.findings_verified, - "agreed": verify_result.agreed, - "disagreed": verify_result.disagreed, - "confirmed_vulnerabilities": verify_result.confirmed_vulnerabilities, - } - ctx.outputs = { - "verified_results_path": verify_result.verified_results_path, - } - - result.verified_results_path = verify_result.verified_results_path - active_results_path = verify_result.verified_results_path - collected_step_reports.append(_load_step_report(output_dir, "verify")) + ctx.summary = { + "findings_input": verify_result.findings_input, + "findings_verified": verify_result.findings_verified, + "agreed": verify_result.agreed, + "disagreed": verify_result.disagreed, + "confirmed_vulnerabilities": verify_result.confirmed_vulnerabilities, + } + ctx.outputs = { + "verified_results_path": verify_result.verified_results_path, + } - print(f" Confirmed: {verify_result.confirmed_vulnerabilities} vulnerabilities", - file=sys.stderr) + result.verified_results_path = verify_result.verified_results_path + active_results_path = verify_result.verified_results_path + + print(f" Confirmed: {verify_result.confirmed_vulnerabilities} vulnerabilities", + file=sys.stderr) + + # Update metrics from verified results + result.metrics = AnalysisMetrics( + total=analyze_result.metrics.total, + vulnerable=verify_result.confirmed_vulnerabilities, + bypassable=0, + inconclusive=analyze_result.metrics.inconclusive, + protected=analyze_result.metrics.protected, + safe=analyze_result.metrics.safe + verify_result.disagreed, + errors=analyze_result.metrics.errors, + verified=verify_result.findings_verified, + stage2_agreed=verify_result.agreed, + stage2_disagreed=verify_result.disagreed, + ) + except Exception as e: + print(f" WARNING: Verification failed: {e}", file=sys.stderr) + print(" Continuing with unverified Stage 1 results.", file=sys.stderr) + ctx.summary = {"skipped": True, "reason": str(e)} + _record_skip(result, "verify", "failed") - # Update metrics from verified results - result.metrics = AnalysisMetrics( - total=analyze_result.metrics.total, - vulnerable=verify_result.confirmed_vulnerabilities, - bypassable=0, - inconclusive=analyze_result.metrics.inconclusive, - protected=analyze_result.metrics.protected, - safe=analyze_result.metrics.safe + verify_result.disagreed, - errors=analyze_result.metrics.errors, - verified=verify_result.findings_verified, - stage2_agreed=verify_result.agreed, - stage2_disagreed=verify_result.disagreed, - ) + collected_step_reports.append(_load_step_report(output_dir, "verify")) elif verify and not has_findings: print(_step_label("Skipping verification (no vulnerable findings)."), file=sys.stderr) - result.skipped_steps.append("verify") + _record_skip(result, "verify", "no_candidates") else: print(_step_label("Skipping verification (--no-verify or not requested)."), file=sys.stderr) - result.skipped_steps.append("verify") + _record_skip(result, "verify", "not_requested") print(file=sys.stderr) # --------------------------------------------------------------- @@ -552,7 +571,7 @@ def _step_label(name: str) -> str: if not shutil.which("docker"): print(_step_label("Skipping dynamic test (Docker not found)."), file=sys.stderr) - result.skipped_steps.append("dynamic-test") + _record_skip(result, "dynamic-test", "docker_unavailable") else: from core.dynamic_tester import run_tests @@ -561,38 +580,47 @@ def _step_label(name: str) -> str: with step_context("dynamic-test", output_dir, inputs={ "pipeline_output_path": pipeline_output_path, }) as ctx: - dt_result = run_tests( - pipeline_output_path=pipeline_output_path, - output_dir=output_dir, - ) + # Dynamic test is OPTIONAL: a failure here must not discard + # completed work (step_context re-raises otherwise). + try: + dt_result = run_tests( + pipeline_output_path=pipeline_output_path, + output_dir=output_dir, + ) - ctx.summary = { - "findings_tested": dt_result.findings_tested, - "confirmed": dt_result.confirmed, - "not_reproduced": dt_result.not_reproduced, - "blocked": dt_result.blocked, - "inconclusive": dt_result.inconclusive, - "errors": dt_result.errors, - } - ctx.outputs = { - "results_json_path": dt_result.results_json_path, - "results_md_path": dt_result.results_md_path, - } + ctx.summary = { + "findings_tested": dt_result.findings_tested, + "confirmed": dt_result.confirmed, + "not_reproduced": dt_result.not_reproduced, + "blocked": dt_result.blocked, + "inconclusive": dt_result.inconclusive, + "errors": dt_result.errors, + } + ctx.outputs = { + "results_json_path": dt_result.results_json_path, + "results_md_path": dt_result.results_md_path, + } + + result.dynamic_test_path = dt_result.results_json_path + + print(f" Dynamic test: {dt_result.confirmed} confirmed, " + f"{dt_result.not_reproduced} not reproduced", file=sys.stderr) + except Exception as e: + print(f" WARNING: Dynamic test failed: {e}", file=sys.stderr) + print(" Continuing without dynamic-test results.", file=sys.stderr) + ctx.summary = {"skipped": True, "reason": str(e)} + _record_skip(result, "dynamic-test", "failed") - result.dynamic_test_path = dt_result.results_json_path collected_step_reports.append( _load_step_report(output_dir, "dynamic-test"), ) - - print(f" Dynamic test: {dt_result.confirmed} confirmed, " - f"{dt_result.not_reproduced} not reproduced", file=sys.stderr) elif dynamic_test and not has_findings: print(_step_label("Skipping dynamic test (no findings to test)."), file=sys.stderr) - result.skipped_steps.append("dynamic-test") + _record_skip(result, "dynamic-test", "no_candidates") else: print(_step_label("Skipping dynamic test (not enabled)."), file=sys.stderr) - result.skipped_steps.append("dynamic-test") + _record_skip(result, "dynamic-test", "not_requested") print(file=sys.stderr) # --------------------------------------------------------------- @@ -639,7 +667,7 @@ def _step_label(name: str) -> str: collected_step_reports.append(_load_step_report(output_dir, "report")) else: print(_step_label("Skipping report generation (--no-report)."), file=sys.stderr) - result.skipped_steps.append("report") + _record_skip(result, "report", "not_requested") print(file=sys.stderr) # --------------------------------------------------------------- @@ -683,6 +711,19 @@ def _count_steps( return count +def _record_skip(result: ScanResult, step: str, reason: str) -> None: + """Record that ``step`` was skipped. + + Appends the bare step name to ``result.skipped_steps`` (UNCHANGED behaviour + — telemetry consumers read this flat list) and ADDITIVELY records the + disambiguated cause in ``result.skipped_step_reasons`` so distinct causes + (e.g. verify auto-skip 'no_candidates' vs opt-out 'not_requested') are no + longer conflated to one bare string. + """ + result.skipped_steps.append(step) + result.skipped_step_reasons[step] = reason + + def _load_step_report(output_dir: str, step: str) -> dict: """Load a step report JSON from disk. Returns empty dict on failure.""" path = os.path.join(output_dir, f"{step}.report.json") @@ -724,6 +765,9 @@ def _write_scan_report( "metrics": result.metrics.to_dict(), "steps_completed": [sr.get("step") for sr in step_reports], "steps_skipped": result.skipped_steps, + # ADDITIVE / non-breaking: disambiguated skip cause per step. + # `steps_skipped` above stays a flat bare list (consumers read it). + "steps_skipped_reasons": result.skipped_step_reasons, }, inputs={"repo_path": result.output_dir.replace(os.path.abspath("."), ".")}, outputs={ diff --git a/libs/openant-core/core/schemas.py b/libs/openant-core/core/schemas.py index 43886ebf..beaf0d2a 100644 --- a/libs/openant-core/core/schemas.py +++ b/libs/openant-core/core/schemas.py @@ -135,6 +135,11 @@ class ScanResult: usage: UsageInfo = field(default_factory=UsageInfo) step_reports: list = field(default_factory=list) skipped_steps: list = field(default_factory=list) + # Disambiguated skip cause per skipped step. ADDITIVE / non-breaking: + # `skipped_steps` stays a flat bare list of step names (telemetry consumers + # read it). This map records WHY each step was skipped (e.g. 'verify' -> + # 'no_candidates' for an auto-skip vs 'not_requested' for an opt-out). + skipped_step_reasons: dict = field(default_factory=dict) def to_dict(self) -> dict: return { @@ -155,6 +160,7 @@ def to_dict(self) -> dict: "usage": self.usage.to_dict(), "step_reports": self.step_reports, "skipped_steps": self.skipped_steps, + "skipped_step_reasons": self.skipped_step_reasons, } diff --git a/libs/openant-core/tests/test_scanner.py b/libs/openant-core/tests/test_scanner.py new file mode 100644 index 00000000..d07485d3 --- /dev/null +++ b/libs/openant-core/tests/test_scanner.py @@ -0,0 +1,287 @@ +"""Unit tests for scanner optional-stage isolation and skip-cause reporting. + +Covers two areas: + +- error-handling: the OPTIONAL stages (enhance, verify, dynamic-test) were + wrapped in ``step_context`` but had NO inner try/except. ``step_context`` + re-raises (core/step_report.py:57), so an optional-stage error escaped + ``scan_repository`` to cli.py's blanket ``except`` and discarded all completed + parse/analyze work. The fix adds an inner warn-and-continue try/except around + each optional stage, matching the existing app-context / llm-reachability + pattern. + +- skip-cause conflation: ``skipped_steps`` recorded the IDENTICAL bare string + for distinct skip causes (verify auto-skip vs opt-out both -> 'verify'; + dynamic-test collapsed ~3 causes -> 'dynamic-test'). The fix ADDITIVELY records + a disambiguated reason per skipped step in a NEW ``skipped_step_reasons`` dict, + WITHOUT changing the existing bare ``skipped_steps`` list (telemetry consumers + read it). + +These tests monkeypatch the heavy LLM/Docker stages so we exercise the +orchestration control-flow without real API/Docker calls. +""" + +import sys +from pathlib import Path + +import pytest + +# Project root must be importable (mirrors conftest). core.scanner is a unique +# module name, so a normal import is safe; we do NOT import any parser modules +# here, so we cannot pollute sys.modules with shared parser basenames. +PROJECT_ROOT = Path(__file__).parent.parent +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from core import scanner as scanner_mod # noqa: E402 +from core.schemas import AnalysisMetrics, ScanResult # noqa: E402 + + +# --------------------------------------------------------------------------- +# Shared stubs — make parse + analyze succeed cheaply, with no LLM/network. +# --------------------------------------------------------------------------- + +class _ParseResult: + def __init__(self, output_dir): + self.dataset_path = str(Path(output_dir) / "dataset.json") + self.analyzer_output_path = str(Path(output_dir) / "analyzer.json") + self.units_count = 3 + self.language = "python" + self.processing_level = "all" + + +def _install_minimal_pipeline(monkeypatch, *, vulnerable=0, bypassable=0): + """Stub parse + analyze + build_pipeline_output so a scan runs offline. + + Returns nothing; the caller drives optional-stage behaviour via flags / + further monkeypatching. + """ + import core.parser_adapter as parser_adapter + import core.analyzer as analyzer + import core.reporter as reporter + import core.tracking as tracking + + def _fake_parse(*, output_dir, **kwargs): + pr = _ParseResult(output_dir) + # Write the files downstream stages expect to exist. + Path(pr.dataset_path).write_text('{"units": []}') + Path(pr.analyzer_output_path).write_text("{}") + return pr + + metrics = AnalysisMetrics( + total=3, + vulnerable=vulnerable, + bypassable=bypassable, + inconclusive=0, + protected=0, + safe=3 - vulnerable - bypassable, + errors=0, + ) + + class _AnalyzeResult: + def __init__(self, output_dir): + self.results_path = str(Path(output_dir) / "results.json") + Path(self.results_path).write_text("[]") + self.metrics = metrics + + def _fake_analysis(*, output_dir, **kwargs): + return _AnalyzeResult(output_dir) + + def _fake_build_output(*, results_path, output_path, **kwargs): + Path(output_path).write_text("{}") + return output_path + + monkeypatch.setattr(parser_adapter, "parse_repository", _fake_parse) + monkeypatch.setattr(analyzer, "run_analysis", _fake_analysis) + monkeypatch.setattr(reporter, "build_pipeline_output", _fake_build_output) + # Keep tracking quiet/deterministic. + tracking.reset_tracking() + + +# --------------------------------------------------------------------------- +# Optional-stage errors must NOT abort the scan. +# --------------------------------------------------------------------------- + +def test_enhance_failure_does_not_abort_scan(monkeypatch, tmp_path): + """An exception in the OPTIONAL enhance stage must be caught (warn+continue), + not propagated out of scan_repository (which would discard parse/analyze).""" + _install_minimal_pipeline(monkeypatch) + + import core.enhancer as enhancer + + def _boom(**kwargs): + raise RuntimeError("enhance blew up") + + monkeypatch.setattr(enhancer, "enhance_dataset", _boom) + + out = tmp_path / "out" + # Must NOT raise. Pre-fix this propagates RuntimeError out of scan_repository. + result = scanner_mod.scan_repository( + repo_path=str(tmp_path), + output_dir=str(out), + generate_context=False, + enhance=True, + verify=False, + generate_report=False, + dynamic_test=False, + ) + assert isinstance(result, ScanResult) + # The completed analyze work survived. + assert result.metrics.total == 3 + + +def test_verify_failure_does_not_abort_scan(monkeypatch, tmp_path): + """An exception in the OPTIONAL verify stage must be caught, not propagated.""" + _install_minimal_pipeline(monkeypatch, vulnerable=1) + + import core.verifier as verifier + + def _boom(**kwargs): + raise RuntimeError("verify blew up") + + monkeypatch.setattr(verifier, "run_verification", _boom) + + out = tmp_path / "out" + result = scanner_mod.scan_repository( + repo_path=str(tmp_path), + output_dir=str(out), + generate_context=False, + enhance=False, + verify=True, + generate_report=False, + dynamic_test=False, + ) + assert isinstance(result, ScanResult) + assert result.metrics.total == 3 + + +def test_dynamic_test_failure_does_not_abort_scan(monkeypatch, tmp_path): + """An exception in the OPTIONAL dynamic-test stage must be caught.""" + _install_minimal_pipeline(monkeypatch, vulnerable=1) + + import shutil as _shutil + import core.dynamic_tester as dynamic_tester + + # Force the "docker present" branch so we reach run_tests. + monkeypatch.setattr(scanner_mod.shutil, "which", lambda name: "/usr/bin/docker") + + def _boom(**kwargs): + raise RuntimeError("docker blew up") + + monkeypatch.setattr(dynamic_tester, "run_tests", _boom) + + out = tmp_path / "out" + result = scanner_mod.scan_repository( + repo_path=str(tmp_path), + output_dir=str(out), + generate_context=False, + enhance=False, + verify=False, + generate_report=False, + dynamic_test=True, + ) + assert isinstance(result, ScanResult) + assert result.metrics.total == 3 + + +def test_required_stage_failure_still_propagates(monkeypatch, tmp_path): + """The REQUIRED analyze stage must still propagate its error (regression + guard so the fix does not over-broadly swallow required-stage failures).""" + _install_minimal_pipeline(monkeypatch) + + import core.analyzer as analyzer + + def _boom(**kwargs): + raise RuntimeError("analyze blew up") + + monkeypatch.setattr(analyzer, "run_analysis", _boom) + + out = tmp_path / "out" + with pytest.raises(RuntimeError, match="analyze blew up"): + scanner_mod.scan_repository( + repo_path=str(tmp_path), + output_dir=str(out), + generate_context=False, + enhance=False, + verify=False, + generate_report=False, + dynamic_test=False, + ) + + +# --------------------------------------------------------------------------- +# Disambiguated skip reasons (ADDITIVE; bare list unchanged). +# --------------------------------------------------------------------------- + +def test_verify_skip_reasons_distinguish_autoskip_vs_optout(monkeypatch, tmp_path): + """verify auto-skip (no findings) vs opt-out (--no-verify) must record + DISTINCT reasons in the new skipped_step_reasons dict, while the existing + bare skipped_steps list stays IDENTICAL ('verify').""" + # Case A: verify requested but no findings -> auto-skip. + _install_minimal_pipeline(monkeypatch, vulnerable=0) + out_a = tmp_path / "out_a" + res_a = scanner_mod.scan_repository( + repo_path=str(tmp_path), output_dir=str(out_a), + generate_context=False, enhance=False, verify=True, + generate_report=False, dynamic_test=False, + ) + + # Case B: verify not requested -> opt-out. + out_b = tmp_path / "out_b" + res_b = scanner_mod.scan_repository( + repo_path=str(tmp_path), output_dir=str(out_b), + generate_context=False, enhance=False, verify=False, + generate_report=False, dynamic_test=False, + ) + + # Bare list unchanged in BOTH cases (consumers read this). + assert "verify" in res_a.skipped_steps + assert "verify" in res_b.skipped_steps + + # New disambiguated reasons differ. + assert res_a.skipped_step_reasons["verify"] != res_b.skipped_step_reasons["verify"] + + +def test_dynamic_test_skip_reasons_distinct(monkeypatch, tmp_path): + """dynamic-test no-findings skip vs not-enabled skip must record DISTINCT + reasons, with the bare list still 'dynamic-test'.""" + # Case A: dynamic_test requested but no findings. + _install_minimal_pipeline(monkeypatch, vulnerable=0) + out_a = tmp_path / "dt_a" + res_a = scanner_mod.scan_repository( + repo_path=str(tmp_path), output_dir=str(out_a), + generate_context=False, enhance=False, verify=False, + generate_report=False, dynamic_test=True, + ) + + # Case B: dynamic_test not enabled. + out_b = tmp_path / "dt_b" + res_b = scanner_mod.scan_repository( + repo_path=str(tmp_path), output_dir=str(out_b), + generate_context=False, enhance=False, verify=False, + generate_report=False, dynamic_test=False, + ) + + assert "dynamic-test" in res_a.skipped_steps + assert "dynamic-test" in res_b.skipped_steps + assert res_a.skipped_step_reasons["dynamic-test"] != res_b.skipped_step_reasons["dynamic-test"] + + +def test_skipped_step_reasons_serialized_in_scan_report(monkeypatch, tmp_path): + """The new reason map must be emitted in scan.report.json alongside the + existing flat steps_skipped, and steps_skipped must be unchanged.""" + import json + _install_minimal_pipeline(monkeypatch, vulnerable=0) + out = tmp_path / "out" + scanner_mod.scan_repository( + repo_path=str(tmp_path), output_dir=str(out), + generate_context=False, enhance=False, verify=True, + generate_report=False, dynamic_test=False, + ) + report = json.loads((out / "scan.report.json").read_text()) + summary = report["summary"] + # Existing flat list still present and bare. + assert "verify" in summary["steps_skipped"] + # New disambiguated map present. + assert "steps_skipped_reasons" in summary + assert "verify" in summary["steps_skipped_reasons"]