From 5610637dc95b7e911a64837f95cbbe97246f4737 Mon Sep 17 00:00:00 2001 From: Christopher Kevin Date: Wed, 16 Sep 2026 19:36:25 -0700 Subject: [PATCH 1/4] feat(static): bound path postprocessing Signed-off-by: Christopher Kevin --- .../nodes/analyzers/static_runner.py | 258 ++++++++++- .../analyzers/test_static_runner_filtering.py | 413 ++++++++++++++++++ 2 files changed, 658 insertions(+), 13 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_runner.py b/src/skillspector/nodes/analyzers/static_runner.py index e13cacb75..5ac0a4b23 100644 --- a/src/skillspector/nodes/analyzers/static_runner.py +++ b/src/skillspector/nodes/analyzers/static_runner.py @@ -17,6 +17,7 @@ from __future__ import annotations +import inspect import json import math import os @@ -583,6 +584,35 @@ def _uses_python_ast(module: object) -> bool: return getattr(module, "USES_PYTHON_AST", False) is True +def _requires_python_ast(pattern_modules: list) -> bool: + """Return whether an analyzer or its postprocessor consumes the shared AST.""" + return any(_uses_python_ast(module) for module in pattern_modules) or bool( + pattern_modules + and _explicit_module_hook(pattern_modules[0], "POSTPROCESS_USES_PYTHON_AST") is True + ) + + +def _python_ast_for_path( + path: str, + content: str, + pattern_modules: list, + python_ast_cache_key: str | None, +) -> ParsedPythonFile | None: + """Return the shared parse needed by analyzer or postprocessor hooks.""" + if len(content) > MAX_FILE_CHARS or not _requires_python_ast(pattern_modules): + return None + if _infer_file_type(path) != "python": + return None + return get_python_ast(python_ast_cache_key, content, path) + + +def _explicit_module_hook(module: object, name: str) -> object | None: + """Return a hook only when the module or its class actually declares it.""" + if inspect.getattr_static(module, name, None) is None: + return None + return getattr(module, name, None) + + def _uses_runtime_check(module: object) -> bool: """Return whether a pattern module accepts the runner-owned deadline hook.""" return getattr(module, "USES_RUNTIME_CHECK", False) is True @@ -769,6 +799,7 @@ def _scan_path( pattern_modules: list, finding_budget: _FindingBudget, python_ast_cache_key: str | None = None, + python_ast: ParsedPythonFile | None = None, ) -> tuple[list[Finding], _StaticResourceLimitError | None]: """Run pattern modules with construction, emission, and runtime guards.""" findings: list[Finding] = [] @@ -779,10 +810,9 @@ def _scan_path( if _is_license_basename(path, file_type) else None ) - python_ast: ParsedPythonFile | None = None if file_type == "python" and any(_uses_python_ast(module) for module in pattern_modules): finding_budget.check_runtime() - python_ast = get_python_ast(python_ast_cache_key, content, path) + python_ast = python_ast or get_python_ast(python_ast_cache_key, content, path) finding_budget.check_runtime() line_starts = logical_line_starts(content) @@ -1436,13 +1466,29 @@ def _scan_all_views_detailed( *, max_findings: int = MAX_FINDINGS_PER_ARTIFACT, timeout_seconds: float | None = None, + started_at: float | None = None, + python_ast: ParsedPythonFile | None = None, ) -> tuple[list[Finding], LedgerReason | None, dict[str, int | float]]: """Scan bounded raw windows and return any limit with observed/limit metrics.""" + started_at = time.monotonic() if started_at is None else started_at ast_modules = [module for module in pattern_modules if _uses_python_ast(module)] lexical_modules = [module for module in pattern_modules if not _uses_python_ast(module)] + if python_ast is None: + python_ast = _python_ast_for_path( + path, + content, + pattern_modules, + python_ast_cache_key, + ) + python_syntax_error = bool( + _infer_file_type(path) == "python" + and len(content) <= MAX_FILE_CHARS + and _requires_python_ast(pattern_modules) + and python_ast is not None + and python_ast.tree is None + ) findings: list[Finding] = [] seen_findings: set[_ViewFindingKey] = set() - started_at = time.monotonic() runtime_limit = MAX_STATIC_ANALYSIS_SECONDS_PER_ARTIFACT if timeout_seconds is not None: runtime_limit = min(runtime_limit, max(0.0, timeout_seconds)) @@ -1539,6 +1585,7 @@ def _scan_all_views_detailed( ast_modules, finding_budget, python_ast_cache_key, + python_ast, ) except _StaticResourceLimitError as exc: return _deduplicate_view_findings(findings), exc.reason, exc.metrics @@ -1771,7 +1818,9 @@ def _scan_all_views_detailed( return ( deduplicated, ( - LedgerReason.STATIC_PARSE_LIMIT + LedgerReason.SYNTAX_ERROR + if python_syntax_error + else LedgerReason.STATIC_PARSE_LIMIT if bounded_parse_limited else LedgerReason.OBFUSCATED_INSTRUCTION_TEXT if marker_projection_limited @@ -1789,6 +1838,8 @@ def _scan_all_views( *, max_findings: int = MAX_FINDINGS_PER_ARTIFACT, timeout_seconds: float | None = None, + started_at: float | None = None, + python_ast: ParsedPythonFile | None = None, ) -> list[Finding]: findings, _, _ = _scan_all_views_detailed( path, @@ -1797,10 +1848,74 @@ def _scan_all_views( python_ast_cache_key, max_findings=max_findings, timeout_seconds=timeout_seconds, + started_at=started_at, + python_ast=python_ast, ) return findings +def _postprocess_path_findings( + content: str, + pattern_modules: list, + findings: list[Finding], + *, + python_ast: ParsedPythonFile | None = None, + started_at: float | None = None, + timeout_seconds: float | None = None, +) -> list[Finding]: + """Let one analyzer family reconcile findings after every view has run.""" + hook = ( + _explicit_module_hook(pattern_modules[0], "postprocess_path_findings") + if pattern_modules + else None + ) + if not callable(hook): + return findings + uses_python_ast = bool( + pattern_modules + and _explicit_module_hook(pattern_modules[0], "POSTPROCESS_USES_PYTHON_AST") is True + ) + uses_runtime_budget = bool( + pattern_modules + and _explicit_module_hook(pattern_modules[0], "POSTPROCESS_USES_RUNTIME_BUDGET") is True + ) + if uses_python_ast or uses_runtime_budget: + kwargs: dict[str, object] = {} + if uses_python_ast: + kwargs["python_ast"] = python_ast + if uses_runtime_budget: + kwargs.update( + { + "started_at": started_at, + "timeout_seconds": timeout_seconds, + } + ) + return cast(list[Finding], hook(content, findings, **kwargs)) + return cast(list[Finding], hook(content, findings)) + + +def _cleanup_expired_path_findings( + pattern_modules: list, + findings: list[Finding], +) -> list[Finding]: + """Run only a module's bounded private-evidence cleanup after a deadline.""" + hook = ( + _explicit_module_hook(pattern_modules[0], "cleanup_path_findings") + if pattern_modules + else None + ) + if callable(hook): + return cast(list[Finding], hook(findings)) + has_postprocessor = bool( + pattern_modules + and callable(_explicit_module_hook(pattern_modules[0], "postprocess_path_findings")) + ) + # A module requiring postprocessing owns the contract that turns its private + # intermediate findings into public objects. Without an explicit bounded + # cleanup hook, dropping that partial prefix is safer than leaking it. + return [] if has_postprocessor else findings + + def run_static_patterns( state: Mapping[str, object], pattern_modules: list, @@ -1847,19 +1962,51 @@ def run_static_patterns( remaining = MAX_FINDINGS_PER_ANALYZER - len(findings) if remaining <= 0: break + path_started_at = time.monotonic() shared_remaining = transitive_remaining_seconds(cast(SkillspectorState, state)) if shared_remaining is not None and shared_remaining <= 0: break - findings.extend( - _scan_all_views( - path, + python_ast = _python_ast_for_path( + path, + content, + pattern_modules, + python_ast_cache_key, + ) + path_limit = min(MAX_FINDINGS_PER_ARTIFACT, remaining) + path_findings, resource_limit, _ = _scan_all_views_detailed( + path, + content, + pattern_modules, + python_ast_cache_key, + max_findings=path_limit, + timeout_seconds=shared_remaining, + started_at=path_started_at, + python_ast=python_ast, + ) + runtime_limit = MAX_STATIC_ANALYSIS_SECONDS_PER_ARTIFACT + if shared_remaining is not None: + runtime_limit = min(runtime_limit, max(0.0, shared_remaining)) + expired = ( + resource_limit is LedgerReason.RUNTIME_LIMIT + or time.monotonic() - path_started_at >= runtime_limit + ) + if expired: + path_findings = _cleanup_expired_path_findings(pattern_modules, path_findings) + else: + path_findings = _postprocess_path_findings( content, pattern_modules, - python_ast_cache_key, - max_findings=min(MAX_FINDINGS_PER_ARTIFACT, remaining), - timeout_seconds=shared_remaining, + path_findings, + python_ast=python_ast, + started_at=path_started_at, + timeout_seconds=runtime_limit, ) - ) + if time.monotonic() - path_started_at >= runtime_limit: + path_findings = _cleanup_expired_path_findings( + pattern_modules, + path_findings, + ) + findings.extend(path_findings[:path_limit]) return findings @@ -1946,6 +2093,7 @@ def run_static_patterns_with_ledger( ) else: remaining = MAX_FINDINGS_PER_ANALYZER - len(findings) + path_started_at = time.monotonic() shared_remaining = transitive_remaining_seconds(cast(SkillspectorState, state)) path_findings: list[Finding] resource_limit: LedgerReason | None @@ -1959,14 +2107,98 @@ def run_static_patterns_with_ledger( } else: try: + python_ast = _python_ast_for_path( + path, + content, + pattern_modules, + python_ast_cache_key, + ) + path_limit = min(MAX_FINDINGS_PER_ARTIFACT, remaining) path_findings, resource_limit, resource_metrics = _scan_all_views_detailed( path, content, pattern_modules, python_ast_cache_key, - max_findings=min(MAX_FINDINGS_PER_ARTIFACT, remaining), + max_findings=path_limit, timeout_seconds=shared_remaining, + started_at=path_started_at, + python_ast=python_ast, ) + has_postprocessor = bool( + pattern_modules + and callable( + _explicit_module_hook( + pattern_modules[0], + "postprocess_path_findings", + ) + ) + ) + runtime_limit = MAX_STATIC_ANALYSIS_SECONDS_PER_ARTIFACT + if shared_remaining is not None: + runtime_limit = min(runtime_limit, max(0.0, shared_remaining)) + observed_seconds = ( + float(resource_metrics.get("observed_seconds", 0.0)) + if resource_limit is LedgerReason.RUNTIME_LIMIT + else max(0.0, time.monotonic() - path_started_at) + ) + expired = ( + resource_limit is LedgerReason.RUNTIME_LIMIT + or observed_seconds >= runtime_limit + ) + if expired: + resource_limit = LedgerReason.RUNTIME_LIMIT + resource_metrics = { + "observed_seconds": observed_seconds, + "limit_seconds": runtime_limit, + } + path_findings = _cleanup_expired_path_findings( + pattern_modules, + path_findings, + ) + elif has_postprocessor: + path_findings = _postprocess_path_findings( + content, + pattern_modules, + path_findings, + python_ast=python_ast, + started_at=path_started_at, + timeout_seconds=runtime_limit, + ) + observed_seconds = max(0.0, time.monotonic() - path_started_at) + if observed_seconds >= runtime_limit: + resource_limit = LedgerReason.RUNTIME_LIMIT + resource_metrics = { + "observed_seconds": observed_seconds, + "limit_seconds": runtime_limit, + } + path_findings = _cleanup_expired_path_findings( + pattern_modules, + path_findings, + ) + if len(path_findings) > path_limit: + postprocessed_count = len(path_findings) + path_findings = path_findings[:path_limit] + if resource_limit is not LedgerReason.RUNTIME_LIMIT: + if remaining < MAX_FINDINGS_PER_ARTIFACT: + observed_findings = len(findings) + postprocessed_count + limit_findings = MAX_FINDINGS_PER_ANALYZER + else: + observed_findings = postprocessed_count + limit_findings = MAX_FINDINGS_PER_ARTIFACT + if resource_limit is LedgerReason.OUTPUT_LIMIT: + observed_findings = max( + observed_findings, + int(resource_metrics.get("observed_findings", 0)), + ) + resource_limit = LedgerReason.OUTPUT_LIMIT + resource_metrics = { + "observed_findings": observed_findings, + "limit_findings": limit_findings, + } + except _StaticResourceLimitError as exc: + path_findings = [] + resource_limit = exc.reason + resource_metrics = exc.metrics except Exception as exc: logger.warning("%s: scan error on %s: %s", analyzer_id, path, exc) event = ledger_event( @@ -1990,7 +2222,7 @@ def run_static_patterns_with_ledger( partial = resource_limit is not None or ( _infer_file_type(path) == "python" and len(content) > MAX_FILE_CHARS - and any(_uses_python_ast(module) for module in pattern_modules) + and _requires_python_ast(pattern_modules) ) partial_reason = resource_limit or LedgerReason.SIZE_LIMIT event = ledger_event( diff --git a/tests/nodes/analyzers/test_static_runner_filtering.py b/tests/nodes/analyzers/test_static_runner_filtering.py index c792f5fb5..005db28d4 100644 --- a/tests/nodes/analyzers/test_static_runner_filtering.py +++ b/tests/nodes/analyzers/test_static_runner_filtering.py @@ -1107,6 +1107,419 @@ def test_non_documentation_paths_not_matched(self, path: str) -> None: class TestInspectionLedgerResponse: + def test_postprocessor_time_is_included_in_runtime_ledger( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + now = [0.0] + monkeypatch.setattr(static_runner.time, "monotonic", lambda: now[0]) + monkeypatch.setattr(static_runner, "MAX_STATIC_ANALYSIS_SECONDS_PER_ARTIFACT", 30.0) + + class SlowPostprocessingModule: + ANALYZER_ID = "slow_postprocessed_static" + + @staticmethod + def analyze(*, content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: + del content, file_type + return [ + AnalyzerFinding( + rule_id="T1", + message="candidate", + severity=Severity.HIGH, + location=Location(file=file_path, start_line=1), + ) + ] + + @staticmethod + def postprocess_path_findings(content: str, findings: list) -> list: + del content + now[0] = 31.0 + return findings + + response = static_runner.run_static_patterns_with_ledger( + {"components": ["input.md"], "file_cache": {"input.md": "input"}}, + [SlowPostprocessingModule], + ) + event = response["inspection_ledger"][0] + + assert event["outcome"] == "partial" + assert event["reason_code"] == "runtime_limit" + assert event["observed_seconds"] == 31.0 + assert event["limit_seconds"] == 30.0 + + def test_postprocessor_deadline_runs_private_evidence_cleanup( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + now = [0.0] + cleanup_calls = 0 + monkeypatch.setattr(static_runner.time, "monotonic", lambda: now[0]) + monkeypatch.setattr(static_runner, "MAX_STATIC_ANALYSIS_SECONDS_PER_ARTIFACT", 30.0) + + class SlowPostprocessingModule: + ANALYZER_ID = "slow_postprocessed_static" + + @staticmethod + def analyze(*, content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: + del content, file_type + return [ + AnalyzerFinding( + rule_id="T1", + message="candidate", + severity=Severity.HIGH, + location=Location(file=file_path, start_line=1), + evidence={"_private_intermediate": "scan"}, + ) + ] + + @staticmethod + def postprocess_path_findings(content: str, findings: list) -> list: + del content + findings[0].evidence["_private_intermediate"] = "postprocess" + now[0] = 31.0 + return findings + + @staticmethod + def cleanup_path_findings(findings: list) -> list: + nonlocal cleanup_calls + cleanup_calls += 1 + for finding in findings: + finding.evidence.pop("_private_intermediate", None) + return findings + + response = static_runner.run_static_patterns_with_ledger( + {"components": ["input.md"], "file_cache": {"input.md": "input"}}, + [SlowPostprocessingModule], + ) + event = response["inspection_ledger"][0] + + assert cleanup_calls == 1 + assert response["findings"][0].evidence == {} + assert event["outcome"] == "partial" + assert event["reason_code"] == "runtime_limit" + + def test_postprocessor_runtime_limit_supersedes_prior_output_limit( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + now = [0.0] + monkeypatch.setattr(static_runner.time, "monotonic", lambda: now[0]) + monkeypatch.setattr(static_runner, "MAX_STATIC_ANALYSIS_SECONDS_PER_ARTIFACT", 30.0) + monkeypatch.setattr(static_runner, "MAX_FINDINGS_PER_ARTIFACT", 1) + + class SlowLimitedModule: + ANALYZER_ID = "slow_limited_static" + + @staticmethod + def analyze(*, content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: + del content, file_type + return [ + AnalyzerFinding( + rule_id="T1", + message="candidate", + severity=Severity.HIGH, + location=Location(file=file_path, start_line=line), + ) + for line in (1, 2) + ] + + @staticmethod + def postprocess_path_findings(content: str, findings: list) -> list: + del content + now[0] = 31.0 + return findings + + response = static_runner.run_static_patterns_with_ledger( + {"components": ["input.md"], "file_cache": {"input.md": "input"}}, + [SlowLimitedModule], + ) + event = response["inspection_ledger"][0] + + assert event["outcome"] == "partial" + assert event["reason_code"] == "runtime_limit" + assert event["observed_seconds"] == 31.0 + assert event["limit_seconds"] == 30.0 + + def test_postprocessor_is_skipped_after_scan_runtime_limit( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + now = [0.0] + postprocess_called = False + monkeypatch.setattr(static_runner.time, "monotonic", lambda: now[0]) + monkeypatch.setattr(static_runner, "MAX_STATIC_ANALYSIS_SECONDS_PER_ARTIFACT", 30.0) + + class ExpiredBeforePostprocessingModule: + ANALYZER_ID = "expired_before_postprocessing_static" + + @staticmethod + def analyze(*, content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: + del content, file_type + finding = AnalyzerFinding( + rule_id="T1", + message="candidate", + severity=Severity.HIGH, + location=Location(file=file_path, start_line=1), + ) + now[0] = 31.0 + return [finding] + + @staticmethod + def postprocess_path_findings(content: str, findings: list) -> list: + nonlocal postprocess_called + del content + postprocess_called = True + raise AssertionError("postprocessor must not run after the shared deadline") + + response = static_runner.run_static_patterns_with_ledger( + {"components": ["input.md"], "file_cache": {"input.md": "input"}}, + [ExpiredBeforePostprocessingModule], + ) + event = response["inspection_ledger"][0] + + assert not postprocess_called + assert event["outcome"] == "partial" + assert event["reason_code"] == "runtime_limit" + assert event["observed_seconds"] == 31.0 + assert event["limit_seconds"] == 30.0 + + def test_postprocessor_runs_before_findings_and_ledger_ids_are_committed(self) -> None: + class PostprocessingModule: + ANALYZER_ID = "postprocessed_static" + + @staticmethod + def analyze(*, content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: + del content, file_type + return [ + AnalyzerFinding( + rule_id="T1", + message="candidate", + severity=Severity.HIGH, + location=Location(file=file_path, start_line=line), + matched_text=f"match-{line}", + ) + for line in (1, 2) + ] + + @staticmethod + def postprocess_path_findings(content: str, findings: list) -> list: + assert content == "input\nsecond" + return findings[1:] + + state = { + "components": ["input.md"], + "file_cache": {"input.md": "input\nsecond"}, + } + response = static_runner.run_static_patterns_with_ledger( + state, + [PostprocessingModule], + ) + findings = response["findings"] + + assert len(findings) == 1 + assert findings[0].start_line == 2 + assert response["inspection_ledger"][0]["emitted_finding_ids"] == [findings[0].finding_id] + assert static_runner.run_static_patterns(state, [PostprocessingModule])[0].start_line == 2 + + def test_ast_aware_postprocessor_requests_shared_parse_without_ast_analyzer(self) -> None: + class AstPostprocessingModule: + ANALYZER_ID = "ast_postprocessed_static" + POSTPROCESS_USES_PYTHON_AST = True + + @staticmethod + def analyze(*, content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: + del content, file_type + return [ + AnalyzerFinding( + rule_id="T1", + message="candidate", + severity=Severity.HIGH, + location=Location(file=file_path, start_line=1), + ) + ] + + @staticmethod + def postprocess_path_findings(content: str, findings: list, *, python_ast) -> list: + del content + assert python_ast is not None + assert python_ast.tree is not None + return findings + + state = {"components": ["input.py"], "file_cache": {"input.py": "value = 1\n"}} + + response = static_runner.run_static_patterns_with_ledger( + state, + [AstPostprocessingModule], + ) + + assert len(response["findings"]) == 1 + assert response["inspection_ledger"][0]["outcome"] == "completed" + + def test_ast_aware_postprocessor_marks_oversized_python_partial( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + path = "input.py" + content = "value = 1\n" + monkeypatch.setattr(static_runner, "MAX_FILE_CHARS", 4) + + class AstPostprocessingModule: + ANALYZER_ID = "ast_postprocessed_static" + POSTPROCESS_USES_PYTHON_AST = True + + @staticmethod + def analyze(*, content: str, file_path: str, file_type: str) -> list: + del content, file_path, file_type + return [] + + @staticmethod + def postprocess_path_findings(content: str, findings: list, *, python_ast) -> list: + del content + assert python_ast is None + return findings + + response = static_runner.run_static_patterns_with_ledger( + {"components": [path], "file_cache": {path: content}}, + [AstPostprocessingModule], + ) + event = response["inspection_ledger"][0] + + assert event["outcome"] == "partial" + assert event["reason_code"] == "size_limit" + assert event["observed_characters"] == len(content) + assert event["limit_characters"] == 4 + + def test_ast_aware_postprocessor_marks_invalid_python_partial(self) -> None: + class AstPostprocessingModule: + ANALYZER_ID = "ast_postprocessed_static" + POSTPROCESS_USES_PYTHON_AST = True + + @staticmethod + def analyze(*, content: str, file_path: str, file_type: str) -> list: + del content, file_path, file_type + return [] + + @staticmethod + def postprocess_path_findings(content: str, findings: list, *, python_ast) -> list: + del content + assert python_ast is not None + assert python_ast.tree is None + return findings + + response = static_runner.run_static_patterns_with_ledger( + {"components": ["input.py"], "file_cache": {"input.py": "if:\n"}}, + [AstPostprocessingModule], + ) + event = response["inspection_ledger"][0] + + assert event["outcome"] == "partial" + assert event["reason_code"] == "syntax_error" + + def test_nonledger_runner_counts_shared_python_parse_against_deadline( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + now = [0.0] + analyzed = False + + class AstModule: + USES_PYTHON_AST = True + + @staticmethod + def analyze(*, content: str, file_path: str, file_type: str, python_ast) -> list: + nonlocal analyzed + del content, file_path, file_type, python_ast + analyzed = True + return [] + + def delayed_parse(*_args, **_kwargs): + now[0] = 31.0 + return type("Parsed", (), {"tree": object()})() + + monkeypatch.setattr(static_runner.time, "monotonic", lambda: now[0]) + monkeypatch.setattr(static_runner, "get_python_ast", delayed_parse) + monkeypatch.setattr(static_runner, "MAX_STATIC_ANALYSIS_SECONDS_PER_ARTIFACT", 30.0) + + findings = static_runner.run_static_patterns( + {"components": ["input.py"], "file_cache": {"input.py": "value = 1\n"}}, + [AstModule], + ) + + assert findings == [] + assert analyzed is False + + def test_nonledger_runner_discards_unfinished_postprocessing_after_deadline( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + now = [0.0] + postprocessed = False + + class PostprocessingModule: + @staticmethod + def analyze(*, content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: + del content, file_type + return [ + AnalyzerFinding( + rule_id="T1", + message="candidate", + severity=Severity.HIGH, + location=Location(file=file_path, start_line=1), + ) + ] + + @staticmethod + def postprocess_path_findings(content: str, findings: list) -> list: + nonlocal postprocessed + del content + postprocessed = True + now[0] = 31.0 + return findings + + monkeypatch.setattr(static_runner.time, "monotonic", lambda: now[0]) + monkeypatch.setattr(static_runner, "MAX_STATIC_ANALYSIS_SECONDS_PER_ARTIFACT", 30.0) + + findings = static_runner.run_static_patterns( + {"components": ["input.md"], "file_cache": {"input.md": "input"}}, + [PostprocessingModule], + ) + + assert postprocessed is True + assert findings == [] + + def test_postprocessor_cannot_expand_past_per_artifact_output_limit( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(static_runner, "MAX_FINDINGS_PER_ARTIFACT", 1) + + class ExpandingPostprocessorModule: + ANALYZER_ID = "expanding_postprocessed_static" + + @staticmethod + def analyze(*, content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: + del content, file_type + return [ + AnalyzerFinding( + rule_id="T1", + message="candidate", + severity=Severity.HIGH, + location=Location(file=file_path, start_line=1), + ) + ] + + @staticmethod + def postprocess_path_findings(content: str, findings: list) -> list: + del content + return findings * 4 + + state = {"components": ["input.md"], "file_cache": {"input.md": "input"}} + response = static_runner.run_static_patterns_with_ledger( + state, + [ExpandingPostprocessorModule], + ) + event = response["inspection_ledger"][0] + + assert len(response["findings"]) == 1 + assert event["outcome"] == "partial" + assert event["reason_code"] == "output_limit" + assert event["observed_findings"] == 4 + assert event["limit_findings"] == 1 + assert len(static_runner.run_static_patterns(state, [ExpandingPostprocessorModule])) == 1 + def test_static_runner_records_and_recovers_from_pattern_failure(self) -> None: class FailingPatternModule: ANALYZER_ID = "failing_static" From 2eaf7e46367b3f9f75aba9b026ed2fc158f25939 Mon Sep 17 00:00:00 2001 From: Christopher Kevin Date: Wed, 16 Sep 2026 19:36:30 -0700 Subject: [PATCH 2/4] fix(cli): preserve recursive failure reporting Signed-off-by: Christopher Kevin --- src/skillspector/cli.py | 545 ++++++++++--- src/skillspector/inspection_ledger.py | 4 + tests/unit/test_cli.py | 1078 ++++++++++++++++++++++++- 3 files changed, 1483 insertions(+), 144 deletions(-) diff --git a/src/skillspector/cli.py b/src/skillspector/cli.py index 4cb6ef33c..ec849a341 100644 --- a/src/skillspector/cli.py +++ b/src/skillspector/cli.py @@ -176,6 +176,7 @@ class _CachedTransitiveResult: artifact_inventory: list[dict[str, object]] artifact_references: list[dict[str, object]] has_executable_scripts: bool + execution_successful: bool refs: list[str] @@ -188,23 +189,30 @@ class _TransitiveTraversalState: scanned_bytes: int = 0 scanned_artifacts: int = 0 truncation_reasons: list[str] = field(default_factory=list) + resource_limit_reached: bool = False budget_exhausted: bool = False paused_at: float | None = None def note_truncation(self, reason: str) -> None: + self.resource_limit_reached = True + self._note_incomplete(reason) + + def exhaust_traversal(self, reason: str) -> None: + """Record a limit that prevents additional target execution.""" + self.budget_exhausted = True + self.note_truncation(reason) + + def _note_incomplete(self, reason: str) -> None: if len(self.truncation_reasons) >= 256: sentinel = "additional transitive limitations omitted" if self.truncation_reasons[-1] != sentinel: self.truncation_reasons[-1] = sentinel - self.budget_exhausted = True return if reason not in self.truncation_reasons: self.truncation_reasons.append(reason) - if "budget" in reason or "time budget" in reason: - self.budget_exhausted = True def note_child_scan_failure(self, target: str) -> None: - self.note_truncation(f"transitive child scan failed for {target}") + self._note_incomplete(f"transitive child scan failed for {target}") def _ensure_started(self) -> None: if self.started_at is None: @@ -215,16 +223,16 @@ def can_scan_more(self) -> bool: if self.budget_exhausted: return False if self.scanned_targets >= self.budget.max_targets: - self.note_truncation(f"target budget {self.budget.max_targets} reached") + self.exhaust_traversal(f"target budget {self.budget.max_targets} reached") return False if self.remaining_bytes() <= 0: - self.note_truncation(f"byte budget {self.budget.max_bytes} reached") + self.exhaust_traversal(f"byte budget {self.budget.max_bytes} reached") return False if self.remaining_artifacts() <= 0: - self.note_truncation(f"artifact budget {self.budget.max_artifacts} reached") + self.exhaust_traversal(f"artifact budget {self.budget.max_artifacts} reached") return False if self.remaining_seconds() <= 0: - self.note_truncation(f"time budget {self.budget.max_seconds:.0f}s reached") + self.exhaust_traversal(f"time budget {self.budget.max_seconds:.0f}s reached") return False return True @@ -232,9 +240,9 @@ def record_scan(self) -> None: self._ensure_started() self.scanned_targets += 1 if self.remaining_bytes() <= 0: - self.note_truncation(f"byte budget {self.budget.max_bytes} reached") + self.exhaust_traversal(f"byte budget {self.budget.max_bytes} reached") if self.remaining_seconds() <= 0: - self.note_truncation(f"time budget {self.budget.max_seconds:.0f}s reached") + self.exhaust_traversal(f"time budget {self.budget.max_seconds:.0f}s reached") def record_bytes(self, bytes_scanned: int) -> None: self._ensure_started() @@ -256,7 +264,7 @@ def record_artifacts(self, artifacts: int) -> None: self._ensure_started() self.scanned_artifacts += max(0, artifacts) if self.scanned_artifacts > self.budget.max_artifacts: - self.note_truncation(f"artifact budget {self.budget.max_artifacts} reached") + self.exhaust_traversal(f"artifact budget {self.budget.max_artifacts} reached") def pause_deadline(self) -> None: if self.started_at is not None and self.paused_at is None: @@ -629,6 +637,9 @@ def scan( raise typer.Exit(code=2) from exc yara_dir = str(yara_rules_dir.resolve()) if yara_rules_dir else None pre_scan_ledger_events: list[dict[str, object]] = [] + discovery_console = ( + err_console if output is None and format is not FormatChoice.terminal else console + ) if recursive and resolved_path.is_dir(): detection = detect_skills(resolved_path) if not detection.complete: @@ -662,7 +673,7 @@ def scan( ) return if detection.complete and not detection.has_root_skill and len(detection.skills) == 0: - (err_console if format in _MACHINE_READABLE_FORMATS else console).print( + discovery_console.print( "[yellow]Warning:[/yellow] --recursive specified but no sub-skills " "detected. Scanning as single skill." ) @@ -675,7 +686,7 @@ def scan( "with a bounded scan and reporting partial coverage." ) if detection.is_multi_skill: - (err_console if format in _MACHINE_READABLE_FORMATS else console).print( + discovery_console.print( f"[yellow]Warning:[/yellow] Found {len(detection.skills)} skills in " f"this directory. Use --recursive to scan each independently." ) @@ -1189,6 +1200,10 @@ def _scope_finding(finding: Finding) -> Finding: source_digest=source_digest, finding_id_map=finding_id_map, ) + required_failure_event = next( + (event for event in scoped_ledger if _is_failed_ledger_event(event)), + None, + ) retained_finding_ids = {item.finding_id for item in scoped_findings} for event in scoped_ledger: for id_field in ("input_finding_ids", "emitted_finding_ids"): @@ -1220,6 +1235,13 @@ def _scope_finding(finding: Finding) -> Finding: limit=traversal.budget.max_ledger_events, traversal=traversal, ) + if required_failure_event is not None: + scoped_ledger = _ensure_required_failure_event( + scoped_ledger, + required_failure_event, + limit=traversal.budget.max_ledger_events, + traversal=traversal, + ) retained_work_ids = { str(event.get("work_id", "")) for event in scoped_ledger if event.get("work_id") } @@ -1280,6 +1302,7 @@ def _scope_finding(finding: Finding) -> Finding: ), has_executable_scripts=bool(child_result.get("has_executable_scripts", False)) or any(bool(entry.get("executable", False)) for entry in child_metadata), + execution_successful=child_result.get("execution_successful") is not False, refs=extraction.references, ) @@ -1677,6 +1700,64 @@ def _merge_bounded_ledger( ] +def _is_failed_ledger_event(event: dict[str, object]) -> bool: + outcome = event.get("outcome") + return getattr(outcome, "value", outcome) == LedgerOutcome.FAILED.value + + +def _transitive_child_failure_event(source_identity: str) -> dict[str, object]: + """Return one deterministic, payload-free fatal fact for an opaque child failure.""" + return dict( + ledger_event( + outcome=LedgerOutcome.FAILED, + record_type=LedgerRecordType.SYSTEM, + phase="transitive_child_scan", + path=f"{source_identity}/SKILL.md", + reason=LedgerReason.TRANSITIVE_CHILD_SCAN_FAILED, + ) + ) + + +def _ensure_required_failure_event( + events: list[dict[str, object]], + failure: dict[str, object], + *, + limit: int, + traversal: _TransitiveTraversalState, +) -> list[dict[str, object]]: + """Retain a fatal child fact even when the shared ledger reaches its bound.""" + if any( + _is_failed_ledger_event(event) and event.get("work_id") == failure.get("work_id") + for event in events + ): + return events + effective_limit = max(1, limit) + if len(events) < effective_limit: + return [*events, failure] + traversal.note_truncation(f"inspection ledger budget {effective_limit} reached") + if effective_limit == 1: + return [failure] + prior_sentinel = next( + (event for event in reversed(events) if event.get("phase") == "ledger_output"), + None, + ) + observed_value = prior_sentinel.get("observed_records") if prior_sentinel else None + observed_records = observed_value if isinstance(observed_value, int) else len(events) + sentinel = dict( + ledger_event( + outcome=LedgerOutcome.PARTIAL, + record_type=LedgerRecordType.SYSTEM, + phase="ledger_output", + path=str(failure.get("path", "SKILL.md")), + reason=LedgerReason.OUTPUT_LIMIT, + observed_records=max(observed_records, len(events)) + 1, + limit_records=effective_limit, + ) + ) + retained = [event for event in events if event.get("phase") != "ledger_output"] + return [*retained[: effective_limit - 2], failure, sentinel] + + def _scan_transitive( initial_result: dict[str, object], format: FormatChoice, @@ -1729,12 +1810,24 @@ def _scan_transitive( merged_effective_finding_ids = _effective_finding_ids(initial_result)[ : traversal.budget.max_findings ] + root_inspection_ledger = _coerce_dict_list(initial_result.get("inspection_ledger")) + required_root_failure_event = next( + (event for event in root_inspection_ledger if _is_failed_ledger_event(event)), + None, + ) merged_inspection_ledger = _merge_bounded_ledger( [], - _coerce_dict_list(initial_result.get("inspection_ledger")), + root_inspection_ledger, limit=traversal.budget.max_ledger_events, traversal=traversal, ) + if required_root_failure_event is not None: + merged_inspection_ledger = _ensure_required_failure_event( + merged_inspection_ledger, + required_root_failure_event, + limit=traversal.budget.max_ledger_events, + traversal=traversal, + ) retained_work_ids = { str(event.get("work_id", "")) for event in merged_inspection_ledger if event.get("work_id") } @@ -1853,14 +1946,17 @@ def _scan_transitive( cached = _cache_transitive_result(target, child_result, traversal) traversal.cache[cache_key] = cached traversal.record_scan() - if child_result.get("execution_successful") is False: - traversal.note_child_scan_failure(target) - child_completeness = child_result.get("analysis_completeness") - if ( - isinstance(child_completeness, dict) - and child_completeness.get("is_complete") is False + if not cached.execution_successful: + traversal.note_child_scan_failure(target) + if not any( + _is_failed_ledger_event(event) for event in cached.inspection_ledger ): - traversal.note_truncation(f"transitive child scan incomplete for {target}") + cached.inspection_ledger = _ensure_required_failure_event( + cached.inspection_ledger, + _transitive_child_failure_event(cached.source_identity), + limit=traversal.budget.max_ledger_events, + traversal=traversal, + ) transitive_sources.add(target) merged_inspection_ledger = _merge_bounded_ledger( merged_inspection_ledger, @@ -1868,6 +1964,17 @@ def _scan_transitive( limit=traversal.budget.max_ledger_events, traversal=traversal, ) + child_failure_event = next( + (event for event in cached.inspection_ledger if _is_failed_ledger_event(event)), + None, + ) + if child_failure_event is not None: + merged_inspection_ledger = _ensure_required_failure_event( + merged_inspection_ledger, + child_failure_event, + limit=traversal.budget.max_ledger_events, + traversal=traversal, + ) global_work_ids = { str(event.get("work_id", "")) for event in merged_inspection_ledger @@ -2022,7 +2129,15 @@ def _scan_transitive( except Exception: transitive_sources.add(target) traversal.note_child_scan_failure(target) - if format in _MACHINE_READABLE_FORMATS: + merged_inspection_ledger = _ensure_required_failure_event( + merged_inspection_ledger, + _transitive_child_failure_event( + _source_identity(target, "transitive-child-scan-failed") + ), + limit=traversal.budget.max_ledger_events, + traversal=traversal, + ) + if format in _MACHINE_READABLE_FORMATS or format is FormatChoice.markdown: logger.warning("Transitive scan failed for %s", target) else: console.print(f"[yellow]Warning:[/yellow] Transitive scan failed for {target}") @@ -2045,7 +2160,11 @@ def _scan_transitive( traversal=traversal, ) - if traversal.truncation_reasons: + if traversal.resource_limit_reached: + required_failure_event = next( + (event for event in merged_inspection_ledger if _is_failed_ledger_event(event)), + None, + ) traversal_event = ledger_event( outcome=LedgerOutcome.PARTIAL, record_type=LedgerRecordType.SYSTEM, @@ -2059,6 +2178,22 @@ def _scan_transitive( limit=traversal.budget.max_ledger_events, traversal=traversal, ) + if required_failure_event is not None: + merged_inspection_ledger = _ensure_required_failure_event( + merged_inspection_ledger, + required_failure_event, + limit=traversal.budget.max_ledger_events, + traversal=traversal, + ) + + retained_work_ids = { + str(event.get("work_id", "")) for event in merged_inspection_ledger if event.get("work_id") + } + merged_analyzer_status_events = _bounded_root_status_events( + merged_analyzer_status_events, + retained_work_ids=retained_work_ids, + limit=traversal.budget.max_status_events, + ) merged_result: dict[str, object] = { **initial_result, @@ -2095,6 +2230,10 @@ def _scan_transitive( result=merged_result, discovered_modules=ANALYZER_MODULES, ) + pre_runtime_failure_event = next( + (event for event in merged_inspection_ledger if _is_failed_ledger_event(event)), + None, + ) if runtime_event is not None and not has_semantic_runtime_event( merged_inspection_ledger, runtime_event ): @@ -2104,6 +2243,13 @@ def _scan_transitive( limit=traversal.budget.max_ledger_events, traversal=traversal, ) + if pre_runtime_failure_event is not None: + merged_inspection_ledger = _ensure_required_failure_event( + merged_inspection_ledger, + pre_runtime_failure_event, + limit=traversal.budget.max_ledger_events, + traversal=traversal, + ) merged_result["inspection_ledger"] = merged_inspection_ledger if merged_inspection_ledger or merged_analyzer_status_events: completeness, effective_ids = finalize_ledger(merged_result) @@ -2154,9 +2300,8 @@ def _scan_skill( yara_dir = str(yara_rules_dir.resolve()) if yara_rules_dir else None active_visited: set[str] = set() if verbose: - (err_console if format in _MACHINE_READABLE_FORMATS else console).print( - "[dim]Running scan...[/dim]" - ) + progress_console = console if format is FormatChoice.terminal else err_console + progress_console.print("[dim]Running scan...[/dim]") logger.debug( "Scan started: input_path=%s, format=%s, use_llm=%s, transitive=%s", input_path, @@ -2270,10 +2415,121 @@ def _multi_skill_analysis_completeness( } +_RECURSIVE_SERIALIZED_OUTPUT_LIMIT_PREFIX = "recursive serialized report character budget " +_RECURSIVE_CHILD_SCAN_FAILED_MESSAGE = "A recursive child scan failed before complete inspection." + + +def _multi_skill_sarif_notifications( + completeness: dict[str, object], +) -> list[dict[str, object]]: + """Project exact aggregate outcomes without inventing an output-limit cause.""" + notifications: list[dict[str, object]] = [] + status = str(completeness.get("status", "partial")) + raw_limitations = completeness.get("limitations") + limitations = ( + [str(item) for item in raw_limitations if isinstance(item, str)] + if isinstance(raw_limitations, list) + else [] + ) + if status == "failed": + notifications.append( + { + "message": {"text": "One or more recursive skill scans failed."}, + "level": "error", + "properties": {"kind": "inspection_failure"}, + } + ) + for limitation in limitations: + properties: dict[str, object] = {"kind": "inspection_limitation"} + if limitation.startswith(_RECURSIVE_SERIALIZED_OUTPUT_LIMIT_PREFIX): + properties["reasonCode"] = LedgerReason.OUTPUT_LIMIT.value + notifications.append( + { + "message": {"text": limitation}, + "level": "warning", + "properties": properties, + } + ) + if status == "partial" and not limitations: + notifications.append( + { + "message": { + "text": "One or more recursive skill scans were incomplete; " + "see child run notifications." + }, + "level": "warning", + "properties": {"kind": "inspection_limitation"}, + } + ) + return notifications + + +def _multi_skill_text_completeness(completeness: dict[str, object]) -> str: + """Render the aggregate state without downgrading a failed child to partial.""" + status = str(completeness.get("status", "partial")) + raw_limitations = completeness.get("limitations") + limitations = ( + [str(item) for item in raw_limitations if isinstance(item, str)] + if isinstance(raw_limitations, list) + else [] + ) + if status == "failed": + limitations.insert(0, "One or more recursive skill scans failed.") + elif status == "partial" and not limitations: + limitations.append("One or more recursive skill scans were incomplete.") + details = "\n".join(f"- {item}" for item in limitations) + return f"--- Recursive Inspection Completeness ---\n\nStatus: {status}\n\n{details}" + + +def _multi_skill_risk_assessment( + max_score: int, + *, + execution_failed: bool, + analysis_incomplete: bool, +) -> dict[str, object]: + """Return bounded aggregate risk evidence independent of child retention.""" + if max_score >= 81: + severity = "CRITICAL" + elif max_score >= 51: + severity = "HIGH" + elif max_score >= 21: + severity = "MEDIUM" + else: + severity = "LOW" + recommendation = ( + "DO_NOT_INSTALL" + if execution_failed or max_score > RISK_THRESHOLD + else "CAUTION" + if analysis_incomplete + else "SAFE" + ) + return { + "max_risk_score": max_score, + "severity": severity, + "recommendation": recommendation, + } + + +def _multi_skill_text_summary( + completeness: dict[str, object], + risk_assessment: dict[str, object], +) -> str: + """Render aggregate risk and completeness even when child bodies are omitted.""" + recommendation = str(risk_assessment.get("recommendation", "CAUTION")).replace("_", " ") + risk = ( + "--- Recursive Risk Assessment ---\n\n" + f"Maximum score: {risk_assessment.get('max_risk_score', 0)}/100\n\n" + f"Severity: {risk_assessment.get('severity', 'LOW')}\n\n" + f"Recommendation: {recommendation}" + ) + return f"{risk}\n\n{_multi_skill_text_completeness(completeness)}" + + def _multi_skill_sarif_report( processed_skills: list[SkillDirectory], results: list[dict[str, object]], completeness: dict[str, object], + risk_assessment: dict[str, object] | None = None, ) -> dict[str, object]: """Merge bounded child SARIF runs and append one aggregate invocation run.""" runs: list[dict[str, object]] = [] @@ -2300,23 +2556,20 @@ def _multi_skill_sarif_report( run["properties"] = run_properties runs.append(run) + invocation_properties: dict[str, object] = {"analysisCompleteness": completeness} + if risk_assessment is not None: + invocation_properties["riskAssessment"] = { + "maxRiskScore": risk_assessment.get("max_risk_score", 0), + "severity": risk_assessment.get("severity", "LOW"), + "recommendation": risk_assessment.get("recommendation", "CAUTION"), + } aggregate_invocation: dict[str, object] = { "executionSuccessful": bool(completeness.get("execution_successful", False)), - "properties": {"analysisCompleteness": completeness}, + "properties": invocation_properties, } - if not bool(completeness.get("is_complete", False)): - aggregate_invocation["toolExecutionNotifications"] = [ - { - "message": { - "text": "Recursive analysis was incomplete after an aggregate safety limit." - }, - "level": "warning", - "properties": { - "kind": "inspection_limitation", - "reasonCode": "output_limit", - }, - } - ] + notifications = _multi_skill_sarif_notifications(completeness) + if notifications: + aggregate_invocation["toolExecutionNotifications"] = notifications runs.append( { "tool": {"driver": {"name": "skillspector", "version": __version__}}, @@ -2380,10 +2633,10 @@ def _scan_multi_skill( if yara_dir is None and isinstance(legacy_kwargs.get("yara_rules_dir"), Path): yara_dir = str(legacy_kwargs["yara_rules_dir"]) skills = detection.skills - status_console = ( - err_console if format in _MACHINE_READABLE_FORMATS and output is None else console + progress_console = ( + err_console if output is None and format is not FormatChoice.terminal else console ) - status_console.print( + progress_console.print( f"[bold]Multi-skill directory detected:[/bold] {len(skills)} skills found\n" ) @@ -2435,7 +2688,7 @@ def _scan_multi_skill( analysis_incomplete = True aggregate_limitations.extend(shared_transitive_traversal.truncation_reasons) break - status_console.print( + progress_console.print( f" [{i}/{len(skills)}] Scanning [bold]{skill.name}[/bold] ({skill.relative_path}/)" ) try: @@ -2455,6 +2708,35 @@ def _scan_multi_skill( transitive_traversal=shared_transitive_traversal, source_local_only=skill.local_only, ) + child_failed = result.get("execution_successful") is False + if child_failed: + execution_failed = True + failed_skill_count += 1 + completeness_value = result.get("analysis_completeness") + if ( + not child_failed + and isinstance(completeness_value, dict) + and not bool(completeness_value.get("is_complete", True)) + ): + analysis_incomplete = True + partial_skill_count += 1 + elif not child_failed: + complete_skill_count += 1 + score = result.get("risk_score") or 0 + try: + score = int(score) + except (TypeError, ValueError): + score = 0 + if score > max_score: + max_score = score + child_transitive_count = result.get("transitive_finding_count") + if isinstance(child_transitive_count, int): + transitive_finding_count += child_transitive_count + for source in _coerce_str_path_list(result.get("transitive_sources")): + transitive_sources.add(source) + severity = result.get("risk_severity") or "LOW" + progress_console.print(f" Score: {score}/100 ({severity})\n") + result_body = _result_body(result) result_characters = len(result_body) result_records = _multi_skill_public_record_count(result) @@ -2484,47 +2766,30 @@ def _scan_multi_skill( processed_skills.append(skill) retained_public_records += result_records retained_report_characters += result_characters - child_failed = result.get("execution_successful") is False - if child_failed: - execution_failed = True - failed_skill_count += 1 - completeness_value = result.get("analysis_completeness") - if ( - not child_failed - and isinstance(completeness_value, dict) - and not bool(completeness_value.get("is_complete", True)) - ): - analysis_incomplete = True - partial_skill_count += 1 - elif not child_failed: - complete_skill_count += 1 - score = result.get("risk_score") or 0 - try: - score = int(score) - except (TypeError, ValueError): - score = 0 - if score > max_score: - max_score = score - child_transitive_count = result.get("transitive_finding_count") - if isinstance(child_transitive_count, int): - transitive_finding_count += child_transitive_count - for source in _coerce_str_path_list(result.get("transitive_sources")): - transitive_sources.add(source) - severity = result.get("risk_severity") or "LOW" - status_console.print(f" Score: {score}/100 ({severity})\n") - except Exception as e: - error_message = str(e)[:1_024] + except Exception: + error_message = _RECURSIVE_CHILD_SCAN_FAILED_MESSAGE err_console.print(f" [red]Error:[/red] {error_message}\n") execution_failed = True failed_skill_count += 1 results.append({"skill_name": skill.name, "error": error_message}) processed_skills.append(skill) - omitted_skill_count = len(skills) - len(processed_skills) - if omitted_skill_count: + scanned_skill_count = complete_skill_count + partial_skill_count + failed_skill_count + unscanned_skill_count = max( + 0, + len(skills) - scanned_skill_count, + ) + output_omitted_skill_count = max(0, scanned_skill_count - len(processed_skills)) + if output_omitted_skill_count: + analysis_incomplete = True + aggregate_limitations.append( + f"{output_omitted_skill_count} scanned recursive skill report(s) omitted " + "after an aggregate output limit" + ) + if unscanned_skill_count: analysis_incomplete = True aggregate_limitations.append( - f"{omitted_skill_count} recursive skill(s) omitted after an aggregate limit" + f"{unscanned_skill_count} recursive skill(s) unscanned after an aggregate limit" ) aggregate_limitations = list(dict.fromkeys(aggregate_limitations))[:256] aggregate_completeness = _multi_skill_analysis_completeness( @@ -2532,20 +2797,25 @@ def _scan_multi_skill( complete_skills=complete_skill_count, partial_skills=partial_skill_count, failed_skills=failed_skill_count, - omitted_skills=omitted_skill_count, + omitted_skills=unscanned_skill_count, limitations=aggregate_limitations, ) analysis_incomplete = not bool(aggregate_completeness["is_complete"]) + aggregate_risk_assessment = _multi_skill_risk_assessment( + max_score, + execution_failed=execution_failed, + analysis_incomplete=analysis_incomplete, + ) - status_console.print("\n[bold]═══ Multi-Skill Summary ═══[/bold]\n") - status_console.print( + progress_console.print("\n[bold]═══ Multi-Skill Summary ═══[/bold]\n") + progress_console.print( f" {'Skill':<30} {'Score':<8} {'Severity':<12} {'Findings':<10} {'Execution':<10}" ) - status_console.print(f" {'─' * 30} {'─' * 8} {'─' * 12} {'─' * 10} {'─' * 10}") + progress_console.print(f" {'─' * 30} {'─' * 8} {'─' * 12} {'─' * 10} {'─' * 10}") for skill, result in zip(processed_skills, results, strict=True): if "error" in result: - status_console.print( + progress_console.print( f" {skill.name:<30} {'ERROR':<8} {'—':<12} {'—':<10} {'error':<10}" ) continue @@ -2553,14 +2823,20 @@ def _scan_multi_skill( severity = result.get("risk_severity", "LOW") finding_count = len(effective_findings(result)) execution = "failed" if result.get("execution_successful") is False else "successful" - status_console.print( + progress_console.print( f" {skill.name:<30} {score:<8} {severity:<12} {finding_count:<10} {execution:<10}" ) - if omitted_skill_count: - status_console.print( - f" {'':<30} {'—':<8} {'—':<12} {omitted_skill_count:<10} {'partial':<10}" + if output_omitted_skill_count: + progress_console.print( + f" {'':<30} {'—':<8} {'—':<12} " + f"{output_omitted_skill_count:<10} {'partial':<10}" + ) + if unscanned_skill_count: + progress_console.print( + f" {'':<30} {'—':<8} {'—':<12} {unscanned_skill_count:<10} {'partial':<10}" ) - status_console.print( + if output_omitted_skill_count or unscanned_skill_count: + progress_console.print( "[yellow]Recursive scan incomplete:[/yellow] one or more skills were omitted " "after an aggregate safety limit." ) @@ -2570,17 +2846,13 @@ def _scan_multi_skill( "multi_skill": True, "skill_count": len(skills), "max_risk_score": max_score, + "risk_severity": aggregate_risk_assessment["severity"], "execution_successful": not execution_failed, - "risk_recommendation": ( - "DO_NOT_INSTALL" - if execution_failed or max_score > RISK_THRESHOLD - else "CAUTION" - if analysis_incomplete - else "SAFE" - ), + "risk_recommendation": aggregate_risk_assessment["recommendation"], "analysis_completeness": aggregate_completeness, - "skills_scanned": len(processed_skills), - "skills_omitted": omitted_skill_count, + "skills_scanned": scanned_skill_count, + "skills_omitted": unscanned_skill_count, + "skills_output_omitted": output_omitted_skill_count, "public_finding_records": retained_public_records, "report_characters": retained_report_characters, "transitive_finding_count": transitive_finding_count, @@ -2614,11 +2886,19 @@ def _scan_multi_skill( combined_skills.append(entry) entry["transitive_finding_count"] = result.get("transitive_finding_count", 0) entry["transitive_sources"] = result.get("transitive_sources", []) - if omitted_skill_count: + if output_omitted_skill_count: combined_skills.append( { "omitted": True, - "omitted_count": omitted_skill_count, + "omitted_count": output_omitted_skill_count, + "reason": "aggregate_output_limit", + } + ) + if unscanned_skill_count: + combined_skills.append( + { + "omitted": True, + "omitted_count": unscanned_skill_count, "reason": "aggregate_scan_limit", } ) @@ -2632,39 +2912,41 @@ def _scan_multi_skill( "multi_skill": True, "skill_count": len(skills), "max_risk_score": max_score, + "risk_severity": aggregate_risk_assessment["severity"], "execution_successful": not execution_failed, - "risk_recommendation": ( - "DO_NOT_INSTALL" - if execution_failed or max_score > RISK_THRESHOLD - else "CAUTION" - ), + "risk_recommendation": _multi_skill_risk_assessment( + max_score, + execution_failed=execution_failed, + analysis_incomplete=True, + )["recommendation"], "analysis_completeness": aggregate_completeness, - "skills_scanned": len(processed_skills), - "skills_omitted": omitted_skill_count, - "skills_output_omitted": len(processed_skills), + "skills_scanned": scanned_skill_count, + "skills_omitted": unscanned_skill_count, + "skills_output_omitted": scanned_skill_count, "public_finding_records": 0, "transitive_finding_count": transitive_finding_count, "transitive_sources": [], "skills": [ { "omitted": True, - "omitted_count": len(processed_skills), + "omitted_count": scanned_skill_count, "reason": "aggregate_output_limit", } ], } rendered = json.dumps(combined, indent=2) _ensure_recursive_output_bound(rendered) - if output: + if output is not None: Path(output).write_text(rendered, encoding="utf-8") - console.print(f"[green]Combined report saved to:[/green] {output}") + progress_console.print(f"[green]Combined report saved to:[/green] {output}") else: - print(rendered) + sys.stdout.write(rendered) elif format == FormatChoice.sarif: merged_sarif = _multi_skill_sarif_report( processed_skills, results, aggregate_completeness, + aggregate_risk_assessment, ) rendered = json.dumps(merged_sarif, indent=2) if len(rendered) > _MULTI_SKILL_MAX_REPORT_CHARACTERS: @@ -2672,23 +2954,32 @@ def _scan_multi_skill( aggregate_completeness, aggregate_limitations = _mark_recursive_output_limited( aggregate_completeness, ) - merged_sarif = _multi_skill_sarif_report([], [], aggregate_completeness) + aggregate_risk_assessment = _multi_skill_risk_assessment( + max_score, + execution_failed=execution_failed, + analysis_incomplete=True, + ) + merged_sarif = _multi_skill_sarif_report( + [], [], aggregate_completeness, aggregate_risk_assessment + ) rendered = json.dumps(merged_sarif, indent=2) _ensure_recursive_output_bound(rendered) - if output: + if output is not None: Path(output).write_text(rendered, encoding="utf-8") - console.print(f"[green]Combined report saved to:[/green] {output}") + progress_console.print(f"[green]Combined report saved to:[/green] {output}") else: - print(rendered) - elif output: + sys.stdout.write(rendered) + else: sections: list[str] = [] for skill, result in zip(processed_skills, results, strict=True): if "error" not in result: sections.append(f"--- {skill.relative_path} ---\n\n{_result_body(result)}") if analysis_incomplete: sections.append( - "--- Recursive Inspection Completeness ---\n\n" - "Status: partial\n\n" + "\n".join(f"- {item}" for item in aggregate_limitations) + _multi_skill_text_summary( + aggregate_completeness, + aggregate_risk_assessment, + ) ) rendered = "\n\n".join(sections) if len(rendered) > _MULTI_SKILL_MAX_REPORT_CHARACTERS: @@ -2696,13 +2987,23 @@ def _scan_multi_skill( aggregate_completeness, aggregate_limitations = _mark_recursive_output_limited( aggregate_completeness, ) - rendered = ( - "--- Recursive Inspection Completeness ---\n\n" - "Status: partial\n\n" + "\n".join(f"- {item}" for item in aggregate_limitations) + aggregate_risk_assessment = _multi_skill_risk_assessment( + max_score, + execution_failed=execution_failed, + analysis_incomplete=True, + ) + rendered = _multi_skill_text_summary( + aggregate_completeness, + aggregate_risk_assessment, ) _ensure_recursive_output_bound(rendered) - Path(output).write_text(rendered, encoding="utf-8") - console.print(f"[green]Combined report saved to:[/green] {output}") + if output is not None: + Path(output).write_text(rendered, encoding="utf-8") + progress_console.print(f"[green]Combined report saved to:[/green] {output}") + elif format is FormatChoice.terminal: + console.print(rendered) + else: + sys.stdout.write(rendered) for result in results: cleanup_result(result) diff --git a/src/skillspector/inspection_ledger.py b/src/skillspector/inspection_ledger.py index 045e34f2b..4fe3f4a0c 100644 --- a/src/skillspector/inspection_ledger.py +++ b/src/skillspector/inspection_ledger.py @@ -93,6 +93,7 @@ class LedgerReason(StrEnum): RUNTIME_LIMIT = "runtime_limit" EXCLUDED_EXECUTABLE_CONTENT = "excluded_executable_content" OUTPUT_LIMIT = "output_limit" + TRANSITIVE_CHILD_SCAN_FAILED = "transitive_child_scan_failed" STATIC_PARSE_LIMIT = "static_parse_limit" OBFUSCATED_INSTRUCTION_TEXT = "obfuscated_instruction_text" @@ -186,6 +187,9 @@ class LedgerReason(StrEnum): "Executable content was inventoried but excluded from content analysis." ), LedgerReason.OUTPUT_LIMIT: "Inspection reached its configured output limit.", + LedgerReason.TRANSITIVE_CHILD_SCAN_FAILED: ( + "A transitive child scan failed before complete inspection." + ), LedgerReason.STATIC_PARSE_LIMIT: ( "A security-relevant expression exceeded a bounded static parser's span limit." ), diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 3208b9872..7a54192c2 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -474,7 +474,7 @@ def test_recursive_scan_exception_marks_combined_execution_as_failed(tmp_path: P "skillspector.cli.graph.invoke", side_effect=[ {"report_body": '{"skill": {"name": "one"}}', "risk_score": 0}, - RuntimeError("child scan crashed"), + RuntimeError("TOKEN=child-scan-secret"), ], ): with pytest.raises(typer.Exit) as exit_info: @@ -490,7 +490,11 @@ def test_recursive_scan_exception_marks_combined_execution_as_failed(tmp_path: P assert exit_info.value.exit_code == 2 payload = json.loads(output.read_text()) assert payload["execution_successful"] is False - assert payload["skills"][1] == {"name": "two", "error": "child scan crashed"} + assert payload["skills"][1] == { + "name": "two", + "error": "A recursive child scan failed before complete inspection.", + } + assert "TOKEN=child-scan-secret" not in output.read_text() def test_recursive_scan_string_risk_score_counts_toward_exit_code(tmp_path: Path) -> None: @@ -632,7 +636,11 @@ def test_recursive_dot_child_static_finding_never_reaches_a_provider( transports: list[MagicMock] = [] def structured_output(schema: type) -> MagicMock: - response = schema(findings=[]) + response = ( + schema(is_mismatch=False) + if "is_mismatch" in schema.model_fields + else schema(findings=[]) + ) transport = MagicMock( invoke=MagicMock(return_value=response), ainvoke=AsyncMock(return_value=response), @@ -1360,8 +1368,9 @@ def test_scan_multi_skill_json_stdout_survives_child_failure( captured = capsys.readouterr() payload = json.loads(captured.out) assert payload["execution_successful"] is False - assert payload["skills"][1] == {"name": "broken", "error": "boom"} - assert "Error: boom" in captured.err + expected_error = "A recursive child scan failed before complete inspection." + assert payload["skills"][1] == {"name": "broken", "error": expected_error} + assert f"Error: {expected_error}" in captured.err @pytest.mark.parametrize("output_format", ["json", "sarif"]) @@ -1567,6 +1576,195 @@ def fake_invoke(*_args, **_kwargs) -> dict[str, object]: } +def test_recursive_oversized_failed_child_preserves_fatal_aggregate( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """A child failure is classified before its oversized body is omitted.""" + skill = SkillDirectory(tmp_path / "failed", "failed", "failed") + output = tmp_path / "combined.json" + monkeypatch.setattr(cli, "_MULTI_SKILL_MAX_REPORT_CHARACTERS", 1_000) + child = { + **_bounded_recursive_result("failed", finding_count=0), + "report_body": "x" * 1_001, + "execution_successful": False, + "analysis_completeness": { + "is_complete": False, + "status": "failed", + "execution_successful": False, + }, + } + monkeypatch.setattr(cli, "_scan_skill", lambda *args, **kwargs: child) + + with pytest.raises(typer.Exit) as exit_info: + cli._scan_multi_skill( + MultiSkillDetectionResult(is_multi_skill=True, skills=[skill]), + FormatChoice.json, + output, + no_llm=True, + ) + + assert exit_info.value.exit_code == 2 + payload = json.loads(output.read_text(encoding="utf-8")) + assert payload["execution_successful"] is False + assert payload["risk_recommendation"] == "DO_NOT_INSTALL" + assert payload["analysis_completeness"]["status"] == "failed" + assert payload["analysis_completeness"]["execution_successful"] is False + assert payload["skills_scanned"] == 1 + assert payload["skills_omitted"] == 0 + assert payload["skills_output_omitted"] == 1 + assert "x" * 1_001 not in output.read_text(encoding="utf-8") + + +def test_recursive_over_record_budget_child_preserves_risk_and_exit( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Risk is aggregated before an over-record-budget child body is omitted.""" + skill = SkillDirectory(tmp_path / "critical", "critical", "critical") + output = tmp_path / "combined.json" + monkeypatch.setattr(cli, "_MULTI_SKILL_MAX_PUBLIC_RECORDS", 1) + child = { + **_bounded_recursive_result("critical", finding_count=2), + "risk_score": 100, + "risk_severity": "CRITICAL", + "risk_recommendation": "DO_NOT_INSTALL", + } + monkeypatch.setattr(cli, "_scan_skill", lambda *args, **kwargs: child) + + with pytest.raises(typer.Exit) as exit_info: + cli._scan_multi_skill( + MultiSkillDetectionResult(is_multi_skill=True, skills=[skill]), + FormatChoice.json, + output, + no_llm=True, + ) + + assert exit_info.value.exit_code == 1 + payload = json.loads(output.read_text(encoding="utf-8")) + assert payload["max_risk_score"] == 100 + assert payload["risk_recommendation"] == "DO_NOT_INSTALL" + assert payload["analysis_completeness"]["is_complete"] is False + assert payload["skills_scanned"] == 1 + assert payload["skills_omitted"] == 0 + assert payload["skills_output_omitted"] == 1 + assert payload["skills"][-1] == { + "omitted": True, + "omitted_count": 1, + "reason": "aggregate_output_limit", + } + + +@pytest.mark.parametrize( + "output_format", + list(FormatChoice), +) +@pytest.mark.parametrize("cap_kind", ["child-retention", "serialized-output"]) +def test_recursive_non_json_caps_preserve_aggregate_risk( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + output_format: FormatChoice, + cap_kind: str, +) -> None: + """Omitted child details never erase known high aggregate risk.""" + relative_path = "critical" + child = { + **_bounded_recursive_result("critical", finding_count=2), + "risk_score": 100, + "risk_severity": "CRITICAL", + "risk_recommendation": "DO_NOT_INSTALL", + } + if cap_kind == "child-retention": + monkeypatch.setattr(cli, "_MULTI_SKILL_MAX_PUBLIC_RECORDS", 1) + elif output_format in {FormatChoice.terminal, FormatChoice.markdown}: + monkeypatch.setattr(cli, "_MULTI_SKILL_MAX_REPORT_CHARACTERS", 1_000) + relative_path = "p" * 400 + child["report_body"] = "x" * 700 + elif output_format is FormatChoice.json: + monkeypatch.setattr(cli, "_MULTI_SKILL_MAX_REPORT_CHARACTERS", 900) + child["report_body"] = json.dumps({"padding": "x" * 700}) + else: + monkeypatch.setattr(cli, "_MULTI_SKILL_MAX_REPORT_CHARACTERS", 1_800) + sarif = cast(dict[str, object], child["sarif_report"]) + runs = cast(list[dict[str, object]], sarif["runs"]) + runs[0]["properties"] = {"padding": "x" * 2_000} + + skill = SkillDirectory(tmp_path / "critical", "critical", relative_path) + output = tmp_path / f"combined.{output_format.value}" + monkeypatch.setattr(cli, "_scan_skill", lambda *args, **kwargs: child) + + with pytest.raises(typer.Exit) as exit_info: + cli._scan_multi_skill( + MultiSkillDetectionResult(is_multi_skill=True, skills=[skill]), + output_format, + output, + no_llm=True, + ) + + assert exit_info.value.exit_code == 1 + body = output.read_text(encoding="utf-8") + if output_format is FormatChoice.json: + payload = json.loads(body) + assert payload["max_risk_score"] == 100 + assert payload["risk_severity"] == "CRITICAL" + assert payload["risk_recommendation"] == "DO_NOT_INSTALL" + elif output_format is FormatChoice.sarif: + payload = json.loads(body) + validate_sarif_report(payload) + aggregate = payload["runs"][-1]["invocations"][0]["properties"] + risk = aggregate["riskAssessment"] + assert risk == { + "maxRiskScore": 100, + "severity": "CRITICAL", + "recommendation": "DO_NOT_INSTALL", + } + else: + assert "Maximum score: 100/100" in body + assert "Severity: CRITICAL" in body + # Human-readable report formats follow the single-skill convention. + assert "Recommendation: DO NOT INSTALL" in body + + +@pytest.mark.parametrize( + "limit_kind", + ["public_records", "child_report_characters"], +) +def test_recursive_sarif_retention_caps_keep_exact_reason_without_output_limit( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + limit_kind: str, +) -> None: + """Pre-serialization retention caps are not mislabeled as output limits.""" + skill = SkillDirectory(tmp_path / "one", "one", "one") + output = tmp_path / "combined.sarif" + child = _bounded_recursive_result( + "one", + finding_count=2 if limit_kind == "public_records" else 0, + ) + if limit_kind == "public_records": + monkeypatch.setattr(cli, "_MULTI_SKILL_MAX_PUBLIC_RECORDS", 1) + expected = "recursive public finding record budget 1 reached" + else: + monkeypatch.setattr(cli, "_MULTI_SKILL_MAX_REPORT_CHARACTERS", 2_000) + child["report_body"] = "x" * 2_001 + expected = "recursive report character budget 2000 reached" + monkeypatch.setattr(cli, "_scan_skill", lambda *args, **kwargs: child) + + cli._scan_multi_skill( + MultiSkillDetectionResult(is_multi_skill=True, skills=[skill]), + FormatChoice.sarif, + output, + no_llm=True, + ) + + payload = json.loads(output.read_text(encoding="utf-8")) + validate_sarif_report(payload) + notifications = payload["runs"][-1]["invocations"][0]["toolExecutionNotifications"] + exact = next(item for item in notifications if item["message"]["text"] == expected) + assert exact["properties"] == {"kind": "inspection_limitation"} + assert not any( + item.get("properties", {}).get("reasonCode") == "output_limit" for item in notifications + ) + + def test_recursive_markdown_report_character_limit_is_explicit( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: @@ -1669,6 +1867,330 @@ def test_recursive_sarif_is_valid_and_carries_aggregate_completeness( assert completeness["is_complete"] is True +@pytest.mark.parametrize( + ("status", "reason_code", "level", "execution_successful"), + [ + ("partial", "static_parse_limit", "warning", True), + ("failed", "analyzer_runtime_error", "error", False), + ], +) +def test_recursive_sarif_preserves_intrinsic_child_state_without_output_limit( + tmp_path: Path, + status: str, + reason_code: str, + level: str, + execution_successful: bool, +) -> None: + """Intrinsic child outcomes remain exact and are not mislabeled as output caps.""" + skill = SkillDirectory(tmp_path / "one", "one", "one") + child = _bounded_recursive_result("one", finding_count=0) + sarif = cast(dict[str, object], child["sarif_report"]) + child_run = cast(list[dict[str, object]], sarif["runs"])[0] + child_run["invocations"] = [ + { + "executionSuccessful": execution_successful, + "toolExecutionNotifications": [ + { + "message": {"text": f"Exact child {status} reason."}, + "level": level, + "properties": { + "kind": "inspection_failure" + if status == "failed" + else "inspection_limitation", + "reasonCode": reason_code, + }, + } + ], + } + ] + completeness = cli._multi_skill_analysis_completeness( + total_skills=1, + complete_skills=0, + partial_skills=int(status == "partial"), + failed_skills=int(status == "failed"), + omitted_skills=0, + limitations=[], + ) + + payload = cli._multi_skill_sarif_report([skill], [child], completeness) + + validate_sarif_report(payload) + child_notifications = payload["runs"][0]["invocations"][0]["toolExecutionNotifications"] + assert child_notifications[0]["properties"]["reasonCode"] == reason_code + aggregate = payload["runs"][-1]["invocations"][0] + assert aggregate["executionSuccessful"] is execution_successful + aggregate_notifications = aggregate["toolExecutionNotifications"] + assert not any( + item.get("properties", {}).get("reasonCode") == "output_limit" + for item in aggregate_notifications + ) + assert "aggregate safety limit" not in json.dumps(aggregate_notifications) + + +def test_recursive_sarif_labels_actual_serialized_output_cap() -> None: + """Only a real recursive output bound uses the output-limit reason code.""" + reason = "recursive serialized report character budget 1800 reached" + completeness = cli._multi_skill_analysis_completeness( + total_skills=1, + complete_skills=0, + partial_skills=1, + failed_skills=0, + omitted_skills=0, + limitations=[reason], + ) + + payload = cli._multi_skill_sarif_report([], [], completeness) + + validate_sarif_report(payload) + aggregate = payload["runs"][-1]["invocations"][0] + notifications = aggregate["toolExecutionNotifications"] + assert notifications == [ + { + "message": {"text": reason}, + "level": "warning", + "properties": { + "kind": "inspection_limitation", + "reasonCode": "output_limit", + }, + } + ] + + +@pytest.mark.parametrize("output_format", [FormatChoice.terminal, FormatChoice.markdown]) +def test_recursive_text_report_labels_failed_aggregate( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + output_format: FormatChoice, +) -> None: + """A failed child is never rendered as merely partial in combined text output.""" + skill = SkillDirectory(tmp_path / "one", "one", "one") + child = _bounded_recursive_result("one", finding_count=0) + child.update( + { + "execution_successful": False, + "analysis_completeness": { + "is_complete": False, + "status": "failed", + "execution_successful": False, + }, + } + ) + monkeypatch.setattr(cli.graph, "invoke", lambda *_args, **_kwargs: child) + output = tmp_path / f"combined.{output_format.value}" + + with pytest.raises(typer.Exit) as exit_info: + _scan_multi_skill( + MultiSkillDetectionResult(is_multi_skill=True, skills=[skill]), + output_format, + output, + no_llm=True, + ) + + assert exit_info.value.exit_code == 2 + body = output.read_text(encoding="utf-8") + assert "Status: failed" in body + assert "Status: partial" not in body + assert "One or more recursive skill scans failed" in body + + +@pytest.mark.parametrize("output_format", list(FormatChoice)) +def test_recursive_no_output_emits_selected_format_with_intrinsic_partial_reason( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + output_format: FormatChoice, +) -> None: + """Without --output, stdout is still the selected bounded recursive report.""" + skill = SkillDirectory(tmp_path / "one", "one", "one") + child = _bounded_recursive_result("one", finding_count=0) + child_completeness = { + "is_complete": False, + "status": "partial", + "execution_successful": True, + "ledger_exceptions": [ + { + "outcome": "partial", + "reason_code": "static_parse_limit", + "message": "A security expression exceeded its bounded parser span.", + "path": "runner", + } + ], + } + child["analysis_completeness"] = child_completeness + child["risk_recommendation"] = "CAUTION" + if output_format is FormatChoice.json: + child["report_body"] = json.dumps({"analysis_completeness": child_completeness}) + elif output_format is FormatChoice.sarif: + sarif = cast(dict[str, object], child["sarif_report"]) + run = cast(list[dict[str, object]], sarif["runs"])[0] + run["invocations"] = [ + { + "executionSuccessful": True, + "toolExecutionNotifications": [ + { + "message": { + "text": "A security expression exceeded its bounded parser span." + }, + "level": "warning", + "properties": { + "kind": "inspection_limitation", + "reasonCode": "static_parse_limit", + }, + } + ], + } + ] + child["report_body"] = json.dumps(sarif) + else: + child["report_body"] = "A security expression exceeded its bounded parser span." + + monkeypatch.setattr(cli, "_scan_skill", lambda *args, **kwargs: child) + cli._scan_multi_skill( + MultiSkillDetectionResult(is_multi_skill=True, skills=[skill]), + output_format, + None, + no_llm=True, + ) + + captured = capsys.readouterr() + if output_format is FormatChoice.json: + payload = json.loads(captured.out) + assert payload["analysis_completeness"]["status"] == "partial" + assert ( + payload["skills"][0]["analysis_completeness"]["ledger_exceptions"][0]["reason_code"] + == "static_parse_limit" + ) + elif output_format is FormatChoice.sarif: + payload = json.loads(captured.out) + validate_sarif_report(payload) + aggregate = payload["runs"][-1]["invocations"][0] + assert aggregate["properties"]["analysisCompleteness"]["status"] == "partial" + assert "static_parse_limit" in captured.out + else: + assert "Recursive Inspection Completeness" in captured.out + assert "Status: partial" in captured.out + assert "A security expression exceeded its bounded parser span" in captured.out + if output_format is not FormatChoice.terminal: + assert "Multi-skill directory detected" not in captured.out + assert "Multi-skill directory detected" in captured.err + + +@pytest.mark.parametrize( + ("output_format", "report_body"), + [ + (FormatChoice.json, "{}"), + (FormatChoice.sarif, '{"version":"2.1.0","runs":[]}'), + (FormatChoice.markdown, "# Report"), + ], +) +@pytest.mark.parametrize("warning_kind", ["recursive-empty", "multi-skill"]) +def test_directory_discovery_warnings_do_not_corrupt_machine_stdout( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + output_format: FormatChoice, + report_body: str, + warning_kind: str, +) -> None: + """Directory discovery diagnostics stay off report-only machine stdout.""" + if warning_kind == "recursive-empty": + detection = MultiSkillDetectionResult( + is_multi_skill=False, + skills=[], + has_root_skill=False, + ) + recursive_args = ["--recursive"] + expected_warning = "no sub-skills detected" + else: + detection = MultiSkillDetectionResult( + is_multi_skill=True, + skills=[ + SkillDirectory(tmp_path / "one", "one", "one"), + SkillDirectory(tmp_path / "two", "two", "two"), + ], + has_root_skill=False, + ) + recursive_args = [] + expected_warning = "Found 2 skills" + + monkeypatch.setattr(cli, "detect_skills", lambda _path: detection) + monkeypatch.setattr( + cli, + "_scan_skill", + lambda *args, **kwargs: { + "report_body": report_body, + "execution_successful": True, + "risk_score": 0, + }, + ) + + result = runner.invoke( + app, + [ + "scan", + str(tmp_path), + *recursive_args, + "--format", + output_format.value, + "--no-llm", + ], + ) + + assert result.exit_code == 0 + assert result.stdout == report_body + "\n" + assert expected_warning not in result.stdout + assert expected_warning in result.stderr + + +@pytest.mark.parametrize("output_format", list(FormatChoice)) +@pytest.mark.parametrize("write_file", [False, True], ids=["stdout", "file"]) +def test_recursive_child_exception_is_sanitized_across_public_formats( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + output_format: FormatChoice, + write_file: bool, +) -> None: + """Recursive child exceptions expose a generic failure, never their payload.""" + skill = SkillDirectory(tmp_path / "one", "one", "one") + output = tmp_path / f"combined.{output_format.value}" if write_file else None + secret = "TOKEN=secret-child-payload" + + def fail_child(*_args: object, **_kwargs: object) -> dict[str, object]: + raise RuntimeError(secret) + + monkeypatch.setattr(cli, "_scan_skill", fail_child) + with pytest.raises(typer.Exit) as exit_info: + cli._scan_multi_skill( + MultiSkillDetectionResult(is_multi_skill=True, skills=[skill]), + output_format, + output, + no_llm=True, + ) + + assert exit_info.value.exit_code == 2 + captured = capsys.readouterr() + public = captured.out + captured.err + if output is not None: + public += output.read_text(encoding="utf-8") + assert secret not in public + assert "A recursive child scan failed before complete inspection." in public + if output_format is FormatChoice.json: + report = output.read_text(encoding="utf-8") if output else captured.out + payload = json.loads(report) + assert payload["execution_successful"] is False + assert payload["skills"][0]["error"] == ( + "A recursive child scan failed before complete inspection." + ) + elif output_format is FormatChoice.sarif: + report = output.read_text(encoding="utf-8") if output else captured.out + payload = json.loads(report) + validate_sarif_report(payload) + assert payload["runs"][-1]["invocations"][0]["executionSuccessful"] is False + else: + report = output.read_text(encoding="utf-8") if output else captured.out + assert "Status: failed" in report + + def test_recursive_sarif_without_output_writes_only_the_log_to_stdout( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture ) -> None: @@ -2463,6 +2985,61 @@ def test_transitive_artifact_budget_allows_exact_limit() -> None: assert traversal.truncation_reasons == ["artifact budget 2 reached"] +@pytest.mark.parametrize( + "reason", + [ + "inspection ledger budget 1 reached", + "analyzer status budget 1 reached", + "finding budget 1 reached", + "component budget 1 reached", + "provider cache budget 1 reached", + "report output budget 1 reached", + ], +) +def test_transitive_retention_limits_do_not_exhaust_execution(reason: str) -> None: + """Public storage truncation remains partial without skipping planned work.""" + traversal = cli._TransitiveTraversalState() + + traversal.note_truncation(reason) + + assert traversal.resource_limit_reached is True + assert traversal.budget_exhausted is False + assert traversal.can_scan_more() is True + assert traversal.truncation_reasons == [reason] + + +def test_child_failure_target_text_cannot_exhaust_traversal_by_substring() -> None: + """An untrusted target containing 'budget' cannot control traversal state.""" + traversal = cli._TransitiveTraversalState() + + traversal.note_child_scan_failure("https://github.com/org/budget") + + assert traversal.budget_exhausted is False + assert traversal.can_scan_more() is True + + +@pytest.mark.parametrize( + ("budget_kwargs", "expected_reason"), + [ + ({"max_targets": 0}, "target budget 0 reached"), + ({"max_bytes": 0}, "byte budget 0 reached"), + ({"max_artifacts": 0}, "artifact budget 0 reached"), + ({"max_seconds": 0.0}, "time budget 0s reached"), + ], +) +def test_transitive_execution_limits_still_stop_planned_work( + budget_kwargs: dict[str, object], expected_reason: str +) -> None: + """Only executable target, byte, artifact, and time ceilings stop traversal.""" + traversal = cli._TransitiveTraversalState( + budget=cli._TransitiveBudget(**budget_kwargs), + ) + + assert traversal.can_scan_more() is False + assert traversal.budget_exhausted is True + assert traversal.truncation_reasons == [expected_reason] + + def test_scan_transitive_depth_one_merges_provenance(tmp_path: Path, monkeypatch) -> None: """--transitive-depth 1 follows one approved external target and merges provenance.""" direct_output = "See dependency: https://github.com/org/transitive.git" @@ -2769,8 +3346,11 @@ def fake_run_graph_scan( assert len(recursive_calls) == 2 -def test_transitive_resolver_failure_preserves_direct_report(tmp_path: Path, monkeypatch) -> None: - """A transitive resolver failure should preserve the direct report result.""" +@pytest.mark.parametrize("strict", [False, True], ids=["default", "strict"]) +def test_transitive_resolver_failure_preserves_fatal_report( + tmp_path: Path, monkeypatch, strict: bool +) -> None: + """A transitive resolver failure writes a sanitized report and exits two.""" target = "https://github.com/org/broken.git" file_cache = {"SKILL.md": f"deps {target}"} @@ -2792,21 +3372,70 @@ def fake_run_graph_scan( raise ValueError("resolver failure") monkeypatch.setattr(cli, "_run_graph_scan", fake_run_graph_scan) - result = runner.invoke( - app, - [ - "scan", - str(tmp_path), - "--format", - "json", - "--transitive", - "--no-llm", - ], - ) - assert result.exit_code == 0 + arguments = [ + "scan", + str(tmp_path), + "--format", + "json", + "--transitive", + "--no-llm", + ] + if strict: + arguments.append("--fail-on-incomplete") + result = runner.invoke(app, arguments) + assert result.exit_code == 2 data = json.loads(result.output) assert len(data["issues"]) == 1 assert data["issues"][0]["id"] == "D1" + assert data["execution_successful"] is False + assert data["analysis_completeness"]["status"] == "failed" + failures = [ + item + for item in data["analysis_completeness"]["ledger_exceptions"] + if item["reason_code"] == "transitive_child_scan_failed" + ] + assert len(failures) == 1 + assert failures[0]["fatal"] is True + assert failures[0]["path"].startswith("external/") + assert target not in failures[0]["path"] + assert "resolver failure" not in result.output + + +def test_transitive_resolver_failure_keeps_sarif_stdout_parseable( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A child warning is diagnostic stderr, never a prefix before SARIF JSON.""" + target = "https://github.com/org/broken-sarif.git" + + def fake_run_graph_scan( + input_path: str, + format, + no_llm: bool, + yara_dir: str | None = None, + baseline=None, + show_suppressed: bool = False, + transitive_traversal=None, + ) -> dict[str, object]: + if input_path == str(tmp_path): + return _mock_graph_result( + file_cache={"SKILL.md": target}, + output_format=format.value, + ) + raise RuntimeError("TOKEN=private-resolver-detail") + + monkeypatch.setattr(cli, "_run_graph_scan", fake_run_graph_scan) + result = runner.invoke( + app, + ["scan", str(tmp_path), "--format", "sarif", "--transitive", "--no-llm"], + ) + + assert result.exit_code == 2 + payload = json.loads(result.stdout) + validate_sarif_report(payload) + invocation = payload["runs"][0]["invocations"][0] + assert invocation["executionSuccessful"] is False + assert "TOKEN=private-resolver-detail" not in result.stdout + assert "Transitive scan failed" not in result.stdout def test_transitive_failure_warning_stays_off_sarif_stdout(tmp_path: Path, monkeypatch) -> None: @@ -2828,7 +3457,7 @@ def fake_run_graph_scan(input_path: str, format, no_llm: bool, **_kwargs) -> dic ["scan", str(tmp_path), "--format", "sarif", "--transitive", "--no-llm"], ) - assert result.exit_code == 0, result.output + assert result.exit_code == 2, result.output assert "Transitive scan failed" not in result.stdout payload = json.loads(result.stdout) validate_sarif_report(payload) @@ -3528,8 +4157,106 @@ def fake_run_graph_scan( assert merged["transitive_finding_count"] == 1 +@pytest.mark.parametrize("output_format", list(cli.FormatChoice)) +def test_scan_transitive_intrinsic_child_partial_is_not_traversal_truncation( + monkeypatch: pytest.MonkeyPatch, output_format: cli.FormatChoice +) -> None: + """A fully traversed partial child keeps its exact cause without a false limit.""" + target = "https://github.com/org/partial" + root_event = ledger_event( + outcome=LedgerOutcome.COMPLETED, + phase="static", + path="SKILL.md", + analyzer_id="root-analyzer", + ) + child_event = ledger_event( + outcome=LedgerOutcome.PARTIAL, + phase="static", + path="runner", + analyzer_id="child-analyzer", + reason=LedgerReason.STATIC_PARSE_LIMIT, + ) + initial_result: dict[str, object] = { + **_mock_graph_result(file_cache={"SKILL.md": target}, output_format=output_format.value), + "local_file_cache": {"SKILL.md": target}, + "component_metadata": [ + { + "path": "SKILL.md", + "type": "markdown", + "lines": 1, + "executable": False, + "size_bytes": len(target), + } + ], + "inspection_ledger": [root_event], + "analyzer_status_events": [analyzer_status_for_events("root-analyzer", [root_event])], + } + child_result: dict[str, object] = { + **_mock_graph_result(file_cache={"runner": "pass\n"}, output_format=output_format.value), + "components": ["runner"], + "local_file_cache": {"runner": "pass\n"}, + "component_metadata": [ + { + "path": "runner", + "type": "other", + "lines": 1, + "executable": True, + "size_bytes": 5, + } + ], + "inspection_ledger": [child_event], + "analyzer_status_events": [analyzer_status_for_events("child-analyzer", [child_event])], + "analysis_completeness": { + "is_complete": False, + "status": "partial", + "execution_successful": True, + }, + "execution_successful": True, + } + + monkeypatch.setattr(cli, "_run_graph_scan", lambda *args, **kwargs: child_result) + merged = cli._scan_transitive( + initial_result=initial_result, + format=output_format, + no_llm=True, + max_depth=1, + transitive_allow_prefix=(), + transitive_deny_prefix=(), + baseline=None, + show_suppressed=False, + visited=set(), + ) + + assert merged["transitive_truncated"] is False + assert merged["transitive_truncation_reasons"] == [] + completeness = cast(dict[str, object], merged["analysis_completeness"]) + assert completeness["is_complete"] is False + assert completeness["status"] == "partial" + assert completeness["execution_successful"] is True + exceptions = cast(list[dict[str, object]], completeness["ledger_exceptions"]) + assert {item["reason_code"] for item in exceptions} == {"static_parse_limit"} + report_body = cast(str, merged["report_body"]) + assert "Inspection reached its configured output limit." not in report_body + assert "Transitive traversal truncated" not in report_body + if output_format is cli.FormatChoice.json: + payload = json.loads(report_body) + assert "transitive_truncated" not in payload["metadata"] + elif output_format is cli.FormatChoice.sarif: + payload = json.loads(report_body) + invocation = payload["runs"][0]["invocations"][0] + assert invocation["executionSuccessful"] is True + reason_codes = { + item["properties"].get("reasonCode") + for item in invocation.get("toolExecutionNotifications", []) + if item["properties"].get("reasonCode") is not None + } + assert reason_codes == {"static_parse_limit"} + else: + assert "bounded static parser's span limit" in report_body + + def test_scan_transitive_child_failure_stays_visible_and_fail_closed(monkeypatch) -> None: - """Child scan exceptions should degrade the report without leaking raw error text.""" + """Child scan exceptions fail execution without leaking raw error text.""" failed_target = "https://github.com/org/broken" initial_result = { "findings": [_finding("D1", "direct finding")], @@ -3585,6 +4312,9 @@ def fake_run_graph_scan( ] assert merged["risk_recommendation"] == "CAUTION" assert body["analysis_completeness"]["is_complete"] is False + assert body["analysis_completeness"]["status"] == "failed" + assert body["analysis_completeness"]["execution_successful"] is False + assert merged["execution_successful"] is False assert body["metadata"]["transitive_truncated"] is True assert any( "transitive child scan failed for https://github.com/org/broken" in limitation @@ -3592,6 +4322,306 @@ def fake_run_graph_scan( ) assert "secret token should stay private" not in merged["transitive_truncation_reasons"][0] assert "secret token should stay private" not in merged["report_body"] + failures = [ + item + for item in body["analysis_completeness"]["ledger_exceptions"] + if item["reason_code"] == "transitive_child_scan_failed" + ] + assert len(failures) == 1 + assert failures[0]["outcome"] == "failed" + assert failures[0]["fatal"] is True + assert failures[0]["path"].startswith("external/") + assert failed_target not in failures[0]["path"] + assert not any( + item["reason_code"] == "output_limit" + for item in body["analysis_completeness"]["ledger_exceptions"] + ) + + +def test_scan_transitive_preserves_returned_child_failure_without_synthetic_duplicate( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An exact FAILED child ledger row remains the sole fatal diagnostic.""" + target = "https://github.com/org/failed-result" + child_event = ledger_event( + outcome=LedgerOutcome.FAILED, + phase="static", + path="runner.py", + analyzer_id="child-analyzer", + reason=LedgerReason.ANALYZER_RUNTIME_ERROR, + ) + initial_result: dict[str, object] = { + **_mock_graph_result(file_cache={"SKILL.md": target}), + "local_file_cache": {"SKILL.md": target}, + } + child_result: dict[str, object] = { + **_mock_graph_result(file_cache={"runner.py": "pass\n"}), + "components": ["runner.py"], + "local_file_cache": {"runner.py": "pass\n"}, + "inspection_ledger": [child_event], + "analyzer_status_events": [analyzer_status_for_events("child-analyzer", [child_event])], + "analysis_completeness": { + "is_complete": False, + "status": "failed", + "execution_successful": False, + }, + "execution_successful": False, + } + + monkeypatch.setattr(cli, "_run_graph_scan", lambda *args, **kwargs: child_result) + merged = cli._scan_transitive( + initial_result=initial_result, + format=cli.FormatChoice.json, + no_llm=True, + max_depth=1, + transitive_allow_prefix=(), + transitive_deny_prefix=(), + baseline=None, + show_suppressed=False, + visited=set(), + ) + + body = json.loads(cast(str, merged["report_body"])) + assert merged["execution_successful"] is False + exceptions = body["analysis_completeness"]["ledger_exceptions"] + assert [item["reason_code"] for item in exceptions] == ["analyzer_runtime_error"] + assert exceptions[0]["fatal"] is True + assert not any(item["reason_code"] == "output_limit" for item in exceptions) + + +@pytest.mark.parametrize("ledger_cap", [1, 2]) +def test_transitive_child_failure_survives_shared_ledger_cap( + monkeypatch: pytest.MonkeyPatch, ledger_cap: int +) -> None: + """A required fatal row wins over an output sentinel at the smallest caps.""" + target = "https://github.com/org/capped-failure" + root_event = ledger_event( + outcome=LedgerOutcome.COMPLETED, + phase="static", + path="SKILL.md", + analyzer_id="root-analyzer", + ) + initial_result: dict[str, object] = { + **_mock_graph_result(file_cache={"SKILL.md": target}), + "local_file_cache": {"SKILL.md": target}, + "inspection_ledger": [root_event], + "analyzer_status_events": [analyzer_status_for_events("root-analyzer", [root_event])], + } + + def fail_child(*_args: object, **_kwargs: object) -> dict[str, object]: + raise RuntimeError("private child error") + + monkeypatch.setattr(cli, "_run_graph_scan", fail_child) + merged = cli._scan_transitive( + initial_result=initial_result, + format=cli.FormatChoice.json, + no_llm=True, + max_depth=1, + transitive_allow_prefix=(), + transitive_deny_prefix=(), + baseline=None, + show_suppressed=False, + visited=set(), + budget=cli._TransitiveBudget(max_ledger_events=ledger_cap), + ) + + body = json.loads(cast(str, merged["report_body"])) + assert merged["execution_successful"] is False + exceptions = body["analysis_completeness"]["ledger_exceptions"] + assert any(item["reason_code"] == "transitive_child_scan_failed" for item in exceptions) + assert not any(item["reason_code"] == "output_limit" for item in exceptions) + assert not any(item["reason_code"] == "unaccounted_work" for item in exceptions) + assert "private child error" not in merged["report_body"] + + +@pytest.mark.parametrize("ledger_cap", [1, 2]) +def test_transitive_child_failure_runs_after_real_root_ledger_overflow( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + ledger_cap: int, +) -> None: + """A root output cap cannot prevent planned transitive execution.""" + target = "https://github.com/org/real-root-capped-failure" + (tmp_path / "SKILL.md").write_text( + f"---\nname: capped-root\ndescription: Root cap regression\n---\n\n{target}\n", + encoding="utf-8", + ) + initial_result = cli._run_graph_scan( + input_path=str(tmp_path), + format=cli.FormatChoice.json, + no_llm=True, + ) + root_ledger = cast(list[dict[str, object]], initial_result["inspection_ledger"]) + root_statuses = cast(list[dict[str, object]], initial_result["analyzer_status_events"]) + assert len(root_ledger) > ledger_cap + assert len(root_statuses) > ledger_cap + + calls: list[str] = [] + secret = "TOKEN=private-real-root-child-error" + + def fail_child(*args: object, **kwargs: object) -> dict[str, object]: + input_path = kwargs.get("input_path") if kwargs else args[0] + calls.append(str(input_path)) + raise RuntimeError(secret) + + monkeypatch.setattr(cli, "_run_graph_scan", fail_child) + merged = cli._scan_transitive( + initial_result=initial_result, + format=cli.FormatChoice.json, + no_llm=True, + max_depth=1, + transitive_allow_prefix=(), + transitive_deny_prefix=(), + baseline=None, + show_suppressed=False, + visited=set(), + budget=cli._TransitiveBudget(max_ledger_events=ledger_cap), + ) + + assert calls == [target] + assert merged["execution_successful"] is False + completeness = cast(dict[str, object], merged["analysis_completeness"]) + assert completeness["status"] == "failed" + assert completeness["execution_successful"] is False + exceptions = cast(list[dict[str, object]], completeness["ledger_exceptions"]) + reasons = {str(item["reason_code"]) for item in exceptions} + if ledger_cap == 1: + assert reasons == {"transitive_child_scan_failed"} + else: + assert reasons == {"output_limit", "transitive_child_scan_failed"} + assert "unaccounted_work" not in reasons + assert secret not in cast(str, merged["report_body"]) + + +@pytest.mark.parametrize("ledger_cap", [1, 2]) +def test_transitive_child_exact_failure_survives_pre_cache_ledger_cap( + monkeypatch: pytest.MonkeyPatch, ledger_cap: int +) -> None: + """Bounding a child ledger cannot replace an available exact fatal reason.""" + target = "https://github.com/org/exact-capped-failure" + initial_result: dict[str, object] = { + **_mock_graph_result(file_cache={"SKILL.md": target}), + "local_file_cache": {"SKILL.md": target}, + } + child_events = [ + ledger_event( + outcome=LedgerOutcome.COMPLETED, + phase="static", + path=path, + analyzer_id="child-analyzer", + ) + for path in ("one.py", "two.py") + ] + child_events.append( + ledger_event( + outcome=LedgerOutcome.FAILED, + phase="static", + path="failed.py", + analyzer_id="child-analyzer", + reason=LedgerReason.ANALYZER_RUNTIME_ERROR, + ) + ) + child_result: dict[str, object] = { + **_mock_graph_result( + file_cache={"one.py": "pass\n", "two.py": "pass\n", "failed.py": "pass\n"} + ), + "components": ["one.py", "two.py", "failed.py"], + "local_file_cache": { + "one.py": "pass\n", + "two.py": "pass\n", + "failed.py": "pass\n", + }, + "inspection_ledger": child_events, + "analyzer_status_events": [analyzer_status_for_events("child-analyzer", child_events)], + "analysis_completeness": { + "is_complete": False, + "status": "failed", + "execution_successful": False, + }, + "execution_successful": False, + } + + monkeypatch.setattr(cli, "_run_graph_scan", lambda *args, **kwargs: child_result) + merged = cli._scan_transitive( + initial_result=initial_result, + format=cli.FormatChoice.json, + no_llm=True, + max_depth=1, + transitive_allow_prefix=(), + transitive_deny_prefix=(), + baseline=None, + show_suppressed=False, + visited=set(), + budget=cli._TransitiveBudget(max_ledger_events=ledger_cap), + ) + + body = json.loads(cast(str, merged["report_body"])) + reasons = {item["reason_code"] for item in body["analysis_completeness"]["ledger_exceptions"]} + assert "analyzer_runtime_error" in reasons + assert "transitive_child_scan_failed" not in reasons + assert "unaccounted_work" not in reasons + + +@pytest.mark.parametrize("ledger_cap", [1, 2]) +@pytest.mark.parametrize("no_llm", [True, False]) +def test_transitive_root_exact_failure_survives_initial_ledger_cap( + ledger_cap: int, + no_llm: bool, +) -> None: + """Initialization and late semantic accounting cannot cap away a root fatal fact.""" + root_events = [ + ledger_event( + outcome=LedgerOutcome.COMPLETED, + phase="static", + path="one.py", + analyzer_id="root-analyzer", + ), + ledger_event( + outcome=LedgerOutcome.COMPLETED, + phase="static", + path="two.py", + analyzer_id="root-analyzer", + ), + ledger_event( + outcome=LedgerOutcome.FAILED, + phase="static", + path="failed.py", + analyzer_id="root-analyzer", + reason=LedgerReason.ANALYZER_RUNTIME_ERROR, + ), + ] + initial_result: dict[str, object] = { + **_mock_graph_result(file_cache={}), + "components": ["one.py", "two.py", "failed.py"], + "local_file_cache": {}, + "inspection_ledger": root_events, + "analyzer_status_events": [analyzer_status_for_events("root-analyzer", root_events)], + "analysis_completeness": { + "is_complete": False, + "status": "failed", + "execution_successful": False, + }, + "execution_successful": False, + } + + merged = cli._scan_transitive( + initial_result=initial_result, + format=cli.FormatChoice.json, + no_llm=no_llm, + max_depth=1, + transitive_allow_prefix=(), + transitive_deny_prefix=(), + baseline=None, + show_suppressed=False, + visited=set(), + budget=cli._TransitiveBudget(max_ledger_events=ledger_cap), + ) + + body = json.loads(cast(str, merged["report_body"])) + reasons = {item["reason_code"] for item in body["analysis_completeness"]["ledger_exceptions"]} + assert merged["execution_successful"] is False + assert "analyzer_runtime_error" in reasons + assert "unaccounted_work" not in reasons def test_scan_transitive_keeps_source_aware_component_coverage(monkeypatch) -> None: @@ -4424,7 +5454,11 @@ def _mcp_module_missing(d: Path) -> FatalPath: "not supported for recursive", id="recursive-baseline", ), - pytest.param(_multi_skill_child_crashes, "child scan crashed", id="multi-skill-child"), + pytest.param( + _multi_skill_child_crashes, + "A recursive child scan failed before complete inspection.", + id="multi-skill-child", + ), pytest.param(_scan_input_missing, "skill vanished", id="scan-input-missing"), pytest.param(_scan_crashes, "scan crashed", id="scan-crashes"), pytest.param(_scan_crashes_verbose, "RuntimeError", id="scan-crashes-verbose"), From 10362c948e69ffd8d8df225e7921bb373b04cf93 Mon Sep 17 00:00:00 2001 From: Christopher Kevin Date: Wed, 16 Sep 2026 20:04:26 -0700 Subject: [PATCH 3/4] fix(cli): close recursive reporting edge cases Signed-off-by: Christopher Kevin --- src/skillspector/cli.py | 121 ++++++++++++++++++++++------------ tests/unit/test_cli.py | 142 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 221 insertions(+), 42 deletions(-) diff --git a/src/skillspector/cli.py b/src/skillspector/cli.py index ec849a341..90df5dd9d 100644 --- a/src/skillspector/cli.py +++ b/src/skillspector/cli.py @@ -1725,22 +1725,39 @@ def _ensure_required_failure_event( limit: int, traversal: _TransitiveTraversalState, ) -> list[dict[str, object]]: - """Retain a fatal child fact even when the shared ledger reaches its bound.""" - if any( - _is_failed_ledger_event(event) and event.get("work_id") == failure.get("work_id") - for event in events - ): - return events + """Retain distinct fatal facts before non-fatal rows at the shared bound.""" effective_limit = max(1, limit) - if len(events) < effective_limit: - return [*events, failure] - traversal.note_truncation(f"inspection ledger budget {effective_limit} reached") - if effective_limit == 1: - return [failure] + failure_work_id = str(failure.get("work_id", "")) + required_failures: list[dict[str, object]] = [] + required_work_ids: set[str] = set() + for event in events: + work_id = str(event.get("work_id", "")) + if not _is_failed_ledger_event(event) or work_id in required_work_ids: + continue + required_failures.append(event) + required_work_ids.add(work_id) + failure_missing = failure_work_id not in required_work_ids + if failure_missing: + required_failures.append(failure) + required_work_ids.add(failure_work_id) + prior_sentinel = next( (event for event in reversed(events) if event.get("phase") == "ledger_output"), None, ) + would_overflow = len(events) + int(failure_missing) > effective_limit + if prior_sentinel is None and not would_overflow: + combined = [*events, *([failure] if failure_missing else [])] + if len(combined) < effective_limit: + return combined + non_failures = [ + event for event in combined if str(event.get("work_id", "")) not in required_work_ids + ] + return [*required_failures, *non_failures][:effective_limit] + + traversal.note_truncation(f"inspection ledger budget {effective_limit} reached") + if effective_limit == 1: + return [failure] observed_value = prior_sentinel.get("observed_records") if prior_sentinel else None observed_records = observed_value if isinstance(observed_value, int) else len(events) sentinel = dict( @@ -1750,12 +1767,23 @@ def _ensure_required_failure_event( phase="ledger_output", path=str(failure.get("path", "SKILL.md")), reason=LedgerReason.OUTPUT_LIMIT, - observed_records=max(observed_records, len(events)) + 1, + observed_records=max(observed_records, len(events)) + int(failure_missing), limit_records=effective_limit, ) ) - retained = [event for event in events if event.get("phase") != "ledger_output"] - return [*retained[: effective_limit - 2], failure, sentinel] + failure_slots = effective_limit - 1 + retained_failures = required_failures[:failure_slots] + if failure_work_id not in {str(event.get("work_id", "")) for event in retained_failures}: + retained_failures = [*required_failures[: failure_slots - 1], failure] + retained_work_ids = {str(event.get("work_id", "")) for event in retained_failures} + retained_non_failures = [ + event + for event in events + if event.get("phase") != "ledger_output" + and str(event.get("work_id", "")) not in retained_work_ids + and not _is_failed_ledger_event(event) + ][: failure_slots - len(retained_failures)] + return [*retained_failures, *retained_non_failures, sentinel] def _scan_transitive( @@ -1930,6 +1958,7 @@ def _scan_transitive( cache_key = (target, source_local_only) cached = traversal.cache.get(cache_key) if cached is None: + traversal.record_scan() child_result = _run_graph_scan_for_source( input_path=target, format=format, @@ -1945,7 +1974,6 @@ def _scan_transitive( ) cached = _cache_transitive_result(target, child_result, traversal) traversal.cache[cache_key] = cached - traversal.record_scan() if not cached.execution_successful: traversal.note_child_scan_failure(target) if not any( @@ -2691,6 +2719,7 @@ def _scan_multi_skill( progress_console.print( f" [{i}/{len(skills)}] Scanning [bold]{skill.name}[/bold] ({skill.relative_path}/)" ) + result: dict[str, object] | None = None try: result = _scan_skill( input_path=str(skill.path), @@ -2709,64 +2738,74 @@ def _scan_multi_skill( source_local_only=skill.local_only, ) child_failed = result.get("execution_successful") is False - if child_failed: - execution_failed = True - failed_skill_count += 1 completeness_value = result.get("analysis_completeness") - if ( + child_partial = ( not child_failed and isinstance(completeness_value, dict) and not bool(completeness_value.get("is_complete", True)) - ): - analysis_incomplete = True - partial_skill_count += 1 - elif not child_failed: - complete_skill_count += 1 + ) score = result.get("risk_score") or 0 try: score = int(score) except (TypeError, ValueError): score = 0 - if score > max_score: - max_score = score child_transitive_count = result.get("transitive_finding_count") - if isinstance(child_transitive_count, int): - transitive_finding_count += child_transitive_count - for source in _coerce_str_path_list(result.get("transitive_sources")): - transitive_sources.add(source) + child_transitive_increment = ( + child_transitive_count if isinstance(child_transitive_count, int) else 0 + ) + child_transitive_sources = _coerce_str_path_list(result.get("transitive_sources")) severity = result.get("risk_severity") or "LOW" progress_console.print(f" Score: {score}/100 ({severity})\n") result_body = _result_body(result) result_characters = len(result_body) result_records = _multi_skill_public_record_count(result) - has_findings = has_findings or bool(effective_findings(result)) - if ( + child_has_findings = bool(effective_findings(result)) + exceeds_record_limit = ( retained_public_records + result_records > _MULTI_SKILL_MAX_PUBLIC_RECORDS - or retained_report_characters + result_characters - > _MULTI_SKILL_MAX_REPORT_CHARACTERS - ): + ) + exceeds_character_limit = ( + retained_report_characters + result_characters > _MULTI_SKILL_MAX_REPORT_CHARACTERS + ) + if exceeds_record_limit or exceeds_character_limit: + cleanup_result(result) + + if child_failed: + execution_failed = True + failed_skill_count += 1 + elif child_partial: analysis_incomplete = True - if retained_public_records + result_records > _MULTI_SKILL_MAX_PUBLIC_RECORDS: + partial_skill_count += 1 + else: + complete_skill_count += 1 + max_score = max(max_score, score) + transitive_finding_count += child_transitive_increment + transitive_sources.update(child_transitive_sources) + has_findings = has_findings or child_has_findings + + if exceeds_record_limit or exceeds_character_limit: + analysis_incomplete = True + if exceeds_record_limit: aggregate_limitations.append( "recursive public finding record budget " f"{_MULTI_SKILL_MAX_PUBLIC_RECORDS} reached" ) - if ( - retained_report_characters + result_characters - > _MULTI_SKILL_MAX_REPORT_CHARACTERS - ): + if exceeds_character_limit: aggregate_limitations.append( "recursive report character budget " f"{_MULTI_SKILL_MAX_REPORT_CHARACTERS} reached" ) - cleanup_result(result) break results.append(result) processed_skills.append(skill) retained_public_records += result_records retained_report_characters += result_characters except Exception: + if result is not None and not any(item is result for item in results): + try: + cleanup_result(result) + except Exception: + logger.warning("Recursive child result cleanup failed") error_message = _RECURSIVE_CHILD_SCAN_FAILED_MESSAGE err_console.print(f" [red]Error:[/red] {error_message}\n") execution_failed = True diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 7a54192c2..60dc871e6 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -1615,6 +1615,40 @@ def test_recursive_oversized_failed_child_preserves_fatal_aggregate( assert "x" * 1_001 not in output.read_text(encoding="utf-8") +def test_recursive_postscan_failure_counts_child_once_and_cleans_result( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A failure after scanning replaces, rather than duplicates, child accounting.""" + skill = SkillDirectory(tmp_path / "one", "one", "one") + output = tmp_path / "combined.json" + child = _bounded_recursive_result("one", finding_count=0) + child["report_body"] = "" + child["sarif_report"] = {"not_json_serializable": object()} + cleaned: list[dict[str, object]] = [] + monkeypatch.setattr(cli, "_scan_skill", lambda *args, **kwargs: child) + monkeypatch.setattr(cli, "cleanup_result", lambda result: cleaned.append(result)) + + with pytest.raises(typer.Exit) as exit_info: + cli._scan_multi_skill( + MultiSkillDetectionResult(is_multi_skill=True, skills=[skill]), + FormatChoice.json, + output, + no_llm=True, + ) + + assert exit_info.value.exit_code == 2 + payload = json.loads(output.read_text(encoding="utf-8")) + completeness = payload["analysis_completeness"] + assert payload["skill_count"] == 1 + assert payload["skills_scanned"] == 1 + assert payload["skills_output_omitted"] == 0 + assert completeness["fully_inspected_files"] == 0 + assert completeness["entirely_uninspected_files"] == 1 + assert completeness["total_files"] == 1 + assert sum(result is child for result in cleaned) == 1 + + def test_recursive_over_record_budget_child_preserves_risk_and_exit( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: @@ -3018,6 +3052,46 @@ def test_child_failure_target_text_cannot_exhaust_traversal_by_substring() -> No assert traversal.can_scan_more() is True +def test_transitive_failed_attempt_consumes_shared_target_budget( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failed target attempt still consumes one shared execution slot.""" + traversal = cli._TransitiveTraversalState( + budget=cli._TransitiveBudget(max_targets=1), + ) + attempted_targets: list[str] = [] + + def fail_child(*args: object, **kwargs: object) -> dict[str, object]: + input_path = kwargs.get("input_path") if kwargs else args[0] + attempted_targets.append(str(input_path)) + raise RuntimeError("TOKEN=private-child-failure") + + monkeypatch.setattr(cli, "_run_graph_scan", fail_child) + targets = ["https://github.com/org/failed-one", "https://github.com/org/failed-two"] + for target in targets: + root = { + **_mock_graph_result(file_cache={"SKILL.md": target}), + "local_file_cache": {"SKILL.md": target}, + } + cli._scan_transitive( + initial_result=root, + format=cli.FormatChoice.json, + no_llm=True, + max_depth=1, + transitive_allow_prefix=(), + transitive_deny_prefix=(), + baseline=None, + show_suppressed=False, + visited=set(), + traversal=traversal, + ) + + assert attempted_targets == [targets[0]] + assert traversal.scanned_targets == 1 + assert traversal.budget_exhausted is True + assert traversal.truncation_reasons[-1] == "target budget 1 reached" + + @pytest.mark.parametrize( ("budget_kwargs", "expected_reason"), [ @@ -4305,7 +4379,7 @@ def fake_run_graph_scan( body = json.loads(merged["report_body"]) assert merged["temp_dir_for_cleanup"] == "root-temp" assert merged["transitive_sources"] == [failed_target] - assert merged["transitive_targets_scanned"] == 0 + assert merged["transitive_targets_scanned"] == 1 assert merged["transitive_truncated"] is True assert merged["transitive_truncation_reasons"] == [ f"transitive child scan failed for {failed_target}" @@ -4562,6 +4636,72 @@ def test_transitive_child_exact_failure_survives_pre_cache_ledger_cap( assert "unaccounted_work" not in reasons +def test_transitive_ledger_cap_preserves_distinct_root_and_child_failures( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A ledger cap retains both fatal scopes before completed work.""" + target = "https://github.com/org/root-and-child-failures" + root_events = [ + ledger_event( + outcome=LedgerOutcome.COMPLETED, + phase="static", + path=path, + analyzer_id="root-analyzer", + ) + for path in ("one.py", "two.py") + ] + root_events.append( + ledger_event( + outcome=LedgerOutcome.FAILED, + phase="static", + path="root-failed.py", + analyzer_id="root-analyzer", + reason=LedgerReason.READ_ERROR, + ) + ) + initial_result: dict[str, object] = { + **_mock_graph_result(file_cache={"SKILL.md": target}), + "components": ["one.py", "two.py", "root-failed.py", "SKILL.md"], + "local_file_cache": {"SKILL.md": target}, + "inspection_ledger": root_events, + "analyzer_status_events": [analyzer_status_for_events("root-analyzer", root_events)], + "execution_successful": False, + } + child_event = ledger_event( + outcome=LedgerOutcome.FAILED, + phase="static", + path="child-failed.py", + analyzer_id="child-analyzer", + reason=LedgerReason.ANALYZER_RUNTIME_ERROR, + ) + child_result: dict[str, object] = { + **_mock_graph_result(file_cache={"child-failed.py": "pass\n"}), + "components": ["child-failed.py"], + "local_file_cache": {"child-failed.py": "pass\n"}, + "inspection_ledger": [child_event], + "analyzer_status_events": [analyzer_status_for_events("child-analyzer", [child_event])], + "execution_successful": False, + } + + monkeypatch.setattr(cli, "_run_graph_scan", lambda *args, **kwargs: child_result) + merged = cli._scan_transitive( + initial_result=initial_result, + format=cli.FormatChoice.json, + no_llm=True, + max_depth=1, + transitive_allow_prefix=(), + transitive_deny_prefix=(), + baseline=None, + show_suppressed=False, + visited=set(), + budget=cli._TransitiveBudget(max_ledger_events=3), + ) + + body = json.loads(cast(str, merged["report_body"])) + reasons = {item["reason_code"] for item in body["analysis_completeness"]["ledger_exceptions"]} + assert reasons == {"read_error", "analyzer_runtime_error", "output_limit"} + + @pytest.mark.parametrize("ledger_cap", [1, 2]) @pytest.mark.parametrize("no_llm", [True, False]) def test_transitive_root_exact_failure_survives_initial_ledger_cap( From fb86cf753c8e4a15cbaca12f39f106bc4a6cf8df Mon Sep 17 00:00:00 2001 From: Christopher Kevin Date: Sun, 20 Sep 2026 19:49:34 -0700 Subject: [PATCH 4/4] fix(cli): retain distinct recursive failures Signed-off-by: Christopher Kevin --- src/skillspector/cli.py | 154 +++++++++++++++++------------- tests/unit/test_cli.py | 202 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 291 insertions(+), 65 deletions(-) diff --git a/src/skillspector/cli.py b/src/skillspector/cli.py index 90df5dd9d..3a4d1b520 100644 --- a/src/skillspector/cli.py +++ b/src/skillspector/cli.py @@ -1200,10 +1200,7 @@ def _scope_finding(finding: Finding) -> Finding: source_digest=source_digest, finding_id_map=finding_id_map, ) - required_failure_event = next( - (event for event in scoped_ledger if _is_failed_ledger_event(event)), - None, - ) + required_failure_events = _distinct_failed_ledger_events(scoped_ledger) retained_finding_ids = {item.finding_id for item in scoped_findings} for event in scoped_ledger: for id_field in ("input_finding_ids", "emitted_finding_ids"): @@ -1235,12 +1232,13 @@ def _scope_finding(finding: Finding) -> Finding: limit=traversal.budget.max_ledger_events, traversal=traversal, ) - if required_failure_event is not None: - scoped_ledger = _ensure_required_failure_event( + if required_failure_events: + scoped_ledger = _ensure_required_failure_events( scoped_ledger, - required_failure_event, + required_failure_events, limit=traversal.budget.max_ledger_events, traversal=traversal, + failures_already_observed=True, ) retained_work_ids = { str(event.get("work_id", "")) for event in scoped_ledger if event.get("work_id") @@ -1705,6 +1703,21 @@ def _is_failed_ledger_event(event: dict[str, object]) -> bool: return getattr(outcome, "value", outcome) == LedgerOutcome.FAILED.value +def _distinct_failed_ledger_events( + events: list[dict[str, object]], +) -> list[dict[str, object]]: + """Return failed work items once each, preserving their observed order.""" + failures: list[dict[str, object]] = [] + work_ids: set[str] = set() + for event in events: + work_id = str(event.get("work_id", "")) + if not _is_failed_ledger_event(event) or work_id in work_ids: + continue + failures.append(event) + work_ids.add(work_id) + return failures + + def _transitive_child_failure_event(source_identity: str) -> dict[str, object]: """Return one deterministic, payload-free fatal fact for an opaque child failure.""" return dict( @@ -1718,36 +1731,37 @@ def _transitive_child_failure_event(source_identity: str) -> dict[str, object]: ) -def _ensure_required_failure_event( +def _ensure_required_failure_events( events: list[dict[str, object]], - failure: dict[str, object], + failures: list[dict[str, object]], *, limit: int, traversal: _TransitiveTraversalState, + failures_already_observed: bool = False, ) -> list[dict[str, object]]: - """Retain distinct fatal facts before non-fatal rows at the shared bound.""" + """Retain distinct fatal facts before non-fatal rows at the shared bound. + + Pre-observed failures came from the input that produced an existing sentinel; + synthesized failures are new observations and remain the required fatal facts. + """ effective_limit = max(1, limit) - failure_work_id = str(failure.get("work_id", "")) - required_failures: list[dict[str, object]] = [] - required_work_ids: set[str] = set() - for event in events: - work_id = str(event.get("work_id", "")) - if not _is_failed_ledger_event(event) or work_id in required_work_ids: - continue - required_failures.append(event) - required_work_ids.add(work_id) - failure_missing = failure_work_id not in required_work_ids - if failure_missing: - required_failures.append(failure) - required_work_ids.add(failure_work_id) + incoming_failures = _distinct_failed_ledger_events(failures) + required_failures = _distinct_failed_ledger_events([*events, *failures]) + if not required_failures: + return events[:effective_limit] + required_work_ids = {str(event.get("work_id", "")) for event in required_failures} + event_work_ids = {str(event.get("work_id", "")) for event in events} + missing_failures = [ + event for event in required_failures if str(event.get("work_id", "")) not in event_work_ids + ] prior_sentinel = next( (event for event in reversed(events) if event.get("phase") == "ledger_output"), None, ) - would_overflow = len(events) + int(failure_missing) > effective_limit + would_overflow = len(events) + len(missing_failures) > effective_limit if prior_sentinel is None and not would_overflow: - combined = [*events, *([failure] if failure_missing else [])] + combined = [*events, *missing_failures] if len(combined) < effective_limit: return combined non_failures = [ @@ -1757,24 +1771,40 @@ def _ensure_required_failure_event( traversal.note_truncation(f"inspection ledger budget {effective_limit} reached") if effective_limit == 1: - return [failure] + return (incoming_failures or required_failures)[:1] observed_value = prior_sentinel.get("observed_records") if prior_sentinel else None observed_records = observed_value if isinstance(observed_value, int) else len(events) + newly_observed = ( + 0 if prior_sentinel is not None and failures_already_observed else len(missing_failures) + ) + sentinel_path = ( + str(prior_sentinel.get("path", "SKILL.md")) + if prior_sentinel is not None and failures_already_observed + else str((incoming_failures or required_failures)[0].get("path", "SKILL.md")) + ) sentinel = dict( ledger_event( outcome=LedgerOutcome.PARTIAL, record_type=LedgerRecordType.SYSTEM, phase="ledger_output", - path=str(failure.get("path", "SKILL.md")), + path=sentinel_path, reason=LedgerReason.OUTPUT_LIMIT, - observed_records=max(observed_records, len(events)) + int(failure_missing), + observed_records=max(observed_records, len(events)) + newly_observed, limit_records=effective_limit, ) ) failure_slots = effective_limit - 1 - retained_failures = required_failures[:failure_slots] - if failure_work_id not in {str(event.get("work_id", "")) for event in retained_failures}: - retained_failures = [*required_failures[: failure_slots - 1], failure] + if failures_already_observed: + retained_failures = required_failures[:failure_slots] + else: + incoming_work_ids = {str(event.get("work_id", "")) for event in incoming_failures} + retained_incoming = incoming_failures[:failure_slots] + retained_existing = [ + event + for event in required_failures + if str(event.get("work_id", "")) not in incoming_work_ids + ][: failure_slots - len(retained_incoming)] + retained_failures = [*retained_existing, *retained_incoming] retained_work_ids = {str(event.get("work_id", "")) for event in retained_failures} retained_non_failures = [ event @@ -1839,22 +1869,20 @@ def _scan_transitive( : traversal.budget.max_findings ] root_inspection_ledger = _coerce_dict_list(initial_result.get("inspection_ledger")) - required_root_failure_event = next( - (event for event in root_inspection_ledger if _is_failed_ledger_event(event)), - None, - ) + required_root_failure_events = _distinct_failed_ledger_events(root_inspection_ledger) merged_inspection_ledger = _merge_bounded_ledger( [], root_inspection_ledger, limit=traversal.budget.max_ledger_events, traversal=traversal, ) - if required_root_failure_event is not None: - merged_inspection_ledger = _ensure_required_failure_event( + if required_root_failure_events: + merged_inspection_ledger = _ensure_required_failure_events( merged_inspection_ledger, - required_root_failure_event, + required_root_failure_events, limit=traversal.budget.max_ledger_events, traversal=traversal, + failures_already_observed=True, ) retained_work_ids = { str(event.get("work_id", "")) for event in merged_inspection_ledger if event.get("work_id") @@ -1979,9 +2007,9 @@ def _scan_transitive( if not any( _is_failed_ledger_event(event) for event in cached.inspection_ledger ): - cached.inspection_ledger = _ensure_required_failure_event( + cached.inspection_ledger = _ensure_required_failure_events( cached.inspection_ledger, - _transitive_child_failure_event(cached.source_identity), + [_transitive_child_failure_event(cached.source_identity)], limit=traversal.budget.max_ledger_events, traversal=traversal, ) @@ -1992,16 +2020,14 @@ def _scan_transitive( limit=traversal.budget.max_ledger_events, traversal=traversal, ) - child_failure_event = next( - (event for event in cached.inspection_ledger if _is_failed_ledger_event(event)), - None, - ) - if child_failure_event is not None: - merged_inspection_ledger = _ensure_required_failure_event( + child_failure_events = _distinct_failed_ledger_events(cached.inspection_ledger) + if child_failure_events: + merged_inspection_ledger = _ensure_required_failure_events( merged_inspection_ledger, - child_failure_event, + child_failure_events, limit=traversal.budget.max_ledger_events, traversal=traversal, + failures_already_observed=True, ) global_work_ids = { str(event.get("work_id", "")) @@ -2157,11 +2183,13 @@ def _scan_transitive( except Exception: transitive_sources.add(target) traversal.note_child_scan_failure(target) - merged_inspection_ledger = _ensure_required_failure_event( + merged_inspection_ledger = _ensure_required_failure_events( merged_inspection_ledger, - _transitive_child_failure_event( - _source_identity(target, "transitive-child-scan-failed") - ), + [ + _transitive_child_failure_event( + _source_identity(target, "transitive-child-scan-failed") + ) + ], limit=traversal.budget.max_ledger_events, traversal=traversal, ) @@ -2189,10 +2217,7 @@ def _scan_transitive( ) if traversal.resource_limit_reached: - required_failure_event = next( - (event for event in merged_inspection_ledger if _is_failed_ledger_event(event)), - None, - ) + required_failure_events = _distinct_failed_ledger_events(merged_inspection_ledger) traversal_event = ledger_event( outcome=LedgerOutcome.PARTIAL, record_type=LedgerRecordType.SYSTEM, @@ -2206,12 +2231,13 @@ def _scan_transitive( limit=traversal.budget.max_ledger_events, traversal=traversal, ) - if required_failure_event is not None: - merged_inspection_ledger = _ensure_required_failure_event( + if required_failure_events: + merged_inspection_ledger = _ensure_required_failure_events( merged_inspection_ledger, - required_failure_event, + required_failure_events, limit=traversal.budget.max_ledger_events, traversal=traversal, + failures_already_observed=True, ) retained_work_ids = { @@ -2258,10 +2284,7 @@ def _scan_transitive( result=merged_result, discovered_modules=ANALYZER_MODULES, ) - pre_runtime_failure_event = next( - (event for event in merged_inspection_ledger if _is_failed_ledger_event(event)), - None, - ) + pre_runtime_failure_events = _distinct_failed_ledger_events(merged_inspection_ledger) if runtime_event is not None and not has_semantic_runtime_event( merged_inspection_ledger, runtime_event ): @@ -2271,12 +2294,13 @@ def _scan_transitive( limit=traversal.budget.max_ledger_events, traversal=traversal, ) - if pre_runtime_failure_event is not None: - merged_inspection_ledger = _ensure_required_failure_event( + if pre_runtime_failure_events: + merged_inspection_ledger = _ensure_required_failure_events( merged_inspection_ledger, - pre_runtime_failure_event, + pre_runtime_failure_events, limit=traversal.budget.max_ledger_events, traversal=traversal, + failures_already_observed=True, ) merged_result["inspection_ledger"] = merged_inspection_ledger if merged_inspection_ledger or merged_analyzer_status_events: diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 279d65bea..96dd9203a 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -4540,6 +4540,63 @@ def fail_child(*_args: object, **_kwargs: object) -> dict[str, object]: assert "private child error" not in merged["report_body"] +def test_new_child_failure_remains_required_when_only_one_fatal_slot_fits( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A new opaque child failure keeps the prior tight-cap replacement contract.""" + target = "https://github.com/org/root-and-child-tight-cap" + root_events = [ + ledger_event( + outcome=LedgerOutcome.COMPLETED, + phase="static", + path=path, + analyzer_id="root-analyzer", + ) + for path in ("one.py", "two.py") + ] + root_events.append( + ledger_event( + outcome=LedgerOutcome.FAILED, + phase="static", + path="root-failed.py", + analyzer_id="root-analyzer", + reason=LedgerReason.READ_ERROR, + ) + ) + initial_result: dict[str, object] = { + **_mock_graph_result(file_cache={"SKILL.md": target}), + "components": ["one.py", "two.py", "root-failed.py", "SKILL.md"], + "local_file_cache": {"SKILL.md": target}, + "inspection_ledger": root_events, + "execution_successful": False, + } + + def fail_child(*_args: object, **_kwargs: object) -> dict[str, object]: + raise RuntimeError("private child failure") + + monkeypatch.setattr(cli, "_run_graph_scan", fail_child) + merged = cli._scan_transitive( + initial_result=initial_result, + format=cli.FormatChoice.json, + no_llm=True, + max_depth=1, + transitive_allow_prefix=(), + transitive_deny_prefix=(), + baseline=None, + show_suppressed=False, + visited=set(), + budget=cli._TransitiveBudget(max_ledger_events=2), + ) + + body = json.loads(cast(str, merged["report_body"])) + exceptions = body["analysis_completeness"]["ledger_exceptions"] + assert {item["reason_code"] for item in exceptions} == { + "output_limit", + "transitive_child_scan_failed", + } + assert "private child failure" not in merged["report_body"] + + @pytest.mark.parametrize("ledger_cap", [1, 2]) def test_transitive_child_failure_runs_after_real_root_ledger_overflow( monkeypatch: pytest.MonkeyPatch, @@ -4668,6 +4725,151 @@ def test_transitive_child_exact_failure_survives_pre_cache_ledger_cap( assert "unaccounted_work" not in reasons +def test_transitive_root_ledger_cap_preserves_all_distinct_failures() -> None: + """All pre-cap root failures displace completed work before the sentinel.""" + root_events = [ + ledger_event( + outcome=LedgerOutcome.COMPLETED, + phase="static", + path=path, + analyzer_id="root-analyzer", + ) + for path in ("one.py", "two.py") + ] + root_events.extend( + [ + ledger_event( + outcome=LedgerOutcome.FAILED, + phase="static", + path="read-failed.py", + analyzer_id="root-reader", + reason=LedgerReason.READ_ERROR, + ), + ledger_event( + outcome=LedgerOutcome.FAILED, + phase="static", + path="runtime-failed.py", + analyzer_id="root-runtime", + reason=LedgerReason.ANALYZER_RUNTIME_ERROR, + ), + ] + ) + initial_result: dict[str, object] = { + **_mock_graph_result(file_cache={}), + "components": ["one.py", "two.py", "read-failed.py", "runtime-failed.py"], + "local_file_cache": {}, + "inspection_ledger": root_events, + "execution_successful": False, + } + + merged = cli._scan_transitive( + initial_result=initial_result, + format=cli.FormatChoice.json, + no_llm=True, + max_depth=1, + transitive_allow_prefix=(), + transitive_deny_prefix=(), + baseline=None, + show_suppressed=False, + visited=set(), + budget=cli._TransitiveBudget(max_ledger_events=3), + ) + + body = json.loads(cast(str, merged["report_body"])) + exceptions = body["analysis_completeness"]["ledger_exceptions"] + assert [(item["reason_code"], item["path"]) for item in exceptions] == [ + ("output_limit", "read-failed.py"), + ("read_error", "read-failed.py"), + ("analyzer_runtime_error", "runtime-failed.py"), + ] + assert merged["execution_successful"] is False + + +def test_transitive_child_ledger_cap_preserves_all_distinct_failures( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Child failures are scoped, deduplicated, and retained before completed work.""" + target = "https://github.com/org/multiple-child-failures" + initial_result: dict[str, object] = { + **_mock_graph_result(file_cache={"SKILL.md": target}), + "local_file_cache": {"SKILL.md": target}, + } + completed_events = [ + ledger_event( + outcome=LedgerOutcome.COMPLETED, + phase="static", + path=path, + analyzer_id="child-analyzer", + ) + for path in ("one.py", "two.py") + ] + read_failure = ledger_event( + outcome=LedgerOutcome.FAILED, + phase="static", + path="read-failed.py", + analyzer_id="child-reader", + reason=LedgerReason.READ_ERROR, + ) + runtime_failure = ledger_event( + outcome=LedgerOutcome.FAILED, + phase="static", + path="runtime-failed.py", + analyzer_id="child-runtime", + reason=LedgerReason.ANALYZER_RUNTIME_ERROR, + ) + child_result: dict[str, object] = { + **_mock_graph_result( + file_cache={ + "one.py": "pass\n", + "two.py": "pass\n", + "read-failed.py": "pass\n", + "runtime-failed.py": "pass\n", + } + ), + "components": ["one.py", "two.py", "read-failed.py", "runtime-failed.py"], + "local_file_cache": { + "one.py": "pass\n", + "two.py": "pass\n", + "read-failed.py": "pass\n", + "runtime-failed.py": "pass\n", + }, + "inspection_ledger": [ + *completed_events, + read_failure, + dict(read_failure), + runtime_failure, + ], + "execution_successful": False, + } + + monkeypatch.setattr(cli, "_run_graph_scan", lambda *args, **kwargs: child_result) + merged = cli._scan_transitive( + initial_result=initial_result, + format=cli.FormatChoice.json, + no_llm=True, + max_depth=1, + transitive_allow_prefix=(), + transitive_deny_prefix=(), + baseline=None, + show_suppressed=False, + visited=set(), + budget=cli._TransitiveBudget(max_ledger_events=3), + ) + + body = json.loads(cast(str, merged["report_body"])) + exceptions = body["analysis_completeness"]["ledger_exceptions"] + reasons = [item["reason_code"] for item in exceptions] + assert reasons.count("read_error") == 1 + assert reasons.count("analyzer_runtime_error") == 1 + assert reasons.count("output_limit") == 1 + paths_by_reason = {item["reason_code"]: item["path"] for item in exceptions} + assert paths_by_reason["read_error"].startswith("external/") + assert paths_by_reason["read_error"].endswith("/read-failed.py") + assert paths_by_reason["analyzer_runtime_error"].startswith("external/") + assert paths_by_reason["analyzer_runtime_error"].endswith("/runtime-failed.py") + assert merged["execution_successful"] is False + + def test_transitive_ledger_cap_preserves_distinct_root_and_child_failures( monkeypatch: pytest.MonkeyPatch, ) -> None: