#!/usr/bin/env python3 """Deterministic execution-target selector CLI over the pure route policy. The selector consumes a static routing task file (``PLAN-*`` or ``CODE_REVIEW-*``), its ``task/plan/tag`` generation header and optional prior decision / quota snapshot, and returns a stable JSON contract that the dispatcher can persist. This module exposes the schema/invalid-input, worker/review grade matrix, resume, failover, and policy-owned promotion transitions at the selector surface. It also owns shell-less quota probe normalization for live dispatcher routing. """ from __future__ import annotations import argparse import importlib.util import json import re import shlex import subprocess import sys from datetime import datetime from pathlib import Path SCHEMA_VERSION = "1.0" TIMEZONE_NAME = "Asia/Seoul" DEFAULT_QUOTA_PROBE_COMMAND = "iop-node quota-probe" _FILENAME_RE = re.compile(r"^(PLAN|CODE_REVIEW)-(local|cloud)-G(\d{2})\.md$") _HEADER_RE = re.compile(r"") _STAGE_BY_KIND = {"PLAN": "worker", "CODE_REVIEW": "review"} _VALID_TRANSITIONS = {"initial", "resume", "failover", "promotion"} _VALID_EXECUTION_CLASSES = {"local_model", "cloud_model"} _VALID_QUOTA_MODES = {"unbounded", "bounded"} _QUALIFIED_FAILOVER_FAILURES = { "provider-quota", "context-limit", "model-unavailable", "provider-stream-disconnect", } _QUALIFIED_PROMOTION_FAILURES = _QUALIFIED_FAILOVER_FAILURES | { "provider-connection" } _VALID_QUOTA_STATUSES = {"not_applicable", "available", "exhausted", "unknown"} _VALID_ELIGIBILITY = {"eligible", "ineligible"} _VALID_REJECTION_REASONS = {"quota_exhausted"} # Local G07~G08 initial decisions record their KST window; resume preserves it. _VALID_TIME_WINDOWS = { "kst-day-[07:00,23:00)", "kst-night-[23:00,07:00)", "not_applicable", } def _load_policy(): path = Path(__file__).resolve().parent / "execution_target_policy.py" spec = importlib.util.spec_from_file_location("execution_target_policy", path) module = importlib.util.module_from_spec(spec) assert spec.loader is not None sys.modules[spec.name] = module spec.loader.exec_module(module) return module policy = _load_policy() class SelectorInputError(Exception): """Input contract violation returned as stderr JSON with a non-zero exit.""" def __init__(self, code: str, message: str) -> None: super().__init__(message) self.code = code def _parse_filename(task_file: Path) -> tuple[str, str, int]: name = Path(task_file).name match = _FILENAME_RE.match(name) if match is None: raise SelectorInputError( "invalid_task_filename", f"task file must match (PLAN|CODE_REVIEW)-(local|cloud)-GNN.md: {name!r}", ) kind, lane, grade_str = match.group(1), match.group(2), match.group(3) grade = int(grade_str) if not 1 <= grade <= 10: raise SelectorInputError( "invalid_grade", f"grade must be G01..G10: G{grade_str}" ) return kind, lane, grade def _parse_header(task_file: Path) -> tuple[str, int, str]: try: with Path(task_file).open("rb") as handle: head = handle.read(1024) except OSError as exc: raise SelectorInputError("task_file_unreadable", str(exc)) from exc text = head.decode("utf-8", errors="replace") match = _HEADER_RE.search(text) if match is None: raise SelectorInputError( "malformed_header", "first 1KiB must contain ", ) return match.group(1), int(match.group(2)), match.group(3) def _work_unit_id(header: tuple[str, int, str]) -> str: task, plan, tag = header return f"{task}::plan-{plan}::tag-{tag}" def _validate_evaluated_at(evaluated_at: datetime) -> None: if evaluated_at.tzinfo is None or evaluated_at.utcoffset() is None: raise SelectorInputError( "naive_evaluated_at", "evaluated_at must be timezone-aware" ) def _validate_prior_selected(selected: object) -> None: code = "malformed_prior_decision" if not isinstance(selected, dict): raise SelectorInputError(code, "prior_decision.selected must be an object") for field in ("adapter", "target", "execution_class"): candidate = selected.get(field) if not isinstance(candidate, str) or not candidate: raise SelectorInputError( code, f"prior_decision.selected.{field} must be a non-empty string", ) if selected["execution_class"] not in _VALID_EXECUTION_CLASSES: raise SelectorInputError( code, "prior_decision.selected.execution_class must be one of " f"{sorted(_VALID_EXECUTION_CLASSES)}", ) if not isinstance(selected.get("selfcheck_required"), bool): raise SelectorInputError( code, "prior_decision.selected.selfcheck_required must be a boolean", ) def _require_non_empty_string( container: dict, field: str, prefix: str, code: str ) -> None: value = container.get(field) if not isinstance(value, str) or not value: raise SelectorInputError( code, f"{prefix}.{field} must be a non-empty string" ) def _require_string_enum( container: dict, field: str, allowed: set, prefix: str, code: str ) -> None: value = container.get(field) if not isinstance(value, str) or value not in allowed: raise SelectorInputError( code, f"{prefix}.{field} must be one of {sorted(allowed)}" ) def _require_nullable_string( container: dict, field: str, prefix: str, code: str ) -> None: if field not in container: raise SelectorInputError(code, f"{prefix}.{field} is required") value = container[field] if value is not None and not isinstance(value, str): raise SelectorInputError( code, f"{prefix}.{field} must be null or a string" ) def _validate_prior_candidates(candidates: object) -> None: """Validate every reused candidate against the initial output schema. Each entry must carry the full ``_initial`` candidate field set with the correct types and enum values, and ``candidate_rank`` must be 1-based and consecutive so a resume cannot re-emit a partial ranking. """ code = "malformed_prior_decision" if not isinstance(candidates, list) or not candidates: raise SelectorInputError( code, "prior_decision.candidates must be a non-empty list" ) for index, entry in enumerate(candidates): prefix = f"prior_decision.candidates[{index}]" if not isinstance(entry, dict): raise SelectorInputError(code, f"{prefix} must be an object") rank = entry.get("candidate_rank") if ( isinstance(rank, bool) or not isinstance(rank, int) or rank != index + 1 ): raise SelectorInputError( code, f"{prefix}.candidate_rank must be {index + 1} " "(1-based and consecutive)", ) _require_non_empty_string(entry, "adapter", prefix, code) _require_non_empty_string(entry, "target", prefix, code) _require_string_enum( entry, "execution_class", _VALID_EXECUTION_CLASSES, prefix, code ) if not isinstance(entry.get("selfcheck_required"), bool): raise SelectorInputError( code, f"{prefix}.selfcheck_required must be a boolean" ) _require_string_enum(entry, "quota_mode", _VALID_QUOTA_MODES, prefix, code) _require_string_enum( entry, "quota_status", _VALID_QUOTA_STATUSES, prefix, code ) _require_string_enum(entry, "eligibility", _VALID_ELIGIBILITY, prefix, code) if "rejection_reason" not in entry: raise SelectorInputError( code, f"{prefix}.rejection_reason is required" ) rejection = entry["rejection_reason"] if rejection is not None and ( not isinstance(rejection, str) or rejection not in _VALID_REJECTION_REASONS ): raise SelectorInputError( code, f"{prefix}.rejection_reason must be null or one of " f"{sorted(_VALID_REJECTION_REASONS)}", ) def _validate_prior_decision_evidence(decision: object) -> None: """Validate the reused decision evidence block against initial output.""" code = "malformed_prior_decision" prefix = "prior_decision.decision" if not isinstance(decision, dict): raise SelectorInputError(code, f"{prefix} must be an object") _require_non_empty_string(decision, "rule_id", prefix, code) priority = decision.get("policy_priority") if isinstance(priority, bool) or not isinstance(priority, int): raise SelectorInputError( code, f"{prefix}.policy_priority must be an integer" ) reason_codes = decision.get("reason_codes") if not isinstance(reason_codes, list) or not all( isinstance(item, str) and item for item in reason_codes ): raise SelectorInputError( code, f"{prefix}.reason_codes must be a list of non-empty strings" ) _require_non_empty_string(decision, "evaluated_at", prefix, code) if decision.get("timezone") != TIMEZONE_NAME: raise SelectorInputError( code, f"{prefix}.timezone must be {TIMEZONE_NAME!r}" ) _require_string_enum( decision, "time_window", _VALID_TIME_WINDOWS, prefix, code ) if not isinstance(decision.get("pinned"), bool): raise SelectorInputError(code, f"{prefix}.pinned must be a boolean") def _validate_prior_quota(quota: object) -> None: """Validate the reused quota block against the initial output schema.""" code = "malformed_prior_decision" prefix = "prior_decision.quota" if not isinstance(quota, dict): raise SelectorInputError(code, f"{prefix} must be an object") _require_nullable_string(quota, "snapshot_id", prefix, code) _require_string_enum(quota, "mode", _VALID_QUOTA_MODES, prefix, code) _require_string_enum(quota, "status", _VALID_QUOTA_STATUSES, prefix, code) _require_non_empty_string(quota, "source", prefix, code) _require_nullable_string(quota, "checked_at", prefix, code) def _validate_prior_decision(value: object) -> dict: """Validate the nested prior-decision schema before it is reused on resume. Only container/field/type/enum shape is enforced here; identity equality against the current work unit stays in ``_resume``. Unknown extra keys are tolerated so forward-compatible producers are not rejected. """ code = "malformed_prior_decision" if not isinstance(value, dict): raise SelectorInputError(code, "prior_decision must be an object") if value.get("schema_version") != SCHEMA_VERSION: raise SelectorInputError( code, f"prior_decision.schema_version must be {SCHEMA_VERSION!r}" ) required = ( "work_unit_id", "stage", "lane", "grade", "selected", "candidates", "decision", "quota", ) missing = [key for key in required if key not in value] if missing: raise SelectorInputError( code, f"prior_decision missing keys: {missing}" ) if not isinstance(value["work_unit_id"], str): raise SelectorInputError( code, "prior_decision.work_unit_id must be a string" ) stage = value["stage"] if not isinstance(stage, str) or stage not in policy.VALID_STAGES: raise SelectorInputError( code, f"prior_decision.stage must be one of {sorted(policy.VALID_STAGES)}", ) lane = value["lane"] if not isinstance(lane, str) or lane not in policy.VALID_LANES: raise SelectorInputError( code, f"prior_decision.lane must be one of {sorted(policy.VALID_LANES)}", ) grade = value["grade"] if isinstance(grade, bool) or not isinstance(grade, int) or not 1 <= grade <= 10: raise SelectorInputError( code, "prior_decision.grade must be an integer in G01..G10" ) _validate_prior_selected(value["selected"]) _validate_prior_candidates(value["candidates"]) _validate_prior_decision_evidence(value["decision"]) _validate_prior_quota(value["quota"]) return value def _validate_quota_snapshot( value: object | None, *, require_producer_shape: bool = False, ) -> dict | None: """Validate the optional quota snapshot container before it is reflected. Required-cap tri-state normalization and admission stay with ``02+01_quota_input``; here we only reject malformed containers and target entries so ``_snapshot_status`` never dereferences a non-object. """ if value is None: return None code = "malformed_quota_snapshot" if not isinstance(value, dict): raise SelectorInputError(code, "quota_snapshot must be an object") if ( require_producer_shape and value.get("schema_version") != SCHEMA_VERSION ): raise SelectorInputError(code, "quota_snapshot.schema_version must be '1.0'") for field in ("snapshot_id", "checked_at"): if field in value and value[field] is not None and not isinstance( value[field], str ): raise SelectorInputError( code, f"quota_snapshot.{field} must be null or a string" ) if require_producer_shape and ( field not in value or not value[field] ): raise SelectorInputError( code, f"quota_snapshot.{field} must be a non-empty string" ) if "source" in value: _require_non_empty_string(value, "source", "quota_snapshot", code) elif require_producer_shape: raise SelectorInputError( code, "quota_snapshot.source must be a non-empty string" ) targets = value.get("targets", []) if not isinstance(targets, list): raise SelectorInputError(code, "quota_snapshot.targets must be a list") if require_producer_shape and not targets: raise SelectorInputError( code, "quota_snapshot.targets must be a non-empty list" ) for index, entry in enumerate(targets): if not isinstance(entry, dict): raise SelectorInputError( code, f"quota_snapshot.targets[{index}] must be an object" ) for field in ("adapter", "target", "status"): candidate = entry.get(field) if not isinstance(candidate, str) or not candidate: raise SelectorInputError( code, f"quota_snapshot.targets[{index}].{field} must be a " "non-empty string", ) if entry["status"] not in {"available", "exhausted", "unknown"}: raise SelectorInputError( code, f"quota_snapshot.targets[{index}].status must be available, " "exhausted, or unknown", ) if require_producer_shape: caps = value.get("required_caps") if not isinstance(caps, list) or not caps: raise SelectorInputError( code, "quota_snapshot.required_caps must be a non-empty list" ) for index, cap in enumerate(caps): prefix = f"quota_snapshot.required_caps[{index}]" if not isinstance(cap, dict): raise SelectorInputError(code, f"{prefix} must be an object") _require_non_empty_string(cap, "name", prefix, code) _require_string_enum( cap, "status", {"available", "exhausted", "unknown"}, prefix, code, ) remaining = cap.get("remaining_percent") if remaining is not None and ( isinstance(remaining, bool) or not isinstance(remaining, (int, float)) ): raise SelectorInputError( code, f"{prefix}.remaining_percent must be null or numeric" ) reasons = value.get("reason_codes") if not isinstance(reasons, list) or not all( isinstance(reason, str) and reason for reason in reasons ): raise SelectorInputError( code, "quota_snapshot.reason_codes must be a list of non-empty strings", ) return value def _snapshot_status(target, quota_snapshot: dict | None) -> str: """Reflect an already-normalized snapshot status for a cloud target. The required-cap tri-state derivation from the Go usage checker is owned by ``02+01_quota_input``. Here we only surface an injected, pre-normalized status; an absent or unmatched snapshot is reported as ``unknown``. """ if quota_snapshot is None: return "unknown" for entry in quota_snapshot.get("targets", []): if ( entry.get("adapter") == target.adapter and entry.get("target") == target.target ): status = entry.get("status", "unknown") if status in {"available", "exhausted", "unknown"}: return status return "unknown" return "unknown" def probe_candidate_quota( *, target: str, adapter: str, required_caps: tuple[str, ...] | list[str], checked_at: datetime, quota_probe_command: str = DEFAULT_QUOTA_PROBE_COMMAND, probe_command: str | None = None, ) -> dict: """Execute shell-less quota probe command and return a snapshot dict.""" checked_at_iso = checked_at.astimezone(policy.KST).isoformat() try: cmd = shlex.split(quota_probe_command) cmd.extend(["--target", target, "--command", probe_command or adapter]) for cap in required_caps: cmd.extend(["--required-cap", cap]) cmd.extend(["--checked-at", checked_at_iso]) res = subprocess.run(cmd, capture_output=True, text=True, timeout=5) if res.returncode == 0 and res.stdout: snapshot = _validate_quota_snapshot( json.loads(res.stdout), require_producer_shape=True ) assert snapshot is not None if not any( entry["adapter"] == adapter and entry["target"] == target for entry in snapshot["targets"] ): raise SelectorInputError( "malformed_quota_snapshot", "quota snapshot does not contain the requested target identity", ) return snapshot except Exception: pass return { "schema_version": SCHEMA_VERSION, "snapshot_id": None, "source": quota_probe_command, "checked_at": checked_at_iso, "targets": [ {"adapter": adapter, "target": target, "status": "unknown"} ], "required_caps": [ { "name": cap, "status": "unknown", "remaining_percent": None, } for cap in required_caps ], "reason_codes": ["probe_error"], } class QuotaBatchProvider: """Aggregate shell-less quota probes across unique probe keys into a single snapshot.""" def __init__(self, quota_probe_command: str = DEFAULT_QUOTA_PROBE_COMMAND): self.quota_probe_command = quota_probe_command def aggregate( self, *, snapshot_id: str, checked_at: datetime, keys: list | set, ) -> dict | None: if not keys: return None checked_at_iso = checked_at.astimezone(policy.KST).isoformat() targets = [] caps = [] reasons = [] seen_keys = set() unique_keys = [] for key in keys: if len(key) == 3: adapter, target_name, required_caps = key probe_command = adapter elif len(key) >= 4: adapter, target_name, probe_command, required_caps = key[:4] else: continue req_tuple = tuple(required_caps) k = (adapter, target_name, probe_command, req_tuple) if k not in seen_keys: seen_keys.add(k) unique_keys.append((adapter, target_name, probe_command, req_tuple)) for adapter, target_name, probe_command, required_caps in unique_keys: try: child = probe_candidate_quota( target=target_name, adapter=adapter, probe_command=probe_command, required_caps=required_caps, checked_at=checked_at, quota_probe_command=self.quota_probe_command, ) except Exception: child = None if not isinstance(child, dict): child = { "schema_version": SCHEMA_VERSION, "snapshot_id": None, "source": self.quota_probe_command, "checked_at": checked_at_iso, "targets": [ {"adapter": adapter, "target": target_name, "status": "unknown"} ], "required_caps": [ {"name": cap, "status": "unknown", "remaining_percent": None} for cap in required_caps ], "reason_codes": ["probe_error"], } for t in child.get("targets", []): t_entry = { "adapter": t["adapter"], "target": t["target"], "status": t.get("status", "unknown"), } if child.get("snapshot_id"): t_entry["child_snapshot_id"] = child["snapshot_id"] if child.get("reason_codes"): t_entry["child_reason_codes"] = child["reason_codes"] if probe_command: t_entry["command"] = probe_command targets.append(t_entry) for c in child.get("required_caps", []): if c not in caps: caps.append(c) for r in child.get("reason_codes", []): if r not in reasons: reasons.append(r) return { "schema_version": SCHEMA_VERSION, "snapshot_id": snapshot_id, "source": self.quota_probe_command, "checked_at": checked_at_iso, "targets": targets, "required_caps": caps or [ {"name": "overall", "status": "available", "remaining_percent": None} ], "reason_codes": reasons or ["batch_probe"], } quota_provider = QuotaBatchProvider() def derive_work_unit_quota_evidence( prior_decision: dict | None, *, status: str = "exhausted", reason: str = "confirmed_runtime_provider_quota", ) -> dict: """Derive task-local quota evidence inheriting observation identity from a prior decision.""" if not isinstance(prior_decision, dict): return { "schema_version": SCHEMA_VERSION, "snapshot_id": None, "source": DEFAULT_QUOTA_PROBE_COMMAND, "checked_at": datetime.now(policy.KST).isoformat(), "targets": [], "required_caps": [], "reason_codes": [reason], } selected = prior_decision.get("selected") adapter = selected.get("adapter") if isinstance(selected, dict) else None target_name = selected.get("target") if isinstance(selected, dict) else None prior_quota = prior_decision.get("quota") if not isinstance(prior_quota, dict): prior_quota = {} snapshot_id = prior_quota.get("snapshot_id") checked_at = prior_quota.get("checked_at") or datetime.now(policy.KST).isoformat() source = prior_quota.get("source", DEFAULT_QUOTA_PROBE_COMMAND) targets = [] found = False for t_entry in prior_quota.get("targets", []): if isinstance(t_entry, dict): new_entry = dict(t_entry) if ( adapter and target_name and t_entry.get("adapter") == adapter and t_entry.get("target") == target_name ): new_entry["status"] = status found = True targets.append(new_entry) if not found and adapter and target_name: targets.append( { "adapter": adapter, "target": target_name, "status": status, } ) reason_codes = list(prior_quota.get("reason_codes", [])) if reason not in reason_codes: reason_codes.append(reason) return { "schema_version": prior_quota.get("schema_version", SCHEMA_VERSION), "snapshot_id": snapshot_id, "source": source, "checked_at": checked_at, "targets": targets, "required_caps": list(prior_quota.get("required_caps", [])), "reason_codes": reason_codes, } def _candidate_quota( target, quota_snapshot: dict | None, evaluated_at: datetime, quota_probe_command: str, ) -> tuple[str, str, dict | None]: if target.execution_class == "local_model": return "unbounded", "not_applicable", None if quota_snapshot is not None: return "bounded", _snapshot_status(target, quota_snapshot), quota_snapshot probe_spec = policy.quota_probe_spec(target) if probe_spec is not None: snapshot = probe_candidate_quota( target=target.target, adapter=target.adapter, required_caps=probe_spec.required_caps, checked_at=evaluated_at, quota_probe_command=quota_probe_command, ) return "bounded", _snapshot_status(target, snapshot), snapshot return "bounded", "unknown", None def _selected_quota( selected, quota_snapshot: dict | None, quota_probe_command: str, probed_snapshot: dict | None = None, ) -> dict: if selected.execution_class == "local_model": return { "snapshot_id": None, "mode": "unbounded", "status": "not_applicable", "source": "local_unbounded", "checked_at": None, "targets": [], } if quota_snapshot is not None: return { "snapshot_id": quota_snapshot.get("snapshot_id"), "mode": "bounded", "status": _snapshot_status(selected, quota_snapshot), "source": quota_snapshot.get("source", quota_probe_command), "checked_at": quota_snapshot.get("checked_at"), "targets": [ dict(entry) for entry in quota_snapshot.get("targets", []) ], } if probed_snapshot is not None: return { "snapshot_id": probed_snapshot.get("snapshot_id"), "mode": "bounded", "status": _snapshot_status(selected, probed_snapshot), "source": probed_snapshot.get("source", quota_probe_command), "checked_at": probed_snapshot.get("checked_at"), "targets": [ dict(entry) for entry in probed_snapshot.get("targets", []) ], } return { "snapshot_id": None, "mode": "bounded", "status": "unknown", "source": quota_probe_command, "checked_at": None, "targets": [], } def _initial( *, work_unit_id: str, stage: str, lane: str, grade: int, evaluated_at: datetime, quota_snapshot: dict | None, quota_probe_command: str, ) -> dict: try: decision = policy.select_policy( stage=stage, lane=lane, grade=grade, evaluated_at=evaluated_at ) except ValueError as exc: raise SelectorInputError("invalid_route", str(exc)) from exc candidates = [] selected = None selected_probed_snapshot = None for rank, target in enumerate(decision.candidates, start=1): mode, status, probed_snapshot = _candidate_quota( target, quota_snapshot, evaluated_at, quota_probe_command ) eligible = status != "exhausted" candidates.append( { "candidate_rank": rank, "adapter": target.adapter, "target": target.target, "execution_class": target.execution_class, "selfcheck_required": target.selfcheck_required, "quota_mode": mode, "quota_status": status, "eligibility": "eligible" if eligible else "ineligible", "rejection_reason": None if eligible else "quota_exhausted", } ) if eligible and selected is None: selected = target selected_probed_snapshot = probed_snapshot if selected is None: raise SelectorInputError( "no_eligible_target", "all policy candidates are exhausted according to the quota snapshot", ) return { "schema_version": SCHEMA_VERSION, "work_unit_id": work_unit_id, "stage": stage, "lane": lane, "grade": grade, "selected": { "adapter": selected.adapter, "target": selected.target, "execution_class": selected.execution_class, "selfcheck_required": selected.selfcheck_required, }, "candidates": candidates, "decision": { "rule_id": decision.rule_id, "policy_priority": decision.policy_priority, "reason_codes": list(decision.reason_codes), "evaluated_at": evaluated_at.astimezone(policy.KST).isoformat(), "timezone": TIMEZONE_NAME, "time_window": decision.time_window, "pinned": False, }, "quota": _selected_quota( selected, quota_snapshot, quota_probe_command, selected_probed_snapshot ), "transition": { "previous_target": None, "next_target": None, "trigger": "initial", "context_transfer": "none", }, } def _validate_selected_and_used_history( prior_decision: dict, canonical_targets: list, ) -> None: code = "malformed_prior_decision" selected = prior_decision.get("selected") if not isinstance(selected, dict): raise SelectorInputError(code, "prior_decision.selected must be an object") sel_key = (selected.get("adapter"), selected.get("target")) canon_keys_list = [(c.adapter, c.target) for c in canonical_targets] canon_keys_set = set(canon_keys_list) matching_cand = policy.canonical_target(*sel_key) if matching_cand is None: raise SelectorInputError( code, f"prior_decision.selected {sel_key} is not a policy-owned target", ) if ( selected.get("execution_class") != matching_cand.execution_class or selected.get("selfcheck_required") != matching_cand.selfcheck_required ): raise SelectorInputError( code, f"prior_decision.selected attributes do not match canonical target for {sel_key}", ) if sel_key not in canon_keys_set: promotion_path = prior_decision.get("promotion_path") if not isinstance(promotion_path, list) or len(promotion_path) < 2: raise SelectorInputError( code, "promoted prior_decision requires promotion_path evidence", ) path_targets = [] for index, entry in enumerate(promotion_path): if not isinstance(entry, dict): raise SelectorInputError( code, f"promotion_path[{index}] must be an object" ) target = policy.canonical_target( entry.get("adapter"), entry.get("target") ) if target is None: raise SelectorInputError( code, f"promotion_path[{index}] is not policy-owned" ) path_targets.append(target) if (path_targets[0].adapter, path_targets[0].target) not in canon_keys_set: raise SelectorInputError( code, "promotion_path must begin at the initial policy target" ) for previous, current in zip(path_targets, path_targets[1:]): if policy.promotion_target(previous) != current: raise SelectorInputError( code, "promotion_path contains a non-adjacent transition" ) if path_targets[-1] != matching_cand: raise SelectorInputError( code, "promotion_path tail does not match selected target" ) if "used_candidates" in prior_decision: raise SelectorInputError( code, "promotion decision must not carry failover used_candidates" ) return if "used_candidates" in prior_decision: used = prior_decision["used_candidates"] if not isinstance(used, list): raise SelectorInputError(code, "prior_decision.used_candidates must be a list") used_keys = [] for idx, entry in enumerate(used): if not isinstance(entry, dict): raise SelectorInputError( code, f"prior_decision.used_candidates[{idx}] must be an object" ) u_key = (entry.get("adapter"), entry.get("target")) if u_key not in canon_keys_set: raise SelectorInputError( code, f"prior_decision.used_candidates[{idx}] {u_key} is not in canonical policy targets {canon_keys_set}", ) used_keys.append(u_key) if len(used_keys) != len(set(used_keys)): raise SelectorInputError( code, "prior_decision.used_candidates contains duplicate targets" ) indices = [canon_keys_list.index(k) for k in used_keys] if indices != sorted(indices): raise SelectorInputError( code, "prior_decision.used_candidates order does not match candidate rank order" ) if used_keys and sel_key != used_keys[-1]: raise SelectorInputError( code, f"prior_decision.selected {sel_key} does not match tail of used_candidates {used_keys[-1]}", ) else: prior_cands = prior_decision.get("candidates", []) eligible_cands = [ (c.get("adapter"), c.get("target")) for c in prior_cands if isinstance(c, dict) and c.get("eligibility") == "eligible" ] if eligible_cands and sel_key != eligible_cands[0]: raise SelectorInputError( code, f"prior_decision.selected {sel_key} does not match first eligible candidate {eligible_cands[0]} when used_candidates is absent", ) def _validate_prior_candidate_identity( prior_decision: dict, *, stage: str, lane: str, grade: int, ) -> None: code = "malformed_prior_decision" decision_info = prior_decision.get("decision") if not isinstance(decision_info, dict): raise SelectorInputError(code, "prior_decision.decision must be an object") eval_str = decision_info.get("evaluated_at") if not isinstance(eval_str, str): raise SelectorInputError(code, "prior_decision.decision.evaluated_at must be a string") try: prior_eval_at = datetime.fromisoformat(eval_str) except (ValueError, TypeError) as exc: raise SelectorInputError( code, f"prior_decision.decision.evaluated_at is not a valid ISO datetime: {eval_str!r}" ) from exc if prior_eval_at.tzinfo is None or prior_eval_at.utcoffset() is None: raise SelectorInputError( code, f"prior_decision.decision.evaluated_at must be timezone-aware: {eval_str!r}" ) try: canonical_decision = policy.select_policy( stage=stage, lane=lane, grade=grade, evaluated_at=prior_eval_at ) except ValueError as exc: raise SelectorInputError(code, str(exc)) from exc if decision_info.get("rule_id") != canonical_decision.rule_id: raise SelectorInputError( code, f"prior_decision.decision.rule_id ({decision_info.get('rule_id')!r}) " f"does not match canonical policy ({canonical_decision.rule_id!r})", ) if decision_info.get("policy_priority") != canonical_decision.policy_priority: raise SelectorInputError( code, f"prior_decision.decision.policy_priority ({decision_info.get('policy_priority')!r}) " f"does not match canonical policy ({canonical_decision.policy_priority!r})", ) if list(decision_info.get("reason_codes", [])) != list(canonical_decision.reason_codes): raise SelectorInputError( code, f"prior_decision.decision.reason_codes ({decision_info.get('reason_codes')!r}) " f"does not match canonical policy ({list(canonical_decision.reason_codes)!r})", ) if decision_info.get("time_window") != canonical_decision.time_window: raise SelectorInputError( code, f"prior_decision.decision.time_window ({decision_info.get('time_window')!r}) " f"does not match canonical policy ({canonical_decision.time_window!r})", ) canonical_targets = canonical_decision.candidates prior_candidates = prior_decision.get("candidates") if not isinstance(prior_candidates, list) or len(prior_candidates) != len(canonical_targets): raise SelectorInputError( code, f"prior_decision.candidates length ({len(prior_candidates) if isinstance(prior_candidates, list) else 0}) " f"does not match canonical policy candidates length ({len(canonical_targets)})", ) for idx, (p_cand, c_target) in enumerate(zip(prior_candidates, canonical_targets)): if not isinstance(p_cand, dict): raise SelectorInputError(code, f"prior_decision.candidates[{idx}] must be an object") if ( p_cand.get("adapter") != c_target.adapter or p_cand.get("target") != c_target.target or p_cand.get("execution_class") != c_target.execution_class or p_cand.get("selfcheck_required") != c_target.selfcheck_required ): raise SelectorInputError( code, f"prior_decision.candidates[{idx}] identity ({p_cand.get('adapter')}, {p_cand.get('target')}) " f"does not match canonical policy candidate ({c_target.adapter}, {c_target.target})", ) _validate_selected_and_used_history(prior_decision, canonical_targets) def _resume( prior_decision: dict | None, *, work_unit_id: str, stage: str, lane: str, grade: int, ) -> dict: if prior_decision is None: raise SelectorInputError( "resume_requires_prior_decision", "resume transition requires prior_decision", ) prior_decision = _validate_prior_decision(prior_decision) for key, expected in ( ("work_unit_id", work_unit_id), ("stage", stage), ("lane", lane), ("grade", grade), ): if prior_decision[key] != expected: raise SelectorInputError( "resume_work_unit_mismatch", f"prior_decision {key}={prior_decision[key]!r} != {expected!r}", ) _validate_prior_candidate_identity(prior_decision, stage=stage, lane=lane, grade=grade) selected = prior_decision["selected"] decision = dict(prior_decision["decision"]) decision["pinned"] = True target_ref = {"adapter": selected["adapter"], "target": selected["target"]} return { "schema_version": SCHEMA_VERSION, "work_unit_id": work_unit_id, "stage": stage, "lane": lane, "grade": grade, "selected": selected, "candidates": prior_decision["candidates"], "decision": decision, "quota": prior_decision["quota"], **({"used_candidates": _validate_used_candidates(prior_decision.get("used_candidates"))} if "used_candidates" in prior_decision else {}), **({"promotion_path": prior_decision["promotion_path"]} if "promotion_path" in prior_decision else {}), "transition": { "previous_target": target_ref, "next_target": dict(target_ref), "trigger": "resume", "context_transfer": "none", }, } def _target_ref(candidate: dict) -> dict: return {"adapter": candidate["adapter"], "target": candidate["target"]} def _validate_used_candidates(value: object) -> list[dict]: if value is None: return [] if not isinstance(value, list): raise SelectorInputError("malformed_prior_decision", "used_candidates must be a list") refs = [] for index, entry in enumerate(value): if not isinstance(entry, dict): raise SelectorInputError("malformed_prior_decision", f"used_candidates[{index}] must be an object") adapter, target = entry.get("adapter"), entry.get("target") if not isinstance(adapter, str) or not adapter or not isinstance(target, str) or not target: raise SelectorInputError("malformed_prior_decision", f"used_candidates[{index}] needs adapter and target") refs.append({"adapter": adapter, "target": target}) return refs def _failover( prior_decision: dict | None, *, work_unit_id: str, stage: str, lane: str, grade: int, evaluated_at: datetime, quota_snapshot: dict | None, quota_probe_command: str, failure_class: str | None, ) -> dict: if failure_class not in _QUALIFIED_FAILOVER_FAILURES: raise SelectorInputError( "unqualified_failover_trigger", f"failover requires one of {sorted(_QUALIFIED_FAILOVER_FAILURES)}", ) if prior_decision is None: raise SelectorInputError("failover_requires_prior_decision", "failover transition requires prior_decision") prior = _validate_prior_decision(prior_decision) for key, expected in (("work_unit_id", work_unit_id), ("stage", stage), ("lane", lane), ("grade", grade)): if prior[key] != expected: raise SelectorInputError("failover_work_unit_mismatch", f"prior_decision {key}={prior[key]!r} != {expected!r}") _validate_prior_candidate_identity(prior, stage=stage, lane=lane, grade=grade) previous = _target_ref(prior["selected"]) previous_index = next(index for index, candidate in enumerate(prior["candidates"]) if _target_ref(candidate) == previous) used = _validate_used_candidates(prior.get("used_candidates")) if previous not in used: used.append(previous) used_set = {(entry["adapter"], entry["target"]) for entry in used} selected_candidate = None selected_probed_snapshot = None candidates = [] for index, candidate in enumerate(prior["candidates"]): current = dict(candidate) current_snapshot = None if current["execution_class"] != "local_model": if failure_class == "provider-quota" and _target_ref(current) == previous: status = "exhausted" elif quota_snapshot is not None: current_snapshot = quota_snapshot status = _snapshot_status(type("Target", (), current)(), quota_snapshot) else: cand_obj = type("Target", (), current)() probe_spec = policy.quota_probe_spec(cand_obj) if probe_spec is not None: sn = probe_candidate_quota( target=current["target"], adapter=current["adapter"], required_caps=probe_spec.required_caps, checked_at=evaluated_at, quota_probe_command=quota_probe_command, ) current_snapshot = sn status = _snapshot_status(cand_obj, sn) else: status = "unknown" current["quota_status"] = status current["eligibility"] = "ineligible" if status == "exhausted" else "eligible" current["rejection_reason"] = "quota_exhausted" if status == "exhausted" else None candidates.append(current) key = (current["adapter"], current["target"]) if index > previous_index and key not in used_set and current["eligibility"] == "eligible" and selected_candidate is None: selected_candidate = current selected_probed_snapshot = current_snapshot if selected_candidate is None: raise SelectorInputError("no_failover_candidate", "no unused eligible candidate remains for this work unit") selected = {field: selected_candidate[field] for field in ("adapter", "target", "execution_class", "selfcheck_required")} next_target = _target_ref(selected) used.append(next_target) decision = dict(prior["decision"]) decision["pinned"] = True selected_target = type("Target", (), selected)() return { "schema_version": SCHEMA_VERSION, "work_unit_id": work_unit_id, "stage": stage, "lane": lane, "grade": grade, "selected": selected, "candidates": candidates, "decision": decision, "quota": _selected_quota( selected_target, quota_snapshot, quota_probe_command, selected_probed_snapshot, ), "used_candidates": used, "transition": { "previous_target": previous, "next_target": next_target, "trigger": failure_class, "context_transfer": "logical", "evaluated_at": evaluated_at.astimezone(policy.KST).isoformat(), }, } def _promotion( prior_decision: dict | None, *, work_unit_id: str, stage: str, lane: str, grade: int, evaluated_at: datetime, quota_snapshot: dict | None, quota_probe_command: str, failure_class: str | None, ) -> dict: if failure_class not in _QUALIFIED_PROMOTION_FAILURES: raise SelectorInputError( "unqualified_promotion_trigger", "promotion requires one of " f"{sorted(_QUALIFIED_PROMOTION_FAILURES)}", ) if prior_decision is None: raise SelectorInputError( "promotion_requires_prior_decision", "promotion transition requires prior_decision", ) prior = _validate_prior_decision(prior_decision) for key, expected in ( ("work_unit_id", work_unit_id), ("stage", stage), ("lane", lane), ("grade", grade), ): if prior[key] != expected: raise SelectorInputError( "promotion_work_unit_mismatch", f"prior_decision {key}={prior[key]!r} != {expected!r}", ) _validate_prior_candidate_identity( prior, stage=stage, lane=lane, grade=grade ) if len(prior["candidates"]) != 1: raise SelectorInputError( "no_promotion_target", "multi-candidate policy routes use failover instead of promotion", ) current = policy.canonical_target( prior["selected"]["adapter"], prior["selected"]["target"] ) promoted = policy.promotion_target(current) if current is not None else None if promoted is None: raise SelectorInputError( "no_promotion_target", "no unused canonical promotion target remains for this work unit", ) previous_target = {"adapter": current.adapter, "target": current.target} next_target = {"adapter": promoted.adapter, "target": promoted.target} promotion_path = list(prior.get("promotion_path", [previous_target])) if not promotion_path or promotion_path[-1] != previous_target: raise SelectorInputError( "malformed_prior_decision", "promotion_path tail does not match the selected target", ) promotion_path.append(next_target) decision = dict(prior["decision"]) decision["pinned"] = True return { "schema_version": SCHEMA_VERSION, "work_unit_id": work_unit_id, "stage": stage, "lane": lane, "grade": grade, "selected": { "adapter": promoted.adapter, "target": promoted.target, "execution_class": promoted.execution_class, "selfcheck_required": promoted.selfcheck_required, }, "candidates": prior["candidates"], "decision": decision, "promotion_path": promotion_path, "quota": _selected_quota( promoted, quota_snapshot, quota_probe_command ), "transition": { "kind": "promotion", "previous_target": previous_target, "next_target": next_target, "trigger": failure_class, "context_transfer": "logical", "evaluated_at": evaluated_at.astimezone(policy.KST).isoformat(), }, } def select_execution_target( task_file: Path, *, stage: str | None = None, evaluated_at: datetime, transition: str = "initial", prior_decision: dict | None = None, quota_snapshot: dict | None = None, quota_probe_command: str = DEFAULT_QUOTA_PROBE_COMMAND, failure_class: str | None = None, ) -> dict: """Return the stable JSON-serializable selector decision for one call.""" kind, lane, grade = _parse_filename(task_file) prefix_stage = _STAGE_BY_KIND[kind] if stage is not None and stage != prefix_stage: raise SelectorInputError( "stage_mismatch", f"explicit stage {stage!r} conflicts with filename stage {prefix_stage!r}", ) resolved_stage = stage or prefix_stage header = _parse_header(task_file) work_unit_id = _work_unit_id(header) _validate_evaluated_at(evaluated_at) quota_snapshot = _validate_quota_snapshot(quota_snapshot) if not isinstance(quota_probe_command, str) or not quota_probe_command: raise SelectorInputError( "invalid_quota_probe_command", "quota_probe_command must be a non-empty string", ) if transition == "initial": return _initial( work_unit_id=work_unit_id, stage=resolved_stage, lane=lane, grade=grade, evaluated_at=evaluated_at, quota_snapshot=quota_snapshot, quota_probe_command=quota_probe_command, ) if transition == "resume": return _resume( prior_decision, work_unit_id=work_unit_id, stage=resolved_stage, lane=lane, grade=grade, ) if transition == "failover": return _failover( prior_decision, work_unit_id=work_unit_id, stage=resolved_stage, lane=lane, grade=grade, evaluated_at=evaluated_at, quota_snapshot=quota_snapshot, quota_probe_command=quota_probe_command, failure_class=failure_class, ) if transition == "promotion": return _promotion( prior_decision, work_unit_id=work_unit_id, stage=resolved_stage, lane=lane, grade=grade, evaluated_at=evaluated_at, quota_snapshot=quota_snapshot, quota_probe_command=quota_probe_command, failure_class=failure_class, ) raise SelectorInputError( "invalid_transition", f"unknown transition: {transition!r}" ) def to_json(payload: dict) -> str: """Serialize a decision to byte-stable JSON (sorted keys, fixed indent).""" return json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) def _load_json_arg(value: str | None): if value is None: return None candidate = Path(value) if candidate.exists(): text = candidate.read_text(encoding="utf-8") else: text = value return json.loads(text) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Select the deterministic execution target for a task file.", ) parser.add_argument("task_file", type=Path) parser.add_argument("--stage", choices=["worker", "review"]) parser.add_argument("--evaluated-at") parser.add_argument( "--transition", default="initial", choices=sorted(_VALID_TRANSITIONS) ) parser.add_argument("--prior-decision") parser.add_argument("--quota-snapshot") parser.add_argument("--failure-class") parser.add_argument( "--quota-probe-command", default=DEFAULT_QUOTA_PROBE_COMMAND ) args = parser.parse_args(argv) try: if args.evaluated_at is not None: evaluated_at = datetime.fromisoformat(args.evaluated_at) else: evaluated_at = datetime.now(policy.KST) payload = select_execution_target( args.task_file, stage=args.stage, evaluated_at=evaluated_at, transition=args.transition, prior_decision=_load_json_arg(args.prior_decision), quota_snapshot=_load_json_arg(args.quota_snapshot), quota_probe_command=args.quota_probe_command, failure_class=args.failure_class, ) except SelectorInputError as exc: json.dump({"error": exc.code, "message": str(exc)}, sys.stderr) sys.stderr.write("\n") return 2 except (ValueError, OSError, json.JSONDecodeError) as exc: json.dump({"error": "input_error", "message": str(exc)}, sys.stderr) sys.stderr.write("\n") return 2 sys.stdout.write(to_json(payload) + "\n") return 0 if __name__ == "__main__": raise SystemExit(main())