iop/agent-ops/skills/project/orchestrate-agent-task-loop/scripts/execution_target_policy.py

#!/usr/bin/env python3
"""Pure execution-target policy for Agent Task worker and review stages."""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime

from zoneinfo import ZoneInfo


KST = ZoneInfo("Asia/Seoul")

VALID_STAGES = {"worker", "review"}
VALID_LANES = {"local", "cloud"}


@dataclass(frozen=True)
class RouteTarget:
    adapter: str
    target: str
    execution_class: str
    selfcheck_required: bool


@dataclass(frozen=True)
class PolicyDecision:
    rule_id: str
    policy_priority: int
    reason_codes: tuple[str, ...]
    time_window: str
    candidates: tuple[RouteTarget, ...]


PI_ORNITH = RouteTarget("pi", "iop/ornith:35b", "local_model", True)
AGY_GEMINI_LOW = RouteTarget(
    "agy", "Gemini 3.6 Flash (Low)", "cloud_model", False
)
AGY_GEMINI_MEDIUM = RouteTarget(
    "agy", "Gemini 3.6 Flash (Medium)", "cloud_model", False
)
AGY_GEMINI_HIGH = RouteTarget(
    "agy", "Gemini 3.6 Flash (High)", "cloud_model", False
)
PI_LAGUNA = RouteTarget("pi", "iop/laguna-s:2.1", "local_model", True)
CLAUDE_OPUS = RouteTarget("claude", "claude-opus-4-8", "cloud_model", False)
CODEX_SOL_XHIGH = RouteTarget("codex", "gpt-5.6-sol", "cloud_model", False)
CODEX_TERRA_HIGH = RouteTarget("codex", "gpt-5.6-terra", "cloud_model", False)


CANONICAL_TARGETS = (
    PI_ORNITH,
    AGY_GEMINI_LOW,
    AGY_GEMINI_MEDIUM,
    AGY_GEMINI_HIGH,
    PI_LAGUNA,
    CLAUDE_OPUS,
    CODEX_SOL_XHIGH,
    CODEX_TERRA_HIGH,
)


def canonical_target(adapter: str, target: str) -> RouteTarget | None:
    """Resolve one policy-owned adapter + target identity."""
    return next(
        (
            candidate
            for candidate in CANONICAL_TARGETS
            if candidate.adapter == adapter and candidate.target == target
        ),
        None,
    )


def promotion_target(current: RouteTarget) -> RouteTarget | None:
    """Return the next target in the policy-owned cloud promotion chain."""
    if current.adapter == "agy" and current in {
        AGY_GEMINI_LOW,
        AGY_GEMINI_MEDIUM,
        AGY_GEMINI_HIGH,
    }:
        return CLAUDE_OPUS
    if current == CLAUDE_OPUS:
        return CODEX_TERRA_HIGH
    return None


@dataclass(frozen=True)
class QuotaProbeSpec:
    command: str
    target: str
    required_caps: tuple[str, ...]


def quota_probe_spec(target: RouteTarget) -> QuotaProbeSpec | None:
    """Return the policy-owned quota probe spec for a route target."""
    if target.execution_class == "local_model":
        return None
    if target.adapter == "agy":
        return QuotaProbeSpec(
            command="agy",
            target=target.target,
            required_caps=("overall", f"model:{target.target}"),
        )
    if target.adapter in {"claude", "codex"}:
        return QuotaProbeSpec(
            command=target.adapter,
            target=target.target,
            required_caps=("overall",),
        )
    return None


def _validate(stage: str, lane: str, grade: int, evaluated_at: datetime) -> None:
    if stage not in VALID_STAGES:
        raise ValueError(f"unsupported stage: {stage}")
    if lane not in VALID_LANES:
        raise ValueError(f"unsupported lane: {lane}")
    if not 1 <= grade <= 10:
        raise ValueError(f"grade must be in G01..G10: {grade}")
    if evaluated_at.tzinfo is None or evaluated_at.utcoffset() is None:
        raise ValueError("evaluated_at must be timezone-aware")


def _kst_time_window(evaluated_at: datetime) -> str:
    kst_time = evaluated_at.astimezone(KST).time()
    if 7 <= kst_time.hour < 23:
        return "kst-day-[07:00,23:00)"
    return "kst-night-[23:00,07:00)"


def select_policy(
    *, stage: str, lane: str, grade: int, evaluated_at: datetime
) -> PolicyDecision:
    """Return the ordered target policy for one initial route evaluation."""

    _validate(stage, lane, grade, evaluated_at)

    if stage == "review":
        return PolicyDecision(
            rule_id="official-review-codex",
            policy_priority=10,
            reason_codes=("official_review_fixed",),
            time_window="not_applicable",
            candidates=(CODEX_SOL_XHIGH,),
        )

    if lane == "local":
        if grade <= 6:
            return PolicyDecision(
                rule_id="worker-local-g01-g06",
                policy_priority=30,
                reason_codes=("local_low_grade",),
                time_window="not_applicable",
                candidates=(PI_ORNITH,),
            )
        if grade <= 8:
            time_window = _kst_time_window(evaluated_at)
            if time_window == "kst-day-[07:00,23:00)":
                rule_id = "worker-local-g07-g08-kst-day"
                reason_code = "kst_day_gemini_medium"
                candidates = (AGY_GEMINI_MEDIUM, PI_LAGUNA)
            else:
                rule_id = "worker-local-g07-g08-kst-night"
                reason_code = "kst_night_laguna"
                candidates = (PI_LAGUNA, AGY_GEMINI_MEDIUM)
            return PolicyDecision(
                rule_id=rule_id,
                policy_priority=20,
                reason_codes=(reason_code,),
                time_window=time_window,
                candidates=candidates,
            )
        return PolicyDecision(
            rule_id="worker-local-g09-g10",
            policy_priority=30,
            reason_codes=("local_high_grade_cloud_target",),
            time_window="not_applicable",
            candidates=(CLAUDE_OPUS,),
        )

    if grade <= 2:
        target = AGY_GEMINI_LOW
        rule_id = "worker-cloud-g01-g02"
        reason_code = "cloud_gemini_low_grade"
    elif grade <= 4:
        target = AGY_GEMINI_MEDIUM
        rule_id = "worker-cloud-g03-g04"
        reason_code = "cloud_gemini_medium_grade"
    elif grade <= 6:
        target = AGY_GEMINI_HIGH
        rule_id = "worker-cloud-g05-g06"
        reason_code = "cloud_gemini_high_grade"
    elif grade <= 8:
        target = CLAUDE_OPUS
        rule_id = "worker-cloud-g07-g08"
        reason_code = "cloud_opus_grade"
    else:
        target = CODEX_SOL_XHIGH
        rule_id = "worker-cloud-g09-g10"
        reason_code = "cloud_codex_grade"
    return PolicyDecision(
        rule_id=rule_id,
        policy_priority=30,
        reason_codes=(reason_code,),
        time_window="not_applicable",
        candidates=(target,),
    )