From a377ceefbdc713be8f26cb25147d7c631a833f84 Mon Sep 17 00:00:00 2001 From: Joseph Magly <1159087+jmagly@users.noreply.github.com> Date: Sun, 16 Aug 2026 01:29:27 -0400 Subject: [PATCH] test: add service orchestration contracts --- ci/test-risk-map.json | 17 +- obliteratus/bestiary_sync.py | 63 ++- obliteratus/models_client.py | 101 ++++- obliteratus/service_contracts.py | 157 +++++++ obliteratus/watchtower.py | 267 ++++++++---- pyproject.toml | 5 + scripts/run_repeat_gate.py | 3 + tests/test_bestiary_contracts.py | 120 ++++++ tests/test_models_client_contracts.py | 290 +++++++++++++ tests/test_watchtower_contracts.py | 572 ++++++++++++++++++++++++++ 10 files changed, 1464 insertions(+), 131 deletions(-) create mode 100644 obliteratus/service_contracts.py create mode 100644 tests/test_bestiary_contracts.py create mode 100644 tests/test_models_client_contracts.py create mode 100644 tests/test_watchtower_contracts.py diff --git a/ci/test-risk-map.json b/ci/test-risk-map.json index 91fc588..d34ec30 100644 --- a/ci/test-risk-map.json +++ b/ci/test-risk-map.json @@ -104,6 +104,7 @@ "obliteratus/models_client.py", "obliteratus/remote.py", "obliteratus/remote_contracts.py", + "obliteratus/service_contracts.py", "obliteratus/ui_watchtower.py", "obliteratus/watchtower.py" ], @@ -111,8 +112,11 @@ "tests/test_cli.py", "tests/test_cli_boundaries.py", "tests/test_local_ui_portability.py", + "tests/test_bestiary_contracts.py", + "tests/test_models_client_contracts.py", "tests/test_remote_boundaries.py", "tests/test_remote_contracts.py", + "tests/test_watchtower_contracts.py", "tests/conditional/test_network_services.py", "tests/conditional/test_operator_ui.py", "tests/conditional/test_remote_runtime.py" @@ -471,23 +475,30 @@ "path": "obliteratus/bestiary_sync.py", "risk_class": "conditional-runtime", "risk": "external catalog synchronization and malformed service responses", - "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_network_services.py"], + "required_tests": ["tests/test_bestiary_contracts.py", "tests/test_module_imports.py", "tests/conditional/test_network_services.py"], "conditional_gates": ["network-services"] }, { "path": "obliteratus/models_client.py", "risk_class": "conditional-runtime", "risk": "catalog resolution across operator files and network services", - "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_network_services.py"], + "required_tests": ["tests/test_models_client_contracts.py", "tests/test_module_imports.py", "tests/conditional/test_network_services.py"], "conditional_gates": ["network-services"] }, { "path": "obliteratus/watchtower.py", "risk_class": "conditional-runtime", "risk": "scheduled scans, queue state, and live service responses", - "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_network_services.py"], + "required_tests": ["tests/test_watchtower_contracts.py", "tests/test_module_imports.py", "tests/conditional/test_network_services.py"], "conditional_gates": ["network-services"] }, + { + "path": "obliteratus/service_contracts.py", + "risk_class": "cpu-contract", + "risk": "catalog validation, external record normalization, and scheduler lifecycle decisions", + "required_tests": ["tests/test_models_client_contracts.py", "tests/test_watchtower_contracts.py"], + "conditional_gates": [] + }, { "path": "obliteratus/interactive.py", "risk_class": "conditional-runtime", diff --git a/obliteratus/bestiary_sync.py b/obliteratus/bestiary_sync.py index 8db97b9..87c84b8 100644 --- a/obliteratus/bestiary_sync.py +++ b/obliteratus/bestiary_sync.py @@ -60,25 +60,44 @@ def extra_presets(ModelPreset, existing_hf_ids): except Exception: return [] - out = [] - seen = set(existing_hf_ids) - for m in catalog: - hf = m.get("hf_id") - if not hf or hf in seen: - continue - seen.add(hf) - tier, params = _infer_tier_params(hf, m) - caps = ", ".join(m.get("capabilities", [])) or "open-weight" - rel = m.get("released") or "?" - org = hf.split("/")[0] if "/" in hf else "" - out.append(ModelPreset( - name=m.get("name") or hf.split("/")[-1], - hf_id=hf, - description=f"[BESTIARY · {rel}] {caps}.", - tier=tier, - params=params, - recommended_dtype="bfloat16", - recommended_quantization=("4bit" if tier in ("large", "frontier") else None), - gated=(org in _GATED_ORGS), - )) - return out + try: + out = [] + seen = set(existing_hf_ids) + for m in catalog: + if not isinstance(m, dict): + raise ValueError("BESTIARY preset record must be an object") + hf = m.get("hf_id") + if not isinstance(hf, str) or not hf.strip(): + raise ValueError("BESTIARY preset record requires a string hf_id") + if hf in seen: + continue + capabilities = m.get("capabilities", []) + if not isinstance(capabilities, list) or not all( + isinstance(capability, str) for capability in capabilities + ): + raise ValueError("BESTIARY capabilities must be a list of strings") + name = m.get("name") + if name is not None and not isinstance(name, str): + raise ValueError("BESTIARY model name must be a string") + released = m.get("released") + if released is not None and not isinstance(released, str): + raise ValueError("BESTIARY release date must be a string") + + seen.add(hf) + tier, params = _infer_tier_params(hf, m) + caps = ", ".join(capabilities) or "open-weight" + rel = released or "?" + org = hf.split("/")[0] if "/" in hf else "" + out.append(ModelPreset( + name=name or hf.split("/")[-1], + hf_id=hf, + description=f"[BESTIARY · {rel}] {caps}.", + tier=tier, + params=params, + recommended_dtype="bfloat16", + recommended_quantization=("4bit" if tier in ("large", "frontier") else None), + gated=(org.casefold() in _GATED_ORGS), + )) + return out + except Exception: + return [] diff --git a/obliteratus/models_client.py b/obliteratus/models_client.py index 02ca204..eb7d4fc 100644 --- a/obliteratus/models_client.py +++ b/obliteratus/models_client.py @@ -32,72 +32,133 @@ from __future__ import annotations import json import os import urllib.request -from datetime import datetime, timezone, timedelta +from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import List, Optional +from typing import Any + +from .service_contracts import ( + normalized_text_list, + validate_catalog, + validate_day_window, +) _HERE = Path(__file__).resolve().parent _CANONICAL = _HERE.parent / "bestiary" / "state" / "catalog.json" _LOCAL = _HERE / "catalog.json" _CHANNEL_ALIASES = {"api": "openrouter", "hosted": "openrouter", "frontier": "openrouter", "hf": "huggingface"} +_CATALOG_TIMEOUT_SECONDS = 15 +_MAX_CATALOG_BYTES = 8 * 1024 * 1024 -def _load() -> dict: +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _decode_catalog(payload: bytes, source: str) -> dict[str, Any]: + if len(payload) > _MAX_CATALOG_BYTES: + raise ValueError(f"BESTIARY catalog exceeds size limit: {source}") + try: + decoded = payload.decode("utf-8") + parsed = json.loads(decoded) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ValueError(f"BESTIARY catalog is not valid UTF-8 JSON: {source}") from exc + return validate_catalog(parsed) + + +def _read_catalog_path(path: Path) -> dict[str, Any]: + try: + size = path.stat().st_size + except FileNotFoundError: + raise FileNotFoundError(f"BESTIARY catalog not found: {path}") from None + if size > _MAX_CATALOG_BYTES: + raise ValueError(f"BESTIARY catalog exceeds size limit: {path}") + return _decode_catalog(path.read_bytes(), str(path)) + + +def _load() -> dict[str, Any]: src = os.environ.get("BESTIARY_CATALOG", "").strip() if src.startswith("http://") or src.startswith("https://"): - with urllib.request.urlopen(src, timeout=15) as r: - return json.loads(r.read().decode("utf-8")) - for p in (Path(src) if src else None, _LOCAL, _CANONICAL, _HERE / "state" / "catalog.json"): - if p and p.exists(): - return json.loads(p.read_text()) + with urllib.request.urlopen(src, timeout=_CATALOG_TIMEOUT_SECONDS) as response: + return _decode_catalog(response.read(_MAX_CATALOG_BYTES + 1), src) + if src: + return _read_catalog_path(Path(src)) + for path in (_LOCAL, _CANONICAL, _HERE / "state" / "catalog.json"): + if path.exists(): + return _read_catalog_path(path) raise FileNotFoundError( "BESTIARY catalog not found. Run `python3 bestiary.py update`, or set " "$BESTIARY_CATALOG to a catalog.json path or URL." ) -def models(channel: Optional[str] = None, vendor: Optional[str] = None, - open_weight: Optional[bool] = None, capability: Optional[str] = None) -> List[dict]: +def _normalize_filter(value: str | None, name: str) -> str | None: + if value is None: + return None + if not isinstance(value, str): + raise TypeError(f"{name} must be a string") + normalized = value.strip().casefold() + return normalized or None + + +def models(channel: str | None = None, vendor: str | None = None, + open_weight: bool | None = None, capability: str | None = None) -> list[dict]: """Return catalog records filtered by channel / vendor / open_weight / capability.""" + channel = _normalize_filter(channel, "channel") + vendor = _normalize_filter(vendor, "vendor") + capability = _normalize_filter(capability, "capability") if channel in ("open-weight", "openweight", "weights"): open_weight, channel = True, None if channel: channel = _CHANNEL_ALIASES.get(channel, channel) out = [] for m in _load().get("models", []): - if channel and channel not in m.get("channels", []): + channels = normalized_text_list(m, "channels") + capabilities = normalized_text_list(m, "capabilities") + record_vendor = m.get("vendor") + if record_vendor is not None and not isinstance(record_vendor, str): + raise ValueError("BESTIARY catalog vendor must be a string") + if channel and channel not in channels: continue - if vendor and m.get("vendor") != vendor.lower(): + if vendor and (record_vendor or "").strip().casefold() != vendor: continue if open_weight is not None and bool(m.get("open_weight")) != open_weight: continue - if capability and capability not in m.get("capabilities", []): + if capability and capability not in capabilities: continue out.append(m) return out -def model_ids(**kw) -> List[str]: +def model_ids(**kw) -> list[str]: """Just the ids — the common case for populating a dropdown or a config.""" - return [m["id"] for m in models(**kw)] + ids = [] + for record in models(**kw): + model_id = record.get("id") + if not isinstance(model_id, str) or not model_id.strip(): + raise ValueError("BESTIARY catalog model requires a non-empty string id") + ids.append(model_id) + return ids -def newest(days: int = 7, **kw) -> List[dict]: +def newest(days: int = 7, **kw) -> list[dict]: """Models first-seen (≈ released) within the last `days`, newest first.""" - cutoff = datetime.now(timezone.utc) - timedelta(days=days) - out = [] + cutoff = _utc_now() - timedelta(days=validate_day_window(days)) + out: list[tuple[datetime, dict]] = [] for m in models(**kw): fs = m.get("first_seen") or m.get("released") if not fs: continue + if not isinstance(fs, str): + raise ValueError("BESTIARY catalog dates must be strings") try: t = datetime.strptime(fs, "%Y-%m-%d").replace(tzinfo=timezone.utc) except ValueError: continue if t >= cutoff: - out.append(m) - return sorted(out, key=lambda r: r.get("first_seen") or "", reverse=True) + out.append((t, m)) + out.sort(key=lambda item: item[0], reverse=True) + return [record for _, record in out] if __name__ == "__main__": diff --git a/obliteratus/service_contracts.py b/obliteratus/service_contracts.py new file mode 100644 index 0000000..a77bc7c --- /dev/null +++ b/obliteratus/service_contracts.py @@ -0,0 +1,157 @@ +"""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 diff --git a/obliteratus/watchtower.py b/obliteratus/watchtower.py index fca82f0..80e8932 100644 --- a/obliteratus/watchtower.py +++ b/obliteratus/watchtower.py @@ -20,10 +20,21 @@ from __future__ import annotations import json import logging import threading -from dataclasses import dataclass, field, asdict +from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from pathlib import Path -from typing import Any +from typing import Any, Callable + +from .service_contracts import ( + SchedulerEvent, + SchedulerState, + has_open_license, + is_instruction_tuned, + normalize_watchtower_candidate, + scheduler_transition, + validate_scheduler_interval, + validate_watchtower_status, +) logger = logging.getLogger(__name__) @@ -36,7 +47,7 @@ STATE_FILE = STATE_DIR / "watchtower_state.json" MIN_RECENT_DOWNLOADS = 1000 # Open-weight licenses we care about (lowercase, prefix-matched) -OPEN_LICENSES = { +OPEN_LICENSES = frozenset({ "apache-2.0", "mit", "bsd-2-clause", "bsd-3-clause", "llama2", "llama3", "llama3.1", "llama3.2", "llama3.3", "llama4", "gemma", "qwen", "deepseek", @@ -44,13 +55,13 @@ OPEN_LICENSES = { "openrail", "openrail++", "bigscience-openrail-m", "artistic-2.0", "wtfpl", "unlicense", "zlib", "other", # many open models use "other" + a permissive custom license -} +}) # Keywords that indicate instruction/chat tuning -INSTRUCT_KEYWORDS = { +INSTRUCT_KEYWORDS = frozenset({ "instruct", "chat", "it", "rlhf", "dpo", "sft", "aligned", "conversational", "assistant", "dialogue", -} +}) # ── Organizations to watch ──────────────────────────────────────────── @@ -97,8 +108,27 @@ class DiscoveredModel: @classmethod def from_dict(cls, d: dict) -> "DiscoveredModel": # Handle extra/missing keys gracefully + if not isinstance(d, dict): + raise ValueError("model state must be an object") known = {f.name for f in cls.__dataclass_fields__.values()} - return cls(**{k: v for k, v in d.items() if k in known}) + try: + model = cls(**{k: v for k, v in d.items() if k in known}) + except TypeError as exc: + raise ValueError("model state is missing required fields") from exc + for name in ( + "model_id", "name", "org", "size_category", "license", "pipeline_tag", + "discovered_at", "last_updated", + ): + if not isinstance(getattr(model, name), str): + raise ValueError(f"model state {name} must be a string") + for name in ("downloads_7d", "total_downloads", "likes"): + value = getattr(model, name) + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise ValueError(f"model state {name} must be a non-negative integer") + model.status = validate_watchtower_status(model.status) + if not isinstance(model.obliteration_metrics, dict): + raise ValueError("model state obliteration_metrics must be an object") + return model # ── Watchtower class ────────────────────────────────────────────────── @@ -106,14 +136,30 @@ class DiscoveredModel: class Watchtower: """Scans HuggingFace Hub for new popular open-weight instruction models.""" - def __init__(self, state_file: Path | str | None = None): + def __init__( + self, + state_file: Path | str | None = None, + *, + clock: Callable[[], datetime] | None = None, + fetch_models: Callable[[], list[Any]] | None = None, + thread_factory: Callable[..., threading.Thread] | None = None, + event_factory: Callable[[], threading.Event] | None = None, + scheduler_join_timeout: float = 5.0, + ): self.state_file = Path(state_file) if state_file else STATE_FILE + self._clock = clock or (lambda: datetime.now(timezone.utc)) + self._fetch_models = fetch_models or self._fetch_models_from_hf + self._thread_factory = thread_factory or threading.Thread + self._event_factory = event_factory or threading.Event + self._scheduler_join_timeout = validate_scheduler_interval(scheduler_join_timeout) self._models: dict[str, DiscoveredModel] = {} self._last_scan: str | None = None self._scan_count: int = 0 self._lock = threading.Lock() + self._scheduler_lock = threading.Lock() self._scheduler_thread: threading.Thread | None = None - self._scheduler_stop = threading.Event() + self._scheduler_stop = self._event_factory() + self._scheduler_state = SchedulerState.STOPPED self._on_new_model_callbacks: list = [] # Load persisted state @@ -126,10 +172,28 @@ class Watchtower: try: if self.state_file.exists(): data = json.loads(self.state_file.read_text(encoding="utf-8")) - self._last_scan = data.get("last_scan") - self._scan_count = data.get("scan_count", 0) - for mid, mdata in data.get("models", {}).items(): - self._models[mid] = DiscoveredModel.from_dict(mdata) + if not isinstance(data, dict): + raise ValueError("state root must be an object") + last_scan = data.get("last_scan") + if last_scan is not None and not isinstance(last_scan, str): + raise ValueError("last_scan must be a string or null") + scan_count = data.get("scan_count", 0) + if isinstance(scan_count, bool) or not isinstance(scan_count, int) or scan_count < 0: + raise ValueError("scan_count must be a non-negative integer") + raw_models = data.get("models", {}) + if not isinstance(raw_models, dict): + raise ValueError("models must be an object") + loaded: dict[str, DiscoveredModel] = {} + for mid, mdata in raw_models.items(): + if not isinstance(mid, str) or not isinstance(mdata, dict): + raise ValueError("model state entries must be objects keyed by strings") + model = DiscoveredModel.from_dict(mdata) + if not model.model_id or model.model_id != mid: + raise ValueError("model state key must match model_id") + loaded[mid] = model + self._last_scan = last_scan + self._scan_count = scan_count + self._models = loaded logger.info( "Watchtower: loaded %d models from %s", len(self._models), self.state_file, @@ -139,18 +203,23 @@ class Watchtower: def _save_state(self): """Persist watchtower state to disk.""" + tmp = self.state_file.with_suffix(".tmp") try: self.state_file.parent.mkdir(parents=True, exist_ok=True) - data = { - "last_scan": self._last_scan, - "scan_count": self._scan_count, - "models": {mid: m.to_dict() for mid, m in self._models.items()}, - } + with self._lock: + data = { + "last_scan": self._last_scan, + "scan_count": self._scan_count, + "models": {mid: m.to_dict() for mid, m in self._models.items()}, + } # Atomic write via temp file - tmp = self.state_file.with_suffix(".tmp") tmp.write_text(json.dumps(data, indent=2, default=str), encoding="utf-8") tmp.replace(self.state_file) except Exception as e: + try: + tmp.unlink(missing_ok=True) + except OSError: + pass logger.warning("Watchtower: failed to save state: %s", e) # ── HuggingFace API helpers ─────────────────────────────────────── @@ -161,7 +230,7 @@ class Watchtower: orgs: set[str] | None = None, limit_per_org: int = 50, min_downloads: int = MIN_RECENT_DOWNLOADS, - ) -> list[dict[str, Any]]: + ) -> list[Any]: """Fetch model metadata from HuggingFace Hub API. Returns a list of raw model info dicts. Gracefully returns [] @@ -175,7 +244,7 @@ class Watchtower: api = HfApi() results = [] - search_orgs = orgs or ALL_WATCHED_ORGS + search_orgs = sorted(orgs or ALL_WATCHED_ORGS, key=str.casefold) for org in search_orgs: try: @@ -187,7 +256,8 @@ class Watchtower: limit=limit_per_org, ) for m in models: - results.append(m) + if (getattr(m, "downloads", 0) or 0) >= min_downloads: + results.append(m) except Exception as e: logger.debug("Watchtower: error scanning org '%s': %s", org, e) continue @@ -197,32 +267,12 @@ class Watchtower: @staticmethod def _is_instruction_tuned(model_id: str, tags: list[str] | None = None) -> bool: """Heuristic check if a model is instruction/chat tuned.""" - name_lower = model_id.lower() - # Check model name - for kw in INSTRUCT_KEYWORDS: - if kw in name_lower: - return True - # Check tags - if tags: - tags_lower = {t.lower() for t in tags} - for kw in INSTRUCT_KEYWORDS: - if kw in tags_lower: - return True - # Explicit tag checks - if "conversational" in tags_lower: - return True - return False + return is_instruction_tuned(model_id, tags, INSTRUCT_KEYWORDS) @staticmethod def _has_open_license(license_id: str | None) -> bool: """Check if the license is considered open-weight.""" - if not license_id: - return False - lid = license_id.lower().strip() - for allowed in OPEN_LICENSES: - if lid == allowed or lid.startswith(allowed): - return True - return False + return has_open_license(license_id, OPEN_LICENSES) @staticmethod def _estimate_size(model_id: str, config: dict | None = None) -> str: @@ -255,34 +305,23 @@ class Watchtower: new_models = [] try: - raw_models = self._fetch_models_from_hf() + raw_models = self._fetch_models() + if not isinstance(raw_models, list): + raise ValueError("HF API response must be a list") except Exception as e: _log(f"Watchtower: HF API error — {e}") return [] _log(f"Watchtower: fetched {len(raw_models)} candidate models from HF Hub") - now = datetime.now(timezone.utc).isoformat() + now = self._now_iso() processed = 0 for m in raw_models: - model_id = getattr(m, "id", None) or getattr(m, "modelId", "") - if not model_id: + candidate = normalize_watchtower_candidate(m) + if candidate is None: continue - - # Extract metadata safely - downloads = getattr(m, "downloads", 0) or 0 - likes = getattr(m, "likes", 0) or 0 - tags = getattr(m, "tags", []) or [] - license_id = getattr(m, "license", None) or "" - # Some models store license in tags - if not license_id: - for t in tags: - if t.startswith("license:"): - license_id = t.split(":", 1)[1] - break - - pipeline_tag = getattr(m, "pipeline_tag", "") or "" + model_id, downloads, likes, tags, license_id, pipeline_tag = candidate # Filter: must be text-generation if pipeline_tag and pipeline_tag != "text-generation": @@ -345,8 +384,10 @@ class Watchtower: ) # Fire callbacks for new models + with self._lock: + callbacks = list(self._on_new_model_callbacks) for dm in new_models: - for cb in self._on_new_model_callbacks: + for cb in callbacks: try: cb(dm) except Exception as e: @@ -385,14 +426,17 @@ class Watchtower: def set_status(self, model_id: str, status: str, metrics: dict | None = None): """Update a model's status (new/queued/obliterating/obliterated/failed).""" + status = validate_watchtower_status(status) with self._lock: - if model_id in self._models: - m = self._models[model_id] - m.status = status - m.last_updated = datetime.now(timezone.utc).isoformat() - if metrics: - m.obliteration_metrics = metrics + if model_id not in self._models: + return False + m = self._models[model_id] + m.status = status + m.last_updated = self._now_iso() + if metrics is not None: + m.obliteration_metrics = metrics self._save_state() + return True def get_stats(self) -> dict[str, Any]: """Return summary statistics.""" @@ -420,7 +464,8 @@ class Watchtower: Signature: callback(model: DiscoveredModel) """ - self._on_new_model_callbacks.append(callback) + with self._lock: + self._on_new_model_callbacks.append(callback) # ── Scheduler ───────────────────────────────────────────────────── @@ -429,38 +474,88 @@ class Watchtower: Safe to call multiple times — restarts with new interval. """ - self.stop_scheduler() - self._scheduler_stop.clear() + interval_seconds = validate_scheduler_interval(interval) + if not self.stop_scheduler(): + raise RuntimeError("Watchtower scheduler did not stop; refusing duplicate start") + stop_event = self._event_factory() def _run(): - while not self._scheduler_stop.is_set(): + while not stop_event.is_set(): try: self.scan(on_log=on_log) except Exception as e: logger.error("Watchtower scheduler error: %s", e) - self._scheduler_stop.wait(timeout=interval) + stop_event.wait(timeout=interval_seconds) - self._scheduler_thread = threading.Thread( - target=_run, daemon=True, name="watchtower-scheduler" - ) - self._scheduler_thread.start() - logger.info("Watchtower: scheduler started (interval=%ds)", interval) + with self._scheduler_lock: + self._scheduler_stop = stop_event + self._scheduler_state = scheduler_transition( + self._scheduler_state, + SchedulerEvent.START, + ) + thread = self._thread_factory( + target=_run, daemon=True, name="watchtower-scheduler" + ) + self._scheduler_thread = thread + try: + thread.start() + except BaseException: + self._scheduler_state = scheduler_transition( + self._scheduler_state, + SchedulerEvent.REQUEST_STOP, + ) + self._scheduler_state = scheduler_transition( + self._scheduler_state, + SchedulerEvent.STOP_CONFIRMED, + ) + self._scheduler_thread = None + raise + logger.info("Watchtower: scheduler started (interval=%ss)", interval_seconds) - def stop_scheduler(self): + def stop_scheduler(self) -> bool: """Stop the background scheduler if running.""" - if self._scheduler_thread and self._scheduler_thread.is_alive(): + with self._scheduler_lock: + thread = self._scheduler_thread + if thread is None: + return True + if self._scheduler_state is SchedulerState.RUNNING: + self._scheduler_state = scheduler_transition( + self._scheduler_state, + SchedulerEvent.REQUEST_STOP, + ) self._scheduler_stop.set() - self._scheduler_thread.join(timeout=5) + thread.join(timeout=self._scheduler_join_timeout) + if thread.is_alive(): + self._scheduler_state = scheduler_transition( + self._scheduler_state, + SchedulerEvent.STOP_TIMEOUT, + ) + logger.warning("Watchtower: scheduler did not stop before timeout") + return False + self._scheduler_state = scheduler_transition( + self._scheduler_state, + SchedulerEvent.STOP_CONFIRMED, + ) self._scheduler_thread = None logger.info("Watchtower: scheduler stopped") + return True @property def is_scanning(self) -> bool: """True if the scheduler is actively running.""" - return ( - self._scheduler_thread is not None - and self._scheduler_thread.is_alive() - ) + with self._scheduler_lock: + return ( + self._scheduler_state is SchedulerState.RUNNING + and self._scheduler_thread is not None + and self._scheduler_thread.is_alive() + ) + + def _now_iso(self) -> str: + """Return an aware clock reading normalized to UTC ISO format.""" + current = self._clock() + if not isinstance(current, datetime) or current.tzinfo is None: + raise ValueError("Watchtower clock must return a timezone-aware datetime") + return current.astimezone(timezone.utc).isoformat() # ── Table formatting ────────────────────────────────────────────── diff --git a/pyproject.toml b/pyproject.toml index 9af73e6..cdf3039 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -122,6 +122,7 @@ only_mutate = [ "obliteratus/runtime_contracts.py", "obliteratus/persistence_contracts.py", "obliteratus/remote_contracts.py", + "obliteratus/service_contracts.py", "obliteratus/evaluation/lm_eval_integration.py", "scripts/check_coverage_thresholds.py", ] @@ -130,20 +131,24 @@ required_mutation_targets = [ "obliteratus/analysis/whitened_svd.py", "obliteratus/persistence_contracts.py", "obliteratus/runtime_contracts.py", + "obliteratus/service_contracts.py", ] pytest_add_cli_args = ["--no-cov", "-q"] pytest_add_cli_args_test_selection = [ "tests/test_config.py", "tests/test_config_properties.py", + "tests/test_bestiary_contracts.py", "tests/test_coverage_thresholds.py", "tests/test_projection_math_contracts.py", "tests/test_lm_eval_reporting_contracts.py", + "tests/test_models_client_contracts.py", "tests/test_numerical_contracts.py", "tests/test_checkpoint_atomicity.py", "tests/test_persistence_contracts.py", "tests/test_persistence_pipeline.py", "tests/test_remote_contracts.py", "tests/test_runtime_contracts.py", + "tests/test_watchtower_contracts.py", "tests/test_whitened_svd_oracles.py", ] mutate_only_covered_lines = true diff --git a/scripts/run_repeat_gate.py b/scripts/run_repeat_gate.py index 01a12b0..628421f 100644 --- a/scripts/run_repeat_gate.py +++ b/scripts/run_repeat_gate.py @@ -16,6 +16,7 @@ from xml.etree import ElementTree DEFAULT_TESTS = ( "tests/test_bayesian_optimizer_contracts.py", + "tests/test_bestiary_contracts.py", "tests/test_checkpoint_atomicity.py", "tests/test_config.py", "tests/test_config_properties.py", @@ -25,6 +26,7 @@ DEFAULT_TESTS = ( "tests/test_lm_eval_reporting_contracts.py", "tests/test_informed_pipeline_contracts.py", "tests/test_model_profile_contracts.py", + "tests/test_models_client_contracts.py", "tests/test_numerical_contracts.py", "tests/test_package_export_contracts.py", "tests/test_persistence_contracts.py", @@ -38,6 +40,7 @@ DEFAULT_TESTS = ( "tests/test_strategy_navigation_contracts.py", "tests/test_sweep_contracts.py", "tests/test_telemetry_failure_contracts.py", + "tests/test_watchtower_contracts.py", ) HASH_SEEDS = ("0", "1", "8675309") diff --git a/tests/test_bestiary_contracts.py b/tests/test_bestiary_contracts.py new file mode 100644 index 0000000..8384148 --- /dev/null +++ b/tests/test_bestiary_contracts.py @@ -0,0 +1,120 @@ +"""Behavioral contracts for non-destructive BESTIARY preset augmentation.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import pytest + +from obliteratus import bestiary_sync, models_client + + +@dataclass +class Preset: + name: str + hf_id: str + description: str + tier: str + params: str + recommended_dtype: str + recommended_quantization: str | None + gated: bool + + +@pytest.mark.parametrize( + "size, expected", + [ + ("1.99B", ("tiny", "1.99B")), + ("2B", ("small", "2B")), + ("7.99B", ("small", "7.99B")), + ("8B", ("medium", "8B")), + ("15.99B", ("medium", "15.99B")), + ("16B", ("large", "16B")), + ("69.9B", ("large", "69.9B")), + ("70B", ("frontier", "70B")), + ], +) +def test_tier_boundaries_are_explicit(size: str, expected: tuple[str, str]) -> None: + assert bestiary_sync._infer_tier_params(f"org/model-{size}", {}) == expected + + +def test_tier_inference_prefers_total_moe_size_and_fails_heavy_when_unknown() -> None: + assert bestiary_sync._infer_tier_params("org/MoE-A3B-235B", {}) == ( + "frontier", + "235B", + ) + assert bestiary_sync._infer_tier_params("org/model", {"name": "unsized"}) == ( + "large", + "unknown", + ) + + +def test_curated_presets_win_and_catalog_duplicates_are_suppressed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + records = [ + {"hf_id": "curated/model", "name": "curated", "capabilities": []}, + { + "hf_id": "Meta-Llama/Fresh-70B", + "name": "Fresh", + "released": "2026-08-16", + "capabilities": ["tools", "vision"], + }, + {"hf_id": "Meta-Llama/Fresh-70B", "name": "duplicate", "capabilities": []}, + {"hf_id": "org/tiny-1B", "capabilities": []}, + ] + monkeypatch.setattr(models_client, "models", lambda **_kwargs: records) + + presets = bestiary_sync.extra_presets(Preset, {"curated/model"}) + + assert [preset.hf_id for preset in presets] == [ + "Meta-Llama/Fresh-70B", + "org/tiny-1B", + ] + assert presets[0].tier == "frontier" + assert presets[0].recommended_quantization == "4bit" + assert presets[0].gated is True + assert presets[0].description == "[BESTIARY · 2026-08-16] tools, vision." + assert presets[1].name == "tiny-1B" + assert presets[1].tier == "tiny" + assert presets[1].recommended_quantization is None + assert presets[1].description == "[BESTIARY · ?] open-weight." + + +def test_catalog_or_transformation_failure_is_an_empty_safe_noop( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def unavailable(**_kwargs): + raise OSError("offline") + + monkeypatch.setattr(models_client, "models", unavailable) + assert bestiary_sync.extra_presets(Preset, set()) == [] + + for malformed in ( + [None], + [{"hf_id": 7, "capabilities": []}], + [{"hf_id": "org/model", "capabilities": "tools"}], + [{"hf_id": "org/model", "capabilities": [7]}], + [{"hf_id": "org/model", "name": 7, "capabilities": []}], + [{"hf_id": "org/model", "released": 7, "capabilities": []}], + ): + monkeypatch.setattr(models_client, "models", lambda **_kwargs: malformed) + assert bestiary_sync.extra_presets(Preset, set()) == [] + + +def test_preset_constructor_failure_does_not_return_partial_results( + monkeypatch: pytest.MonkeyPatch, +) -> None: + records = [ + {"hf_id": "org/valid-7B", "capabilities": []}, + {"hf_id": "org/explode-8B", "capabilities": []}, + ] + monkeypatch.setattr(models_client, "models", lambda **_kwargs: records) + + class RejectSecond(Preset): + def __init__(self, **kwargs): + if kwargs["hf_id"] == "org/explode-8B": + raise ValueError("rejected") + super().__init__(**kwargs) + + assert bestiary_sync.extra_presets(RejectSecond, set()) == [] diff --git a/tests/test_models_client_contracts.py b/tests/test_models_client_contracts.py new file mode 100644 index 0000000..13af7f8 --- /dev/null +++ b/tests/test_models_client_contracts.py @@ -0,0 +1,290 @@ +"""Deterministic contracts for BESTIARY catalog resolution and filtering.""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +from obliteratus import models_client + + +def _catalog(*records: dict) -> dict: + return {"generated_at": "2026-08-16T00:00:00Z", "models": list(records)} + + +def _write_catalog(path: Path, payload: object) -> Path: + path.write_text(json.dumps(payload), encoding="utf-8") + return path + + +@pytest.fixture +def isolated_sources(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.delenv("BESTIARY_CATALOG", raising=False) + monkeypatch.setattr(models_client, "_LOCAL", tmp_path / "local.json") + monkeypatch.setattr(models_client, "_CANONICAL", tmp_path / "canonical.json") + monkeypatch.setattr(models_client, "_HERE", tmp_path / "package") + + +def test_explicit_path_is_authoritative_and_does_not_fall_back( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + isolated_sources: None, +) -> None: + _write_catalog(models_client._LOCAL, _catalog({"id": "fallback"})) + missing = tmp_path / "operator-selected.json" + monkeypatch.setenv("BESTIARY_CATALOG", str(missing)) + + with pytest.raises(FileNotFoundError, match="operator-selected"): + models_client._load() + + +def test_default_resolution_uses_first_existing_packaged_source( + isolated_sources: None, +) -> None: + models_client._CANONICAL.parent.mkdir(parents=True, exist_ok=True) + _write_catalog(models_client._CANONICAL, _catalog({"id": "canonical"})) + state = models_client._HERE / "state" / "catalog.json" + state.parent.mkdir(parents=True) + _write_catalog(state, _catalog({"id": "state"})) + + assert models_client.model_ids() == ["canonical"] + + +def test_http_resolution_is_timeout_and_size_bounded( + monkeypatch: pytest.MonkeyPatch, + isolated_sources: None, +) -> None: + payload = json.dumps(_catalog({"id": "remote"})).encode() + calls: dict[str, object] = {} + + class Response: + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def read(self, amount: int) -> bytes: + calls["read"] = amount + return payload + + def urlopen(url: str, *, timeout: int): + calls["url"] = url + calls["timeout"] = timeout + return Response() + + monkeypatch.setenv("BESTIARY_CATALOG", "https://catalog.invalid/catalog.json") + monkeypatch.setattr(models_client.urllib.request, "urlopen", urlopen) + + assert models_client.model_ids() == ["remote"] + assert calls == { + "url": "https://catalog.invalid/catalog.json", + "timeout": models_client._CATALOG_TIMEOUT_SECONDS, + "read": models_client._MAX_CATALOG_BYTES + 1, + } + + +def test_http_resolution_rejects_oversized_catalog( + monkeypatch: pytest.MonkeyPatch, + isolated_sources: None, +) -> None: + class Response: + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def read(self, amount: int) -> bytes: + return b"x" * amount + + monkeypatch.setenv("BESTIARY_CATALOG", "https://catalog.invalid/catalog.json") + monkeypatch.setattr( + models_client.urllib.request, + "urlopen", + lambda *_args, **_kwargs: Response(), + ) + + with pytest.raises(ValueError, match="size limit"): + models_client._load() + + +@pytest.mark.parametrize("payload", [b"\xff", b"{not-json}"]) +def test_path_resolution_rejects_invalid_utf8_or_json( + payload: bytes, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + isolated_sources: None, +) -> None: + path = tmp_path / "catalog.json" + path.write_bytes(payload) + monkeypatch.setenv("BESTIARY_CATALOG", str(path)) + with pytest.raises(ValueError, match="not valid UTF-8 JSON"): + models_client._load() + + +def test_path_resolution_is_size_bounded( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + isolated_sources: None, +) -> None: + path = tmp_path / "catalog.json" + path.write_bytes(b"123") + monkeypatch.setattr(models_client, "_MAX_CATALOG_BYTES", 2) + monkeypatch.setenv("BESTIARY_CATALOG", str(path)) + with pytest.raises(ValueError, match="size limit"): + models_client._load() + + +def test_filters_reject_non_string_arguments(isolated_sources: None) -> None: + with pytest.raises(TypeError, match="channel must be a string"): + models_client.models(channel=7) # type: ignore[arg-type] + + +@pytest.mark.parametrize( + "payload, message", + [ + ([], "root must be an object"), + ({}, "models must be a list"), + ({"models": ["not-a-record"]}, "record 0 must be an object"), + ], +) +def test_catalog_schema_fails_closed( + payload: object, + message: str, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + isolated_sources: None, +) -> None: + path = _write_catalog(tmp_path / "catalog.json", payload) + monkeypatch.setenv("BESTIARY_CATALOG", str(path)) + + with pytest.raises(ValueError, match=message) as error: + models_client._load() + assert str(error.value) == f"BESTIARY catalog {message}" + + +def test_filters_normalize_aliases_and_catalog_values( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + isolated_sources: None, +) -> None: + path = _write_catalog( + tmp_path / "catalog.json", + _catalog( + { + "id": "Example/Alpha", + "vendor": "Example", + "channels": ["OpenRouter"], + "open_weight": False, + "capabilities": ["Tools"], + }, + { + "id": "Example/Beta", + "vendor": "example", + "channels": ["huggingface"], + "open_weight": True, + "capabilities": ["vision"], + }, + ), + ) + monkeypatch.setenv("BESTIARY_CATALOG", str(path)) + + assert models_client.model_ids(channel="API", vendor="EXAMPLE", capability="TOOLS") == [ + "Example/Alpha" + ] + assert models_client.model_ids(channel="open-weight") == ["Example/Beta"] + assert models_client.model_ids(channel="HF", open_weight=True) == ["Example/Beta"] + + +@pytest.mark.parametrize( + "record, message", + [ + ({"id": "bad", "channels": "api"}, "channels must be a list"), + ({"id": "bad", "capabilities": [7]}, "capabilities must be a list"), + ({"id": "bad", "vendor": 7}, "vendor must be a string"), + ], +) +def test_filters_reject_malformed_catalog_fields( + record: dict, + message: str, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + isolated_sources: None, +) -> None: + path = _write_catalog(tmp_path / "catalog.json", _catalog(record)) + monkeypatch.setenv("BESTIARY_CATALOG", str(path)) + + with pytest.raises(ValueError, match=message): + models_client.models() + + +@pytest.mark.parametrize("bad_id", [None, "", 7]) +def test_model_ids_rejects_missing_or_non_string_ids( + bad_id: object, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + isolated_sources: None, +) -> None: + path = _write_catalog(tmp_path / "catalog.json", _catalog({"id": bad_id})) + monkeypatch.setenv("BESTIARY_CATALOG", str(path)) + + with pytest.raises(ValueError, match="non-empty string id"): + models_client.model_ids() + + +def test_newest_uses_injected_clock_release_fallback_and_stable_order( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + isolated_sources: None, +) -> None: + path = _write_catalog( + tmp_path / "catalog.json", + _catalog( + {"id": "released", "released": "2026-08-15"}, + {"id": "newest", "first_seen": "2026-08-16"}, + {"id": "cutoff", "first_seen": "2026-08-09"}, + {"id": "old", "first_seen": "2026-08-08"}, + {"id": "invalid", "first_seen": "yesterday"}, + {"id": "undated"}, + ), + ) + monkeypatch.setenv("BESTIARY_CATALOG", str(path)) + monkeypatch.setattr( + models_client, + "_utc_now", + lambda: datetime(2026, 8, 16, tzinfo=timezone.utc), + ) + + assert models_client.model_ids() == [ + "released", "newest", "cutoff", "old", "invalid", "undated" + ] + assert [record["id"] for record in models_client.newest(7)] == [ + "newest", "released", "cutoff" + ] + + +def test_newest_rejects_non_string_catalog_dates( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + isolated_sources: None, +) -> None: + path = _write_catalog( + tmp_path / "catalog.json", + _catalog({"id": "bad-date", "first_seen": 20260816}), + ) + monkeypatch.setenv("BESTIARY_CATALOG", str(path)) + with pytest.raises(ValueError, match="dates must be strings"): + models_client.newest() + + +@pytest.mark.parametrize("days", [-1, True, 1.5, "7"]) +def test_newest_rejects_invalid_day_windows( + days: object, + isolated_sources: None, +) -> None: + with pytest.raises((TypeError, ValueError), match="days"): + models_client.newest(days) # type: ignore[arg-type] diff --git a/tests/test_watchtower_contracts.py b/tests/test_watchtower_contracts.py new file mode 100644 index 0000000..0bc366e --- /dev/null +++ b/tests/test_watchtower_contracts.py @@ -0,0 +1,572 @@ +"""Deterministic transport, state, and scheduler contracts for Watchtower.""" + +from __future__ import annotations + +import json +import math +import sys +import threading +from datetime import datetime, timezone +from pathlib import Path +from types import ModuleType, SimpleNamespace + +import pytest + +from obliteratus.service_contracts import ( + SchedulerEvent, + SchedulerState, + normalize_watchtower_candidate, + scheduler_transition, + validate_day_window, + validate_scheduler_interval, +) +from obliteratus.watchtower import DiscoveredModel, Watchtower + + +NOW = datetime(2026, 8, 16, 12, 30, tzinfo=timezone.utc) + + +def _model(model_id: str, **overrides): + values = { + "id": model_id, + "downloads": 2_000, + "likes": 10, + "tags": ["instruct"], + "license": "apache-2.0", + "pipeline_tag": "text-generation", + } + values.update(overrides) + return SimpleNamespace(**values) + + +def _watchtower(tmp_path: Path, **kwargs) -> Watchtower: + return Watchtower( + tmp_path / "watchtower.json", + clock=lambda: NOW, + fetch_models=lambda: [], + **kwargs, + ) + + +@pytest.mark.parametrize( + "state,event,expected", + [ + (SchedulerState.STOPPED, SchedulerEvent.START, SchedulerState.RUNNING), + (SchedulerState.RUNNING, SchedulerEvent.REQUEST_STOP, SchedulerState.STOPPING), + (SchedulerState.STOPPING, SchedulerEvent.STOP_TIMEOUT, SchedulerState.STOPPING), + (SchedulerState.STOPPING, SchedulerEvent.STOP_CONFIRMED, SchedulerState.STOPPED), + ], +) +def test_scheduler_state_machine_contract(state, event, expected) -> None: + assert scheduler_transition(state, event) is expected + + +@pytest.mark.parametrize( + "state,event", + [ + (SchedulerState.STOPPED, SchedulerEvent.REQUEST_STOP), + (SchedulerState.STOPPED, SchedulerEvent.STOP_CONFIRMED), + (SchedulerState.RUNNING, SchedulerEvent.START), + (SchedulerState.RUNNING, SchedulerEvent.STOP_CONFIRMED), + (SchedulerState.STOPPING, SchedulerEvent.START), + (SchedulerState.STOPPING, SchedulerEvent.REQUEST_STOP), + ], +) +def test_scheduler_state_machine_rejects_invalid_transitions(state, event) -> None: + with pytest.raises(ValueError, match="scheduler transition"): + scheduler_transition(state, event) + + +def test_service_validation_messages_and_zero_boundaries() -> None: + with pytest.raises(TypeError) as interval_type: + validate_scheduler_interval("60") + assert str(interval_type.value) == "scheduler interval must be a number" + with pytest.raises(ValueError) as interval_value: + validate_scheduler_interval(0) + assert str(interval_value.value) == ( + "scheduler interval must be finite and greater than zero" + ) + + with pytest.raises(TypeError) as day_type: + validate_day_window("7") + assert str(day_type.value) == "days must be a non-negative integer" + with pytest.raises(ValueError) as day_value: + validate_day_window(-1) + assert str(day_value.value) == "days must be a non-negative integer" + assert validate_day_window(0) == 0 + + +def test_candidate_normalization_defaults_boundaries_and_license_tags() -> None: + missing_counts = SimpleNamespace( + id="org/model", + tags=[], + license="mit", + pipeline_tag="text-generation", + ) + assert normalize_watchtower_candidate(missing_counts) == ( + "org/model", 0, 0, [], "mit", "text-generation" + ) + assert normalize_watchtower_candidate( + _model("org/zero", downloads=0, likes=0) + ) == ("org/zero", 0, 0, ["instruct"], "apache-2.0", "text-generation") + assert normalize_watchtower_candidate(_model("org/bool", likes=True)) is None + assert normalize_watchtower_candidate(_model("org/tuple", tags=("chat",))) is None + assert normalize_watchtower_candidate(_model("org/license", license=7)) is None + assert normalize_watchtower_candidate(_model("org/pipeline", pipeline_tag=7)) is None + + tagged = SimpleNamespace( + modelId="org/tagged", + downloads=1, + likes=0, + tags=["license:custom:variant"], + pipeline_tag="", + ) + assert normalize_watchtower_candidate(tagged) == ( + "org/tagged", 1, 0, ["license:custom:variant"], "custom:variant", "" + ) + + +def test_state_load_is_transactional_on_malformed_records(tmp_path: Path) -> None: + state = tmp_path / "watchtower.json" + state.write_text( + json.dumps( + { + "last_scan": "2026-08-15T00:00:00+00:00", + "scan_count": 9, + "models": { + "org/valid": {"model_id": "org/valid", "name": "valid", "org": "org"}, + "org/bad": "not-an-object", + }, + } + ), + encoding="utf-8", + ) + + watchtower = Watchtower(state, clock=lambda: NOW, fetch_models=lambda: []) + + assert watchtower.get_all_models() == [] + assert watchtower.get_stats() == { + "total_tracked": 0, + "last_scan": None, + "scan_count": 0, + "by_status": {}, + } + + +def test_model_state_rejects_non_object_and_missing_required_fields() -> None: + with pytest.raises(ValueError, match="must be an object"): + DiscoveredModel.from_dict("bad") # type: ignore[arg-type] + with pytest.raises(ValueError, match="missing required fields"): + DiscoveredModel.from_dict({}) + + +@pytest.mark.parametrize( + "payload", + [ + [], + {"last_scan": 7}, + {"scan_count": True}, + {"scan_count": -1}, + {"models": []}, + { + "models": { + "org/key": {"model_id": "org/other", "name": "other", "org": "org"} + } + }, + ], +) +def test_state_load_rejects_malformed_envelopes_transactionally( + payload: object, + tmp_path: Path, +) -> None: + state = tmp_path / "watchtower.json" + state.write_text(json.dumps(payload), encoding="utf-8") + watchtower = Watchtower(state, clock=lambda: NOW, fetch_models=lambda: []) + assert watchtower.get_all_models() == [] + assert watchtower.get_stats()["last_scan"] is None + assert watchtower.get_stats()["scan_count"] == 0 + + +def test_state_save_failure_cleans_up_without_masking_operation( + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + blocked_parent = tmp_path / "not-a-directory" + blocked_parent.write_text("file", encoding="utf-8") + watchtower = Watchtower( + blocked_parent / "watchtower.json", + clock=lambda: NOW, + fetch_models=lambda: [], + ) + watchtower._save_state() + assert "failed to save state" in caplog.text + + +@pytest.mark.parametrize( + "bad_field,bad_value", + [ + ("status", "invented"), + ("downloads_7d", "many"), + ("obliteration_metrics", []), + ("discovered_at", 7), + ], +) +def test_state_load_rejects_malformed_model_fields_transactionally( + bad_field: str, + bad_value: object, + tmp_path: Path, +) -> None: + state = tmp_path / "watchtower.json" + record = {"model_id": "org/model", "name": "model", "org": "org"} + record[bad_field] = bad_value + state.write_text( + json.dumps({"last_scan": "prior", "scan_count": 2, "models": {"org/model": record}}), + encoding="utf-8", + ) + + watchtower = Watchtower(state, clock=lambda: NOW, fetch_models=lambda: []) + assert watchtower.get_all_models() == [] + assert watchtower.get_stats()["last_scan"] is None + assert watchtower.get_stats()["scan_count"] == 0 + + +def test_valid_state_round_trips_unknown_fields_and_defaults(tmp_path: Path) -> None: + state = tmp_path / "watchtower.json" + state.write_text( + json.dumps( + { + "last_scan": "2026-08-15T00:00:00+00:00", + "scan_count": 2, + "models": { + "org/model": { + "model_id": "org/model", + "name": "model", + "org": "org", + "unknown_future_field": True, + } + }, + } + ), + encoding="utf-8", + ) + + watchtower = Watchtower(state, clock=lambda: NOW, fetch_models=lambda: []) + model = watchtower.get_model("org/model") + + assert model is not None + assert model.status == "new" + assert watchtower.get_stats()["scan_count"] == 2 + + +def test_huggingface_adapter_is_deterministic_bounded_and_fault_isolated( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[dict[str, object]] = [] + + class HfApi: + def list_models(self, **kwargs): + calls.append(kwargs) + if kwargs["author"] == "bad": + raise OSError("one organization is unavailable") + return [ + _model(f"{kwargs['author']}/kept", downloads=1_000), + _model(f"{kwargs['author']}/low", downloads=999), + ] + + fake_module = ModuleType("huggingface_hub") + fake_module.HfApi = HfApi + monkeypatch.setitem(sys.modules, "huggingface_hub", fake_module) + + records = Watchtower._fetch_models_from_hf( + orgs={"zeta", "bad", "Alpha"}, + limit_per_org=3, + min_downloads=1_000, + ) + + assert [call["author"] for call in calls] == ["Alpha", "bad", "zeta"] + assert all( + call == { + "author": call["author"], + "pipeline_tag": "text-generation", + "sort": "downloads", + "direction": -1, + "limit": 3, + } + for call in calls + ) + assert [record.id for record in records] == ["Alpha/kept", "zeta/kept"] + + +@pytest.mark.parametrize( + "license_id, expected", + [ + (None, False), + (" MIT ", True), + ("apache-2.0", True), + ("apache-2.0-custom", True), + ("mitigation-license", False), + ("proprietary", False), + ], +) +def test_open_license_matching_has_identifier_boundaries(license_id, expected) -> None: + assert Watchtower._has_open_license(license_id) is expected + + +@pytest.mark.parametrize( + "model_id,tags,expected", + [ + ("org/model-instruct", None, True), + ("org/model-it", None, True), + ("org/bitnet-base", None, False), + ("org/plain", ["Conversational"], True), + ("org/plain", ["pretraining"], False), + (7, [], False), + ("org/plain", "chat", False), + ("org/plain", [7], False), + ], +) +def test_instruction_matching_has_token_boundaries(model_id, tags, expected) -> None: + assert Watchtower._is_instruction_tuned(model_id, tags) is expected + + +@pytest.mark.parametrize( + "model_id, expected", + [ + ("org/model-0.5B", "500M"), + ("org/model-7B", "7B"), + ("org/model-70b-instruct", "70B"), + ("org/unsized", "unknown"), + ], +) +def test_size_estimation_boundaries(model_id: str, expected: str) -> None: + assert Watchtower._estimate_size(model_id) == expected + + +def test_scan_filters_candidates_updates_existing_and_is_deterministic(tmp_path: Path) -> None: + candidates = [ + object(), + _model("org/bad-downloads", downloads="many"), + _model("org/bad-likes", likes=-1), + _model("org/bad-tags", tags=[7]), + _model("org/bad-license", license=7), + SimpleNamespace(modelId="org/missing-id-fallback-7B", downloads=2_000, likes=1, + tags=["License: mit", "chat"], license="", pipeline_tag=""), + _model("org/wrong-task-7B", pipeline_tag="text-classification"), + _model("org/unpopular-7B", downloads=999), + _model("org/closed-7B", license="proprietary"), + _model("org/base-7B", tags=[]), + _model("Qwen/major-base-8B", tags=[]), + _model("org/fresh-instruct-7B"), + ] + logs: list[str] = [] + callback_ids: list[str] = [] + watchtower = Watchtower( + tmp_path / "watchtower.json", + clock=lambda: NOW, + fetch_models=lambda: candidates, + ) + watchtower.on_new_model(lambda model: callback_ids.append(model.model_id)) + watchtower.on_new_model(lambda _model: (_ for _ in ()).throw(RuntimeError("ignored"))) + + discovered = watchtower.scan(on_log=logs.append) + + assert [model.model_id for model in discovered] == [ + "org/missing-id-fallback-7B", + "Qwen/major-base-8B", + "org/fresh-instruct-7B", + ] + assert callback_ids == [model.model_id for model in discovered] + assert watchtower.get_stats()["scan_count"] == 1 + assert watchtower.get_stats()["last_scan"] == NOW.isoformat() + assert logs[0] == "Watchtower: starting scan..." + assert logs[-1].endswith("3 new discoveries") + + candidates[-1].downloads = 9_000 + candidates[-1].likes = 99 + assert watchtower.scan() == [] + updated = watchtower.get_model("org/fresh-instruct-7B") + assert updated is not None + assert (updated.downloads_7d, updated.likes, updated.last_updated) == ( + 9_000, + 99, + NOW.isoformat(), + ) + + +def test_transport_failure_does_not_advance_scan_state(tmp_path: Path) -> None: + def fail(): + raise OSError("offline") + + watchtower = Watchtower( + tmp_path / "watchtower.json", + clock=lambda: NOW, + fetch_models=fail, + ) + + assert watchtower.scan() == [] + assert watchtower.get_stats()["scan_count"] == 0 + assert not watchtower.state_file.exists() + + malformed = Watchtower( + tmp_path / "malformed.json", + clock=lambda: NOW, + fetch_models=lambda: None, # type: ignore[return-value] + ) + assert malformed.scan() == [] + assert malformed.get_stats()["scan_count"] == 0 + + +def test_status_updates_validate_state_and_preserve_empty_metrics(tmp_path: Path) -> None: + watchtower = Watchtower( + tmp_path / "watchtower.json", + clock=lambda: NOW, + fetch_models=lambda: [_model("org/model-instruct-7B")], + ) + watchtower.scan() + + assert watchtower.set_status("missing", "queued") is False + with pytest.raises(ValueError, match="status"): + watchtower.set_status("org/model-instruct-7B", "invented") + assert watchtower.set_status("org/model-instruct-7B", "obliterated", {}) is True + model = watchtower.get_model("org/model-instruct-7B") + assert model is not None + assert model.status == "obliterated" + assert model.obliteration_metrics == {} + assert model.last_updated == NOW.isoformat() + assert [item.model_id for item in watchtower.get_obliterated()] == [model.model_id] + + +@pytest.mark.parametrize("interval", [0, -1, True, math.inf, math.nan, "1"]) +def test_scheduler_rejects_non_positive_or_non_finite_intervals( + interval: object, + tmp_path: Path, +) -> None: + watchtower = _watchtower(tmp_path) + with pytest.raises((TypeError, ValueError), match="interval"): + watchtower.start_scheduler(interval=interval) # type: ignore[arg-type] + + +def test_scheduler_runs_immediately_and_stops_without_sleep_races(tmp_path: Path) -> None: + scanned = threading.Event() + + def fetch(): + scanned.set() + return [] + + watchtower = Watchtower( + tmp_path / "watchtower.json", + clock=lambda: NOW, + fetch_models=fetch, + ) + + watchtower.start_scheduler(interval=3_600) + assert scanned.wait(timeout=1) + assert watchtower.is_scanning + assert watchtower.stop_scheduler() is True + assert not watchtower.is_scanning + + +def test_scheduler_restart_and_start_failure_are_transactional(tmp_path: Path) -> None: + threads = [] + + class FakeThread: + def __init__(self, **kwargs): + assert kwargs["daemon"] is True + assert kwargs["name"] == "watchtower-scheduler" + self.alive = False + threads.append(self) + + def start(self) -> None: + self.alive = True + + def join(self, timeout: float) -> None: + assert timeout == 5.0 + self.alive = False + + def is_alive(self) -> bool: + return self.alive + + watchtower = _watchtower(tmp_path, thread_factory=FakeThread) + watchtower.start_scheduler(interval=60) + assert watchtower.is_scanning + watchtower.start_scheduler(interval=120) + assert len(threads) == 2 + assert not threads[0].is_alive() + assert threads[1].is_alive() + assert watchtower.stop_scheduler() is True + + class FailingThread(FakeThread): + def start(self) -> None: + raise RuntimeError("cannot start") + + failing = _watchtower(tmp_path, thread_factory=FailingThread) + with pytest.raises(RuntimeError, match="cannot start"): + failing.start_scheduler(interval=60) + assert failing._scheduler_thread is None + assert failing._scheduler_state is SchedulerState.STOPPED + assert not failing.is_scanning + + +def test_stuck_scheduler_is_retained_and_blocks_duplicate_start(tmp_path: Path) -> None: + class StuckThread: + def is_alive(self) -> bool: + return True + + def join(self, timeout: float) -> None: + assert timeout == 0.01 + + watchtower = _watchtower(tmp_path, scheduler_join_timeout=0.01) + original = StuckThread() + watchtower._scheduler_thread = original + watchtower._scheduler_state = SchedulerState.RUNNING + + assert watchtower.stop_scheduler() is False + assert watchtower._scheduler_thread is original + assert watchtower._scheduler_state is SchedulerState.STOPPING + with pytest.raises(RuntimeError, match="did not stop"): + watchtower.start_scheduler(interval=60) + assert watchtower._scheduler_thread is original + + +def test_queries_sort_limit_copy_and_format_rows(tmp_path: Path) -> None: + watchtower = _watchtower(tmp_path) + watchtower._models = { + "slow": DiscoveredModel( + model_id="org/slow", + name="slow", + org="org", + downloads_7d=10, + discovered_at="not-a-date-value", + ), + "fast": DiscoveredModel( + model_id="org/fast", + name="fast", + org="org", + downloads_7d=20, + likes=3, + license="mit", + discovered_at=NOW.isoformat(), + ), + } + + assert [m.model_id for m in watchtower.get_trending(limit=1)] == ["org/fast"] + assert watchtower.get_model_choices() == ["org/fast", "org/slow"] + copied = watchtower.get_all_models() + copied.clear() + assert len(watchtower.get_all_models()) == 2 + assert [m.model_id for m in watchtower.get_new_models()] == ["org/slow", "org/fast"] + rows = watchtower.format_table() + assert rows[0] == [ + "org/fast", "org", "", "20", "3", "mit", "2026-08-16 12:30", "🆕 new" + ] + assert rows[1][6] == "not-a-date-value" + + +def test_clock_must_be_timezone_aware(tmp_path: Path) -> None: + watchtower = Watchtower( + tmp_path / "watchtower.json", + clock=lambda: datetime(2026, 8, 16), + fetch_models=lambda: [], + ) + with pytest.raises(ValueError, match="timezone-aware"): + watchtower.scan()