fold/scripts/fold_quality.py

#!/usr/bin/env python3
"""Run Fold's local code-quality gates and write one machine-readable report."""

from __future__ import annotations

import argparse
import importlib
import json
import os
import shlex
import subprocess
import sys
import time
from pathlib import Path
from typing import Protocol, cast

from benchmarks.fold_eval.token_efficiency import estimate_text_tokens

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_MANIFEST = ROOT / "tests" / "benchmarks" / "projects" / "fold_source.json"
DEFAULT_TEST_IMPACT_MANIFEST = ROOT / "tests" / "quality" / "impacted_tests.json"
SOURCE_SUFFIXES = {".c", ".cc", ".cpp", ".cxx"}
HEADER_SUFFIXES = {".h", ".hh", ".hpp", ".hxx"}
MAX_EVIDENCE_CHARS = 1000
MAX_EVIDENCE_LINES = 5
MAX_FINDING_DETAILS = 20
EXCLUDED_PATH_PARTS = {
    "build",
    "generated",
    "subprojects",
    "third_party",
    "vendor",
    "vendored",
    "external",
}


def positive_integer(value: str) -> int:
    try:
        parsed = int(value)
    except ValueError as error:
        raise argparse.ArgumentTypeError("must be an integer") from error
    if parsed < 1:
        raise argparse.ArgumentTypeError("must be at least 1")
    return parsed


class FlakyEvidenceModule(Protocol):
    def parse_meson_testlog(
        self, path: Path, attempt: int, exit_code: int
    ) -> dict[str, object]: ...

    def summarize_attempts(
        self,
        attempts: list[dict[str, object]],
        *,
        expected_tests: list[str] | None = None,
    ) -> dict[str, object]: ...


def flaky_evidence_module() -> FlakyEvidenceModule:
    module = importlib.import_module("benchmarks.fold_eval.flaky_evidence")
    return cast(FlakyEvidenceModule, module)


def parse_test_attempt(path: Path, attempt: int, exit_code: int) -> dict[str, object]:
    parser = flaky_evidence_module().parse_meson_testlog
    result = parser(path, attempt, exit_code)
    if not isinstance(result, dict):
        raise ValueError("test evidence parser returned a non-object result")
    return result


def summarize_test_attempts(
    attempts: list[dict[str, object]], expected_tests: list[str] | None
) -> dict[str, object]:
    summarizer = flaky_evidence_module().summarize_attempts
    result = summarizer(attempts, expected_tests=expected_tests)
    if not isinstance(result, dict):
        raise ValueError("test evidence summarizer returned a non-object result")
    return result


def select_impacted_tests(changed: list[str], manifest: Path) -> dict[str, object]:
    module = importlib.import_module("benchmarks.fold_eval.test_selection")
    selector = module.select_tests
    result = selector(changed, manifest)
    if not isinstance(result, dict):
        raise ValueError("test selector returned a non-object result")
    return result


def apply_full_test_override(selection: dict[str, object], force_full: bool) -> None:
    if not force_full:
        return
    selection["full_suite"] = True
    reasons_value = selection.get("fallback_reasons", [])
    reasons = reasons_value if isinstance(reasons_value, list) else []
    reasons.append("complete suite requested explicitly")
    selection["fallback_reasons"] = reasons


def compact_evidence(output: str) -> list[str]:
    lines = [line for line in output.splitlines() if line.strip()]
    selected: list[str] = []
    remaining = MAX_EVIDENCE_CHARS
    for line in reversed(lines[-MAX_EVIDENCE_LINES:]):
        if remaining <= 0:
            break
        value = line[-remaining:]
        selected.append(value)
        remaining -= len(value) + (1 if selected else 0)
    selected.reverse()
    return selected


def text_artifact(path: Path, root: Path, kind: str) -> dict[str, object]:
    text = path.read_text(encoding="utf-8", errors="replace")
    return {
        "kind": kind,
        "path": path.resolve().relative_to(root.resolve()).as_posix(),
        "bytes": len(text.encode("utf-8")),
        "estimated_tokens": estimate_text_tokens(text),
    }


def run_step(
    name: str,
    command: list[str],
    *,
    artifact_dir: Path,
    acceptable: set[int] | None = None,
) -> dict[str, object]:
    started = time.monotonic()
    proc = subprocess.run(command, cwd=ROOT, text=True, capture_output=True)
    accepted = {0} if acceptable is None else acceptable
    output = proc.stdout + proc.stderr
    log_path = artifact_dir / "logs" / f"{name}.log"
    log_path.parent.mkdir(parents=True, exist_ok=True)
    log_path.write_text(output, encoding="utf-8")
    return {
        "name": name,
        "command": command,
        "exit_code": proc.returncode,
        "passed": proc.returncode in accepted,
        "duration_seconds": round(time.monotonic() - started, 3),
        "evidence": compact_evidence(output),
        "artifacts": [text_artifact(log_path, artifact_dir, "log")],
    }


def summarize_policy_report(summary_path: Path, output_root: Path) -> dict[str, object]:
    try:
        summary = json.loads(summary_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        return {"policy_summary_error": str(error), "artifacts": []}
    change_time = summary.get("change_time", {})
    if not isinstance(change_time, dict):
        change_time = {}
    counts = change_time.get("counts", {})
    if not isinstance(counts, dict):
        counts = {}
    new = change_time.get("new", [])
    resolved = change_time.get("resolved", [])
    new = new if isinstance(new, list) else []
    resolved = resolved if isinstance(resolved, list) else []
    artifacts = []
    candidates = [summary_path, summary_path.with_suffix(".md")]
    for value in summary.get("run_artifacts", []):
        if isinstance(value, str):
            candidates.append(Path(value))
    for path in candidates:
        if path.is_file():
            try:
                artifacts.append(text_artifact(path, output_root, "policy-report"))
            except ValueError:
                continue
    failures = summary.get("reliability", {}).get("failures", [])
    failures = failures if isinstance(failures, list) else []
    return {
        "policy_status": str(summary.get("status", "unknown")),
        "finding_delta": {
            "counts": {
                "new": counts.get("new", len(new)),
                "resolved": counts.get("resolved", len(resolved)),
            },
            "new": new[:MAX_FINDING_DETAILS],
            "resolved": resolved[:MAX_FINDING_DETAILS],
            "truncated": {
                "new": max(0, len(new) - MAX_FINDING_DETAILS),
                "resolved": max(0, len(resolved) - MAX_FINDING_DETAILS),
            },
        },
        "failures": [str(value) for value in failures[:MAX_EVIDENCE_LINES]],
        "artifacts": artifacts,
    }


def finalize_report(steps: list[dict[str, object]]) -> dict[str, object]:
    artifacts: list[dict[str, object]] = []
    for step in steps:
        value = step.get("artifacts", [])
        if isinstance(value, list):
            artifacts.extend(item for item in value if isinstance(item, dict))
    report: dict[str, object] = {
        "schema_version": 2,
        "status": "ok" if all(step.get("passed", True) for step in steps) else "failed",
        "steps": steps,
        "artifacts": sorted(artifacts, key=lambda item: str(item.get("path", ""))),
    }
    compact_payload = json.dumps(report, indent=2, sort_keys=True) + "\n"
    artifact_bytes = 0
    artifact_tokens = 0
    for item in artifacts:
        byte_count = item.get("bytes")
        token_count = item.get("estimated_tokens")
        if isinstance(byte_count, int) and not isinstance(byte_count, bool):
            artifact_bytes += byte_count
        if isinstance(token_count, int) and not isinstance(token_count, bool):
            artifact_tokens += token_count
    report["token_efficiency"] = {
        "estimator": "ceil(UTF-8 bytes / 4)",
        "compact_payload_bytes": len(compact_payload.encode("utf-8")),
        "compact_payload_estimated_tokens": estimate_text_tokens(compact_payload),
        "artifact_bytes": artifact_bytes,
        "artifact_estimated_tokens": artifact_tokens,
    }
    return report


def git_lines(root: Path, args: list[str]) -> list[str]:
    proc = subprocess.run(
        ["git", *args],
        cwd=root,
        text=True,
        capture_output=True,
    )
    if proc.returncode != 0:
        detail = proc.stderr.strip() or proc.stdout.strip() or f"exit {proc.returncode}"
        raise ValueError(f"git {' '.join(args)} failed: {detail}")
    return proc.stdout.splitlines()


def git_paths(root: Path, args: list[str]) -> list[str]:
    proc = subprocess.run(
        ["git", *args],
        cwd=root,
        capture_output=True,
    )
    if proc.returncode != 0:
        stderr = os.fsdecode(proc.stderr).strip()
        stdout = os.fsdecode(proc.stdout).strip()
        detail = stderr or stdout or f"exit {proc.returncode}"
        raise ValueError(f"git {' '.join(args)} failed: {detail}")
    return [os.fsdecode(path) for path in proc.stdout.split(b"\0") if path]


def changed_paths(
    *, root: Path = ROOT, base_ref: str | None = None, head_ref: str | None = None
) -> list[str]:
    if bool(base_ref) != bool(head_ref):
        raise ValueError("base_ref and head_ref must be provided together")

    if base_ref and head_ref:
        merge_base = git_lines(root, ["merge-base", base_ref, head_ref])
        if len(merge_base) != 1 or not merge_base[0]:
            raise ValueError(
                f"git merge-base returned no revision for {base_ref} and {head_ref}"
            )
        changed = git_paths(
            root,
            [
                "diff",
                "--name-only",
                "-z",
                "--diff-filter=ACMR",
                merge_base[0],
                head_ref,
            ],
        )
    else:
        changed = git_paths(
            root, ["diff", "--name-only", "-z", "--diff-filter=ACMR", "HEAD"]
        )
        changed.extend(
            git_paths(root, ["ls-files", "--others", "--exclude-standard", "-z"])
        )
    return sorted(set(changed))


def changed_test_paths(
    *, root: Path = ROOT, base_ref: str | None = None, head_ref: str | None = None
) -> list[str]:
    if bool(base_ref) != bool(head_ref):
        raise ValueError("base_ref and head_ref must be provided together")
    if base_ref and head_ref:
        merge_base = git_lines(root, ["merge-base", base_ref, head_ref])
        if len(merge_base) != 1 or not merge_base[0]:
            raise ValueError(
                f"git merge-base returned no revision for {base_ref} and {head_ref}"
            )
        changed = git_lines(
            root,
            ["diff", "--name-only", "--diff-filter=ACMRD", merge_base[0], head_ref],
        )
    else:
        changed = git_lines(
            root, ["diff", "--name-only", "--diff-filter=ACMRD", "HEAD"]
        )
        changed.extend(git_lines(root, ["ls-files", "--others", "--exclude-standard"]))
    return sorted(set(changed))


def excluded_project_path(path: Path) -> bool:
    return any(part in EXCLUDED_PATH_PARTS for part in path.parts)


def compile_db_file(path: Path, root: Path = ROOT) -> Path:
    resolved = path if path.is_absolute() else root / path
    return resolved / "compile_commands.json" if resolved.is_dir() else resolved


def compile_db_exists(path: Path, root: Path = ROOT) -> bool:
    return compile_db_file(path, root).is_file()


def load_compile_commands(path: Path, root: Path) -> list[dict[str, object]]:
    database = compile_db_file(path, root)
    try:
        payload = json.loads(database.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise ValueError(
            f"failed to read compile database {database}: {error}"
        ) from error
    if not isinstance(payload, list):
        raise ValueError(f"compile database must contain a JSON array: {database}")
    return [entry for entry in payload if isinstance(entry, dict)]


def entry_arguments(entry: dict[str, object]) -> list[str]:
    arguments = entry.get("arguments")
    if isinstance(arguments, list) and all(
        isinstance(value, str) for value in arguments
    ):
        return arguments
    command = entry.get("command")
    if not isinstance(command, str):
        return []
    try:
        return shlex.split(command)
    except ValueError:
        return []


def entry_depfile(entry: dict[str, object], directory: Path) -> Path | None:
    arguments = entry_arguments(entry)
    for index, argument in enumerate(arguments):
        value: str | None = None
        if argument == "-MF" and index + 1 < len(arguments):
            value = arguments[index + 1]
        elif argument.startswith("-MF") and len(argument) > 3:
            value = argument[3:]
        if value:
            path = Path(value)
            return path if path.is_absolute() else directory / path
    return None


def depfile_dependencies(path: Path, directory: Path) -> set[Path] | None:
    try:
        content = path.read_text(encoding="utf-8").replace("\\\n", " ")
    except OSError:
        return None
    _, separator, dependencies = content.partition(":")
    if not separator:
        return None
    try:
        values = shlex.split(dependencies)
    except ValueError:
        return None
    return {
        (candidate if candidate.is_absolute() else directory / candidate).resolve()
        for value in values
        if (candidate := Path(value))
    }


def parse_ninja_dependencies(output: str, build_dir: Path) -> dict[Path, set[Path]]:
    dependencies: dict[Path, set[Path]] = {}
    target: Path | None = None
    for line in output.splitlines():
        if line and not line[0].isspace():
            name, separator, _ = line.partition(": #deps ")
            target = (build_dir / name).resolve() if separator else None
            if target is not None:
                dependencies[target] = set()
        elif target is not None and line.strip():
            dependency = Path(line.strip())
            resolved = (
                dependency if dependency.is_absolute() else build_dir / dependency
            ).resolve()
            dependencies[target].add(resolved)
    return dependencies


def load_ninja_dependencies(build_dir: Path) -> dict[Path, set[Path]]:
    if not (build_dir / "build.ninja").is_file():
        return {}
    try:
        proc = subprocess.run(
            ["ninja", "-C", str(build_dir), "-t", "deps"],
            text=True,
            capture_output=True,
        )
    except OSError:
        return {}
    if proc.returncode != 0:
        return {}
    return parse_ninja_dependencies(proc.stdout, build_dir)


def entry_output(entry: dict[str, object], directory: Path) -> Path | None:
    value = entry.get("output")
    if not isinstance(value, str):
        return None
    path = Path(value)
    return (path if path.is_absolute() else directory / path).resolve()


def impacted_translation_units(
    headers: list[str], compile_db: Path, root: Path
) -> list[str]:
    changed_headers = {(root / header).resolve() for header in headers}
    impacted: set[str] = set()
    root_resolved = root.resolve()
    database = compile_db_file(compile_db, root)
    ninja_dependencies: dict[Path, set[Path]] | None = None
    for entry in load_compile_commands(database, root):
        directory_value = entry.get("directory")
        file_value = entry.get("file")
        if not isinstance(directory_value, str) or not isinstance(file_value, str):
            continue
        directory = Path(directory_value)
        source = Path(file_value)
        source = source if source.is_absolute() else directory / source
        try:
            relative_source = source.resolve().relative_to(root_resolved)
        except ValueError:
            continue
        if (
            source.suffix not in SOURCE_SUFFIXES
            or excluded_project_path(relative_source)
            or not source.is_file()
        ):
            continue
        depfile = entry_depfile(entry, directory)
        dependencies = (
            depfile_dependencies(depfile, directory) if depfile is not None else None
        )
        if dependencies is None:
            if ninja_dependencies is None:
                ninja_dependencies = load_ninja_dependencies(database.parent)
            output = entry_output(entry, directory)
            dependencies = (
                ninja_dependencies.get(output) if output is not None else None
            )
        if dependencies is None or changed_headers.intersection(dependencies):
            impacted.add(relative_source.as_posix())
    return sorted(impacted)


def changed_source_files(
    *,
    root: Path = ROOT,
    base_ref: str | None = None,
    head_ref: str | None = None,
    compile_db: Path | None = None,
) -> list[str]:
    changed = changed_paths(root=root, base_ref=base_ref, head_ref=head_ref)
    sources = {
        path
        for path in changed
        if Path(path).suffix in SOURCE_SUFFIXES
        and not excluded_project_path(Path(path))
        and (root / path).is_file()
    }
    headers = [
        path
        for path in changed
        if Path(path).suffix in HEADER_SUFFIXES
        and not excluded_project_path(Path(path))
        and (root / path).is_file()
    ]
    if headers and compile_db is not None:
        sources.update(impacted_translation_units(headers, compile_db, root))
    return sorted(sources)


def diff_scan_precondition_errors(
    *,
    ci: bool,
    skip_diff_scan: bool,
    base_ref: str | None,
    head_ref: str | None,
    compile_db: Path | None,
) -> list[str]:
    if skip_diff_scan:
        return []
    errors: list[str] = []
    if bool(base_ref) != bool(head_ref):
        errors.append("--base-ref and --head-ref must be provided together")
    elif ci and not (base_ref and head_ref):
        errors.append("CI differential scan requires base/head refs")
    if ci and compile_db is None:
        errors.append("CI differential scan requires a compile database")
    elif compile_db is not None and not compile_db_exists(compile_db):
        errors.append(f"compile database not found: {compile_db}")
    return errors


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--fold", type=Path, required=True)
    parser.add_argument("--out-dir", type=Path, default=ROOT / "build" / "quality")
    parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST)
    parser.add_argument("--compile-db", type=Path)
    parser.add_argument(
        "--test-impact-manifest", type=Path, default=DEFAULT_TEST_IMPACT_MANIFEST
    )
    parser.add_argument(
        "--test-repetitions",
        type=positive_integer,
        default=1,
        help="run each selected test set N times and retain flake evidence",
    )
    parser.add_argument(
        "--base-ref", help="base revision used to compute the PR merge base"
    )
    parser.add_argument("--head-ref", help="head revision used for the branch diff")
    parser.add_argument("--skip-format", action="store_true")
    parser.add_argument("--skip-tests", action="store_true")
    parser.add_argument(
        "--full-tests",
        action="store_true",
        help="run the complete Meson suite instead of impacted tests",
    )
    parser.add_argument("--skip-policy", action="store_true")
    parser.add_argument("--skip-diff-scan", action="store_true")
    parser.add_argument("--allow-findings", action="store_true")
    args = parser.parse_args()

    fold = args.fold.expanduser().resolve()
    if not fold.is_file():
        print(f"fold binary not found: {fold}", file=sys.stderr)
        return 2
    out_dir = args.out_dir.expanduser().resolve()
    out_dir.mkdir(parents=True, exist_ok=True)
    steps: list[dict[str, object]] = []
    if not args.skip_format:
        steps.append(
            run_step(
                "format",
                ["./scripts/format-code.sh", "--check"],
                artifact_dir=out_dir,
            )
        )
        steps.append(
            run_step(
                "diff-hygiene",
                ["git", "diff", "--check"],
                artifact_dir=out_dir,
            )
        )
    if not args.skip_tests:
        try:
            test_changed_paths = changed_test_paths(
                base_ref=args.base_ref, head_ref=args.head_ref
            )
            test_selection = select_impacted_tests(
                test_changed_paths, args.test_impact_manifest
            )
        except ValueError as error:
            test_selection = {
                "schema_version": 1,
                "manifest": args.test_impact_manifest.as_posix(),
                "changed_paths": [],
                "full_suite": True,
                "fallback_reasons": [f"test selection failed closed: {error}"],
                "unmapped_paths": [],
                "matched_rules": [],
                "selected_tests": [],
            }
        apply_full_test_override(test_selection, args.full_tests)
        selection_path = out_dir / "test-selection.json"
        selection_path.write_text(
            json.dumps(test_selection, indent=2, sort_keys=True) + "\n",
            encoding="utf-8",
        )
        full_suite_value = test_selection.get("full_suite")
        full_suite = full_suite_value if isinstance(full_suite_value, bool) else True
        selected_value = test_selection.get("selected_tests", [])
        selected = selected_value if isinstance(selected_value, list) else []
        unmapped_value = test_selection.get("unmapped_paths", [])
        unmapped_paths = unmapped_value if isinstance(unmapped_value, list) else []
        matched_value = test_selection.get("matched_rules", [])
        matched_rules = matched_value if isinstance(matched_value, list) else []
        fallback_value = test_selection.get("fallback_reasons", [])
        fallback_reasons = fallback_value if isinstance(fallback_value, list) else []
        selected_names = [
            str(item["name"])
            for item in selected
            if isinstance(item, dict) and isinstance(item.get("name"), str)
        ]
        compact_matched_rules = []
        for item in matched_rules:
            if not isinstance(item, dict):
                continue
            paths = item.get("paths", [])
            compact_matched_rules.append(
                {
                    "id": str(item.get("id", "unknown")),
                    "path_count": len(paths) if isinstance(paths, list) else 0,
                }
            )
        attempt_records: list[dict[str, object]] = []
        evidence_dir = out_dir / "test-evidence"
        evidence_dir.mkdir(parents=True, exist_ok=True)
        for attempt in range(1, args.test_repetitions + 1):
            logbase = f"fold-quality-attempt-{attempt}"
            meson_log = ROOT / "build" / "release" / "meson-logs" / f"{logbase}.json"
            meson_log.unlink(missing_ok=True)
            test_command = [
                "meson",
                "test",
                "-C",
                "build/release",
                "--print-errorlogs",
                "--logbase",
                logbase,
            ]
            if not full_suite:
                test_command.extend(selected_names)
            test_step = run_step(
                f"impacted-tests-attempt-{attempt}",
                test_command,
                artifact_dir=out_dir,
            )
            exit_code = test_step.get("exit_code")
            parsed_attempt = parse_test_attempt(
                meson_log,
                attempt,
                exit_code if isinstance(exit_code, int) else 2,
            )
            attempt_path = evidence_dir / f"attempt-{attempt}.json"
            attempt_path.write_text(
                json.dumps(parsed_attempt, indent=2, sort_keys=True) + "\n",
                encoding="utf-8",
            )
            artifacts = test_step.get("artifacts", [])
            if isinstance(artifacts, list):
                artifacts.append(
                    text_artifact(attempt_path, out_dir, "test-attempt-evidence")
                )
            parsed_tests_value = parsed_attempt.get("tests", [])
            parsed_tests = (
                parsed_tests_value if isinstance(parsed_tests_value, list) else []
            )
            test_step["test_attempt"] = {
                "attempt": attempt,
                "result_count": len(parsed_tests),
                "parse_error": parsed_attempt.get("parse_error"),
            }
            steps.append(test_step)
            attempt_records.append(parsed_attempt)

        flaky_summary = summarize_test_attempts(
            attempt_records,
            selected_names if not full_suite else None,
        )
        flaky_path = out_dir / "flaky-test-evidence.json"
        flaky_path.write_text(
            json.dumps(flaky_summary, indent=2, sort_keys=True) + "\n",
            encoding="utf-8",
        )
        passed_value = flaky_summary.get("all_attempts_passed")
        classification_value = flaky_summary.get("counts", {})
        classification_counts = (
            classification_value if isinstance(classification_value, dict) else {}
        )
        flaky_value = flaky_summary.get("flaky_tests", [])
        flaky_tests = flaky_value if isinstance(flaky_value, list) else []
        parse_errors_value = flaky_summary.get("parse_errors", [])
        parse_errors = (
            parse_errors_value if isinstance(parse_errors_value, list) else []
        )
        steps.append(
            {
                "name": "flaky-test-evidence",
                "passed": passed_value if isinstance(passed_value, bool) else False,
                "test_repetitions": args.test_repetitions,
                "classification_counts": classification_counts,
                "flaky_tests": flaky_tests[:MAX_FINDING_DETAILS],
                "parse_error_count": len(parse_errors),
                "test_selection": {
                    "full_suite": full_suite,
                    "selected_tests": selected,
                    "matched_rules": compact_matched_rules,
                    "fallback_reasons": fallback_reasons,
                    "unmapped_path_count": len(unmapped_paths),
                },
                "artifacts": [
                    text_artifact(selection_path, out_dir, "test-selection"),
                    text_artifact(flaky_path, out_dir, "flaky-test-evidence"),
                ],
            }
        )
    if not args.skip_policy:
        policy_step = run_step(
            "policy",
            [
                sys.executable,
                "scripts/benchmarks/run_project_eval.py",
                "--manifest",
                str(args.manifest),
                "--fold",
                str(fold),
                "--out-dir",
                str(out_dir / "policy"),
            ],
            artifact_dir=out_dir,
        )
        policy_details = summarize_policy_report(
            out_dir / "policy" / "summary.json", out_dir
        )
        policy_artifacts = policy_details.pop("artifacts", [])
        if isinstance(policy_artifacts, list):
            artifacts = policy_step.get("artifacts", [])
            if isinstance(artifacts, list):
                artifacts.extend(policy_artifacts)
        policy_step.update(policy_details)
        steps.append(policy_step)
    ci = os.environ.get("CI", "").lower() in {"1", "true", "yes"}
    precondition_errors = diff_scan_precondition_errors(
        ci=ci,
        skip_diff_scan=args.skip_diff_scan,
        base_ref=args.base_ref,
        head_ref=args.head_ref,
        compile_db=args.compile_db,
    )
    changed: list[str] = []
    if precondition_errors:
        steps.append(
            {
                "name": "diff-scan",
                "passed": False,
                "errors": precondition_errors,
                "changed_files": changed,
            }
        )
    elif not args.skip_diff_scan:
        try:
            changed = changed_source_files(
                base_ref=args.base_ref,
                head_ref=args.head_ref,
                compile_db=args.compile_db,
            )
        except ValueError as error:
            steps.append(
                {
                    "name": "diff-scan",
                    "passed": False,
                    "errors": [str(error)],
                    "changed_files": changed,
                }
            )
        else:
            if changed and args.compile_db:
                result = run_step(
                    "diff-scan",
                    [
                        str(fold),
                        "scan",
                        "--mode",
                        "source",
                        "--compile-db",
                        str(args.compile_db),
                        "--format",
                        "sarif",
                        "--output",
                        str(out_dir / "diff-scan.sarif.json"),
                        *changed,
                    ],
                    artifact_dir=out_dir,
                    acceptable={0, 1} if args.allow_findings else {0},
                )
                result["changed_files"] = changed
                scan_report = out_dir / "diff-scan.sarif.json"
                if scan_report.is_file():
                    artifacts = result.get("artifacts", [])
                    if isinstance(artifacts, list):
                        artifacts.append(
                            text_artifact(scan_report, out_dir, "scan-report")
                        )
                steps.append(result)
            else:
                reason = (
                    "no changed C/C++ files"
                    if not changed
                    else "compile database omitted for local differential scan"
                )
                steps.append(
                    {
                        "name": "diff-scan",
                        "skipped": True,
                        "changed_files": changed,
                        "reason": reason,
                    }
                )

    report = finalize_report(steps)
    report_path = out_dir / "summary.json"
    report_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(report, indent=2))
    return 0 if report["status"] == "ok" else 1


if __name__ == "__main__":
    raise SystemExit(main())