diff_preflight.py

"""Phase 5 — Reviewer diff preflight (single policy point).

The reviewer LLM is expensive and learns nothing new by re-reading a
diff that exceeds the configured ``review_max_diff_chars`` ceiling.
This module centralizes the preflight decision so every caller (merge
node, reviewer node, security review) short-circuits oversize diffs to
``REDECOMPOSE`` identically, and so the escalation logic (tighten the
decomposer budget, stop after N attempts) lives in ONE place that can
be unit-tested without the LangGraph machinery.

Public API:

* ``evaluate(result, review_max_diff_chars)`` -> :class:`DiffPreflightVerdict`
* ``tighten_decomposer_budget(current_max_lines)`` -> int
* ``should_escalate_oversize(oversize_attempts)`` -> bool
* ``OVERSIZE_STUCK_CAUSE`` constant for dashboard / retro matching.

See ``platform/tests/engine/test_v4_5_2_diff_preflight.py`` for the
reproduction of the live 524 KB loop.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

OVERSIZE_STUCK_CAUSE: str = "redecompose.oversize_stuck"

MAX_OVERSIZE_ATTEMPTS: int = 2
TIGHTEN_RATIO: float = 0.70
BUDGET_FLOOR: int = 40


@dataclass
class DiffPreflightVerdict:
    """Single decision emitted by :func:`evaluate`.

    * ``decision`` is ``"accept"`` (proceed to reviewer) or ``"reject"``
      (short-circuit before the LLM call).
    * ``action`` is ``"review"`` when accepted; ``"redecompose"`` when
      rejected — routes directly to ``RecoveryAction.REDECOMPOSE``.
    * ``cause`` disambiguates rejection reasons; today only
      ``"oversize_pre_review"`` exists but the field is reserved for
      future preflight rules (e.g. binary-blob detection).
    * ``invoked_reviewer`` is a boolean observers check in tests to
      make sure the LLM was NOT called on a reject.
    * ``payload`` is JSON-serializable and should be passed to
      ``log_event("diff_preflight_reject", payload)``.
    """

    decision: str = "accept"
    action: str = "review"
    cause: str = ""
    oversize_ratio: float = 0.0
    diff_bytes: int = 0
    limit: int = 0
    subtask_id: str = ""
    invoked_reviewer: bool = True
    payload: dict[str, Any] = field(default_factory=dict)


def evaluate(result: dict[str, Any], review_max_diff_chars: int) -> DiffPreflightVerdict:
    """Return a preflight verdict for *result*.

    ``result`` is a mutable subtask-result dict (as built by
    ``worker_session.build_subtask_result``). The function never
    mutates it.
    """
    diff_content = (result or {}).get("diff_content") or ""
    diff_bytes = len(diff_content)
    limit = int(review_max_diff_chars or 0)
    subtask_id = str((result or {}).get("id") or "")

    if not diff_content or limit <= 0 or diff_bytes <= limit:
        verdict = DiffPreflightVerdict(
            decision="accept",
            action="review",
            cause="",
            oversize_ratio=(diff_bytes / limit) if limit > 0 else 0.0,
            diff_bytes=diff_bytes,
            limit=limit,
            subtask_id=subtask_id,
            invoked_reviewer=True,
        )
        verdict.payload = {
            "subtask_id": subtask_id,
            "diff_bytes": diff_bytes,
            "limit": limit,
            "decision": "accept",
            "cause": "",
            "oversize_ratio": verdict.oversize_ratio,
        }
        return verdict

    ratio = diff_bytes / limit
    verdict = DiffPreflightVerdict(
        decision="reject",
        action="redecompose",
        cause="oversize_pre_review",
        oversize_ratio=ratio,
        diff_bytes=diff_bytes,
        limit=limit,
        subtask_id=subtask_id,
        invoked_reviewer=False,
    )
    verdict.payload = {
        "subtask_id": subtask_id,
        "diff_bytes": diff_bytes,
        "limit": limit,
        "decision": "reject",
        "cause": "oversize_pre_review",
        "oversize_ratio": ratio,
        "schema_version": 1,
    }
    return verdict


def tighten_decomposer_budget(current_max_lines: int) -> int:
    """Return a tightened ``max_lines_per_task`` for the next retry.

    Shrinks by ``1 - TIGHTEN_RATIO`` (30 %) per call — strictly more than
    the 25 % minimum the plan requires — and clamps at ``BUDGET_FLOOR``
    so a stuck loop never drives the decomposer to absurdly small
    per-subtask sizes.
    """
    try:
        current = int(current_max_lines)
    except (TypeError, ValueError):
        current = 0
    if current <= 0:
        return BUDGET_FLOOR
    tightened = int(current * TIGHTEN_RATIO)
    return max(tightened, BUDGET_FLOOR)


def should_escalate_oversize(oversize_attempts: int) -> bool:
    """True iff the epic has exhausted the oversize retry budget."""
    try:
        return int(oversize_attempts) >= MAX_OVERSIZE_ATTEMPTS
    except (TypeError, ValueError):
        return False


__all__ = [
    "BUDGET_FLOOR",
    "MAX_OVERSIZE_ATTEMPTS",
    "OVERSIZE_STUCK_CAUSE",
    "TIGHTEN_RATIO",
    "DiffPreflightVerdict",
    "evaluate",
    "should_escalate_oversize",
    "tighten_decomposer_budget",
]