Gitea-Tools/allocator_dependencies.py

"""Canonical dependency parsing and live-state resolution for the allocator (#758).

The work allocator previously inferred dependency state from two lowercase
substrings (``"blocked on #"`` / ``"downstream of #"``). The repository's
canonical declaration form is a ``Depends:`` field inside the issue body's
linkage line, for example::

    * Parent: #631 · Depends: #633, #634 · Related: #630, #434

That form matched neither substring, so dependency-blocked issues were emitted
as eligible candidates. This module replaces substring inference with:

1. structured parsing of ``Depends:`` declarations into issue references, and
2. resolution of each reference against **live issue state**, never body text.

Both halves fail closed: a reference whose state cannot be established makes
the owning candidate ineligible rather than assignable.

No issue number is special-cased here (#758 AC4/AC14); the parser is driven
entirely by the declaration syntax.
"""

from __future__ import annotations

import re
from typing import Callable, Iterable

# Live issue states, as reported by Gitea.
DEP_STATE_OPEN = "open"
DEP_STATE_CLOSED = "closed"

# "Depends:" / "Depends on:" introduces the declaration. "Dependencies" does
# not match: after "depend" it continues with "e", not "s".
_DEPENDS_KEYWORD = re.compile(r"depends(?:\s+on)?\s*:?\s*", re.IGNORECASE)

# Immediately after the keyword, consume only the contiguous run of issue
# references. Anchoring the run this way means the declaration ends naturally
# at the next separator ("·", newline) or sibling field ("Related:"), without
# needing to enumerate separators.
_DEP_RUN = re.compile(
    r"\s*(#\d+(?:\s*(?:,|and|&)\s*#\d+)*)",
    re.IGNORECASE,
)

# Legacy marker retained so previously-recognized bodies keep working.
_LEGACY_BLOCKED = re.compile(r"blocked\s+on\s+#(\d+)", re.IGNORECASE)

_ISSUE_REF = re.compile(r"#(\d+)")


def parse_dependency_refs(body: str | None) -> tuple[int, ...]:
    """Extract declared dependency issue numbers from an issue *body*.

    Recognizes the canonical ``Depends: #N, #N`` field (including the
    ``Depends on`` spelling) plus the legacy ``blocked on #N`` marker.
    Returns references in first-seen order with duplicates removed. Malformed
    or absent declarations yield an empty tuple rather than raising.
    """
    if not body:
        return ()

    refs: list[int] = []

    def _add(value: str) -> None:
        number = int(value)
        if number > 0 and number not in refs:
            refs.append(number)

    for match in _DEPENDS_KEYWORD.finditer(body):
        run = _DEP_RUN.match(body, match.end())
        if not run:
            continue
        for ref in _ISSUE_REF.findall(run.group(1)):
            _add(ref)

    for ref in _LEGACY_BLOCKED.findall(body):
        _add(ref)

    return tuple(refs)


def resolve_dependency_state(
    refs: Iterable[int],
    state_lookup: Callable[[int], str | None],
    *,
    subject: str = "candidate",
) -> dict:
    """Resolve declared *refs* against live issue state.

    *state_lookup* maps an issue number to its live state string, or to
    ``None`` when that evidence could not be obtained. A reference is:

    * **met** when live state is ``closed``;
    * **unmet** when live state is any other live value (``open``, etc.);
    * **unavailable** when state is ``None`` or the lookup raises.

    Unmet *and* unavailable both mark the candidate ineligible (#758 AC6/AC7):
    allocation must never assume a dependency is satisfied.
    """
    unmet: list[int] = []
    unavailable: list[int] = []
    met: list[int] = []

    for ref in refs:
        try:
            number = int(ref)
        except (TypeError, ValueError):
            continue
        try:
            state = state_lookup(number)
        except Exception:  # noqa: BLE001 — unavailable evidence fails closed
            state = None
        normalized = (str(state).strip().lower() if state is not None else "") or None
        if normalized is None:
            unavailable.append(number)
        elif normalized == DEP_STATE_CLOSED:
            met.append(number)
        else:
            unmet.append(number)

    reason: str | None = None
    if unmet and unavailable:
        reason = (
            f"{subject} has unresolved dependencies "
            f"{_fmt(unmet)} and unverifiable dependencies {_fmt(unavailable)} "
            "(fail closed)"
        )
    elif unmet:
        reason = (
            f"{subject} depends on unresolved issue(s) {_fmt(unmet)}; "
            "they are not closed"
        )
    elif unavailable:
        reason = (
            f"{subject} dependency evidence unavailable for {_fmt(unavailable)} "
            "(fail closed)"
        )

    return {
        "refs": tuple(int(x) for x in refs),
        "met": tuple(met),
        "unmet": tuple(unmet),
        "unavailable": tuple(unavailable),
        "dependency_unmet": bool(unmet or unavailable),
        "reason": reason,
    }


def _fmt(numbers: Iterable[int]) -> str:
    return ", ".join(f"#{n}" for n in numbers)