bmc_agent/bmc_engine.py

"""
Phase 2: Compositional BMC Engine [CONVENTIONAL].

Deterministic tool invocation — not agentic. Exposes one interface:
  check(function, spec, callee_specs) -> {verified, counterexample}
The agentic layers use this interface; swapping CBMC for another backend
changes only the harness-synthesis-and-invocation code here.
"""

from __future__ import annotations

import dataclasses
import shutil
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

from bmc_agent.artifacts import ArtifactStore
from bmc_agent.backends import BMCBackend, CBMCBackend
from bmc_agent.cbmc import CBMCResult, Counterexample, run_cbmc
from bmc_agent.config import Config
from bmc_agent.harness_generator import HarnessGenerator
from bmc_agent.logger import get_logger
from bmc_agent.parser import FunctionInfo, ParsedCFile
from bmc_agent.spec import Spec

logger = get_logger("bmc_engine")


# CBMC error substrings that indicate the harness failed to BUILD (parse /
# convert / type-check) rather than a verification outcome (property failure,
# unwind-bound, timeout). Used to trigger the agentic harness-repair fallback.
_HARNESS_BUILD_ERROR_MARKERS = (
    "conversion error",
    "incomplete type",
    "parsing error",
    "redefinition",
    "conflicting types",
    "syntax error",
    "cannot convert",
    "type mismatch",
    "expected ',' or ';'",
    "undeclared",
)


def _is_harness_build_error(err: "str | None") -> bool:
    """True iff a CBMC error string looks like a harness BUILD failure (parse /
    conversion / incomplete-type), as opposed to a verification result or a
    resource limit (unwind bound, timeout)."""
    if not err:
        return False
    e = err.lower()
    # Never treat resource/verification outcomes as build errors.
    if "unwind" in e or "timed out" in e or "timeout" in e:
        return False
    return any(m in e for m in _HARNESS_BUILD_ERROR_MARKERS)


def _harness_entry_of(harness_path) -> "str | None":
    """Read the `/* Harness entry: NAME */` header tag, if present. Returns the
    entry function name CBMC should use via --function, or None for main()."""
    try:
        with open(harness_path, "r") as _hf:
            for _line in _hf:
                if "Harness entry:" in _line:
                    _name = _line.split("Harness entry:", 1)[1].strip().rstrip("*/").strip()
                    return _name if _name and _name != "main" else None
                if not _line.startswith("/*") and "include" in _line:
                    break  # past the header
    except Exception:
        pass
    return None


@dataclass
class BMCVerdict:
    """Result of running BMC on a single function."""

    function_name: str
    verified: bool                          # True = verified up to bound k
    counterexamples: list[Counterexample] = field(default_factory=list)
    harness_path: str = ""
    cbmc_result: Optional[CBMCResult] = None
    error: Optional[str] = None

    def to_dict(self) -> dict:
        d = dataclasses.asdict(self)
        # CBMCResult and Counterexample are already handled by asdict
        return d


class BMCEngine:
    """Runs CBMC on function harnesses to check specs."""

    def __init__(
        self,
        config: Config,
        store: ArtifactStore,
        backend: "BMCBackend | None" = None,
    ) -> None:
        self.config = config
        self.store = store
        self.harness_gen = HarnessGenerator(config)  # kept for backward compat
        self.backend: BMCBackend = backend or CBMCBackend(config)

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    def check_function(
        self,
        func: FunctionInfo,
        spec: Spec,
        parsed_file: ParsedCFile,
        driver_name: str,
        all_funcs: "dict | None" = None,
        flag_selection: "object | None" = None,
    ) -> BMCVerdict:
        """
        Check a single function against its spec using CBMC.

        Steps:
        1. Generate a CBMC harness.
        2. Save harness to the artifact directory.
        3. Run CBMC.
        4. Return a structured BMCVerdict.
        """
        fn_name = func.name
        logger.info("Checking function '%s' (driver '%s')", fn_name, driver_name)

        # ---- Step 1: generate harness ----
        try:
            harness_src = self.backend.generate_harness(func, spec, {}, parsed_file, all_funcs=all_funcs)
        except Exception as exc:
            # Distinguish unresolvable-type skips from genuine generator failures.
            # Unresolvable-type cases (impl-method functions referencing types
            # imported from sibling files Kani can't see) are an expected skip,
            # not an error -- they're noise if logged at ERROR level. Log them
            # at INFO and use a typed error marker so downstream filters can
            # exclude them from the harness-compile-failure tally.
            from bmc_agent.backends.kani_backend import HarnessUnresolvableTypes
            if isinstance(exc, HarnessUnresolvableTypes):
                logger.info(
                    "Skipping harness for '%s' — unresolvable types: %s",
                    fn_name,
                    ", ".join(exc.unresolved_types),
                )
                return BMCVerdict(
                    function_name=fn_name,
                    verified=False,
                    error=f"harness-skipped-unresolvable-types: {', '.join(exc.unresolved_types)}",
                )
            logger.error("Harness generation failed for '%s': %s", fn_name, exc)
            return BMCVerdict(
                function_name=fn_name,
                verified=False,
                error=f"Harness generation failed: {exc}",
            )

        # ---- Step 2: save harness ----
        harness_path = self._save_harness(driver_name, fn_name, harness_src)
        logger.debug("Harness saved to: %s", harness_path)

        # ---- Step 3: run the backend verifier ----
        # The CBMC path threads threat-model + per-function flag selection
        # through to run_cbmc.  Non-C backends (Kani for Rust) don't take
        # those CBMC-specific flags, so for them we use the polymorphic
        # backend.check() method and ignore flag selection.
        if self.backend.language == "c":
            threat_model = getattr(self.config, "threat_model", "security")
            pointer_check    = threat_model in ("security", "safety")
            bounds_check     = threat_model in ("security", "safety")
            div_by_zero_check = threat_model == "safety"

            unsigned_overflow_check = bool(getattr(flag_selection, "unsigned_overflow_check", False))
            signed_overflow_check   = bool(getattr(flag_selection, "signed_overflow_check", False))
            conversion_check        = bool(getattr(flag_selection, "conversion_check", False))
            pointer_overflow_check  = bool(getattr(flag_selection, "pointer_overflow_check", False))
            undefined_shift_check   = bool(getattr(flag_selection, "undefined_shift_check", False))
            # Per-function unwind override (None = use global default).
            unwind_for_this_run     = getattr(flag_selection, "unwind_override", None) or self.config.cbmc_unwind
            # Per-function CBMC timeout override (None = use global default).
            timeout_for_this_run    = getattr(flag_selection, "timeout_override", None) or self.config.cbmc_timeout

            if flag_selection and flag_selection.any_enabled():
                logger.debug(
                    "Flag selection for '%s': %s (%s)",
                    fn_name,
                    ", ".join(flag_selection.enabled_flags()),
                    getattr(flag_selection, "reasoning", ""),
                )
            # Extract the harness entry function name from the harness
            # file header (real-libc mode tags it with `/* Harness entry:
            # NAME */`). When the source already defines `main`, the
            # harness uses a different function name and CBMC needs
            # --function to pick it.
            harness_entry = _harness_entry_of(harness_path)

            def _run_c_cbmc(_hpath, _hentry):
                return run_cbmc(
                    harness_path=_hpath,
                    unwind=unwind_for_this_run,
                    timeout=timeout_for_this_run,
                    cbmc_path=self.config.cbmc_path,
                    include_dirs=getattr(self.config, "include_dirs", None),
                    defines=getattr(self.config, "cbmc_defines", None),
                    unsigned_overflow_check=unsigned_overflow_check,
                    signed_overflow_check=signed_overflow_check,
                    conversion_check=conversion_check,
                    pointer_overflow_check=pointer_overflow_check,
                    undefined_shift_check=undefined_shift_check,
                    pointer_check=pointer_check,
                    bounds_check=bounds_check,
                    div_by_zero_check=div_by_zero_check,
                    object_bits=getattr(self.config, "cbmc_object_bits", None),
                    auto_scale_object_bits=getattr(
                        self.config, "cbmc_auto_scale_object_bits", True
                    ),
                    function=_hentry,
                )

            cbmc_result = _run_c_cbmc(harness_path, harness_entry)

            # Agentic harness-repair fallback (opt-in): the DETERMINISTIC harness
            # failed to BUILD (conversion / incomplete-type / parse error). Let the
            # code-reading AgenticHarnessGen rebuild it, then re-run. Fires only on
            # a build error, so there's no soundness downside — a non-building
            # harness yields no verdict either way.
            if (
                getattr(self.config, "enable_agentic_harness_repair", False)
                and cbmc_result.error
                and _is_harness_build_error(cbmc_result.error)
            ):
                repaired_path = self._agentic_repair_harness(
                    func, parsed_file, all_funcs, driver_name,
                    build_error=cbmc_result.error,
                )
                if repaired_path is not None:
                    repaired_result = _run_c_cbmc(
                        repaired_path, _harness_entry_of(repaired_path)
                    )
                    if not (
                        repaired_result.error
                        and _is_harness_build_error(repaired_result.error)
                    ):
                        logger.info(
                            "agentic harness repair resolved the build error for "
                            "'%s' (verified=%s, cex=%d)",
                            fn_name, repaired_result.verified,
                            len(repaired_result.counterexamples),
                        )
                        harness_path = repaired_path
                        cbmc_result = repaired_result
                    else:
                        logger.info(
                            "agentic harness repair did NOT resolve the build "
                            "error for '%s' — keeping original verdict", fn_name,
                        )
        else:
            # Rust / Kani path: the backend wraps its own verifier
            # invocation and returns a CBMCResult-shaped object.
            cbmc_result = self.backend.check(
                harness_path,
                harness_name=f"check_{fn_name}",
                source_path=getattr(parsed_file, "path", None),
            )

            # Timeout retry: complex Rust functions (UTF-8 validation,
            # allocator-driven Vec/String code) can blow past Kani's
            # 120-s timeout at the default slice_bound=4. Regenerate
            # the harness with progressively smaller buffer bounds
            # (4 → 2 → 1) and a tighter loop unwind, then re-check.
            # Each retry overwrites the saved harness so the artifact
            # reflects the configuration that actually produced the
            # verdict. Regression: CCC encoding.rs run 2026-05-19 —
            # bytes_to_string and encode_non_utf8 timed out at default
            # bound=4; encode_non_utf8 verifies clean at bound=1 in ~40s.
            # Unwind-exhausted retry: when Kani reports "loop unwind
            # bound exhausted" the result is inconclusive — the function
            # would have verified clean if the solver had been allowed
            # more loop iterations. Bump unwind 4 → 16 (one retry) and
            # try again; keep slice_bound the same. Distinct from the
            # timeout-retry path below, which shrinks state.
            if (
                cbmc_result.error
                and "unwind bound exhausted" in cbmc_result.error
                and getattr(self.backend, "generate_harness", None) is not None
            ):
                bumped_unwind = max(self.config.kani_unwind * 4, 16)
                logger.info(
                    "Kani unwind-exhausted for '%s' — retrying with unwind=%d",
                    fn_name, bumped_unwind,
                )
                cbmc_result = self.backend.check(
                    harness_path,
                    harness_name=f"check_{fn_name}",
                    source_path=getattr(parsed_file, "path", None),
                    unwind_override=bumped_unwind,
                )
                if not cbmc_result.error or "unwind bound" not in cbmc_result.error:
                    logger.info(
                        "Kani unwind-bump succeeded for '%s' at unwind=%d",
                        fn_name, bumped_unwind,
                    )

            if (
                cbmc_result.error
                and "timed out" in cbmc_result.error
                and getattr(self.backend, "generate_harness", None) is not None
            ):
                # Use a shorter wall-clock for retries: the whole point of
                # shrinking the bound is to reduce state-space, so if
                # bound=1 doesn't land in 60s it almost certainly won't
                # land in 120s. Keeps total per-function budget bounded
                # (worst case: 120 + 60 + 60 ≈ 4 min, vs 6 min before).
                retry_timeout = min(60, self.config.kani_timeout)
                for retry_bound, retry_unwind in [(2, 4), (1, 2)]:
                    logger.info(
                        "Kani timed out for '%s' — retrying with "
                        "slice_bound=%d, unwind=%d, timeout=%ds",
                        fn_name, retry_bound, retry_unwind, retry_timeout,
                    )
                    try:
                        retry_src = self.backend.generate_harness(
                            func, spec, {}, parsed_file,
                            all_funcs=all_funcs,
                            slice_bound_override=retry_bound,
                        )
                    except Exception as exc:
                        logger.warning(
                            "Retry harness gen failed for '%s' at bound=%d: %s",
                            fn_name, retry_bound, exc,
                        )
                        break
                    retry_path = self._save_harness(driver_name, fn_name, retry_src)
                    cbmc_result = self.backend.check(
                        retry_path,
                        harness_name=f"check_{fn_name}",
                        source_path=getattr(parsed_file, "path", None),
                        unwind_override=retry_unwind,
                        timeout_override=retry_timeout,
                    )
                    harness_path = retry_path
                    if not cbmc_result.error or "timed out" not in cbmc_result.error:
                        logger.info(
                            "Kani retry succeeded for '%s' at slice_bound=%d",
                            fn_name, retry_bound,
                        )
                        break

        # ---- Step 4: build verdict ----
        if cbmc_result.error:
            logger.warning(
                "CBMC error for '%s': %s", fn_name, cbmc_result.error
            )
            verdict = BMCVerdict(
                function_name=fn_name,
                verified=False,
                counterexamples=cbmc_result.counterexamples,
                harness_path=str(harness_path),
                cbmc_result=cbmc_result,
                error=cbmc_result.error,
            )
        else:
            logger.info(
                "CBMC verdict for '%s': verified=%s, counterexamples=%d",
                fn_name,
                cbmc_result.verified,
                len(cbmc_result.counterexamples),
            )
            verdict = BMCVerdict(
                function_name=fn_name,
                verified=cbmc_result.verified,
                counterexamples=cbmc_result.counterexamples,
                harness_path=str(harness_path),
                cbmc_result=cbmc_result,
                error=None,
            )

        # ---- Save results to artifact store ----
        try:
            self.store.save_cbmc_result(driver_name, fn_name, cbmc_result)
            self.store.save_bug_report(driver_name, fn_name, verdict.to_dict())
        except Exception as exc:
            logger.warning("Failed to save artifacts for '%s': %s", fn_name, exc)

        return verdict

    def _agentic_repair_harness(
        self,
        func: FunctionInfo,
        parsed_file: ParsedCFile,
        all_funcs: "dict | None",
        driver_name: str,
        build_error: str,
    ) -> "str | None":
        """Rebuild a non-compiling harness with the agentic, code-reading
        generator (``AgenticHarnessGen``), which reads the real structs/headers
        and compile-checks with retry. Returns the path to a freshly saved
        harness on success, or None. Fail-safe: any error returns None and the
        caller keeps the original (failed) verdict.
        """
        try:
            from pathlib import Path as _Path
            from bmc_agent.agentic_harness_gen import AgenticHarnessGen

            src_path = getattr(parsed_file, "path", "") or ""
            parsed_files = {src_path: parsed_file} if src_path else {}
            corpus_root = _Path(src_path).parent if src_path else _Path(".")
            logger.info(
                "agentic harness repair: rebuilding harness for '%s' "
                "(build error: %s)",
                func.name, (build_error or "")[:140],
            )
            agen = AgenticHarnessGen(
                config=self.config,
                parsed_files=parsed_files,
                corpus_root=corpus_root,
            )
            gen_kwargs = dict(
                func=func,
                all_funcs_global=all_funcs or {},
                include_dirs=list(getattr(self.config, "include_dirs", None) or []),
                defines=list(getattr(self.config, "cbmc_defines", None) or []),
            )
            # Route by the RESOLVED PROVIDER, not just the claude_code_agentic
            # flag: under --agentic the user may pin the agentic roles to an API
            # (e.g. OpenRouter) via per-role overrides. Honour that — use the
            # Claude Code agent ONLY when the resolved provider is actually
            # claude-code; otherwise use bmc's in-process tool loop, which runs on
            # the openai-compatible API. (Previously this hardcoded claude-code
            # whenever claude_code_agentic was set, so "--agentic with API" still
            # fell back to the claude-code subscription for harness repair.)
            try:
                rs = self.config.role_settings("spec_gen") if hasattr(self.config, "role_settings") else None
                prov = (rs or {}).get("provider") or self.config.resolved_provider()
            except Exception:
                prov = "claude-code" if getattr(self.config, "claude_code_agentic", False) else ""
            if prov == "claude-code":
                logger.info("agentic harness repair: using Claude Code agent for '%s'", func.name)
                res = agen.generate_via_claude_code(**gen_kwargs)
            else:
                logger.info("agentic harness repair: using in-process tool loop (provider=%s) for '%s'",
                            prov or "?", func.name)
                res = agen.generate(**gen_kwargs)
            harness = getattr(res, "harness", None)
            if harness and not getattr(res, "last_compile_error", None):
                return str(self._save_harness(driver_name, func.name, harness))
            logger.info(
                "agentic harness repair produced no clean harness for '%s' "
                "(last_compile_error=%s)",
                func.name, str(getattr(res, "last_compile_error", ""))[:140],
            )
        except Exception as exc:
            logger.warning(
                "agentic harness repair failed for '%s': %s", func.name, exc
            )
        return None

    def check_all(
        self,
        funcs: dict[str, FunctionInfo],
        specs: dict[str, Spec],
        parsed_file: ParsedCFile,
        driver_name: str,
        all_funcs: "dict | None" = None,
        flag_selections: "dict | None" = None,
    ) -> dict[str, BMCVerdict]:
        """
        Check all functions in parallel (ThreadPoolExecutor).

        Parameters
        ----------
        funcs:
            Mapping function_name → FunctionInfo.
        specs:
            Mapping function_name → Spec.
        parsed_file:
            The parsed C file object.
        driver_name:
            Driver name for artifact storage.

        Returns
        -------
        Mapping function_name → BMCVerdict.
        """
        verdicts: dict[str, BMCVerdict] = {}

        # Only check functions that have both a FunctionInfo and a Spec
        to_check = {
            name: funcs[name]
            for name in funcs
            if name in specs
        }
        if not to_check:
            logger.warning("No functions to check in driver '%s'", driver_name)
            return verdicts

        max_workers = min(len(to_check), self.config.batch_size, 8)
        logger.info(
            "Checking %d functions in driver '%s' with %d workers",
            len(to_check),
            driver_name,
            max_workers,
        )

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_name = {
                executor.submit(
                    self.check_function,
                    func,
                    specs[name],
                    parsed_file,
                    driver_name,
                    all_funcs,
                    (flag_selections or {}).get(name),
                ): name
                for name, func in to_check.items()
            }
            for future in as_completed(future_to_name):
                name = future_to_name[future]
                try:
                    verdict = future.result()
                    verdicts[name] = verdict
                except Exception as exc:
                    logger.error(
                        "Unexpected error checking '%s': %s", name, exc
                    )
                    verdicts[name] = BMCVerdict(
                        function_name=name,
                        verified=False,
                        error=f"Unexpected error: {exc}",
                    )

        return verdicts

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _save_harness(
        self,
        driver_name: str,
        func_name: str,
        harness_src: str,
    ) -> Path:
        """
        Save the harness source to
        ``{artifact_dir}/{driver_name}/{func_name}/harness.{c,rs}``.

        The extension follows ``self.backend.language`` so Rust harnesses
        get a ``.rs`` extension (required for Kani to parse them) and
        artifact inspection tools can rely on filename to language.
        """
        ext = "rs" if self.backend.language == "rust" else "c"
        fn_dir = (
            Path(self.config.artifact_dir) / driver_name / func_name
        )
        fn_dir.mkdir(parents=True, exist_ok=True)
        harness_path = fn_dir / f"harness.{ext}"
        harness_path.write_text(harness_src, encoding="utf-8")
        return harness_path