# 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.

"""Deterministic output-safety validation for CVE research reports."""

from __future__ import annotations

import logging
import re
from dataclasses import dataclass
from datetime import date

from packaging.version import InvalidVersion
from packaging.version import Version
from univers.versions import EnhancedSemanticVersion

from vuln_analysis.data_models.cve_researcher import RemediationProvenanceStatus

logger = logging.getLogger(__name__)

CVE_REPORT_POLICY_VERSION = 15

AUTHORING_DIRECTIVE_MARKERS = (
    "list each affected branch",
    "list every affected branch separately",
    "case a — backported fixes across branches",
    "case b — no backport to older branches",
    "backported fixes across branches",
    "no backport to older branches",
    "the catch-all row is critical",
    "a single cve may affect multiple packages or modules",
    'check the ghsa "affected versions" section',
    "if only one package is affected",
    "use this section only for distro/vendor package status",
    "if no distro-specific status was researched",
    "the fenced block below is the report structure",
    "replace every bracketed placeholder",
    "never copy these authoring instructions",
    "apply these rules while filling the structure",
    "include every canonical package or module",
    "include only researched distro/vendor package status",
)

_REPORT_HEADING_RE = re.compile(r"(?im)^#\s+CVE Analysis:\s*\S.*$")
_REPORT_START_RE = re.compile(r"\A# CVE Analysis:\s*")
_REPORT_OVERVIEW_RE = re.compile(r"(?im)^##\s+Overview\s*$")
_LEADING_HTML_COMMENT_RE = re.compile(r"\A\s*<!--(?P<body>.*?)-->\s*", re.DOTALL)
_SPDX_COPYRIGHT_LINE_RE = re.compile(
    r"SPDX-FileCopyrightText: Copyright \(c\) \d{4}(?:-\d{4})?, "
    r"NVIDIA CORPORATION(?: & AFFILIATES)?\.(?: All rights reserved\.)?"
)
_ALLOWED_SPDX_COMMENT_LINES = frozenset(
    {
        "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.",
    }
)
_AFFECTED_VERSION_SECTION_RE = re.compile(
    r"(?ims)^###\s+Affected Version Ranges\s*$\n"
    r"(?P<body>.*?)(?=^#{1,3}\s|\Z)",
)
_AFFECTED_PACKAGES_SECTION_RE = re.compile(
    r"(?ims)^###\s+Affected Packages\s*$\n"
    r"(?P<body>.*?)(?=^#{1,3}\s|\Z)",
)
_AFFECTED_VERSION_COLUMNS = (
    "Branch",
    "Vulnerable Range",
    "First Fixed Version",
    "Fix Date",
)
_FIX_DATE_SENTINELS = frozenset({"unknown", "n/a"})
_NO_UPSTREAM_FIX = "no upstream fix"
_UNKNOWN_VERSION = "unknown"
_REQUIRED_OVERVIEW_FIELDS = (
    ("Affected Package", "missing_affected_package"),
    ("Vulnerable Versions", "missing_vulnerable_versions"),
    ("Fixed In", "missing_fixed_in"),
)
_VERSION_TEXT = r"[vV]?\d[0-9A-Za-z]*(?:[-.!+_~:][0-9A-Za-z]+)*"
_MACHINE_VERSION_RE = re.compile(rf"\A{_VERSION_TEXT}\Z")
_VERSION_RE = re.compile(
    rf"(?<![\w.])(?P<version>{_VERSION_TEXT})(?!\w|\.(?=\d))"
)
_INCLUSIVE_VERSION_RANGE_RE = re.compile(
    rf"(?<![\w.])(?P<lower>{_VERSION_TEXT})\s*"
    rf"(?:[–—]|\s+-\s+|\bthrough\b|\bto\b)\s*"
    rf"(?P<upper>{_VERSION_TEXT})(?![\w.])",
    re.IGNORECASE,
)
_VERSION_COMPARATOR_RE = re.compile(
    rf"(?P<operator>>=|<=|==|>|<)\s*(?P<version>{_VERSION_TEXT})",
    re.IGNORECASE,
)
_PATCHLEVEL_VERSION_RE = re.compile(
    r"\A[vV]?(?P<release>\d+(?:\.\d+)+)(?:p(?P<patch>\d+))?\Z",
    re.IGNORECASE,
)
_LETTER_PATCH_VERSION_RE = re.compile(
    r"\A[vV]?(?P<release>\d+(?:\.\d+){2})(?P<patch>[a-z]+)?\Z",
    re.IGNORECASE,
)
_UNRESOLVED_TEMPLATE_RE = re.compile(
    r"(?:"
    r"CVE-YYYY-NNNNN|"
    r"GHSA-xxxx-yyyy-zzzz|"
    r"\[[^\]\n]+\]\(link\)|"
    r"\[(?:"
    r"name|version range|first fixed version|score|severity|vector string|"
    r"CWE-ID|weakness name|date|affected branch|canonical package|ecosystem|"
    r"fixed version|distribution|release|package|version/revision|distro version|"
    r"vendor status|evidence and caveats|concise description|path/to/file|"
    r"function name|detailed explanation|brief explanation|value|"
    r"required conditions|YYYY-MM-DD|source URL or N/A|"
    r"CISA KEV catalog URL or Unknown|FIRST EPSS URL or Unknown|"
    r"value% or Unknown|yes / no / unknown|"
    r"none / low / high|none / required|"
    r"low / medium / high|verified >= lower bound|verified range|"
    r"verified fixed version|verified fix(?:, Unknown,)? or|"
    r"full canonical package name|"
    r"go / pypi / npm / maven / crate / c/deb|"
    r"ubuntu/debian/red hat/suse|affected/fixed/not-affected/unknown|"
    r"network / adjacent / local / physical|low / high|unchanged / changed|"
    r"yes/no|description of the potential impact|"
    r"fixed version and patch reference|available mitigations|"
    r"patch / mitigate / monitor|other relevant links|name/organization|"
    r"if applicable"
    r")[^\]\n]*\]"
    r")",
    re.IGNORECASE,
)
_VERSION_CLAIM_LINE_RE = re.compile(
    r"\b(?:first\s+fixed|fixed\s+in|patched\s+in|shipped\s+in|"
    r"released\s+in|remediated\s+by|upgrade(?:d|s|ing)?\s+to)\b",
    re.IGNORECASE,
)
_VERSION_REMEDIATION_ACTION_RE = re.compile(
    r"\b(?:deploy|install|use|adopt|select|pin|switch|move|migrate|"
    r"update)\b",
    re.IGNORECASE,
)
_VERSION_RECOMMENDATION_RE = re.compile(
    r"\brecommend(?:s|ed|ing|ation(?:s)?)?\b",
    re.IGNORECASE,
)
_VERSION_MARKDOWN_DELIMITER = r"(?:`|\*{1,2}|_{1,2})?"
_RECOMMENDED_VERSION_TEXT = (
    rf"{_VERSION_MARKDOWN_DELIMITER}(?:>=|<=|==|>|<)?\s*"
    rf"{_VERSION_TEXT}{_VERSION_MARKDOWN_DELIMITER}"
)
_VERSION_RECOMMENDATION_TARGET_RE = re.compile(
    rf"\b(?:recommend(?:s|ed|ing)?|recommendation(?:s)?)\b"
    rf"(?:\s+is\b|\s*:)?"
    rf"(?:\s+(?:to\s+)?(?:using|use|upgrading|upgrade|updating|update|deploying|"
    rf"deploy|installing|install|adopting|adopt|pinning|pin|selecting|select|"
    rf"switching|switch|moving|move|migrating|migrate)(?:\s+to)?)?"
    rf"\s+(?:the\s+)?(?:(?:versions?|releases?)\s*:?(?:is\s+)?\s*)?"
    rf"(?P<versions>{_RECOMMENDED_VERSION_TEXT}(?:"
    rf"(?:\s*,\s*(?:(?:or|and)\s+)?|\s+(?:or|and)\s+)"
    rf"(?:(?:version|release)\s+)?{_RECOMMENDED_VERSION_TEXT})*)"
    rf"(?!\w|\.(?=\d))"
    rf"(?=\s*(?:[,.!?;]|$|\b(?:for|or|and|because|as|to|in|with|on)\b))",
    re.IGNORECASE,
)
_RECOMMENDATION_LINE_WRAP_RE = re.compile(
    r"(\brecommend(?:s|ation(?:s)?)?\b"
    r"(?:[ \t]+(?:is|using|use|upgrading|upgrade|updating|update|"
    r"deploying|deploy|installing|install|version|release)){0,2}"
    r"[ \t]*:?)[ \t]*\r?\n"
    r"(?=[ \t]*(?:the[ \t]+)?(?:version|release|[vV]?\d))",
    re.IGNORECASE,
)
_VERSION_REMEDIATION_SUBJECT_RE = re.compile(
    r"\b(?:version|release)\b",
    re.IGNORECASE,
)
_UPSTREAM_REMEDIATION_SUBJECT_RE = re.compile(
    r"\b(?:patch|fix|commit|pull\s+request|PR\s*#?\d+)\b",
    re.IGNORECASE,
)
_UPSTREAM_REMEDIATION_ACTION_RE = re.compile(
    r"\b(?:merged|landed|integrated|shipped|released|include(?:d|s)?|"
    r"contain(?:ed|s)?|available)\b",
    re.IGNORECASE,
)
_NEGATED_REMEDIATION_RE = re.compile(
    r"\b(?:not|never|neither|no|unmerged|open|pending|proposed|unverified|"
    r"unsupported|disputed)\b",
    re.IGNORECASE,
)
_PROSPECTIVE_REMEDIATION_RE = re.compile(
    r"\b(?:monitor|track|watch|wait)\b[^.!?\n]*\b(?:for|until)\b|"
    r"\b(?:if|when|once)\b[^.!?\n]*\b(?:merge|release|ship|land|include)|"
    r"\buntil\b[^.!?\n]*\b(?:merge|release|ship|land|include)|"
    r"\b(?:could|may|might|would)\b[^.!?\n]*\b(?:merge|release|ship|land|include)|"
    r"\b(?:future|subsequent|next)\s+(?:release|version)\b",
    re.IGNORECASE,
)
_REMEDIATION_UNIT_SPLIT_RE = re.compile(
    r"(?<=[.!?])(?:[*_`]+)?\s+|[;\n]+|\s+\bbut\b\s+|\s+(?=\bbefore\b)",
    re.IGNORECASE,
)
_EXPLICIT_REMEDIATION_DIRECTIVE_RE = re.compile(
    r"\b(?:should|must|need\s+to)\s+"
    r"(?:upgrade|update|deploy|install|use|adopt|select|pin|switch|move|migrate)\b|"
    r"^\s*(?:[-*]\s*)?"
    r"(?:upgrade|update|deploy|install|use|adopt|select|pin|switch|move|migrate)\b",
    re.IGNORECASE,
)


def replace_cve_report_placeholders(content: str) -> str:
    """Replace unresolved template tokens with an explicit unknown value."""

    return _UNRESOLVED_TEMPLATE_RE.sub("Unknown", content)


def strip_cve_report_preamble(content: str) -> str:
    """Discard narration before one unambiguous CVE report heading."""

    headings = tuple(_REPORT_HEADING_RE.finditer(content))
    if len(headings) != 1 or headings[0].start() == 0:
        return content
    return content[headings[0].start():]


def _strip_spdx_html_preamble(content: str) -> str:
    """Ignore repository license metadata without accepting model narration."""

    match = _LEADING_HTML_COMMENT_RE.match(content)
    if match is None:
        return content

    lines = tuple(line.strip() for line in match.group("body").splitlines() if line.strip())
    copyright_lines = tuple(line for line in lines if _SPDX_COPYRIGHT_LINE_RE.fullmatch(line))
    if len(copyright_lines) != 1 or lines.count("SPDX-License-Identifier: Apache-2.0") != 1:
        return content
    if any(
        line not in _ALLOWED_SPDX_COMMENT_LINES and line not in copyright_lines
        for line in lines
    ):
        return content
    return content[match.end():]


def _has_nonempty_table_field(content: str, label: str) -> bool:
    match = re.search(
        rf"(?im)^\|\s*(?:\*\*)?{re.escape(label)}(?:\*\*)?\s*"
        r"\|\s*([^|\n]+?)\s*\|",
        content,
    )
    return bool(match and match.group(1).strip())


def _table_field_value(content: str, label: str) -> str | None:
    match = re.search(
        rf"(?im)^\|\s*(?:\*\*)?{re.escape(label)}(?:\*\*)?\s*"
        r"\|\s*([^|\n]+?)\s*\|",
        content,
    )
    if not match:
        return None
    value = match.group(1).strip()
    return value or None


@dataclass(frozen=True)
class _UpstreamVersion:
    """A machine-readable version with independent ordering interpretations."""

    text: str
    normalized_text: str
    pep440: Version | None
    semver: EnhancedSemanticVersion | None


def _parse_version(raw: str) -> _UpstreamVersion | None:
    if not _MACHINE_VERSION_RE.fullmatch(raw):
        return None
    normalized = raw[1:] if raw.startswith(("v", "V")) else raw
    if _valid_iso_date(normalized):
        return None
    try:
        pep440 = Version(normalized)
    except InvalidVersion:
        pep440 = None
    try:
        semver = EnhancedSemanticVersion(normalized)
    except ValueError:
        semver = None
    return _UpstreamVersion(
        text=raw,
        normalized_text=normalized.casefold(),
        pep440=pep440,
        semver=semver,
    )


def _integer_release(raw: str) -> tuple[int, ...]:
    return tuple(int(part) for part in raw.split("."))


def _compare_integer_releases(
    left: tuple[int, ...],
    right: tuple[int, ...],
) -> int:
    width = max(len(left), len(right))
    padded_left = left + (0,) * (width - len(left))
    padded_right = right + (0,) * (width - len(right))
    return (padded_left > padded_right) - (padded_left < padded_right)


def _compare_structured_versions(
    left: _UpstreamVersion,
    right: _UpstreamVersion,
    pattern: re.Pattern[str],
    *,
    numeric_patch: bool,
) -> int | None:
    left_match = pattern.fullmatch(left.text)
    right_match = pattern.fullmatch(right.text)
    if left_match is None or right_match is None:
        return None

    release_order = _compare_integer_releases(
        _integer_release(left_match.group("release")),
        _integer_release(right_match.group("release")),
    )
    if release_order:
        return release_order

    left_patch = left_match.group("patch") or ""
    right_patch = right_match.group("patch") or ""
    if numeric_patch:
        left_value = int(left_patch or "0")
        right_value = int(right_patch or "0")
    else:
        left_value = left_patch.casefold()
        right_value = right_patch.casefold()
    return (left_value > right_value) - (left_value < right_value)


def _compare_versions(
    left: _UpstreamVersion,
    right: _UpstreamVersion,
) -> int | None:
    """Compare versions only when applicable interpretations agree.

    Version syntax is not enough to identify an ecosystem. In particular,
    PEP 440 reads SemVer's ``1.27.0-0`` prerelease as a post release. Treat
    every parser as evidence rather than authority and decline to overrule the
    researcher-authored ecosystem range when interpretations conflict.
    """

    if left.normalized_text == right.normalized_text:
        return 0

    interpretations: dict[str, int] = {}

    patchlevel_order = _compare_structured_versions(
        left,
        right,
        _PATCHLEVEL_VERSION_RE,
        numeric_patch=True,
    )
    if patchlevel_order is not None:
        interpretations["patchlevel"] = patchlevel_order

    letter_patch_order = _compare_structured_versions(
        left,
        right,
        _LETTER_PATCH_VERSION_RE,
        numeric_patch=False,
    )
    if letter_patch_order is not None:
        interpretations["letter-patch"] = letter_patch_order

    if left.semver is not None and right.semver is not None:
        interpretations["semver"] = (
            (left.semver > right.semver) - (left.semver < right.semver)
        )

    if left.pep440 is not None and right.pep440 is not None:
        interpretations["pep440"] = (
            (left.pep440 > right.pep440) - (left.pep440 < right.pep440)
        )

    orders = set(interpretations.values())
    if len(orders) > 1:
        logger.warning(
            "Version ordering interpretations disagree for %r and %r "
            "(%s); deferring to the researcher-authored ecosystem range",
            left.text,
            right.text,
            ", ".join(
                f"{name}={order:+d}"
                for name, order in interpretations.items()
            ),
        )
        return None
    return next(iter(orders), None)


def _versions_equivalent(
    left: _UpstreamVersion,
    right: _UpstreamVersion,
) -> bool:
    return _compare_versions(left, right) == 0


def _is_inclusive_exact_boundary(
    version: _UpstreamVersion,
    comparators: list[tuple[str, _UpstreamVersion]],
) -> bool:
    return any(
        operator in {">=", "<=", "=="}
        and _versions_equivalent(version, boundary)
        for operator, boundary in comparators
    )


def _versions_in(value: str) -> tuple[_UpstreamVersion, ...]:
    versions: list[_UpstreamVersion] = []
    for match in _VERSION_RE.finditer(value):
        version = _parse_version(match.group("version"))
        if version is not None and not any(
            _versions_equivalent(version, existing)
            for existing in versions
        ):
            versions.append(version)
    return tuple(versions)


def _compare(
    version: _UpstreamVersion,
    operator: str,
    boundary: _UpstreamVersion,
) -> bool | None:
    ordering = _compare_versions(version, boundary)
    if ordering is None:
        return None
    if operator == ">=":
        return ordering >= 0
    if operator == "<=":
        return ordering <= 0
    if operator == ">":
        return ordering > 0
    if operator == "<":
        return ordering < 0
    return ordering == 0


def _range_semantics(
    value: str,
    version: _UpstreamVersion,
) -> tuple[bool, bool]:
    """Return whether a range includes version and whether a bound is reversed."""

    includes = False
    invalid = False
    found_range = False

    for match in _INCLUSIVE_VERSION_RANGE_RE.finditer(value):
        lower = _parse_version(match.group("lower"))
        upper = _parse_version(match.group("upper"))
        if lower is None or upper is None:
            continue
        found_range = True
        bound_order = _compare_versions(lower, upper)
        if bound_order is not None and bound_order > 0:
            invalid = True
            continue
        membership = (
            _compare(version, ">=", lower),
            _compare(version, "<=", upper),
        )
        exact_endpoint = (
            _versions_equivalent(version, lower)
            or _versions_equivalent(version, upper)
        )
        includes = includes or all(
            result is True for result in membership
        ) or (
            exact_endpoint
            and not any(result is False for result in membership)
        )

    for expression in re.split(
        r"\s*;\s*|\s+or\s+", value, flags=re.IGNORECASE
    ):
        comparators: list[tuple[str, _UpstreamVersion]] = []
        for match in _VERSION_COMPARATOR_RE.finditer(expression):
            boundary = _parse_version(match.group("version"))
            if boundary is not None:
                comparators.append((match.group("operator"), boundary))
        if comparators:
            found_range = True
            lower_bounds = [
                (boundary, operator == ">=")
                for operator, boundary in comparators
                if operator in {">", ">="}
            ]
            upper_bounds = [
                (boundary, operator == "<=")
                for operator, boundary in comparators
                if operator in {"<", "<="}
            ]
            for lower_version, lower_inclusive in lower_bounds:
                for upper_version, upper_inclusive in upper_bounds:
                    bound_order = _compare_versions(
                        lower_version,
                        upper_version,
                    )
                    if bound_order is not None and (
                        bound_order > 0
                        or (
                            bound_order == 0
                            and not (
                                lower_inclusive and upper_inclusive
                            )
                        )
                    ):
                        invalid = True
            equality_versions = [
                boundary
                for operator, boundary in comparators
                if operator == "=="
            ]
            if equality_versions and any(
                any(
                    _compare(equality_version, operator, boundary)
                    is False
                    for operator, boundary in comparators
                )
                for equality_version in equality_versions
            ):
                invalid = True
            membership = tuple(
                _compare(version, operator, boundary)
                for operator, boundary in comparators
            )
            includes = includes or all(
                result is True for result in membership
            ) or (
                _is_inclusive_exact_boundary(version, comparators)
                and not any(result is False for result in membership)
            )

    if not found_range:
        includes = any(
            _versions_equivalent(version, candidate)
            for candidate in _versions_in(value)
        )
    return includes, invalid


def _markdown_table(
    section: str,
) -> tuple[tuple[str, ...], tuple[dict[str, str], ...], bool]:
    header: list[str] | None = None
    rows: list[dict[str, str]] = []
    malformed_row = False
    for line in section.splitlines():
        stripped = line.strip()
        if not stripped.startswith("|") or not stripped.endswith("|"):
            if header is not None and rows and stripped:
                break
            continue
        cells = [
            cell.strip().replace("**", "")
            for cell in stripped[1:-1].split("|")
        ]
        if not cells or all(
            re.fullmatch(r"[-: ]+", cell) for cell in cells
        ):
            continue
        if header is None:
            header = cells
            continue
        if len(cells) == len(header):
            rows.append(dict(zip(header, cells, strict=True)))
        else:
            malformed_row = True
    return tuple(header or ()), tuple(rows), malformed_row


def _valid_iso_date(value: str) -> bool:
    if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
        return False
    try:
        date.fromisoformat(value)
    except ValueError:
        return False
    return True


def repair_cve_report_version_rows(content: str) -> str:
    """Conservatively repair invalid version/date cells in the branch table.

    This fallback never invents a fixed version or date. Invalid values are
    downgraded to ``Unknown`` while independently verified ISO dates are
    preserved. It is intended for use only after the model's structural repair
    attempt has failed.
    """

    section_match = _AFFECTED_VERSION_SECTION_RE.search(content)
    if section_match is None:
        return content

    lines = section_match.group("body").splitlines(keepends=True)
    found_header = False
    found_separator = False
    changed = False

    for index, line in enumerate(lines):
        line_body = line.rstrip("\r\n")
        line_ending = line[len(line_body):]
        stripped = line_body.strip()
        if not stripped.startswith("|") or not stripped.endswith("|"):
            if found_separator and stripped:
                break
            continue

        cells = [cell.strip() for cell in stripped[1:-1].split("|")]
        normalized_cells = tuple(
            cell.replace("**", "").strip() for cell in cells
        )
        if not found_header:
            if normalized_cells == _AFFECTED_VERSION_COLUMNS:
                found_header = True
            continue
        if all(re.fullmatch(r"[-: ]+", cell) for cell in cells):
            found_separator = True
            continue
        if not found_separator or len(cells) != len(_AFFECTED_VERSION_COLUMNS):
            continue

        first_fixed = cells[2]
        fix_date = cells[3]
        normalized_first_fixed = first_fixed.casefold()
        normalized_fix_date = fix_date.casefold()

        if normalized_first_fixed == _NO_UPSTREAM_FIX:
            repaired_first_fixed = "No upstream fix"
            repaired_fix_date = (
                fix_date
                if normalized_fix_date in _FIX_DATE_SENTINELS
                else "N/A"
            )
        else:
            parsed_first_fixed = (
                None
                if normalized_first_fixed == _UNKNOWN_VERSION
                else _parse_version(first_fixed)
            )
            repaired_first_fixed = (
                first_fixed if parsed_first_fixed is not None else "Unknown"
            )
            repaired_fix_date = fix_date
            if (
                normalized_fix_date == "n/a"
                or (
                    normalized_fix_date != "unknown"
                    and not _valid_iso_date(fix_date)
                )
            ):
                repaired_fix_date = "Unknown"

        if (
            repaired_first_fixed == first_fixed
            and repaired_fix_date == fix_date
        ):
            continue

        cells[2] = repaired_first_fixed
        cells[3] = repaired_fix_date
        indentation = line_body[: len(line_body) - len(line_body.lstrip())]
        lines[index] = (
            f"{indentation}| {' | '.join(cells)} |{line_ending}"
        )
        changed = True

    if not changed:
        return content

    repaired_body = "".join(lines)
    return (
        content[:section_match.start("body")]
        + repaired_body
        + content[section_match.end("body"):]
    )


def _affected_version_table_failures(
    content: str,
) -> tuple[list[dict[str, str]], tuple[str, ...]]:
    section_match = _AFFECTED_VERSION_SECTION_RE.search(content)
    if not section_match:
        return [], ("missing_affected_version_ranges",)

    header, rows, malformed_row = _markdown_table(
        section_match.group("body")
    )
    if not header:
        return [], ("missing_affected_version_table",)
    if header != _AFFECTED_VERSION_COLUMNS:
        return [], ("invalid_affected_version_columns",)

    failures: list[str] = []
    if malformed_row:
        failures.append("invalid_affected_version_row")
    if not rows:
        failures.append("missing_affected_version_rows")

    for row in rows:
        if any(
            not row[column].strip()
            for column in _AFFECTED_VERSION_COLUMNS
        ):
            failures.append("incomplete_affected_version_row")
            continue

        first_fixed = row["First Fixed Version"].strip()
        fix_date = row["Fix Date"].strip()
        normalized_first_fixed = first_fixed.casefold()
        first_fixed_is_missing = (
            normalized_first_fixed == _NO_UPSTREAM_FIX
        )
        first_fixed_is_unknown = normalized_first_fixed == _UNKNOWN_VERSION
        parsed_first_fixed = None
        if not first_fixed_is_missing and not first_fixed_is_unknown:
            parsed_first_fixed = _parse_version(first_fixed)
            if parsed_first_fixed is None:
                failures.append("invalid_first_fixed_version")

        normalized_fix_date = fix_date.casefold()
        if (
            normalized_fix_date not in _FIX_DATE_SENTINELS
            and not _valid_iso_date(fix_date)
        ):
            failures.append("invalid_fix_date")
        elif (
            normalized_first_fixed == _NO_UPSTREAM_FIX
            and normalized_fix_date not in _FIX_DATE_SENTINELS
        ):
            failures.append("invalid_fix_date")
        elif first_fixed_is_unknown and normalized_fix_date == "n/a":
            failures.append("invalid_fix_date")
        elif (
            parsed_first_fixed is not None
            and normalized_fix_date == "n/a"
        ):
            failures.append("invalid_fix_date")

    return list(rows), tuple(dict.fromkeys(failures))


def cve_report_semantic_advisories(content: str) -> tuple[str, ...]:
    """Return non-fatal concerns about affected-version range semantics.

    Overview and Affected Packages values are human-readable summaries. Their
    versions can belong to different packages or independent release lines, so
    comparing them in one global version space creates false contradictions.
    These findings require researcher adjudication and must never make an
    otherwise structurally usable report fail the workflow.
    """

    affected_version_rows, table_failures = (
        _affected_version_table_failures(content)
    )
    if table_failures:
        return ()

    advisories: list[str] = []

    for row in affected_version_rows:
        vulnerable_range = row.get("Vulnerable Range")
        first_fixed = row.get("First Fixed Version")
        if not vulnerable_range or not first_fixed:
            continue
        row_fixed_versions = _versions_in(first_fixed)
        range_probe = (
            row_fixed_versions[0]
            if row_fixed_versions
            else _parse_version("0")
        )
        assert range_probe is not None
        _, invalid = _range_semantics(vulnerable_range, range_probe)
        if invalid and "invalid_affected_version_range" not in advisories:
            advisories.append("invalid_affected_version_range")
        for fixed_version in row_fixed_versions:
            includes, _ = _range_semantics(
                vulnerable_range, fixed_version
            )
            if includes:
                advisories.append(
                    "affected_range_includes_first_fixed_version"
                )
                break
    return tuple(dict.fromkeys(advisories))


def _fixed_version_claims(content: str) -> tuple[_UpstreamVersion, ...]:
    claims: list[_UpstreamVersion] = []

    overview_fixed = _table_field_value(content, "Fixed In")
    if overview_fixed:
        claims.extend(_versions_in(overview_fixed))

    affected_rows, _ = _affected_version_table_failures(content)
    for row in affected_rows:
        claims.extend(_versions_in(row.get("First Fixed Version", "")))

    packages_match = _AFFECTED_PACKAGES_SECTION_RE.search(content)
    if packages_match:
        header, rows, _ = _markdown_table(packages_match.group("body"))
        if "Fixed In" in header:
            for row in rows:
                claims.extend(_versions_in(row.get("Fixed In", "")))

    for sentence in _REMEDIATION_UNIT_SPLIT_RE.split(_remediation_prose(content)):
        versions = _versions_in(sentence)
        if not versions or _NEGATED_REMEDIATION_RE.search(sentence):
            continue
        has_recommendation = _VERSION_RECOMMENDATION_RE.search(sentence)
        action = _VERSION_REMEDIATION_ACTION_RE.search(sentence)
        for match in _VERSION_RECOMMENDATION_TARGET_RE.finditer(sentence):
            claims.extend(_versions_in(match.group("versions")))
        if (
            _VERSION_CLAIM_LINE_RE.search(sentence)
            or (
                action
                and (
                    not has_recommendation
                    or action.start() < has_recommendation.start()
                )
            )
            or (
                _VERSION_REMEDIATION_SUBJECT_RE.search(sentence)
                and _UPSTREAM_REMEDIATION_ACTION_RE.search(sentence)
            )
            or (
                _UPSTREAM_REMEDIATION_SUBJECT_RE.search(sentence)
                and _UPSTREAM_REMEDIATION_ACTION_RE.search(sentence)
            )
        ):
            claims.extend(versions)

    unique_claims: list[_UpstreamVersion] = []
    for claim in claims:
        if not any(
            _versions_equivalent(claim, existing)
            for existing in unique_claims
        ):
            unique_claims.append(claim)
    return tuple(unique_claims)


def _remediation_prose(content: str) -> str:
    """Include named remediation sections and explicit advice elsewhere."""
    content = _RECOMMENDATION_LINE_WRAP_RE.sub(r"\1 ", content)
    sections: list[str] = []
    heading = re.compile(r"(?im)^##\s+(?:Remediation|Patch(?:es)?)\s*$")
    for match in heading.finditer(content):
        next_section = re.search(r"(?m)^##\s+", content[match.end():])
        end = match.end() + next_section.start() if next_section else len(content)
        sections.append(content[match.end():end])
    for sentence in _REMEDIATION_UNIT_SPLIT_RE.split(content):
        if (
            _VERSION_CLAIM_LINE_RE.search(sentence)
            or _EXPLICIT_REMEDIATION_DIRECTIVE_RE.search(sentence)
            or _VERSION_RECOMMENDATION_RE.search(sentence)
        ):
            sections.append(sentence)
    return "\n".join(sections)


def _has_positive_upstream_remediation_claim(content: str) -> bool:
    for sentence in _REMEDIATION_UNIT_SPLIT_RE.split(content):
        action = _UPSTREAM_REMEDIATION_ACTION_RE.search(sentence)
        prospective = _PROSPECTIVE_REMEDIATION_RE.search(sentence)
        if (
            _UPSTREAM_REMEDIATION_SUBJECT_RE.search(sentence)
            and action
            and not _NEGATED_REMEDIATION_RE.search(sentence)
            and not (
                prospective
                and prospective.start() <= action.start()
            )
        ):
            return True
    return False


def remediation_claim_validation_failures(
    content: str,
    *,
    provenance_status: RemediationProvenanceStatus,
    verified_fixed_versions: tuple[str, ...] | list[str] = (),
) -> tuple[str, ...]:
    """Reject remediation claims that lack deterministic provider evidence."""

    if provenance_status in {
        "not_requested",
        "disabled",
        "not_applicable",
    }:
        return ()

    verified = tuple(
        parsed
        for raw in verified_fixed_versions
        if (parsed := _parse_version(raw)) is not None
    )
    claims = _fixed_version_claims(content)
    failures: list[str] = []
    if any(
        not any(
            _versions_equivalent(claim, verified_version)
            for verified_version in verified
        )
        for claim in claims
    ):
        failures.append("unverified_fixed_version_claim")
    if (
        provenance_status != "verified"
        and _has_positive_upstream_remediation_claim(content)
    ):
        failures.append("unverified_upstream_remediation_claim")
    return tuple(failures)


def invalid_first_fixed_version_values(
    content: str | None,
    *,
    max_length: int = 80,
) -> tuple[str, ...]:
    """Return bounded, log-safe values rejected as first fixed versions."""

    if not content or max_length < 4:
        return ()
    section_match = _AFFECTED_VERSION_SECTION_RE.search(content)
    if section_match is None:
        return ()
    header, rows, _ = _markdown_table(section_match.group("body"))
    if header != _AFFECTED_VERSION_COLUMNS:
        return ()

    invalid_values: list[str] = []
    for row in rows:
        first_fixed = row["First Fixed Version"].strip()
        normalized = first_fixed.casefold()
        if normalized in {_NO_UPSTREAM_FIX, _UNKNOWN_VERSION}:
            continue
        if _parse_version(first_fixed) is not None:
            continue

        compact = " ".join(first_fixed.split())
        safe = re.sub(r"[^0-9A-Za-z._+~:/() -]", "?", compact)
        if len(safe) > max_length:
            safe = f"{safe[:max_length - 3]}..."
        if safe and safe not in invalid_values:
            invalid_values.append(safe)
    return tuple(invalid_values)


def cve_report_validation_failures(
    content: str | None,
    *,
    expected_identifier: str | None = None,
) -> tuple[str, ...]:
    """Return stable failure codes for incomplete or unsafe report content."""

    if not content or not content.strip():
        return ("empty_response",)

    content = _strip_spdx_html_preamble(content)
    failures: list[str] = []
    if not _REPORT_HEADING_RE.search(content):
        failures.append("missing_report_heading")
    elif not _REPORT_START_RE.match(content):
        failures.append("unexpected_report_preamble")
    elif expected_identifier is not None:
        first_line = content.partition("\n")[0].removesuffix("\r")
        expected_heading = f"# CVE Analysis: {expected_identifier}"
        if first_line != expected_heading:
            failures.append("report_heading_identifier_mismatch")
    if not _REPORT_OVERVIEW_RE.search(content):
        failures.append("missing_overview")
    for label, failure in _REQUIRED_OVERVIEW_FIELDS:
        if not _has_nonempty_table_field(content, label):
            failures.append(failure)
    _, affected_version_failures = _affected_version_table_failures(content)
    failures.extend(affected_version_failures)

    normalized = " ".join(content.lower().split())
    if any(marker in normalized for marker in AUTHORING_DIRECTIVE_MARKERS):
        failures.append("authoring_directive_leak")
    if _UNRESOLVED_TEMPLATE_RE.search(content):
        failures.append("unresolved_template_placeholder")
    return tuple(failures)


def is_clean_cve_report(
    content: str | None,
    *,
    expected_identifier: str | None = None,
) -> bool:
    """Return whether content is structurally complete and safe to persist."""

    return not cve_report_validation_failures(
        content,
        expected_identifier=expected_identifier,
    )


def append_cve_report_warnings(
    content: str,
    warnings: tuple[str, ...] | list[str],
) -> str:
    """Append sanitized, downstream-visible research warnings to a valid report."""

    if cve_report_validation_failures(content):
        raise ValueError("research warnings can only be added to a valid CVE report")

    safe_warnings: list[str] = []
    for value in warnings:
        warning = " ".join(value.split()).strip()
        if not warning:
            continue
        normalized = warning.casefold()
        if any(marker in normalized for marker in AUTHORING_DIRECTIVE_MARKERS):
            warning = (
                "A report-policy warning was recorded; see the stage diagnostics "
                "for details."
            )
        safe_warnings.append(warning)

    if not safe_warnings:
        return content

    lines = [content.rstrip(), "", "## Research Warnings", ""]
    lines.extend(f"- {warning}" for warning in safe_warnings)
    return "\n".join(lines).rstrip() + "\n"


__all__ = [
    "AUTHORING_DIRECTIVE_MARKERS",
    "CVE_REPORT_POLICY_VERSION",
    "append_cve_report_warnings",
    "cve_report_semantic_advisories",
    "cve_report_validation_failures",
    "invalid_first_fixed_version_values",
    "is_clean_cve_report",
    "repair_cve_report_version_rows",
    "replace_cve_report_placeholders",
    "remediation_claim_validation_failures",
    "strip_cve_report_preamble",
]