cleveragents-core/tools/_implementer_compliance.py

"""Compliance gap detector for the implementer dispatcher.

Walks the pre-cloned worktree and checks PR Compliance Checklist
items mechanically:

- ``worktree_clean`` — no uncommitted changes (``git status
  --porcelain`` empty). When True, the agent doesn't need to
  re-apply the code fix; whatever's at HEAD is already what gets
  pushed.
- ``changelog_unreleased_nonempty`` — the ``[Unreleased]`` section
  in CHANGELOG.md has at least one entry.
- ``contributors_has_author`` — the git author email appears in
  CONTRIBUTORS.md.
- ``commit_has_issues_closed`` — the last commit's message body
  contains an ``ISSUES CLOSED: #N`` footer.

The detector emits a structured dict that gets embedded into the
worker prompt under a ``## Compliance gap report (deterministic)``
stanza. The agent's instruction is: when ALL bools are true, the
PR is done; emit ``{"outcome": "resolved", ...}``. When some are
false, fill ONLY the missing items — do NOT re-touch the code fix.

Motivation (plan §"#2", 2026-05-13): in production on 2026-05-12,
the actual fix on PR #30 was just two missing compliance items
(CHANGELOG + CONTRIBUTORS) — Kimi at Tier 2 spent 76 minutes
DISCOVERING this and applying it. Detection takes <100 ms. Letting
the worker fill in known gaps (instead of having it explore the
worktree to FIND them) collapses the work to a one- or two-turn
fill-in-the-blanks.

The module has no LLM dependency and is pure read-only filesystem
work — easy to unit-test against fixture trees.
"""
from __future__ import annotations

import logging
import re
import subprocess
from pathlib import Path
from typing import Any

_logger = logging.getLogger("implementer_compliance")


def _git(args: list[str], cwd: Path, timeout: int = 10) -> tuple[int, str]:
    """Run a ``git`` subprocess inside ``cwd`` and return
    ``(returncode, stdout)``. Stderr is discarded — for these
    read-only commands we only care about success + stdout."""
    try:
        result = subprocess.run(
            ["git", *args],
            cwd=str(cwd),
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
        )
    except (subprocess.TimeoutExpired, FileNotFoundError):
        return -1, ""
    return result.returncode, (result.stdout or "")


def check_worktree_clean(worktree: Path) -> bool:
    """``True`` iff ``git status --porcelain`` is empty.

    Empty = no uncommitted changes = whatever's at HEAD is what
    would get pushed if the agent ran ``git push``. The agent should
    NOT re-apply code changes when this is True; the diff is already
    present in the committed history.

    Git-unavailable / non-repo handling: a non-zero return code or a
    missing ``git`` binary returns ``True`` (treat as "no observable
    gap"), not ``False``. Returning ``False`` would conflate "we
    couldn't check" with "there IS a gap" — the worker would then
    chase a phantom dirty-worktree and re-apply the code fix on top
    of HEAD. Callers that need to distinguish the masked-True from a
    real-True must use :func:`check_compliance_gaps_with_masking`.
    """
    value, _was_masked = _check_worktree_clean_with_masking(worktree)
    return value


# Regex to find the ``[Unreleased]`` section and the next ``## `` or
# end-of-file. Multi-line, ungreedy. The section is "non-empty" if
# anything but whitespace + horizontal rules appears between the
# header and the next section.
#
# Exact alternation across the two accepted forms:
#
# - Keep-a-Changelog form: ``## [Unreleased]``
# - Bare-header form:      ``## Unreleased``
#
# The previous regex used ``\[?Unreleased\]?`` which silently
# accepted mismatched brackets (``## [Unreleased`` or
# ``## Unreleased]``) — a malformed header that should be flagged
# as missing the section, not silently matched. The non-capturing
# group makes the alternation explicit so a future contributor
# adding a third form (e.g. ``## (Unreleased)``) sees the pattern
# and knows to extend it deliberately.
_UNRELEASED_SECTION_RE = re.compile(
    r"^##\s*(?:\[Unreleased\]|Unreleased)[^\n]*\n(?P<body>.*?)(?=^##\s|\Z)",
    re.MULTILINE | re.DOTALL | re.IGNORECASE,
)


def check_changelog_unreleased_nonempty(worktree: Path) -> bool:
    """``True`` iff ``CHANGELOG.md`` has a ``[Unreleased]`` section
    with at least one non-trivial line.

    "Non-trivial" excludes lines that are just whitespace, comments,
    or ``---`` horizontal-rule separators. A section that only
    contains a placeholder ``###`` subheader with no content under
    it counts as empty.

    Returns ``False`` if the file is missing or unreadable — the
    agent's job in that case is to either create the section or
    flag the structural problem.
    """
    changelog = worktree / "CHANGELOG.md"
    try:
        text = changelog.read_text(encoding="utf-8")
    except (OSError, UnicodeDecodeError):
        return False
    m = _UNRELEASED_SECTION_RE.search(text)
    if m is None:
        return False
    body = m.group("body")
    for line in body.splitlines():
        stripped = line.strip()
        if not stripped:
            continue
        if stripped.startswith("#"):
            # subheader like "### Added" / "### Changed" — only
            # counts as content if accompanied by an entry under it
            continue
        if set(stripped) <= {"-", "="}:
            # horizontal rule
            continue
        # Found a real content line (bullet, paragraph, etc.)
        return True
    return False


def check_contributors_has_author(
    worktree: Path, git_user_email: str
) -> bool:
    """``True`` iff ``CONTRIBUTORS.md`` contains the author's email
    (case-insensitive)."""
    if not git_user_email:
        return False
    contributors = worktree / "CONTRIBUTORS.md"
    try:
        text = contributors.read_text(encoding="utf-8")
    except (OSError, UnicodeDecodeError):
        return False
    return git_user_email.lower() in text.lower()


_ISSUES_CLOSED_RE = re.compile(
    r"^\s*ISSUES\s*CLOSED\s*:\s*#?\d+", re.MULTILINE | re.IGNORECASE
)


def check_commit_has_issues_closed(worktree: Path) -> bool:
    """``True`` iff the HEAD commit's message body contains
    ``ISSUES CLOSED: #N``.

    Reads the full message via ``git log -1 --pretty=%B``. Case-
    insensitive. The actual project convention uses uppercase
    ``ISSUES CLOSED:`` but we accept variants like
    ``Issues closed: 42`` for robustness.

    Git-unavailable / non-repo handling: same rule as
    :func:`check_worktree_clean` — a non-zero git return code returns
    ``True``. Callers that need to distinguish the masked-True from
    a real-True must use :func:`check_compliance_gaps_with_masking`.
    """
    value, _was_masked = _check_commit_has_issues_closed_with_masking(worktree)
    return value


def check_compliance_gaps(
    worktree: Path, git_user_email: str = ""
) -> dict[str, bool]:
    """Run all four compliance checks and return a structured dict.

    Each value is a strict ``bool``. The agent's prompt instruction
    is to inspect this dict and fill in the missing items only.

    See :func:`check_compliance_gaps_with_masking` for the
    masking-aware variant the dispatcher uses to detect when a
    ``True`` value was synthesised because git failed (vs. a real
    passing check). Callers that only need the simple-bool dict
    (most unit tests) can keep using this thin wrapper.
    """
    gaps, _ = check_compliance_gaps_with_masking(worktree, git_user_email)
    return gaps


def check_compliance_gaps_with_masking(
    worktree: Path, git_user_email: str = ""
) -> tuple[dict[str, bool], set[str]]:
    """Run all four checks and report which ones were MASKED.

    Two checks ( :func:`check_worktree_clean` and
    :func:`check_commit_has_issues_closed`) return ``True`` when git
    itself fails — the deliberate "don't conflate masking with a real
    gap" policy documented on those functions. The cost of that
    policy: if BOTH masked checks coincide with a real-content
    CHANGELOG + CONTRIBUTORS the worker sees ``all_gaps_closed=True``
    and would receive a confident "PR resolved, do NOT re-apply"
    directive from the renderer.

    This function returns both the gap dict AND the set of check
    names that were masked this run. Callers that surface the verdict
    to the worker (the dispatcher's ``_compute_deterministic_sections``
    and the pointer-stanza renderer) consult ``masked`` and hedge
    their language when it is non-empty. Pure unit-test callers can
    ignore the second element.
    """
    masked: set[str] = set()
    worktree_clean, was_masked = _check_worktree_clean_with_masking(worktree)
    if was_masked:
        masked.add("worktree_clean")
    issues_closed, was_masked = _check_commit_has_issues_closed_with_masking(
        worktree
    )
    if was_masked:
        masked.add("commit_has_issues_closed")
    gaps = {
        "worktree_clean": worktree_clean,
        "changelog_unreleased_nonempty": check_changelog_unreleased_nonempty(
            worktree
        ),
        "contributors_has_author": check_contributors_has_author(
            worktree, git_user_email
        ),
        "commit_has_issues_closed": issues_closed,
    }
    return gaps, masked


def _check_worktree_clean_with_masking(worktree: Path) -> tuple[bool, bool]:
    """Internal helper: returns ``(value, was_masked)``.

    ``was_masked`` is True when the underlying ``git status`` call
    failed and the function returned ``True`` because of the
    conflation-avoidance policy. Used by
    :func:`check_compliance_gaps_with_masking` to surface the
    masking state to the renderer. The public
    :func:`check_worktree_clean` wraps this and discards the second
    element for backward compat.
    """
    rc, out = _git(["status", "--porcelain"], worktree)
    if rc != 0:
        _logger.info(
            "check_worktree_clean: git status returned rc=%d in %s; "
            "masking failure (returning True to avoid conflation "
            "with a real gap)",
            rc, worktree,
        )
        return True, True
    return out.strip() == "", False


def _check_commit_has_issues_closed_with_masking(
    worktree: Path,
) -> tuple[bool, bool]:
    """Internal helper mirror of
    :func:`_check_worktree_clean_with_masking` for the commit-footer
    check. Same masking contract."""
    rc, body = _git(["log", "-1", "--pretty=%B", "HEAD"], worktree)
    if rc != 0:
        _logger.info(
            "check_commit_has_issues_closed: git log returned rc=%d "
            "in %s; masking failure (returning True to avoid "
            "conflation with a real gap)",
            rc, worktree,
        )
        return True, True
    return _ISSUES_CLOSED_RE.search(body) is not None, False


def check_pr_metadata_gaps(
    pr_details: dict[str, Any] | None,
) -> dict[str, bool]:
    """G6 harvest (2026-05-15) — return Forgejo-metadata compliance
    gaps that complement the worktree-side checks in
    :func:`check_compliance_gaps_with_masking`.

    Each value is a strict ``bool`` (``True`` means the check passes,
    ``False`` means a gap). The two checks ported from
    ``agents/final-working``'s 8-item PR Compliance Checklist
    (items 7 and 8):

    * ``pr_label_set_complete`` — the PR carries at least one label
      in every required family: ``State/``, ``Priority/``, ``Type/``,
      and ``MoSCoW/``. Each family is independent; missing any one
      flips the check to ``False``.
    * ``pr_milestone_assigned`` — the PR has a milestone associated.

    Pure function: ``pr_details`` is the Forgejo pull-request JSON
    payload the dispatcher already pre-fetches; this helper does not
    issue any HTTP calls. ``None`` (e.g. a prefetch failure) returns
    a False/False dict — the conservative "we don't know, treat as a
    gap" position; the caller can mask this in the same way
    :func:`check_compliance_gaps_with_masking` masks git failures.

    Intended consumers: the dispatcher's compliance-gap fixer
    (``_implementer_compliance_apply``) when extended to apply
    label / milestone fixes, and any future grooming driver that
    wants the same gap signal. The dict shape composes cleanly with
    the worktree-side dict via ``{**worktree_gaps, **metadata_gaps}``.
    """
    if not isinstance(pr_details, dict):
        return {
            "pr_label_set_complete": False,
            "pr_milestone_assigned": False,
        }
    label_names: set[str] = set()
    labels = pr_details.get("labels")
    if isinstance(labels, list):
        for lbl in labels:
            if isinstance(lbl, dict):
                name = lbl.get("name")
                if isinstance(name, str):
                    label_names.add(name)
    label_complete = all(
        any(name.startswith(prefix) for name in label_names)
        for prefix in ("State/", "Priority/", "Type/", "MoSCoW/")
    )
    milestone = pr_details.get("milestone")
    milestone_assigned = isinstance(milestone, dict) and bool(
        milestone.get("id") or milestone.get("title")
    )
    return {
        "pr_label_set_complete": label_complete,
        "pr_milestone_assigned": milestone_assigned,
    }


def gaps_open_count(
    gaps: dict[str, bool], masked_checks: set[str] | None = None,
) -> int:
    """Number of compliance checks that returned ``False`` — i.e.
    real observable gaps the worker needs to fill.

    When ``masked_checks`` is provided, MASKED checks are excluded
    from the count entirely (neither "open" nor "closed" — their
    value is unverified, not a real signal). Without the
    ``masked_checks`` argument the function returns the raw count
    of False values, which over-counts the truly-open set when
    masking happened.

    Convenience for telemetry / status comments: lets analysts see
    "how much real compliance debt was discovered this cycle"
    without re-parsing the dict. ``0`` when all checks pass; up to
    ``len(gaps)`` when none do.
    """
    masked = masked_checks or set()
    return sum(
        1 for k, v in gaps.items()
        if not v and k not in masked
    )


def all_gaps_closed(gaps: dict[str, bool]) -> bool:
    """``True`` iff every check passed. Convenience for callers that
    want a single "is the PR done?" signal."""
    return all(bool(v) for v in gaps.values()) and bool(gaps)


def render_prompt_stanza(
    gaps: dict[str, bool],
    pr_number: int | None = None,
    *,
    masked_checks: list[str] | set[str] | None = None,
) -> str:
    """Render the compliance-gap report as a markdown stanza for
    the worker prompt.

    The instruction text tells the agent how to react:

    - All True (no masking) → emit the success-JSON and exit (no
      code changes needed).
    - All True (with masking) → hedge: re-verify the masked check(s)
      before exiting; the dispatcher returned True for them because
      git itself failed, not because they really passed.
    - Some False → fill the missing items only.

    Keeps the rendering co-located with the data shape so a future
    schema change lands in one place. Note: this module-level
    renderer is paralleled by
    :func:`dispatch_implementer._render_compliance_pointer_stanza`
    which produces the short, prompt-survival-optimised pointer that
    production actually sends. Both must stay in sync on the
    all-passed-with-masking hedge — otherwise a worker reading the
    sentinel JSON sees one verdict and a worker reading the prompt
    sees another.
    """
    if not gaps:
        return ""
    masked = set(masked_checks or [])

    lines = ["## Compliance gap report (deterministic)\n"]
    lines.append(
        "The dispatcher checked the PR Compliance Checklist items "
        "mechanically against the pre-cloned worktree. Treat these "
        "as ground truth — they're cheaper to trust than re-deriving."
    )
    lines.append("")
    for key, value in sorted(gaps.items()):
        marker = "✅" if value else "❌"
        masked_note = " _(masked — git failed)_" if key in masked else ""
        lines.append(f"- {marker} `{key}`: **{value}**{masked_note}")
    lines.append("")

    if all_gaps_closed(gaps) and not masked:
        lines.append(
            "**All checks passed.** The PR is already complete — "
            "the code fix is in HEAD and every Compliance Checklist "
            "item is satisfied. Do NOT re-apply or re-edit the "
            "code. Verify the local quality gates (if you have not "
            "already) and emit:"
        )
        lines.append(
            "    {\"outcome\": \"resolved\", \"files_touched\": []}"
        )
    elif all_gaps_closed(gaps) and masked:
        joined_masked = ", ".join(f"`{m}`" for m in sorted(masked))
        lines.append(
            f"**All OBSERVABLE checks passed, but {len(masked)} "
            f"check(s) were MASKED** because `git` itself failed: "
            f"{joined_masked}. Do NOT exit with `resolved` — "
            "re-verify the masked check(s) in-session (run `git "
            "status` / `git log -1`) before deciding."
        )
    else:
        missing = [k for k, v in gaps.items() if not v]
        lines.append(
            f"**Gaps to fill ({len(missing)}):** "
            + ", ".join(f"`{m}`" for m in missing)
        )
        lines.append("")
        lines.append(
            "Fill ONLY the missing items — the existing code fix in "
            "HEAD is correct and should not be re-touched unless a "
            "related-to-diff gate failure tells you otherwise."
        )
        # Per-gap hints
        if not gaps.get("changelog_unreleased_nonempty", True):
            lines.append(
                "- For `changelog_unreleased_nonempty`: add an entry "
                "under the `[Unreleased]` section of `CHANGELOG.md` "
                "describing the change in this PR."
            )
        if not gaps.get("contributors_has_author", True):
            lines.append(
                "- For `contributors_has_author`: add a line to "
                "`CONTRIBUTORS.md` attributing the change to the "
                "author email in your prompt."
            )
        if not gaps.get("commit_has_issues_closed", True):
            tail = (
                f" referencing PR #{pr_number}'s linked issue"
                if pr_number else ""
            )
            lines.append(
                "- For `commit_has_issues_closed`: amend the HEAD "
                f"commit message to add an `ISSUES CLOSED: #N` footer{tail}."
            )
        if not gaps.get("worktree_clean", True):
            lines.append(
                "- For `worktree_clean`: there are uncommitted "
                "changes in the worktree — commit them (or stash "
                "if intentional) before pushing."
            )
    return "\n".join(lines)