"""Pure contracts shared by catalog and Watchtower service orchestration.""" from __future__ import annotations from enum import Enum import math import re from typing import Any class SchedulerState(str, Enum): """Externally observable lifecycle states for the background scheduler.""" STOPPED = "stopped" RUNNING = "running" STOPPING = "stopping" class SchedulerEvent(str, Enum): """Events accepted by the scheduler lifecycle state machine.""" START = "start" REQUEST_STOP = "request_stop" STOP_CONFIRMED = "stop_confirmed" STOP_TIMEOUT = "stop_timeout" _SCHEDULER_TRANSITIONS = { (SchedulerState.STOPPED, SchedulerEvent.START): SchedulerState.RUNNING, (SchedulerState.RUNNING, SchedulerEvent.REQUEST_STOP): SchedulerState.STOPPING, (SchedulerState.STOPPING, SchedulerEvent.STOP_CONFIRMED): SchedulerState.STOPPED, (SchedulerState.STOPPING, SchedulerEvent.STOP_TIMEOUT): SchedulerState.STOPPING, } def scheduler_transition(state: SchedulerState, event: SchedulerEvent) -> SchedulerState: """Return the next scheduler state, rejecting undefined transitions.""" try: return _SCHEDULER_TRANSITIONS[(state, event)] except KeyError as exc: raise ValueError( f"invalid scheduler transition: {state.value} + {event.value}", ) from exc def validate_scheduler_interval(value: object) -> float: """Return a finite positive scheduler interval in seconds.""" if isinstance(value, bool) or not isinstance(value, (int, float)): raise TypeError("scheduler interval must be a number") interval = float(value) if not math.isfinite(interval) or interval <= 0: raise ValueError("scheduler interval must be finite and greater than zero") return interval VALID_WATCHTOWER_STATUSES = frozenset( {"new", "queued", "obliterating", "obliterated", "failed"}, ) def validate_watchtower_status(status: object) -> str: """Return a supported Watchtower status or fail closed.""" if not isinstance(status, str) or status not in VALID_WATCHTOWER_STATUSES: raise ValueError(f"invalid Watchtower status: {status!r}") return status def validate_catalog(payload: object) -> dict[str, Any]: """Validate the stable catalog envelope without over-constraining records.""" if not isinstance(payload, dict): raise ValueError("BESTIARY catalog root must be an object") records = payload.get("models") if not isinstance(records, list): raise ValueError("BESTIARY catalog models must be a list") for index, record in enumerate(records): if not isinstance(record, dict): raise ValueError(f"BESTIARY catalog record {index} must be an object") return payload def normalized_text_list(record: dict[str, Any], field: str) -> tuple[str, ...]: """Return a case-folded list field, rejecting malformed catalog values.""" values = record.get(field, []) if not isinstance(values, list) or not all(isinstance(value, str) for value in values): raise ValueError(f"BESTIARY catalog {field} must be a list of strings") return tuple(value.strip().casefold() for value in values) def validate_day_window(days: object) -> int: """Return a non-negative integer catalog age window.""" if isinstance(days, bool) or not isinstance(days, int): raise TypeError("days must be a non-negative integer") if days < 0: raise ValueError("days must be a non-negative integer") return days def has_open_license(license_id: object, allowed_licenses: frozenset[str]) -> bool: """Match an allow-listed license at an identifier boundary.""" if not isinstance(license_id, str) or not license_id.strip(): return False normalized = license_id.strip().casefold() return any( normalized == allowed or normalized.startswith(f"{allowed}-") for allowed in allowed_licenses ) def is_instruction_tuned( model_id: object, tags: object, keywords: frozenset[str], ) -> bool: """Match instruction indicators as name tokens or exact tags.""" if not isinstance(model_id, str): return False name_tokens = { token for token in re.split(r"[^a-z0-9]+", model_id.casefold()) if token } if name_tokens & keywords: return True if tags is None: return False if not isinstance(tags, list) or not all(isinstance(tag, str) for tag in tags): return False tag_values = {tag.strip().casefold() for tag in tags} return bool(tag_values & keywords) def normalize_watchtower_candidate( candidate: object, ) -> tuple[str, int, int, list[str], str, str] | None: """Normalize one service record or return ``None`` when it is malformed.""" model_id = getattr(candidate, "id", None) or getattr(candidate, "modelId", None) downloads = getattr(candidate, "downloads", None) or 0 likes = getattr(candidate, "likes", None) or 0 tags = getattr(candidate, "tags", None) or [] license_id = getattr(candidate, "license", None) or "" pipeline_tag = getattr(candidate, "pipeline_tag", None) or "" if not isinstance(model_id, str) or not model_id.strip(): return None if isinstance(downloads, bool) or not isinstance(downloads, int) or downloads < 0: return None if isinstance(likes, bool) or not isinstance(likes, int) or likes < 0: return None if not isinstance(tags, list) or not all(isinstance(tag, str) for tag in tags): return None if not isinstance(license_id, str) or not isinstance(pipeline_tag, str): return None if not license_id: for tag in tags: if tag.casefold().startswith("license:"): license_id = tag.split(":", 1)[1].strip() break return model_id, downloads, likes, tags, license_id, pipeline_tag