model_registry¶

def list_bundles(models_dir: Path) -> list[ModelBundle]:
    """Scan *models_dir* for model bundle subdirectories.

    Each subdirectory is parsed independently; failures are captured as
    invalid bundles rather than aborting the scan.

    Args:
        models_dir: Parent directory containing bundle subdirectories
            (e.g. ``Path("models")``).

    Returns:
        A list of :class:`ModelBundle` instances sorted by ``model_id``
        for deterministic ordering.
    """
    if not models_dir.is_dir():
        return []

    bundles: list[ModelBundle] = []
    for candidate in models_dir.iterdir():
        if not candidate.is_dir():
            continue
        bundle = _parse_bundle(candidate)
        bundles.append(bundle)

    bundles.sort(key=lambda b: b.model_id)
    return bundles

def is_compatible(
    bundle: ModelBundle,
    required_schema_hash: str = get_feature_schema("v1").SCHEMA_HASH,
    required_label_set: frozenset[str] = LABEL_SET_V1,
) -> bool:
    """Check whether *bundle* is compatible with the current runtime.

    A bundle is compatible when both hold:

    1. ``metadata.schema_hash`` exactly matches *required_schema_hash*.
    2. ``sorted(metadata.label_set)`` exactly matches ``sorted(required_label_set)``.

    Args:
        bundle: A scanned model bundle.
        required_schema_hash: Expected schema hash (defaults to
            ``FeatureSchemaV1.SCHEMA_HASH``).
        required_label_set: Expected label vocabulary (defaults to
            ``LABEL_SET_V1``).

    Returns:
        ``True`` if the bundle is valid and compatible.
    """
    if not bundle.valid or bundle.metadata is None:
        return False
    if bundle.metadata.schema_hash != required_schema_hash:
        return False
    if sorted(bundle.metadata.label_set) != sorted(required_label_set):
        return False
    return True

def passes_constraints(
    bundle: ModelBundle,
    policy: SelectionPolicy,  # noqa: ARG001 — reserved for future policy versions
) -> bool:
    """Check whether *bundle* passes the hard constraints of *policy*.

    Policy v1 applies no additional constraints beyond validity: all
    acceptance gates are enforced at promotion time by retrain, so any
    promoted bundle that parsed successfully is eligible for ranking.

    Args:
        bundle: A scanned model bundle (must be valid).
        policy: Selection policy configuration.

    Returns:
        ``True`` if the bundle is valid with non-null metrics.
    """
    return bundle.valid and bundle.metrics is not None

def score(
    bundle: ModelBundle,
    policy: SelectionPolicy,  # noqa: ARG001 — reserved for future policy versions
) -> tuple[float, float, str]:
    """Compute a sortable ranking key for *bundle*.

    The tuple sorts **descending** on all three components:

    1. ``macro_f1`` (higher is better)
    2. ``weighted_f1`` (tie-break; higher is better)
    3. ``created_at`` (tie-break; newer is better — ISO8601 strings
       with the same UTC offset sort lexicographically)

    Args:
        bundle: A valid model bundle with non-null metrics and metadata.
        policy: Selection policy configuration.

    Returns:
        A 3-tuple ``(macro_f1, weighted_f1, created_at_raw)`` suitable
        for ``sorted(..., reverse=True)``.

    Raises:
        ValueError: If the bundle is invalid or missing metrics/metadata.
    """
    if not bundle.valid or bundle.metrics is None or bundle.metadata is None:
        raise ValueError(f"Cannot score invalid bundle {bundle.model_id!r}")
    return (
        bundle.metrics.macro_f1,
        bundle.metrics.weighted_f1,
        bundle.metadata.created_at,
    )

def find_best_model(
    models_dir: Path,
    policy: SelectionPolicy | None = None,
    required_schema_hash: str | None = None,
    required_label_set: frozenset[str] | None = None,
) -> SelectionReport:
    """Scan, filter, rank, and select the best model bundle.

    This is the main entry-point for non-mutating model selection.
    It composes :func:`list_bundles`, :func:`is_compatible`,
    :func:`passes_constraints`, and :func:`score` into a single call
    that returns a structured :class:`SelectionReport`.

    Args:
        models_dir: Directory containing promoted model bundle
            subdirectories (e.g. ``Path("models")``).
        policy: Selection policy configuration.  Defaults to
            ``SelectionPolicy()`` (policy v1).
        required_schema_hash: Schema hash that bundles must match.
            Defaults to ``FeatureSchemaV1.SCHEMA_HASH``.
        required_label_set: Label vocabulary that bundles must match.
            Defaults to ``LABEL_SET_V1``.

    Returns:
        A :class:`SelectionReport` with the best bundle (if any),
        the full ranked list of eligible bundles, and exclusion
        records for every bundle that was filtered out.
    """
    explicit_schema_hash = required_schema_hash is not None

    if policy is None:
        policy = SelectionPolicy()
    if required_label_set is None:
        required_label_set = LABEL_SET_V1
    if required_schema_hash is not None:
        schema_hashes: tuple[str, ...] = (required_schema_hash,)
    else:
        schema_hashes = tuple(
            get_feature_schema(version).SCHEMA_HASH
            for version in iter_feature_schema_versions(LATEST_FEATURE_SCHEMA_VERSION)
        )

    last_report: SelectionReport | None = None
    for candidate_schema_hash in schema_hashes:
        bundles = list_bundles(models_dir)

        excluded: list[ExclusionRecord] = []
        eligible: list[ModelBundle] = []

        for bundle in bundles:
            if not bundle.valid:
                excluded.append(
                    ExclusionRecord(
                        model_id=bundle.model_id,
                        path=bundle.path,
                        reason=f"invalid: {bundle.invalid_reason}",
                    )
                )
                continue

            if not is_compatible(bundle, candidate_schema_hash, required_label_set):
                assert bundle.metadata is not None
                if bundle.metadata.schema_hash != candidate_schema_hash:
                    detail = "schema_hash mismatch"
                else:
                    detail = "label_set mismatch"
                excluded.append(
                    ExclusionRecord(
                        model_id=bundle.model_id,
                        path=bundle.path,
                        reason=f"incompatible: {detail}",
                    )
                )
                continue

            if not passes_constraints(bundle, policy):
                excluded.append(
                    ExclusionRecord(
                        model_id=bundle.model_id,
                        path=bundle.path,
                        reason="constraint: failed policy constraints",
                    )
                )
                continue

            eligible.append(bundle)

        ranked = sorted(
            eligible,
            key=lambda b: score(b, policy),
            reverse=True,
        )

        last_report = SelectionReport(
            best=ranked[0] if ranked else None,
            ranked=ranked,
            excluded=excluded,
            policy=policy,
            required_schema_hash=candidate_schema_hash,
        )
        if ranked or explicit_schema_hash:
            return last_report

    assert last_report is not None
    return last_report

def read_active(models_dir: Path) -> ActivePointer | None:
    """Read the active model pointer from ``models_dir/active.json``.

    Returns ``None`` (without raising) when the file is missing,
    contains invalid JSON, or fails :class:`ActivePointer` validation.
    A warning is logged on parse/validation failures so operators can
    notice stale pointer files.
    """
    active_path = models_dir / _ACTIVE_FILE
    if not active_path.is_file():
        return None

    try:
        raw = json.loads(active_path.read_text())
    except (json.JSONDecodeError, OSError) as exc:
        logger.warning("active.json parse error: %s", exc)
        return None

    try:
        return ActivePointer.model_validate(raw)
    except Exception as exc:
        logger.warning("active.json validation error: %s", exc)
        return None

def append_active_history(
    models_dir: Path,
    old: ActivePointer | None,
    new: ActivePointer,
) -> None:
    """Append a transition record to ``models_dir/active_history.jsonl``.

    Creates the file if it does not exist.  Each line is a
    self-contained JSON object matching :class:`ActiveHistoryEntry`.
    """
    entry = ActiveHistoryEntry(at=new.selected_at, old=old, new=new)
    history_path = models_dir / _HISTORY_FILE
    with history_path.open("a") as fh:
        fh.write(entry.model_dump_json() + "\n")

def write_active_atomic(
    models_dir: Path,
    bundle: ModelBundle,
    policy: SelectionPolicy,
    reason: str | None = None,
) -> ActivePointer:
    """Atomically write ``models_dir/active.json`` for *bundle*.

    The pointer is written to a temporary file first, then moved into
    place with :func:`os.replace` to guarantee readers never see a
    partial write.  The previous pointer (if any) is read before the
    overwrite and both old and new are appended to the audit log via
    :func:`append_active_history`.

    Args:
        models_dir: The ``models/`` directory.
        bundle: The bundle to activate (must be valid with metrics).
        policy: The selection policy used to choose this bundle.
        reason: Optional human-readable reason string.

    Returns:
        The newly written :class:`ActivePointer`.
    """
    old = read_active(models_dir)

    reason_dict: dict[str, object] | None = None
    if bundle.metrics is not None:
        reason_dict = {
            "metric": "macro_f1",
            "macro_f1": bundle.metrics.macro_f1,
            "weighted_f1": bundle.metrics.weighted_f1,
        }
        if reason is not None:
            reason_dict["note"] = reason

    model_dir = str(bundle.path.relative_to(models_dir.parent))

    pointer = ActivePointer(
        model_dir=model_dir,
        model_id=bundle.model_id,
        selected_at=datetime.now(UTC).isoformat(),
        policy_version=policy.version,
        reason=reason_dict,
    )

    tmp_path = models_dir / _ACTIVE_TMP
    final_path = models_dir / _ACTIVE_FILE

    tmp_path.write_text(pointer.model_dump_json(indent=2) + "\n")
    os.replace(tmp_path, final_path)

    append_active_history(models_dir, old, pointer)

    return pointer

def resolve_active_model(
    models_dir: Path,
    policy: SelectionPolicy | None = None,
    required_schema_hash: str | None = None,
    required_label_set: frozenset[str] | None = None,
) -> tuple[ModelBundle | None, SelectionReport | None]:
    """Resolve the active model bundle, falling back to selection.

    Resolution order:

    1. Read ``active.json``.  If valid and the pointed-to bundle
       exists, is parseable, and is compatible — return it immediately
       (no full scan).
    2. Otherwise fall back to :func:`find_best_model`.  If a best
       bundle is found, atomically update ``active.json`` to self-heal
       the pointer.

    Returns:
        A 2-tuple ``(bundle, report)``.  *report* is ``None`` when the
        pointer was valid and no scan was needed.
    """
    if policy is None:
        policy = SelectionPolicy()
    if required_label_set is None:
        required_label_set = LABEL_SET_V1

    pointer = read_active(models_dir)
    if pointer is not None:
        bundle_path = models_dir.parent / pointer.model_dir
        if bundle_path.is_dir():
            bundle = _parse_bundle(bundle_path)
            metadata = bundle.metadata
            if bundle.valid and metadata is not None:
                if required_schema_hash is None:
                    if _matches_declared_schema(bundle) and sorted(
                        metadata.label_set
                    ) == sorted(required_label_set):
                        return bundle, None
                elif is_compatible(bundle, required_schema_hash, required_label_set):
                    return bundle, None
            logger.warning(
                "active.json points to invalid/incompatible bundle %s; "
                "falling back to selection",
                pointer.model_dir,
            )
        else:
            logger.warning(
                "active.json points to missing directory %s; falling back to selection",
                pointer.model_dir,
            )

    report = find_best_model(
        models_dir,
        policy,
        required_schema_hash,
        required_label_set,
    )

    if report.best is not None:
        write_active_atomic(models_dir, report.best, policy, reason="auto-repair")

    return report.best, report

def write_index_cache(
    models_dir: Path,
    report: SelectionReport,
) -> IndexCache:
    """Write ``models_dir/index.json`` from a :class:`SelectionReport`.

    The cache is written atomically (temp + :func:`os.replace`).  It is
    informational only — :func:`find_best_model` never reads it.

    Args:
        models_dir: The ``models/`` directory.
        report: A completed selection report.

    Returns:
        The :class:`IndexCache` that was persisted.
    """
    ranked_summaries: list[IndexCacheBundleSummary] = []
    for b in report.ranked:
        ranked_summaries.append(
            IndexCacheBundleSummary(
                model_id=b.model_id,
                path=str(b.path),
                macro_f1=b.metrics.macro_f1 if b.metrics else None,
                weighted_f1=b.metrics.weighted_f1 if b.metrics else None,
                created_at=b.metadata.created_at if b.metadata else None,
                eligible=True,
            )
        )

    cache = IndexCache(
        generated_at=datetime.now(UTC).isoformat(),
        schema_hash=report.required_schema_hash,
        policy_version=report.policy.version,
        ranked=ranked_summaries,
        excluded=report.excluded,
        best_model_id=report.best.model_id if report.best else None,
    )

    tmp_path = models_dir / _INDEX_TMP
    final_path = models_dir / _INDEX_FILE

    tmp_path.write_text(cache.model_dump_json(indent=2) + "\n")
    os.replace(tmp_path, final_path)

    return cache

def read_index_cache(models_dir: Path) -> IndexCache | None:
    """Read the cached index from ``models_dir/index.json``.

    Returns ``None`` (without raising) when the file is missing,
    contains invalid JSON, or fails :class:`IndexCache` validation.
    """
    index_path = models_dir / _INDEX_FILE
    if not index_path.is_file():
        return None

    try:
        raw = json.loads(index_path.read_text())
    except (json.JSONDecodeError, OSError) as exc:
        logger.warning("index.json parse error: %s", exc)
        return None

    try:
        return IndexCache.model_validate(raw)
    except Exception as exc:
        logger.warning("index.json validation error: %s", exc)
        return None

def should_switch_active(
    current: ActivePointer | None,
    candidate: ModelBundle,
    policy: SelectionPolicy,
) -> bool:
    """Decide whether *candidate* should replace *current* as active.

    When ``policy.min_improvement`` is positive, the candidate's
    ``macro_f1`` must exceed the current active model's ``macro_f1``
    by at least that amount.  If ``current`` is ``None`` or has no
    recorded ``macro_f1``, the switch is always allowed.

    Args:
        current: The current active pointer (may be ``None``).
        candidate: The best-ranked bundle from selection.
        policy: Selection policy with hysteresis threshold.

    Returns:
        ``True`` if the active pointer should be updated.
    """
    if current is None or policy.min_improvement <= 0.0:
        return True

    if candidate.metrics is None:
        return False

    current_f1: float | None = None
    if current.reason and "macro_f1" in current.reason:
        try:
            current_f1 = float(current.reason["macro_f1"])  # type: ignore[arg-type]
        except TypeError, ValueError:
            logger.debug(
                "Could not parse macro_f1 from active pointer reason", exc_info=True
            )

    if current_f1 is None:
        return True

    return candidate.metrics.macro_f1 >= current_f1 + policy.min_improvement