core.inference_policy¶

def build_inference_policy(
    *,
    model_dir: str,
    model_schema_hash: str,
    model_label_set: list[str],
    reject_threshold: float = DEFAULT_REJECT_THRESHOLD,
    calibrator_store_dir: str | None = None,
    calibration_method: str | None = None,
    source: str = "manual",
) -> InferencePolicy:
    """Convenience builder that fills in provenance fields automatically.

    Args:
        model_dir: Path to model bundle, relative to ``models_dir.parent``.
        model_schema_hash: Schema hash from the model's metadata.
        model_label_set: Label set from the model's metadata.
        reject_threshold: Reject threshold for this model+calibration pair.
        calibrator_store_dir: Path to calibrator store directory,
            relative to ``models_dir.parent``.  ``None`` for identity.
        calibration_method: ``"temperature"`` or ``"isotonic"``; ``None``
            when no calibrator store is used.
        source: How this policy was created (``"manual"``,
            ``"tune-reject"``, ``"retrain"``, ``"calibrate"``).

    Returns:
        A populated :class:`InferencePolicy`.
    """
    return InferencePolicy(
        model_dir=model_dir,
        model_schema_hash=model_schema_hash,
        model_label_set=sorted(model_label_set),
        reject_threshold=reject_threshold,
        calibrator_store_dir=calibrator_store_dir,
        calibration_method=calibration_method,
        source=source,
        git_commit=_current_git_commit(),
    )

def render_default_inference_policy_template_json() -> str:
    """Return the canonical checked-in starter JSON text for ``configs/``.

    Uses stable ``created_at`` / ``git_commit`` so the file matches the
    repository template byte-for-byte. Runtime starter files written by
    :func:`write_inference_policy_starter_template` use live provenance.
    """
    policy = InferencePolicy(
        model_dir="models/<run_id>",
        model_schema_hash="<schema_hash>",
        model_label_set=["<label>"],
        reject_threshold=DEFAULT_REJECT_THRESHOLD,
        created_at=_INFERENCE_POLICY_TEMPLATE_CANONICAL_CREATED_AT,
        git_commit="",
        source="template",
    )
    data = policy.model_dump()
    data["_help"] = _inference_policy_starter_help(
        _INFERENCE_POLICY_TEMPLATE_PATHS_PLACEHOLDER
    )
    return json.dumps(data, indent=2, ensure_ascii=False) + "\n"

def write_inference_policy_starter_template(models_dir: Path) -> Path:
    """Atomically write a placeholder ``inference_policy.json`` for manual editing.

    Same content shape as :func:`render_default_inference_policy_template_json`,
    but with live ``created_at`` / ``git_commit`` and
    ``_help.paths_are_relative_to`` set to ``str(models_dir.parent)``.

    Args:
        models_dir: The ``models/`` directory (under TASKCLF_HOME).

    Returns:
        Path to the written policy file.
    """
    policy = build_inference_policy(
        model_dir="models/<run_id>",
        model_schema_hash="<schema_hash>",
        model_label_set=["<label>"],
        reject_threshold=DEFAULT_REJECT_THRESHOLD,
        source="starter-template",
    )
    data = policy.model_dump()
    data["_help"] = _inference_policy_starter_help(str(models_dir.parent))

    models_dir.mkdir(parents=True, exist_ok=True)
    tmp_path = models_dir / _POLICY_STARTER_TMP
    final_path = models_dir / DEFAULT_INFERENCE_POLICY_FILE
    tmp_path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n")
    os.replace(tmp_path, final_path)
    logger.info("Wrote inference policy starter template to %s", final_path)
    return final_path

def save_inference_policy(
    policy: InferencePolicy,
    models_dir: Path,
) -> Path:
    """Atomically persist *policy* as ``models_dir/inference_policy.json``.

    Uses temp-file + :func:`os.replace` so readers never see a partial
    write.

    Args:
        policy: The policy to persist.
        models_dir: The ``models/`` directory.

    Returns:
        Path to the written policy file.
    """
    models_dir.mkdir(parents=True, exist_ok=True)

    tmp_path = models_dir / _POLICY_TMP
    final_path = models_dir / DEFAULT_INFERENCE_POLICY_FILE

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

    logger.info(
        "Wrote inference policy to %s (model=%s, threshold=%.4f, source=%s)",
        final_path,
        policy.model_dir,
        policy.reject_threshold,
        policy.source,
    )
    return final_path

def load_inference_policy(models_dir: Path) -> InferencePolicy | None:
    """Load the inference policy from ``models_dir/inference_policy.json``.

    Returns ``None`` (without raising) when the file is missing, contains
    invalid JSON, or fails validation.

    Args:
        models_dir: The ``models/`` directory.

    Returns:
        A validated :class:`InferencePolicy`, or ``None``.
    """
    policy_path = models_dir / DEFAULT_INFERENCE_POLICY_FILE
    if not policy_path.is_file():
        return None

    try:
        raw = json.loads(policy_path.read_text())
        return InferencePolicy.model_validate(raw)
    except (json.JSONDecodeError, ValueError, OSError) as exc:
        logger.warning(
            "Could not load inference policy from %s: %s",
            policy_path,
            exc,
        )
        return None

def remove_inference_policy(models_dir: Path) -> bool:
    """Delete ``models_dir/inference_policy.json`` if it exists.

    Args:
        models_dir: The ``models/`` directory.

    Returns:
        ``True`` if the file was removed, ``False`` if it did not exist.
    """
    policy_path = models_dir / DEFAULT_INFERENCE_POLICY_FILE
    try:
        policy_path.unlink()
        logger.info("Removed inference policy at %s", policy_path)
        return True
    except FileNotFoundError:
        return False

def validate_policy(
    policy: InferencePolicy,
    models_dir: Path,
) -> None:
    """Validate that *policy* references artifacts that exist and are compatible.

    Checks:

    1. ``model_dir`` resolves to an existing directory with ``metadata.json``.
    2. ``model_schema_hash`` matches the bundle's recorded schema hash.
    3. ``model_label_set`` matches the bundle's recorded label set.
    4. If ``calibrator_store_dir`` is set, it resolves to an existing
       directory with ``store.json``.
    5. If the calibrator store has model-binding metadata, it matches the
       policy's model binding.

    Args:
        policy: The policy to validate.
        models_dir: The ``models/`` directory (paths are resolved
            relative to ``models_dir.parent``).

    Raises:
        PolicyValidationError: When any check fails.
    """
    base = models_dir.parent
    bundle_path = base / policy.model_dir
    if not bundle_path.is_dir():
        raise PolicyValidationError(
            f"Model directory does not exist: {bundle_path}",
            {"model_dir": policy.model_dir},
        )

    meta_path = bundle_path / "metadata.json"
    if not meta_path.is_file():
        raise PolicyValidationError(
            f"Model metadata not found: {meta_path}",
            {"model_dir": policy.model_dir},
        )

    try:
        meta = json.loads(meta_path.read_text())
    except (json.JSONDecodeError, OSError) as exc:
        raise PolicyValidationError(
            f"Could not read model metadata: {exc}",
            {"model_dir": policy.model_dir},
        ) from exc

    bundle_hash = meta.get("schema_hash", "")
    if bundle_hash != policy.model_schema_hash:
        raise PolicyValidationError(
            f"Schema hash mismatch: policy has {policy.model_schema_hash!r}, "
            f"bundle has {bundle_hash!r}",
            {
                "policy_hash": policy.model_schema_hash,
                "bundle_hash": bundle_hash,
            },
        )

    bundle_labels = sorted(meta.get("label_set", []))
    if bundle_labels != sorted(policy.model_label_set):
        raise PolicyValidationError(
            f"Label set mismatch: policy has {sorted(policy.model_label_set)!r}, "
            f"bundle has {bundle_labels!r}",
        )

    if policy.calibrator_store_dir is not None:
        store_path = base / policy.calibrator_store_dir
        if not store_path.is_dir():
            raise PolicyValidationError(
                f"Calibrator store directory does not exist: {store_path}",
                {"calibrator_store_dir": policy.calibrator_store_dir},
            )
        store_meta_path = store_path / "store.json"
        if not store_meta_path.is_file():
            raise PolicyValidationError(
                f"Calibrator store metadata not found: {store_meta_path}",
                {"calibrator_store_dir": policy.calibrator_store_dir},
            )

        try:
            store_meta = json.loads(store_meta_path.read_text())
        except (json.JSONDecodeError, OSError) as exc:
            raise PolicyValidationError(
                f"Could not read calibrator store metadata: {exc}",
                {"calibrator_store_dir": policy.calibrator_store_dir},
            ) from exc

        store_model_hash = store_meta.get("model_schema_hash")
        if (
            store_model_hash is not None
            and store_model_hash != policy.model_schema_hash
        ):
            raise PolicyValidationError(
                f"Calibrator store was fitted against a different schema: "
                f"store has {store_model_hash!r}, "
                f"policy expects {policy.model_schema_hash!r}",
                {
                    "store_schema_hash": store_model_hash,
                    "policy_schema_hash": policy.model_schema_hash,
                },
            )