SupplyAgent/supply_infra/pipeline/gates.py

from __future__ import annotations

from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class GateDecision:
    passed: bool
    error_code: str | None = None
    error_message: str | None = None


def evaluate_step_gate(step_key: str, payload: dict[str, Any]) -> GateDecision:
    if payload.get("success") is False:
        return GateDecision(
            passed=False,
            error_code=str(payload.get("error_code") or "step_reported_failure"),
            error_message=str(payload.get("error") or "Step returned success=false"),
        )

    if step_key in {"demand_grade", "demand_expand", "video_discovery"}:
        failed = int(payload.get("failed", payload.get("failed_count", 0)) or 0)
        if failed > 0:
            return GateDecision(
                False,
                "business_failures",
                f"{step_key} reported {failed} failed item(s)",
            )

    if step_key == "aigc_write_record":
        request_made = payload.get("external_request_made") is True
        if request_made and (
            payload.get("effect_recorded") is not True
            or not payload.get("outbox_ids")
        ):
            return GateDecision(
                False,
                "aigc_record_incomplete",
                "AIGC publish batches were not durably recorded before dispatch",
            )
        failed_batches = int(payload.get("failed_batch_count", 0) or 0)
        if failed_batches > 0:
            return GateDecision(
                False,
                "aigc_publish_failed",
                f"AIGC publish reported {failed_batches} failed batch(es)",
            )

    return GateDecision(True)