932 lines
35 KiB
Python
932 lines
35 KiB
Python
#!/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, and failover transitions at the selector
|
|
surface. Quota probe execution and live dispatcher failover integration are
|
|
owned by later tasks.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import importlib.util
|
|
import json
|
|
import re
|
|
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"<!--\s*task=(\S+)\s+plan=(\d+)\s+tag=(\S+)\s*-->")
|
|
_STAGE_BY_KIND = {"PLAN": "worker", "CODE_REVIEW": "review"}
|
|
_VALID_TRANSITIONS = {"initial", "resume", "failover"}
|
|
_VALID_EXECUTION_CLASSES = {"local_model", "cloud_model"}
|
|
_VALID_QUOTA_MODES = {"unbounded", "bounded"}
|
|
_QUALIFIED_FAILOVER_FAILURES = {
|
|
"provider-quota",
|
|
"context-limit",
|
|
"model-unavailable",
|
|
"provider-stream-disconnect",
|
|
}
|
|
|
|
_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 <!-- task=... plan=N tag=... -->",
|
|
)
|
|
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) -> 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")
|
|
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 "source" in value:
|
|
_require_non_empty_string(value, "source", "quota_snapshot", code)
|
|
targets = value.get("targets", [])
|
|
if not isinstance(targets, list):
|
|
raise SelectorInputError(code, "quota_snapshot.targets must be a 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",
|
|
)
|
|
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 _candidate_quota(target, quota_snapshot: dict | None) -> tuple[str, str]:
|
|
if target.execution_class == "local_model":
|
|
return "unbounded", "not_applicable"
|
|
return "bounded", _snapshot_status(target, quota_snapshot)
|
|
|
|
|
|
def _selected_quota(
|
|
selected, quota_snapshot: dict | None, quota_probe_command: str
|
|
) -> dict:
|
|
if selected.execution_class == "local_model":
|
|
return {
|
|
"snapshot_id": None,
|
|
"mode": "unbounded",
|
|
"status": "not_applicable",
|
|
"source": "local_unbounded",
|
|
"checked_at": None,
|
|
}
|
|
if quota_snapshot is None:
|
|
return {
|
|
"snapshot_id": None,
|
|
"mode": "bounded",
|
|
"status": "unknown",
|
|
"source": quota_probe_command,
|
|
"checked_at": 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"),
|
|
}
|
|
|
|
|
|
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
|
|
for rank, target in enumerate(decision.candidates, start=1):
|
|
mode, status = _candidate_quota(target, quota_snapshot)
|
|
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
|
|
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),
|
|
"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)
|
|
|
|
if sel_key not in canon_keys_set:
|
|
raise SelectorInputError(
|
|
code,
|
|
f"prior_decision.selected {sel_key} is not in canonical policy targets {canon_keys_set}",
|
|
)
|
|
|
|
matching_cand = next(
|
|
(c for c in canonical_targets if (c.adapter, c.target) == sel_key), None
|
|
)
|
|
if matching_cand is not None:
|
|
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 candidate for {sel_key}",
|
|
)
|
|
|
|
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 {}),
|
|
"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
|
|
candidates = []
|
|
for index, candidate in enumerate(prior["candidates"]):
|
|
current = dict(candidate)
|
|
if current["execution_class"] != "local_model":
|
|
status = (
|
|
"exhausted"
|
|
if failure_class == "provider-quota" and _target_ref(current) == previous
|
|
else _snapshot_status(type("Target", (), current)(), quota_snapshot)
|
|
)
|
|
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
|
|
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),
|
|
"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 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,
|
|
)
|
|
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())
|