Gitea-Tools/issue_workflow_labels.py

"""Canonical issue type/status label taxonomy and transition helpers (#513)."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable, Mapping, Sequence


@dataclass(frozen=True)
class LabelSpec:
    name: str
    color: str
    description: str


TYPE_LABEL_SPECS: tuple[LabelSpec, ...] = (
    LabelSpec("type:bug", "b60205", "Bug or defect"),
    LabelSpec("type:feature", "a2eeef", "Feature or enhancement"),
    LabelSpec("type:process", "5319e7", "Process or policy work"),
    LabelSpec("type:workflow", "c2e0c6", "Workflow automation or guidance"),
    LabelSpec("type:guardrail", "d93f0b", "Safety gate or guardrail"),
    LabelSpec("type:docs", "006b75", "Documentation work"),
    LabelSpec("type:test", "0e8a16", "Tests or test infrastructure"),
    LabelSpec("type:discussion", "bfdadc", "Discussion-only issue"),
    LabelSpec("type:umbrella", "c5def5", "Umbrella or tracker issue"),
    LabelSpec("type:cleanup", "ededed", "Cleanup or hygiene work"),
)

STATUS_LABEL_SPECS: tuple[LabelSpec, ...] = (
    LabelSpec("status:triage", "fbca04", "Issue needs triage"),
    LabelSpec("status:ready", "0e8a16", "Issue is ready for work"),
    LabelSpec("status:claimed", "fefe2e", "Issue is claimed"),
    LabelSpec("status:in-progress", "fefe2e", "Issue is being worked on"),
    LabelSpec("status:blocked", "b60205", "Issue is blocked"),
    LabelSpec("status:needs-review", "0052cc", "Issue work needs review"),
    LabelSpec("status:pr-open", "1d76db", "A linked PR is open"),
    LabelSpec(
        "status:changes-requested",
        "e11d21",
        "Reviewer requested changes on the linked PR",
    ),
    LabelSpec("status:approved", "0e8a16", "Linked PR is approved"),
    LabelSpec("status:merged", "5319e7", "Linked PR is merged"),
    LabelSpec("status:reconcile", "d93f0b", "Issue needs reconciliation"),
    LabelSpec("status:done", "0e8a16", "Issue workflow is complete"),
    LabelSpec("status:duplicate", "cccccc", "Issue is a duplicate"),
    LabelSpec("status:wontfix", "000000", "Issue will not be fixed"),
)

# Durable validation-outcome labels (#529): the four canonical distinctions a
# reviewer report can carry. These are orthogonal to the single active status:*
# label, so they use their own prefix and are not subject to the one-status
# invariant.
VALIDATION_LABEL_SPECS: tuple[LabelSpec, ...] = (
    LabelSpec("validation:clean-pass", "0e8a16", "Validation was a clean pass"),
    LabelSpec(
        "validation:baseline-accepted",
        "fbca04",
        "Validation passed with a baseline-proven unrelated failure",
    ),
    LabelSpec(
        "validation:blocked",
        "b60205",
        "Validation blocked by an unresolved failure",
    ),
    LabelSpec(
        "validation:post-merge-moot",
        "5319e7",
        "Validation is post-merge moot (PR already merged/closed before review)",
    ),
)

# Lifecycle role-ownership labels (#603): which workflow role currently owns the
# item. Advisory visibility only — the control-plane lease (#601) remains the
# source of truth for mutation authority. Only one role:* label is active at a
# time, mirroring the single-active-status invariant.
ROLE_LABEL_SPECS: tuple[LabelSpec, ...] = (
    LabelSpec("role:author", "1d76db", "Author currently owns the item"),
    LabelSpec("role:reviewer", "5319e7", "Reviewer currently owns the item"),
    LabelSpec("role:merger", "0e8a16", "Merger currently owns the item"),
)

# Lifecycle hazard labels (#603): orthogonal warning flags surfacing dangerous
# coordination conditions. Unlike status/role, multiple hazard:* labels may be
# active at once, and they never substitute for live lease / PR state checks.
HAZARD_LABEL_SPECS: tuple[LabelSpec, ...] = (
    LabelSpec("hazard:stale-lease", "d93f0b", "A stale or expired lease references this item"),
    LabelSpec(
        "hazard:workflow-contaminated",
        "b60205",
        "Session/workflow state is contaminated and must not mutate",
    ),
    LabelSpec("hazard:conflicted", "e11d21", "Linked PR has merge conflicts"),
    LabelSpec("hazard:root-mutation", "b60205", "Work was mutated in the project root checkout"),
    LabelSpec("hazard:manual-state", "d93f0b", "Session or lease state was edited manually"),
    LabelSpec(
        "hazard:terminal-blocker",
        "000000",
        "A terminal review/merge lock blocks progress (#332/#602)",
    ),
)

CANONICAL_LABEL_SPECS: tuple[LabelSpec, ...] = (
    TYPE_LABEL_SPECS
    + STATUS_LABEL_SPECS
    + VALIDATION_LABEL_SPECS
    + ROLE_LABEL_SPECS
    + HAZARD_LABEL_SPECS
)

TYPE_LABELS: frozenset[str] = frozenset(spec.name for spec in TYPE_LABEL_SPECS)
STATUS_LABELS: frozenset[str] = frozenset(spec.name for spec in STATUS_LABEL_SPECS)
VALIDATION_LABELS: frozenset[str] = frozenset(
    spec.name for spec in VALIDATION_LABEL_SPECS
)
ROLE_LABELS: frozenset[str] = frozenset(spec.name for spec in ROLE_LABEL_SPECS)
HAZARD_LABELS: frozenset[str] = frozenset(spec.name for spec in HAZARD_LABEL_SPECS)
CANONICAL_LABELS: frozenset[str] = frozenset(
    spec.name for spec in CANONICAL_LABEL_SPECS
)

STATUS_TRANSITIONS: dict[str, str] = {
    "triage": "status:triage",
    "ready": "status:ready",
    "claim": "status:claimed",
    "claimed": "status:claimed",
    "start": "status:in-progress",
    "start_work": "status:in-progress",
    "in_progress": "status:in-progress",
    "in-progress": "status:in-progress",
    "block": "status:blocked",
    "blocked": "status:blocked",
    "needs_review": "status:needs-review",
    "needs-review": "status:needs-review",
    "reviewing": "status:needs-review",
    "pr_open": "status:pr-open",
    "pr-open": "status:pr-open",
    "changes_requested": "status:changes-requested",
    "changes-requested": "status:changes-requested",
    "changes": "status:changes-requested",
    "approved": "status:approved",
    "merge_ready": "status:approved",
    "merge-ready": "status:approved",
    "merge": "status:reconcile",
    "merged": "status:reconcile",
    "reconcile": "status:reconcile",
    "done": "status:done",
    "complete": "status:done",
    "duplicate": "status:duplicate",
    "wontfix": "status:wontfix",
    "abandoned": "status:wontfix",
    # #603 requested state:* synonyms folded into the canonical status vocabulary
    "needs_triage": "status:triage",
    "needs-triage": "status:triage",
    "authoring": "status:in-progress",
}

# #603: single-active role ownership. Maps role kinds / role:* labels to the
# canonical role label. Mirrors STATUS_TRANSITIONS for the role dimension.
ROLE_TRANSITIONS: dict[str, str] = {
    "author": "role:author",
    "reviewer": "role:reviewer",
    "merger": "role:merger",
}

# #603: hazard flag synonyms. Hazards are additive (not single-active), so this
# only normalizes names; it does not drive replacement.
HAZARD_TRANSITIONS: dict[str, str] = {
    "stale_lease": "hazard:stale-lease",
    "stale-lease": "hazard:stale-lease",
    "workflow_contaminated": "hazard:workflow-contaminated",
    "workflow-contaminated": "hazard:workflow-contaminated",
    "contaminated": "hazard:workflow-contaminated",
    "conflicted": "hazard:conflicted",
    "conflict": "hazard:conflicted",
    "root_mutation": "hazard:root-mutation",
    "root-mutation": "hazard:root-mutation",
    "manual_state": "hazard:manual-state",
    "manual-state": "hazard:manual-state",
    "terminal_blocker": "hazard:terminal-blocker",
    "terminal-blocker": "hazard:terminal-blocker",
}


def label_name(label: str | Mapping[str, object]) -> str:
    """Return a normalized label name from a Gitea label object or string."""
    if isinstance(label, str):
        return label.strip()
    value = label.get("name")
    return str(value or "").strip()


def label_names(labels: Iterable[str | Mapping[str, object]] | Mapping[str, object]) -> list[str]:
    """Extract label names from a label list or issue-like mapping."""
    raw: object
    if isinstance(labels, Mapping):
        raw = labels.get("labels", [])
    else:
        raw = labels
    return [name for name in (label_name(label) for label in raw or []) if name]


def type_labels(labels: Iterable[str | Mapping[str, object]] | Mapping[str, object]) -> list[str]:
    return [name for name in label_names(labels) if name.startswith("type:")]


def status_labels(labels: Iterable[str | Mapping[str, object]] | Mapping[str, object]) -> list[str]:
    return [name for name in label_names(labels) if name.startswith("status:")]


def canonical_status_label(status_or_transition: str) -> str:
    status = status_or_transition.strip()
    if status in STATUS_LABELS:
        return status
    normalized = status.lower().replace(" ", "-")
    try:
        return STATUS_TRANSITIONS[normalized]
    except KeyError as exc:
        raise ValueError(
            f"unknown workflow status or transition '{status_or_transition}'"
        ) from exc


def transition_status_labels(
    existing_labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
    status_or_transition: str,
) -> list[str]:
    """Replace all active status labels with the requested canonical status."""
    new_status = canonical_status_label(status_or_transition)
    kept = [name for name in label_names(existing_labels) if not name.startswith("status:")]
    if new_status not in kept:
        kept.append(new_status)
    return kept


def role_labels(labels: Iterable[str | Mapping[str, object]] | Mapping[str, object]) -> list[str]:
    return [name for name in label_names(labels) if name.startswith("role:")]


def hazard_labels(labels: Iterable[str | Mapping[str, object]] | Mapping[str, object]) -> list[str]:
    return [name for name in label_names(labels) if name.startswith("hazard:")]


def canonical_role_label(role_or_transition: str) -> str:
    role = role_or_transition.strip()
    if role in ROLE_LABELS:
        return role
    normalized = role.lower().replace(" ", "-")
    try:
        return ROLE_TRANSITIONS[normalized]
    except KeyError as exc:
        raise ValueError(
            f"unknown workflow role or transition '{role_or_transition}'"
        ) from exc


def transition_role_labels(
    existing_labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
    role_or_transition: str,
) -> list[str]:
    """Replace all active role labels with the requested canonical role."""
    new_role = canonical_role_label(role_or_transition)
    kept = [name for name in label_names(existing_labels) if not name.startswith("role:")]
    if new_role not in kept:
        kept.append(new_role)
    return kept


def canonical_hazard_label(hazard: str) -> str:
    name = hazard.strip()
    if name in HAZARD_LABELS:
        return name
    normalized = name.lower().replace(" ", "-")
    try:
        return HAZARD_TRANSITIONS[normalized]
    except KeyError as exc:
        raise ValueError(f"unknown workflow hazard '{hazard}'") from exc


def add_hazard_label(
    existing_labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
    hazard: str,
) -> list[str]:
    """Add a hazard flag without disturbing status/role/type labels (additive)."""
    new_hazard = canonical_hazard_label(hazard)
    kept = label_names(existing_labels)
    if new_hazard not in kept:
        kept.append(new_hazard)
    return kept


def clear_hazard_label(
    existing_labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
    hazard: str,
) -> list[str]:
    """Remove a single hazard flag, leaving all other labels intact."""
    target = canonical_hazard_label(hazard)
    return [name for name in label_names(existing_labels) if name != target]


def is_discussion(labels: Iterable[str | Mapping[str, object]] | Mapping[str, object]) -> bool:
    return "type:discussion" in type_labels(labels)


def is_implementation_candidate(
    labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
) -> bool:
    """Whether an item may enter an implementation queue on labels alone.

    Discussion issues are excluded (#603 AC3) unless a controller explicitly
    selects them; the allocator still cross-checks live lease/PR state and never
    trusts labels alone (#603 AC2).
    """
    return not is_discussion(labels)


def requires_blocking_reason(
    labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
) -> bool:
    """Whether the item must carry a blocking-reason / next-action comment (AC4).

    True when blocked or when any hazard flag is present.
    """
    names = label_names(labels)
    if "status:blocked" in names:
        return True
    return any(name.startswith("hazard:") for name in names)


def labels_for_new_issue(
    issue_type: str | None = None,
    initial_status: str | None = None,
    extra_labels: Sequence[str] | None = None,
    *,
    discussion: bool = False,
) -> list[str]:
    names = list(extra_labels or [])
    if issue_type:
        type_name = issue_type if issue_type.startswith("type:") else f"type:{issue_type}"
        names.append(type_name)
    if discussion:
        names.append("type:discussion")
    if initial_status:
        names.append(canonical_status_label(initial_status))

    result: list[str] = []
    for name in names:
        if name not in result:
            result.append(name)
    return result


def assess_issue_labels(
    labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
    *,
    discussion: bool = False,
    require_type: bool = True,
    require_status: bool = True,
) -> dict:
    names = label_names(labels)
    found_types = type_labels(names)
    found_statuses = status_labels(names)
    found_roles = role_labels(names)
    found_hazards = hazard_labels(names)
    errors: list[str] = []
    warnings: list[str] = []

    if require_type and not found_types:
        errors.append("issue is missing a type:* label")
    if require_status and not found_statuses:
        errors.append("issue is missing a status:* label")
    if discussion and "type:discussion" not in found_types:
        errors.append("discussion issue is missing type:discussion")
    if len(found_statuses) > 1:
        errors.append(
            "issue has multiple active status:* labels: "
            + ", ".join(found_statuses)
        )
    if len(found_roles) > 1:
        errors.append(
            "issue has multiple active role:* labels: " + ", ".join(found_roles)
        )

    for name in found_types:
        if name not in TYPE_LABELS:
            warnings.append(f"unknown type label '{name}'")
    for name in found_statuses:
        if name not in STATUS_LABELS:
            warnings.append(f"unknown status label '{name}'")
    for name in found_roles:
        if name not in ROLE_LABELS:
            warnings.append(f"unknown role label '{name}'")
    for name in found_hazards:
        if name not in HAZARD_LABELS:
            warnings.append(f"unknown hazard label '{name}'")

    return {
        "valid": not errors,
        "labels": names,
        "type_labels": found_types,
        "status_labels": found_statuses,
        "role_labels": found_roles,
        "hazard_labels": found_hazards,
        "errors": errors,
        "warnings": warnings,
    }