#!/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 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", "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)}" ) selfcheck = item.get("selfcheck_required") if not isinstance(selfcheck, bool): raise CatalogError(f"{path}.selfcheck_required must be a boolean") 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, 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 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, )