toki
b383d20a1b
작업 모델별로 전체 검토와 코드리뷰 체크 검토를 독립적으로 켜고 끌 수 있어야 하며, 다음 단계 진입 시 변경된 카탈로그 설정을 재시작 없이 반영해야 한다.
597 lines
22 KiB
Python
597 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""Catalog-backed execution-target policy for Agent Task stages."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
from typing import Any
|
|
|
|
from zoneinfo import ZoneInfo
|
|
|
|
|
|
KST = ZoneInfo("Asia/Seoul")
|
|
|
|
VALID_STAGES = {"worker", "review"}
|
|
VALID_LANES = {"local", "cloud"}
|
|
VALID_PI_THINKING_LEVELS = frozenset({"low", "medium", "high"})
|
|
VALID_REASONING_EFFORTS = frozenset({"medium", "high", "max", "xhigh"})
|
|
VALID_EXECUTION_CLASSES = frozenset({"local_model", "cloud_model"})
|
|
CATALOG_SCHEMA_VERSION = 1
|
|
CATALOG_PATH = Path(__file__).with_name("execution_target_catalog.json")
|
|
TIME_WINDOWS = frozenset(
|
|
{"kst-day-[07:00,23:00)", "kst-night-[23:00,07:00)"}
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RouteTarget:
|
|
adapter: str
|
|
target: str
|
|
execution_class: str
|
|
selfcheck_required: bool
|
|
selfcheck_full_review: bool = False
|
|
selfcheck_checklist_review: bool = False
|
|
thinking_level: str | None = None
|
|
reasoning_effort: str | None = None
|
|
command_model: str | None = None
|
|
catalog_id: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LanePolicy:
|
|
candidates: tuple[str, ...]
|
|
policy_priority: int
|
|
rule_id: str | None = None
|
|
reason_codes: tuple[str, ...] = ()
|
|
time_windows: dict[str, tuple[str, tuple[str, ...]]] | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ExecutionTargetCatalog:
|
|
schema_version: int
|
|
revision: str
|
|
targets: dict[str, RouteTarget]
|
|
lanes: dict[str, dict[str, LanePolicy]]
|
|
promotions: dict[str, str]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PolicyDecision:
|
|
rule_id: str
|
|
policy_priority: int
|
|
reason_codes: tuple[str, ...]
|
|
time_window: str
|
|
candidates: tuple[RouteTarget, ...]
|
|
route_id: str
|
|
catalog_revision: str
|
|
|
|
|
|
class CatalogError(ValueError):
|
|
"""Raised when the operator-owned model catalog is malformed."""
|
|
|
|
|
|
def _load_target_contract():
|
|
module_name = "execution_target_contract"
|
|
loaded = sys.modules.get(module_name)
|
|
if loaded is not None:
|
|
return loaded
|
|
path = Path(__file__).with_name("execution_target_contract.py")
|
|
spec = importlib.util.spec_from_file_location(module_name, path)
|
|
if spec is None or spec.loader is None:
|
|
raise CatalogError(f"target contract load failed: {path}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
target_contract = _load_target_contract()
|
|
VALID_ADAPTERS = target_contract.VALID_ADAPTERS
|
|
|
|
|
|
def _object(value: object, path: str) -> dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
raise CatalogError(f"{path} must be an object")
|
|
return value
|
|
|
|
|
|
def _non_empty_string(value: object, path: str) -> str:
|
|
if not isinstance(value, str) or not value:
|
|
raise CatalogError(f"{path} must be a non-empty string")
|
|
return value
|
|
|
|
|
|
def _optional_enum(
|
|
value: object, allowed: frozenset[str], path: str
|
|
) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, str) or value not in allowed:
|
|
raise CatalogError(f"{path} must be null or one of {sorted(allowed)}")
|
|
return value
|
|
|
|
|
|
def _target_from_config(target_id: str, value: object) -> RouteTarget:
|
|
path = f"targets.{target_id}"
|
|
item = _object(value, path)
|
|
allowed = {
|
|
"adapter",
|
|
"target",
|
|
"execution_class",
|
|
"selfcheck_required",
|
|
"selfcheck",
|
|
"thinking_level",
|
|
"reasoning_effort",
|
|
"command_model",
|
|
}
|
|
unknown = sorted(set(item) - allowed)
|
|
if unknown:
|
|
raise CatalogError(f"{path} has unknown fields: {unknown}")
|
|
execution_class = _non_empty_string(
|
|
item.get("execution_class"), f"{path}.execution_class"
|
|
)
|
|
if execution_class not in VALID_EXECUTION_CLASSES:
|
|
raise CatalogError(
|
|
f"{path}.execution_class must be one of {sorted(VALID_EXECUTION_CLASSES)}"
|
|
)
|
|
legacy_selfcheck = item.get("selfcheck_required")
|
|
raw_selfcheck = item.get("selfcheck")
|
|
if raw_selfcheck is not None and legacy_selfcheck is not None:
|
|
raise CatalogError(
|
|
f"{path} cannot combine selfcheck and selfcheck_required"
|
|
)
|
|
if raw_selfcheck is not None:
|
|
selfcheck_config = _object(raw_selfcheck, f"{path}.selfcheck")
|
|
expected_fields = {"full_review", "checklist_review"}
|
|
if set(selfcheck_config) != expected_fields:
|
|
raise CatalogError(
|
|
f"{path}.selfcheck must contain exactly {sorted(expected_fields)}"
|
|
)
|
|
full_review = selfcheck_config.get("full_review")
|
|
checklist_review = selfcheck_config.get("checklist_review")
|
|
if not isinstance(full_review, bool):
|
|
raise CatalogError(
|
|
f"{path}.selfcheck.full_review must be a boolean"
|
|
)
|
|
if not isinstance(checklist_review, bool):
|
|
raise CatalogError(
|
|
f"{path}.selfcheck.checklist_review must be a boolean"
|
|
)
|
|
# Keep the old decision field stable for persisted-state compatibility.
|
|
# Runtime self-check admission uses the two explicit stage flags below.
|
|
selfcheck_required = item.get("adapter") == "pi"
|
|
else:
|
|
if not isinstance(legacy_selfcheck, bool):
|
|
raise CatalogError(
|
|
f"{path}.selfcheck must be an object with boolean stages"
|
|
)
|
|
full_review = legacy_selfcheck
|
|
checklist_review = legacy_selfcheck
|
|
selfcheck_required = legacy_selfcheck
|
|
command_model = item.get("command_model")
|
|
if command_model is not None:
|
|
command_model = _non_empty_string(command_model, f"{path}.command_model")
|
|
target = RouteTarget(
|
|
adapter=_non_empty_string(item.get("adapter"), f"{path}.adapter"),
|
|
target=_non_empty_string(item.get("target"), f"{path}.target"),
|
|
execution_class=execution_class,
|
|
selfcheck_required=selfcheck_required,
|
|
selfcheck_full_review=full_review,
|
|
selfcheck_checklist_review=checklist_review,
|
|
thinking_level=_optional_enum(
|
|
item.get("thinking_level"),
|
|
VALID_PI_THINKING_LEVELS,
|
|
f"{path}.thinking_level",
|
|
),
|
|
reasoning_effort=_optional_enum(
|
|
item.get("reasoning_effort"),
|
|
VALID_REASONING_EFFORTS,
|
|
f"{path}.reasoning_effort",
|
|
),
|
|
command_model=command_model,
|
|
catalog_id=target_id,
|
|
)
|
|
target_contract.validate_target_contract(target, path, CatalogError)
|
|
return target
|
|
|
|
|
|
def _string_list(value: object, path: str) -> tuple[str, ...]:
|
|
if not isinstance(value, list) or not value:
|
|
raise CatalogError(f"{path} must be a non-empty array")
|
|
values = tuple(
|
|
_non_empty_string(entry, f"{path}[{index}]")
|
|
for index, entry in enumerate(value)
|
|
)
|
|
if len(values) != len(set(values)):
|
|
raise CatalogError(f"{path} must not contain duplicate ids")
|
|
return values
|
|
|
|
|
|
def _lane_from_config(
|
|
stage: str,
|
|
lane_id: str,
|
|
value: object,
|
|
targets: dict[str, RouteTarget],
|
|
) -> LanePolicy:
|
|
path = f"lanes.{stage}.{lane_id}"
|
|
item = _object(value, path)
|
|
allowed = {
|
|
"candidates",
|
|
"rule_id",
|
|
"policy_priority",
|
|
"reason_codes",
|
|
"time_windows",
|
|
}
|
|
unknown = sorted(set(item) - allowed)
|
|
if unknown:
|
|
raise CatalogError(f"{path} has unknown fields: {unknown}")
|
|
candidates = _string_list(item.get("candidates"), f"{path}.candidates")
|
|
missing_targets = [target_id for target_id in candidates if target_id not in targets]
|
|
if missing_targets:
|
|
raise CatalogError(f"{path} references unknown targets: {missing_targets}")
|
|
execution_classes = {targets[target_id].execution_class for target_id in candidates}
|
|
if len(execution_classes) != 1:
|
|
raise CatalogError(
|
|
f"{path}.candidates cannot mix local_model and cloud_model targets"
|
|
)
|
|
priority = item.get("policy_priority")
|
|
if isinstance(priority, bool) or not isinstance(priority, int) or priority < 0:
|
|
raise CatalogError(f"{path}.policy_priority must be a non-negative integer")
|
|
raw_reasons = item.get("reason_codes", [])
|
|
if not isinstance(raw_reasons, list) or any(
|
|
not isinstance(reason, str) or not reason for reason in raw_reasons
|
|
):
|
|
raise CatalogError(f"{path}.reason_codes must be an array of strings")
|
|
rule_id = item.get("rule_id")
|
|
if rule_id is not None:
|
|
rule_id = _non_empty_string(rule_id, f"{path}.rule_id")
|
|
raw_windows = item.get("time_windows")
|
|
windows: dict[str, tuple[str, tuple[str, ...]]] | None = None
|
|
if raw_windows is not None:
|
|
if rule_id is not None or raw_reasons:
|
|
raise CatalogError(
|
|
f"{path}: time_windows cannot be combined with base rule metadata"
|
|
)
|
|
windows_obj = _object(raw_windows, f"{path}.time_windows")
|
|
if set(windows_obj) != TIME_WINDOWS:
|
|
raise CatalogError(
|
|
f"{path}.time_windows must define exactly {sorted(TIME_WINDOWS)}"
|
|
)
|
|
windows = {}
|
|
for window_name, raw_window in windows_obj.items():
|
|
window = _object(raw_window, f"{path}.time_windows.{window_name}")
|
|
if set(window) != {"rule_id", "reason_codes"}:
|
|
raise CatalogError(
|
|
f"{path}.time_windows.{window_name} must contain rule_id and reason_codes"
|
|
)
|
|
reasons = window.get("reason_codes")
|
|
if not isinstance(reasons, list) or not reasons or any(
|
|
not isinstance(reason, str) or not reason for reason in reasons
|
|
):
|
|
raise CatalogError(
|
|
f"{path}.time_windows.{window_name}.reason_codes must be a non-empty string array"
|
|
)
|
|
windows[window_name] = (
|
|
_non_empty_string(
|
|
window.get("rule_id"),
|
|
f"{path}.time_windows.{window_name}.rule_id",
|
|
),
|
|
tuple(reasons),
|
|
)
|
|
elif rule_id is None:
|
|
raise CatalogError(f"{path}.rule_id is required without time_windows")
|
|
return LanePolicy(
|
|
candidates=candidates,
|
|
policy_priority=priority,
|
|
rule_id=rule_id,
|
|
reason_codes=tuple(raw_reasons),
|
|
time_windows=windows,
|
|
)
|
|
|
|
|
|
def _read_catalog_root(path: Path) -> dict[str, Any]:
|
|
try:
|
|
raw = path.read_text(encoding="utf-8")
|
|
except OSError as exc:
|
|
raise CatalogError(f"cannot read execution target catalog {path}: {exc}") from exc
|
|
try:
|
|
data = json.loads(raw)
|
|
except json.JSONDecodeError as exc:
|
|
raise CatalogError(f"invalid JSON in execution target catalog {path}: {exc}") from exc
|
|
root = _object(data, "catalog")
|
|
if set(root) != {"schema_version", "targets", "lanes", "promotions"}:
|
|
raise CatalogError(
|
|
"catalog must contain exactly schema_version, targets, lanes, promotions"
|
|
)
|
|
if root.get("schema_version") != CATALOG_SCHEMA_VERSION:
|
|
raise CatalogError(
|
|
f"catalog.schema_version must be {CATALOG_SCHEMA_VERSION}"
|
|
)
|
|
return root
|
|
|
|
|
|
def load_catalog(path: Path = CATALOG_PATH) -> ExecutionTargetCatalog:
|
|
root = _read_catalog_root(path)
|
|
raw_targets = _object(root.get("targets"), "targets")
|
|
if not raw_targets:
|
|
raise CatalogError("targets must not be empty")
|
|
targets = {
|
|
_non_empty_string(target_id, "targets key"): _target_from_config(
|
|
target_id, value
|
|
)
|
|
for target_id, value in raw_targets.items()
|
|
}
|
|
identities: dict[tuple[object, ...], str] = {}
|
|
for target_id, target in targets.items():
|
|
identity = (
|
|
target.adapter,
|
|
target.target,
|
|
target.thinking_level,
|
|
target.reasoning_effort,
|
|
)
|
|
if identity in identities:
|
|
raise CatalogError(
|
|
f"targets {identities[identity]!r} and {target_id!r} have duplicate runtime identity"
|
|
)
|
|
identities[identity] = target_id
|
|
|
|
raw_lanes = _object(root.get("lanes"), "lanes")
|
|
if set(raw_lanes) != VALID_STAGES:
|
|
raise CatalogError(f"lanes must define exactly {sorted(VALID_STAGES)}")
|
|
expected_lane_ids = {
|
|
f"{lane}-G{grade:02d}" for lane in VALID_LANES for grade in range(1, 11)
|
|
}
|
|
lanes: dict[str, dict[str, LanePolicy]] = {}
|
|
for stage in sorted(VALID_STAGES):
|
|
stage_lanes = _object(raw_lanes.get(stage), f"lanes.{stage}")
|
|
if set(stage_lanes) != expected_lane_ids:
|
|
missing = sorted(expected_lane_ids - set(stage_lanes))
|
|
extra = sorted(set(stage_lanes) - expected_lane_ids)
|
|
raise CatalogError(
|
|
f"lanes.{stage} must define every grade independently; missing={missing}, extra={extra}"
|
|
)
|
|
lanes[stage] = {
|
|
lane_id: _lane_from_config(stage, lane_id, value, targets)
|
|
for lane_id, value in stage_lanes.items()
|
|
}
|
|
|
|
raw_promotions = _object(root.get("promotions"), "promotions")
|
|
promotions: dict[str, str] = {}
|
|
for source, destination in raw_promotions.items():
|
|
source_id = _non_empty_string(source, "promotions key")
|
|
destination_id = _non_empty_string(
|
|
destination, f"promotions.{source_id}"
|
|
)
|
|
if source_id not in targets or destination_id not in targets:
|
|
raise CatalogError(
|
|
f"promotions.{source_id} references an unknown target"
|
|
)
|
|
if source_id == destination_id:
|
|
raise CatalogError(f"promotions.{source_id} cannot point to itself")
|
|
if (
|
|
targets[source_id].execution_class != "cloud_model"
|
|
or targets[destination_id].execution_class != "cloud_model"
|
|
):
|
|
raise CatalogError("promotions may contain only cloud_model targets")
|
|
promotions[source_id] = destination_id
|
|
for source_id in promotions:
|
|
seen: set[str] = set()
|
|
current = source_id
|
|
while current in promotions:
|
|
if current in seen:
|
|
raise CatalogError(f"promotions contain a cycle at {current!r}")
|
|
seen.add(current)
|
|
current = promotions[current]
|
|
normalized = json.dumps(root, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
return ExecutionTargetCatalog(
|
|
schema_version=CATALOG_SCHEMA_VERSION,
|
|
revision=hashlib.sha256(normalized.encode("utf-8")).hexdigest(),
|
|
targets=targets,
|
|
lanes=lanes,
|
|
promotions=promotions,
|
|
)
|
|
|
|
|
|
CATALOG = load_catalog()
|
|
CATALOG_REVISION = CATALOG.revision
|
|
CATALOG_TARGETS_BY_ID = CATALOG.targets
|
|
CANONICAL_TARGETS = tuple(CATALOG.targets.values())
|
|
|
|
|
|
def reload_catalog(path: Path = CATALOG_PATH) -> ExecutionTargetCatalog:
|
|
"""Atomically publish the latest operator-owned catalog.
|
|
|
|
The dispatcher calls this before each scheduler admission and immediately
|
|
before a self-check stage starts. Existing model invocations keep their
|
|
pinned decision; the next stage observes the newest self-check switches.
|
|
"""
|
|
catalog = load_catalog(path)
|
|
compatibility_ids = (
|
|
"pi-ornith-high",
|
|
"pi-laguna-high",
|
|
"agy-gemini-low",
|
|
"agy-gemini-medium",
|
|
"agy-gemini-high",
|
|
"opencode-glm-medium",
|
|
"opencode-glm-high",
|
|
"opencode-glm-max",
|
|
"legacy-claude-glm",
|
|
"claude-opus-xhigh",
|
|
"claude-haiku-xhigh",
|
|
"codex-spark-xhigh",
|
|
"codex-sol-xhigh",
|
|
"codex-terra-high",
|
|
)
|
|
missing = [
|
|
target_id
|
|
for target_id in compatibility_ids
|
|
if target_id not in catalog.targets
|
|
]
|
|
if missing:
|
|
raise CatalogError(
|
|
"catalog is missing compatibility targets required by the "
|
|
f"dispatcher: {missing}"
|
|
)
|
|
compatibility_targets = {
|
|
target_id: catalog.targets[target_id] for target_id in compatibility_ids
|
|
}
|
|
global CATALOG, CATALOG_REVISION, CATALOG_TARGETS_BY_ID, CANONICAL_TARGETS
|
|
global PI_ORNITH, PI_LAGUNA, AGY_GEMINI_LOW, AGY_GEMINI_MEDIUM
|
|
global AGY_GEMINI_HIGH, OPENCODE_GLM_MEDIUM, OPENCODE_GLM_HIGH
|
|
global OPENCODE_GLM_MAX, CLAUDE_GLM, CLAUDE_OPUS, CLAUDE_HAIKU_XHIGH
|
|
global CODEX_SPARK_XHIGH, CODEX_SOL_XHIGH, CODEX_TERRA_HIGH
|
|
CATALOG = catalog
|
|
CATALOG_REVISION = catalog.revision
|
|
CATALOG_TARGETS_BY_ID = catalog.targets
|
|
CANONICAL_TARGETS = tuple(catalog.targets.values())
|
|
PI_ORNITH = compatibility_targets["pi-ornith-high"]
|
|
PI_LAGUNA = compatibility_targets["pi-laguna-high"]
|
|
AGY_GEMINI_LOW = compatibility_targets["agy-gemini-low"]
|
|
AGY_GEMINI_MEDIUM = compatibility_targets["agy-gemini-medium"]
|
|
AGY_GEMINI_HIGH = compatibility_targets["agy-gemini-high"]
|
|
OPENCODE_GLM_MEDIUM = compatibility_targets["opencode-glm-medium"]
|
|
OPENCODE_GLM_HIGH = compatibility_targets["opencode-glm-high"]
|
|
OPENCODE_GLM_MAX = compatibility_targets["opencode-glm-max"]
|
|
CLAUDE_GLM = compatibility_targets["legacy-claude-glm"]
|
|
CLAUDE_OPUS = compatibility_targets["claude-opus-xhigh"]
|
|
CLAUDE_HAIKU_XHIGH = compatibility_targets["claude-haiku-xhigh"]
|
|
CODEX_SPARK_XHIGH = compatibility_targets["codex-spark-xhigh"]
|
|
CODEX_SOL_XHIGH = compatibility_targets["codex-sol-xhigh"]
|
|
CODEX_TERRA_HIGH = compatibility_targets["codex-terra-high"]
|
|
return catalog
|
|
|
|
|
|
def catalog_target(target_id: str) -> RouteTarget:
|
|
try:
|
|
return CATALOG.targets[target_id]
|
|
except KeyError as exc:
|
|
raise CatalogError(f"unknown catalog target: {target_id}") from exc
|
|
|
|
|
|
# Compatibility names remain for persisted-state recovery and focused driver tests.
|
|
# Lane membership and order live only in execution_target_catalog.json.
|
|
PI_ORNITH = catalog_target("pi-ornith-high")
|
|
PI_LAGUNA = catalog_target("pi-laguna-high")
|
|
AGY_GEMINI_LOW = catalog_target("agy-gemini-low")
|
|
AGY_GEMINI_MEDIUM = catalog_target("agy-gemini-medium")
|
|
AGY_GEMINI_HIGH = catalog_target("agy-gemini-high")
|
|
OPENCODE_GLM_MEDIUM = catalog_target("opencode-glm-medium")
|
|
OPENCODE_GLM_HIGH = catalog_target("opencode-glm-high")
|
|
OPENCODE_GLM_MAX = catalog_target("opencode-glm-max")
|
|
CLAUDE_GLM = catalog_target("legacy-claude-glm")
|
|
CLAUDE_OPUS = catalog_target("claude-opus-xhigh")
|
|
CLAUDE_HAIKU_XHIGH = catalog_target("claude-haiku-xhigh")
|
|
CODEX_SPARK_XHIGH = catalog_target("codex-spark-xhigh")
|
|
CODEX_SOL_XHIGH = catalog_target("codex-sol-xhigh")
|
|
CODEX_TERRA_HIGH = catalog_target("codex-terra-high")
|
|
|
|
|
|
def canonical_target(
|
|
adapter: str,
|
|
target: str,
|
|
thinking_level: str | None = None,
|
|
reasoning_effort: str | None = None,
|
|
) -> RouteTarget | None:
|
|
"""Resolve one catalog target, accepting pre-catalog implicit defaults."""
|
|
if adapter == "pi" and thinking_level is None:
|
|
thinking_level = "high"
|
|
if adapter in {"claude", "claude-glm", "codex"} and reasoning_effort is None:
|
|
reasoning_effort = "xhigh"
|
|
return next(
|
|
(
|
|
candidate
|
|
for candidate in CANONICAL_TARGETS
|
|
if (
|
|
candidate.adapter == adapter
|
|
and candidate.target == target
|
|
and candidate.thinking_level == thinking_level
|
|
and candidate.reasoning_effort == reasoning_effort
|
|
)
|
|
),
|
|
None,
|
|
)
|
|
|
|
|
|
def promotion_target(current: RouteTarget) -> RouteTarget | None:
|
|
"""Return a legacy promotion target declared by the catalog."""
|
|
if current.catalog_id is None:
|
|
return None
|
|
destination = CATALOG.promotions.get(current.catalog_id)
|
|
return CATALOG.targets.get(destination) if destination else None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class QuotaProbeSpec:
|
|
command: str
|
|
target: str
|
|
required_caps: tuple[str, ...]
|
|
|
|
|
|
def quota_probe_spec(target: RouteTarget) -> QuotaProbeSpec | None:
|
|
"""Return the driver-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 targets for one explicit stage/lane/grade entry."""
|
|
_validate(stage, lane, grade, evaluated_at)
|
|
lane_id = f"{lane}-G{grade:02d}"
|
|
lane_policy = CATALOG.lanes[stage][lane_id]
|
|
if lane_policy.time_windows is not None:
|
|
time_window = _kst_time_window(evaluated_at)
|
|
rule_id, reason_codes = lane_policy.time_windows[time_window]
|
|
else:
|
|
time_window = "not_applicable"
|
|
assert lane_policy.rule_id is not None
|
|
rule_id, reason_codes = lane_policy.rule_id, lane_policy.reason_codes
|
|
return PolicyDecision(
|
|
rule_id=rule_id,
|
|
policy_priority=lane_policy.policy_priority,
|
|
reason_codes=reason_codes,
|
|
time_window=time_window,
|
|
candidates=tuple(
|
|
CATALOG.targets[target_id] for target_id in lane_policy.candidates
|
|
),
|
|
route_id=f"{stage}:{lane_id}",
|
|
catalog_revision=CATALOG.revision,
|
|
)
|