exploitability_analyzer.py

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Exploitability Analyzer agent function for NAT."""

import logging
import re
from datetime import UTC
from datetime import datetime
from pathlib import Path

from langchain_core.messages import AIMessage
from langchain_core.messages import HumanMessage
from langchain_core.messages import SystemMessage
from nat.builder.builder import Builder
from nat.builder.framework_enum import LLMFrameworkEnum
from nat.builder.function_info import FunctionInfo
from nat.cli.register_workflow import register_function
from nat.data_models.component_ref import FunctionRef
from nat.data_models.function import FunctionBaseConfig
from pydantic import Field

from vuln_analysis.data_models.exploitability_analyzer import ExploitabilityAnalyzerInput
from vuln_analysis.data_models.exploitability_analyzer import ExploitabilityAnalyzerResult
from vuln_analysis.data_models.exploitability_analyzer import PackageVerification
from vuln_analysis.data_models.exploitability_analyzer import PreTriageResult
from vuln_analysis.data_models.exploitability_analyzer import extract_container_name_from_path
from vuln_analysis.data_models.exploitability_analyzer import extract_cve_id_from_path
from vuln_analysis.prompts.exploitability_analyzer import PHASE2_SYSTEM_PROMPT
from vuln_analysis.prompts.exploitability_analyzer import build_phase2_user_message
from vuln_analysis.semantic_failure import serialize_result_for_cli
from vuln_analysis.tools.filesystem_tools import configure_allowed_paths
from vuln_analysis.utils.agentic_loop import run_agentic_loop
from vuln_analysis.utils.concurrency import make_lazy_rate_limiter
from vuln_analysis.utils.exploitability_report_metadata import EXPLOITABILITY_REPORT_POLICY_VERSION
from vuln_analysis.utils.exploitability_report_metadata import ExploitabilityReportMetadata
from vuln_analysis.utils.exploitability_report_metadata import ExploitabilityReportMetadataError
from vuln_analysis.utils.exploitability_report_metadata import build_exploitability_report_metadata
from vuln_analysis.utils.exploitability_report_metadata import exploitability_report_structure_failures
from vuln_analysis.utils.exploitability_report_metadata import normalize_exploitability_report_metadata
from vuln_analysis.utils.model_errors import is_retryable_model_error
from vuln_analysis.utils.report_envelope import strip_report_preamble
from vuln_analysis.utils.report_generator import generate_early_exit_report
from vuln_analysis.utils.report_parser import extract_markdown_table_field

logger = logging.getLogger(__name__)

_STRUCTURAL_REPAIR_SYSTEM_PROMPT = """You repair the structure of a completed container
exploitability report. Preserve its analysis conclusions and evidence. Return only the
corrected Markdown report: no narration, preamble, explanation, or code fence. The first
line must be exactly `# Container Exploitability Analysis`. Include exactly one complete
report and exactly one `## Summary` section."""


class ExploitabilityAnalyzerConfig(FunctionBaseConfig, name="exploitability_analyzer"):
    llm_name: str = Field(description="LLM for Phase 2 deep analysis")
    tool_names: list[FunctionRef] = Field(description="Filesystem + binary tools")
    phase2_max_iterations: int = Field(default=40)
    context_window_tokens: int | None = Field(
        default=None,
        description=(
            "Optional deployment-specific context cap. When omitted, derive the limit from "
            "the configured model's metadata and hosted-endpoint profile."
        ),
    )
    wrap_up_fraction: float = Field(default=0.80)
    loop_detection_threshold: int = Field(default=2)
    llm_max_rate: int | None = Field(
        default=None,
        description="Per-function LLM rate limit (requests/second). Overrides the workflow setting.",
    )
    output_subdir: str = Field(default="exploitability_reports")
    description: str = Field(default="Analyzes exploitability of a CVE in a container")


@register_function(config_type=ExploitabilityAnalyzerConfig, framework_wrappers=[LLMFrameworkEnum.LANGCHAIN])
async def exploitability_analyzer(config: ExploitabilityAnalyzerConfig, builder: Builder):
    llm = await builder.get_llm(llm_name=config.llm_name, wrapper_type=LLMFrameworkEnum.LANGCHAIN)
    tools = await builder.get_tools(tool_names=config.tool_names, wrapper_type=LLMFrameworkEnum.LANGCHAIN)
    get_rate_limiter = make_lazy_rate_limiter(config.llm_max_rate, builder)

    async def _arun(inp: ExploitabilityAnalyzerInput) -> ExploitabilityAnalyzerResult:
        rate_limiter = get_rate_limiter()
        for label, path in [
            ("Vulnerability report", inp.vulnerability_report_path),
            ("Container report", inp.container_report_path),
        ]:
            if not Path(path).is_file():
                return ExploitabilityAnalyzerResult(
                    cve_id="unknown", container_name="unknown",
                    success=False, error=f"{label} not found: {path}",
                )

        if not Path(inp.filesystem_path).is_dir():
            return ExploitabilityAnalyzerResult(
                cve_id="unknown", container_name="unknown",
                success=False, error=f"Filesystem directory not found: {inp.filesystem_path}",
            )

        cve_id = _extract_cve_id(inp.vulnerability_report_path)
        container_name = extract_container_name_from_path(inp.container_report_path)
        report_metadata = build_exploitability_report_metadata(
            analysis_date=datetime.now(UTC).date(),
        )

        logger.info("Starting exploitability analysis: %s × %s", cve_id, container_name)

        # Pre-triage (deterministic)
        try:
            from vuln_analysis.utils.pre_triage import run_pre_triage
            pre_triage_result = run_pre_triage(inp)
        except Exception as exc:
            logger.error("Pre-triage failed: %s", exc, exc_info=True)
            return ExploitabilityAnalyzerResult(
                cve_id=cve_id, container_name=container_name,
                success=False, error=f"Pre-triage failed: {exc}",
            )

        if pre_triage_result.early_exit:
            report_content = generate_early_exit_report(
                pre_triage_result,
                inp,
                analysis_date=report_metadata.analysis_date,
            )
            verdict = pre_triage_result.early_exit_verdict
            final_validation_failures = list(
                exploitability_report_validation_failures(
                    report_content,
                    pre_triage_result,
                )
            )
            logger.info(
                "Pre-triage early exit: verdict=%s, reason=%s",
                verdict,
                pre_triage_result.early_exit_reason,
            )
            return ExploitabilityAnalyzerResult(
                cve_id=cve_id,
                container_name=container_name,
                report_content=report_content,
                verdict=verdict,
                confidence="High",
                success=not final_validation_failures,
                error=(
                    "Generated early-exit report failed final validation: "
                    + "; ".join(final_validation_failures)
                    if final_validation_failures
                    else None
                ),
                final_validation_failures=final_validation_failures,
                final_report_validated=not final_validation_failures,
                report_policy_version=EXPLOITABILITY_REPORT_POLICY_VERSION,
                tool_call_count=0,
                llm_call_count=pre_triage_result.llm_call_count,
            )

        pkg_status = "found" if pre_triage_result.package_verification.found else "not confirmed"
        extra_found = sum(
            1 for a in pre_triage_result.additional_package_verifications
            if a.verification.found
        )
        extra_msg = f", additional_found={extra_found}" if extra_found else ""
        logger.info(
            "Pre-triage complete (primary_package=%s%s). Proceeding to Phase 2.",
            pkg_status,
            extra_msg,
        )

        configure_allowed_paths(
            roots=[inp.filesystem_path],
            files=[inp.vulnerability_report_path, inp.container_report_path],
            filesystem_root=inp.filesystem_path,
        )

        # Phase 2: Deep Analysis
        logger.info("Phase 2: starting deep analysis")
        messages = [
            SystemMessage(content=PHASE2_SYSTEM_PROMPT),
            HumanMessage(
                content=build_phase2_user_message(
                    inp,
                    cve_id,
                    container_name,
                    pre_triage_result,
                    report_metadata=report_metadata,
                )
            ),
        ]

        total_tool_calls = 0
        total_llm_calls = 0
        content = ""
        validation_errors: list[str] = []
        rejected_attempt_failures: list[str] = []
        rejected_syntheses: list[str] = []
        repair_attempted = False

        for attempt in range(2):
            try:
                content, tool_calls, llm_calls = await run_agentic_loop(
                    llm=llm,
                    messages=messages,
                    tools=tools,
                    max_iterations=config.phase2_max_iterations,
                    context_window_tokens=config.context_window_tokens,
                    wrap_up_fraction=config.wrap_up_fraction,
                    loop_threshold=config.loop_detection_threshold,
                    phase_label="Phase 2" if attempt == 0 else "Phase 2 correction",
                    rate_limiter=rate_limiter,
                )
                total_tool_calls += tool_calls
                total_llm_calls += llm_calls
            except Exception as exc:
                logger.error("Phase 2 failed: %s", exc, exc_info=True)
                return ExploitabilityAnalyzerResult(
                    cve_id=cve_id, container_name=container_name,
                    success=False, error=str(exc),
                    retryable=is_retryable_model_error(exc),
                    tool_call_count=total_tool_calls,
                    llm_call_count=total_llm_calls,
                )

            if not content:
                return ExploitabilityAnalyzerResult(
                    cve_id=cve_id, container_name=container_name,
                    success=False, error="No report content produced by the LLM.",
                    rejected_attempt_failures=rejected_attempt_failures,
                    rejected_syntheses=rejected_syntheses,
                    final_validation_failures=["empty_response"],
                    report_policy_version=EXPLOITABILITY_REPORT_POLICY_VERSION,
                    repair_attempted=repair_attempted,
                    tool_call_count=total_tool_calls, llm_call_count=total_llm_calls,
                )

            try:
                content = _normalize_exploitability_report(content, report_metadata)
            except ExploitabilityReportMetadataError as exc:
                failure = str(exc)
                rejected_attempt_failures.append(failure)
                rejected_syntheses.append(content)
                if repair_attempted:
                    logger.error(
                        "Exploitability report structural repair failed validation: %s",
                        failure,
                    )
                    return ExploitabilityAnalyzerResult(
                        cve_id=cve_id,
                        container_name=container_name,
                        success=False,
                        error=f"Exploitability report validation failed after repair: {failure}",
                        retryable=False,
                        rejected_attempt_failures=rejected_attempt_failures,
                        rejected_syntheses=rejected_syntheses,
                        final_validation_failures=[failure],
                        report_policy_version=EXPLOITABILITY_REPORT_POLICY_VERSION,
                        repair_attempted=True,
                        tool_call_count=total_tool_calls,
                        llm_call_count=total_llm_calls,
                    )

                repair_attempted = True
                logger.warning(
                    "Exploitability report failed structural validation; attempting one "
                    "tool-free repair (failure=%s)",
                    failure,
                )
                try:
                    content, repair_tool_calls, repair_llm_calls = await run_agentic_loop(
                        llm=llm,
                        messages=[
                            SystemMessage(content=_STRUCTURAL_REPAIR_SYSTEM_PROMPT),
                            HumanMessage(
                                content=_build_structural_repair_message(
                                    failure,
                                    rejected_syntheses[-1],
                                )
                            ),
                        ],
                        tools=[],
                        max_iterations=1,
                        context_window_tokens=config.context_window_tokens,
                        wrap_up_fraction=None,
                        loop_threshold=0,
                        phase_label="Exploitability structural repair",
                        rate_limiter=rate_limiter,
                    )
                    total_tool_calls += repair_tool_calls
                    total_llm_calls += repair_llm_calls
                except Exception as exc:
                    logger.error("Exploitability structural repair failed: %s", exc, exc_info=True)
                    return ExploitabilityAnalyzerResult(
                        cve_id=cve_id,
                        container_name=container_name,
                        success=False,
                        error=f"Exploitability structural repair failed: {exc}",
                        retryable=is_retryable_model_error(exc),
                        rejected_attempt_failures=rejected_attempt_failures,
                        rejected_syntheses=rejected_syntheses,
                        report_policy_version=EXPLOITABILITY_REPORT_POLICY_VERSION,
                        repair_attempted=True,
                        tool_call_count=total_tool_calls,
                        llm_call_count=total_llm_calls,
                    )

                try:
                    content = _normalize_exploitability_report(content, report_metadata)
                except ExploitabilityReportMetadataError as exc:
                    failure = str(exc)
                    rejected_attempt_failures.append(failure)
                    rejected_syntheses.append(content)
                    logger.error(
                        "Exploitability report structural repair failed validation: %s",
                        failure,
                    )
                    return ExploitabilityAnalyzerResult(
                        cve_id=cve_id,
                        container_name=container_name,
                        success=False,
                        error=f"Exploitability report validation failed after repair: {failure}",
                        retryable=False,
                        rejected_attempt_failures=rejected_attempt_failures,
                        rejected_syntheses=rejected_syntheses,
                        final_validation_failures=[failure],
                        report_policy_version=EXPLOITABILITY_REPORT_POLICY_VERSION,
                        repair_attempted=True,
                        tool_call_count=total_tool_calls,
                        llm_call_count=total_llm_calls,
                    )

            validation_errors = _validate_against_pre_triage(content, pre_triage_result)
            if not validation_errors:
                break

            failure = "pre_triage_contradiction:" + "; ".join(validation_errors)
            rejected_attempt_failures.append(failure)
            rejected_syntheses.append(content)

            logger.warning(
                "Phase 2 report contradicted pre-triage (%s); attempt %d/2",
                "; ".join(validation_errors),
                attempt + 1,
            )
            if attempt == 0:
                messages = [
                    *messages,
                    AIMessage(content=content),
                    HumanMessage(content=_build_pre_triage_correction(validation_errors, pre_triage_result)),
                ]
                continue

            return ExploitabilityAnalyzerResult(
                cve_id=cve_id,
                container_name=container_name,
                success=False,
                error=(
                    "Exploitability report contradicted deterministic pre-triage: "
                    + "; ".join(validation_errors)
                ),
                retryable=False,
                rejected_attempt_failures=rejected_attempt_failures,
                rejected_syntheses=rejected_syntheses,
                final_validation_failures=[failure],
                report_policy_version=EXPLOITABILITY_REPORT_POLICY_VERSION,
                repair_attempted=repair_attempted,
                tool_call_count=total_tool_calls,
                llm_call_count=total_llm_calls,
            )

        final_validation_failures = list(
            exploitability_report_validation_failures(content, pre_triage_result)
        )
        if final_validation_failures:
            logger.error(
                "Exploitability report failed terminal validation: %s",
                "; ".join(final_validation_failures),
            )
            rejected_syntheses.append(content)
            return ExploitabilityAnalyzerResult(
                cve_id=cve_id,
                container_name=container_name,
                report_content=content,
                success=False,
                error=(
                    "Exploitability report failed terminal validation: "
                    + "; ".join(final_validation_failures)
                ),
                retryable=False,
                rejected_attempt_failures=rejected_attempt_failures,
                rejected_syntheses=rejected_syntheses,
                final_validation_failures=final_validation_failures,
                report_policy_version=EXPLOITABILITY_REPORT_POLICY_VERSION,
                repair_attempted=repair_attempted,
                tool_call_count=total_tool_calls,
                llm_call_count=total_llm_calls,
            )

        verdict = extract_markdown_table_field(content, "Verdict")
        confidence = extract_markdown_table_field(content, "Confidence")

        logger.info(
            "Analysis complete: %s × %s (verdict=%s, confidence=%s, %d tool calls, %d LLM calls)",
            cve_id,
            container_name,
            verdict,
            confidence,
            total_tool_calls,
            total_llm_calls,
        )
        return ExploitabilityAnalyzerResult(
            cve_id=cve_id, container_name=container_name,
            report_content=content, verdict=verdict, confidence=confidence,
            rejected_attempt_failures=rejected_attempt_failures,
            rejected_syntheses=rejected_syntheses,
            final_validation_failures=[],
            final_report_validated=True,
            report_policy_version=EXPLOITABILITY_REPORT_POLICY_VERSION,
            repair_attempted=repair_attempted,
            success=True, tool_call_count=total_tool_calls, llm_call_count=total_llm_calls,
        )

    def _str_to_input(s: str) -> ExploitabilityAnalyzerInput:
        return ExploitabilityAnalyzerInput.model_validate_json(s)

    def _result_to_str(r: ExploitabilityAnalyzerResult) -> str:
        return serialize_result_for_cli(r)

    yield FunctionInfo.from_fn(
        _arun, input_schema=ExploitabilityAnalyzerInput, description=config.description,
        converters=[_str_to_input, _result_to_str],
    )


def _extract_cve_id(vulnerability_report_path: str) -> str:
    cve_id = extract_cve_id_from_path(vulnerability_report_path)
    if re.match(r"CVE-\d{4}-\d+", cve_id):
        return cve_id
    try:
        content = Path(vulnerability_report_path).read_text(encoding="utf-8")[:2000]
    except (OSError, UnicodeDecodeError):
        return cve_id
    match = re.search(r"(CVE-\d{4}-\d+)", content)
    if match:
        return match.group(1)
    return cve_id


def _normalize_exploitability_report(
    content: str,
    metadata: ExploitabilityReportMetadata,
) -> str:
    content = strip_report_preamble(content, "# Container Exploitability Analysis")
    failures = exploitability_report_structure_failures(content)
    if failures:
        raise ExploitabilityReportMetadataError("+".join(failures))
    return normalize_exploitability_report_metadata(content, metadata)


def exploitability_report_validation_failures(
    content: str | None,
    pre_triage: PreTriageResult,
) -> tuple[str, ...]:
    """Validate the promoted artifact, excluding rejected-attempt history."""
    structural_failures = exploitability_report_structure_failures(content)
    if structural_failures:
        return structural_failures

    contradictions = _validate_against_pre_triage(content or "", pre_triage)
    if contradictions:
        return ("pre_triage_contradiction:" + "; ".join(contradictions),)
    return ()


def _build_structural_repair_message(failure: str, rejected_synthesis: str) -> str:
    return (
        "Repair the report below without changing its findings or per-installation facts. "
        f"Validation failure: {failure}. Remove any duplicate report or Summary section. "
        "Ensure the Summary table retains Filesystem Path, Analysis Date, Verdict, and "
        "Confidence rows. Return only one corrected Markdown report.\n\n"
        "<rejected_report>\n"
        f"{rejected_synthesis}\n"
        "</rejected_report>"
    )


def _validate_against_pre_triage(content: str, pre_triage: PreTriageResult) -> list[str]:
    """Catch direct contradictions between deterministic package triage and the LLM report."""
    pkg = pre_triage.package_verification
    if not pkg.found:
        return []

    primary = pkg.active_env_installation or (pkg.installations[0] if pkg.installations else None)
    errors: list[str] = []

    package_in_fs = extract_markdown_table_field(content, "Package in Filesystem")
    if _looks_negative(package_in_fs):
        expected = primary.location if primary else "a deterministic filesystem hit"
        errors.append(
            "report says Package in Filesystem is absent, but pre-triage found "
            f"{pre_triage.vuln_report.package_raw} at {expected}"
        )

    installed_version = extract_markdown_table_field(content, "Installed Version")
    if primary and _looks_absent_version(installed_version):
        errors.append(
            "report says Installed Version is absent, but pre-triage found "
            f"version {primary.version} at {primary.location}"
        )

    any_installation_affected = extract_markdown_table_field(
        content, "Any Installation Version Affected"
    )
    if any_installation_affected is None:
        # Backward compatibility for reports produced with the legacy template.
        any_installation_affected = extract_markdown_table_field(
            content, "Version is Vulnerable"
        )
    if (
        pkg.aggregate_version_status == "affected"
        and _looks_negative(any_installation_affected)
    ):
        errors.append(
            "report says no installation version is affected, but pre-triage "
            f"classified {len(pkg.affected_installation_ids)} installation(s) "
            "as affected"
        )
    elif (
        pkg.aggregate_version_status == "not_affected"
        and _looks_positive(any_installation_affected)
    ):
        errors.append(
            "report says an installation version is affected, but pre-triage "
            "classified every installation as not affected"
        )

    default_installation_affected = extract_markdown_table_field(
        content, "Default Installation Version Affected"
    )
    if (
        pkg.default_environment_affected is False
        and _looks_positive(default_installation_affected)
    ):
        errors.append(
            "report says the default installation is affected, but pre-triage "
            "classified the default installation as not affected"
        )
    elif (
        pkg.default_environment_affected is True
        and _looks_negative(default_installation_affected)
    ):
        errors.append(
            "report says the default installation is not affected, but "
            "pre-triage classified it as affected"
        )

    errors.extend(_validate_installation_assessments(content, pkg))
    return errors


def _validate_installation_assessments(
    content: str,
    pkg: PackageVerification,
) -> list[str]:
    """Ensure every generated installation row preserves deterministic facts."""
    if not pkg.installations:
        return []

    rows, table_error = _extract_installation_assessment_rows(content)
    if table_error:
        return [table_error]

    expected = {inst.installation_id: inst for inst in pkg.installations}
    actual = {row["installation id"]: row for row in rows}
    errors: list[str] = []

    duplicate_ids = _duplicate_installation_ids(rows)
    if duplicate_ids:
        errors.append(
            "Installation Assessments table repeats installation ID(s): "
            + _summarize_ids(duplicate_ids)
        )

    missing_ids = sorted(set(expected) - set(actual))
    if missing_ids:
        errors.append(
            "Installation Assessments table omitted deterministic installation(s): "
            + _summarize_ids(missing_ids)
        )

    extra_ids = sorted(set(actual) - set(expected))
    if extra_ids:
        errors.append(
            "Installation Assessments table added installation(s) not present in "
            "pre-triage: " + _summarize_ids(extra_ids)
        )

    for installation_id in sorted(set(expected) & set(actual)):
        inst = expected[installation_id]
        row = actual[installation_id]
        authoritative = {
            "version": inst.version,
            "source": inst.source,
            "environment": inst.environment or "N/A",
            "activation": inst.assessment.activation_status,
        }
        for field, expected_value in authoritative.items():
            if _normalized_installation_value(row[field]) != (
                _normalized_installation_value(expected_value)
            ):
                errors.append(
                    f"installation {installation_id} changed {field} from "
                    f"{expected_value} to {row[field]}"
                )

        expected_status = inst.assessment.version_status
        actual_status = _normalized_installation_value(row["version status"])
        valid_statuses = {"affected", "not_affected", "unknown"}
        if actual_status not in valid_statuses:
            errors.append(
                f"installation {installation_id} used invalid version status "
                f"{row['version status']}"
            )
        elif expected_status != "unknown" and actual_status != expected_status:
            errors.append(
                f"installation {installation_id} changed version status from "
                f"{expected_status} to {row['version status']}"
            )

    return errors


def _extract_installation_assessment_rows(
    content: str,
) -> tuple[list[dict[str, str]], str | None]:
    heading = re.search(
        r"(?im)^#{2,4}\s+Installation Assessments(?:\s+\(.*\))?\s*$",
        content,
    )
    if not heading:
        return [], "report is missing the required Installation Assessments table"

    section = content[heading.end():]
    next_heading = re.search(r"(?m)^#{1,4}\s+", section)
    if next_heading:
        section = section[:next_heading.start()]

    table_lines = [line for line in section.splitlines() if line.strip().startswith("|")]
    header_index = next(
        (
            index
            for index, line in enumerate(table_lines)
            if "installation id" in _normalized_installation_value(line)
        ),
        None,
    )
    if header_index is None:
        return [], "Installation Assessments section has no table header"

    headers = [
        _normalized_installation_value(cell)
        for cell in _split_markdown_table_row(table_lines[header_index])
    ]
    required = {
        "installation id",
        "version",
        "source",
        "environment",
        "activation",
        "version status",
        "location",
    }
    missing_columns = sorted(required - set(headers))
    if missing_columns:
        return (
            [],
            "Installation Assessments table is missing required column(s): "
            + ", ".join(missing_columns),
        )

    rows: list[dict[str, str]] = []
    for line in table_lines[header_index + 1:]:
        if re.match(r"^\s*\|[-:\s|]+\|\s*$", line):
            continue
        cells = _split_markdown_table_row(line)
        if len(cells) < len(headers):
            continue
        row = dict(zip(headers, cells, strict=False))
        if row.get("installation id"):
            rows.append(row)

    if not rows:
        return [], "Installation Assessments table contains no installation rows"
    return rows, None


def _split_markdown_table_row(line: str) -> list[str]:
    cells = re.split(r"(?<!\\)\|", line.strip())
    if cells and not cells[0]:
        cells = cells[1:]
    if cells and not cells[-1]:
        cells = cells[:-1]
    return [_clean_markdown_cell(cell) for cell in cells]


def _clean_markdown_cell(value: str) -> str:
    return value.replace(r"\|", "|").strip().strip("`*_ ")


def _normalized_installation_value(value: str) -> str:
    return " ".join(_clean_markdown_cell(value).lower().split())


def _duplicate_installation_ids(rows: list[dict[str, str]]) -> list[str]:
    seen: set[str] = set()
    duplicates: set[str] = set()
    for row in rows:
        installation_id = row["installation id"]
        if installation_id in seen:
            duplicates.add(installation_id)
        seen.add(installation_id)
    return sorted(duplicates)


def _summarize_ids(installation_ids: list[str], limit: int = 5) -> str:
    shown = installation_ids[:limit]
    summary = ", ".join(shown)
    remaining = len(installation_ids) - len(shown)
    if remaining:
        summary += f" (+{remaining} more)"
    return summary


def _looks_negative(value: str | None) -> bool:
    if not value:
        return False
    cleaned = value.strip().strip("`*_ ").lower()
    return (
        cleaned == "no"
        or cleaned.startswith("no ")
        or cleaned.startswith("no |")
        or "not installed" in cleaned
        or "not found" in cleaned
        or "package absent" in cleaned
    )


def _looks_positive(value: str | None) -> bool:
    if not value:
        return False
    cleaned = value.strip().strip("`*_ ").lower()
    return cleaned == "yes" or cleaned.startswith("yes ") or cleaned.startswith("yes |")


def _looks_absent_version(value: str | None) -> bool:
    if not value:
        return False
    cleaned = value.strip().strip("`*_ ").lower()
    absent_markers = {"n/a", "na", "none", "not installed", "not found", "absent"}
    return cleaned in absent_markers or "not installed" in cleaned


def _build_pre_triage_correction(errors: list[str], pre_triage: PreTriageResult) -> str:
    pkg = pre_triage.package_verification
    installation_lines = []
    for inst in pkg.installations:
        installation_lines.append(
            f"- {inst.installation_id}: version={inst.version}, "
            f"source={inst.source}, environment={inst.environment or 'N/A'}, "
            f"activation={inst.assessment.activation_status}, "
            f"version_status={inst.assessment.version_status}, "
            f"discovered_location={inst.location}"
        )
    installation_facts = ""
    if installation_lines:
        installation_facts = (
            "\n\nInstallation identity facts and discovery evidence:\n"
            + "\n".join(installation_lines)
            + "\nAny Installation Version Affected: "
            + ("Yes" if pkg.aggregate_version_status == "affected" else "No"
               if pkg.aggregate_version_status == "not_affected" else "Unknown")
            + "\nDefault Installation Version Affected: "
            + ("Yes" if pkg.default_environment_affected is True else "No"
               if pkg.default_environment_affected is False else "Unknown")
        )

    return (
        "Your previous exploitability report contradicted deterministic pre-triage. "
        "Regenerate the full markdown report using the required template, preserving "
        "the pre-triage package presence/version facts and redoing only reachability, "
        "data-flow, trigger, and control analysis.\n\n"
        "Contradictions:\n- "
        + "\n- ".join(errors)
        + installation_facts
        + "\n\nCopy installation identity, version, source, environment, and activation "
        "unchanged. Treat Location as supporting filesystem evidence: use the supplied "
        "path or an equivalent verified path for the same installation. Preserve "
        "affected/not_affected version statuses. An unknown "
        "version status is inconclusive, not authoritative; you may resolve it to affected "
        "or not_affected when the supplied reports or filesystem evidence support that result. "
        "Redo only reachability, data-flow, trigger, and control analysis. Do not "
        "mark the package as absent or not installed unless you can cite a more "
        "precise filesystem finding that reconciles the deterministic result."
    )