#!/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, )