| 1 |
"""Typed errors raised by `dlm.base_models`.""" |
| 2 |
|
| 3 |
from __future__ import annotations |
| 4 |
|
| 5 |
from dataclasses import dataclass, field |
| 6 |
|
| 7 |
|
| 8 |
class BaseModelError(Exception): |
| 9 |
"""Base class for every `dlm.base_models` error.""" |
| 10 |
|
| 11 |
|
| 12 |
class UnknownBaseModelError(BaseModelError): |
| 13 |
"""Spec didn't resolve — not in the registry and not a valid `hf:` escape. |
| 14 |
|
| 15 |
Carries a short list of the registry keys we do know so the caller |
| 16 |
can render a helpful diagnostic. |
| 17 |
""" |
| 18 |
|
| 19 |
def __init__(self, spec: str, known_keys: tuple[str, ...]) -> None: |
| 20 |
self.spec = spec |
| 21 |
self.known_keys = known_keys |
| 22 |
preview = ", ".join(known_keys[:5]) |
| 23 |
tail = "" |
| 24 |
if len(known_keys) > 5: |
| 25 |
tail = f", … ({len(known_keys) - 5} more)" |
| 26 |
super().__init__( |
| 27 |
f"unknown base model {spec!r}. Known keys: {preview}{tail}. " |
| 28 |
"Use `hf:org/name` for models outside the registry." |
| 29 |
) |
| 30 |
|
| 31 |
|
| 32 |
@dataclass(frozen=True) |
| 33 |
class ProbeResult: |
| 34 |
"""Outcome of a single compatibility probe. |
| 35 |
|
| 36 |
`skipped=True` signals the probe couldn't run (e.g., vendored |
| 37 |
llama.cpp not yet installed) but we're choosing not to block — the |
| 38 |
aggregate verdict treats skipped as pass so offline dev stays |
| 39 |
unblocked. `detail` carries the human-readable reason. |
| 40 |
""" |
| 41 |
|
| 42 |
name: str |
| 43 |
passed: bool |
| 44 |
detail: str |
| 45 |
skipped: bool = False |
| 46 |
|
| 47 |
|
| 48 |
class ProbeFailedError(BaseModelError): |
| 49 |
"""One or more compatibility probes failed for a resolved spec. |
| 50 |
|
| 51 |
The error carries every probe's result (pass and fail) so the CLI |
| 52 |
can render a complete picture, not just the first failure. |
| 53 |
""" |
| 54 |
|
| 55 |
def __init__(self, hf_id: str, results: list[ProbeResult]) -> None: |
| 56 |
self.hf_id = hf_id |
| 57 |
self.results = tuple(results) |
| 58 |
failed = [r for r in results if not r.passed] |
| 59 |
failed_summary = "; ".join(f"{r.name}: {r.detail}" for r in failed) |
| 60 |
super().__init__( |
| 61 |
f"{hf_id}: {len(failed)} of {len(results)} probes failed: {failed_summary}" |
| 62 |
) |
| 63 |
|
| 64 |
|
| 65 |
class GatedModelError(BaseModelError): |
| 66 |
"""Model requires license acceptance and the user hasn't accepted. |
| 67 |
|
| 68 |
Lives here because registry probes catch it first; the acceptance |
| 69 |
record is written elsewhere, but the error shape is owned here. |
| 70 |
""" |
| 71 |
|
| 72 |
def __init__(self, hf_id: str, license_url: str | None) -> None: |
| 73 |
self.hf_id = hf_id |
| 74 |
self.license_url = license_url |
| 75 |
where = f" License: {license_url}" if license_url else "" |
| 76 |
super().__init__( |
| 77 |
f"{hf_id} requires license acceptance. Accept the license and " |
| 78 |
f"pass --i-accept-license (or via `dlm init`).{where}" |
| 79 |
) |
| 80 |
|
| 81 |
|
| 82 |
@dataclass(frozen=True) |
| 83 |
class ProbeReport: |
| 84 |
"""Aggregate of probe results; useful for `dlm doctor <base>` reporting.""" |
| 85 |
|
| 86 |
hf_id: str |
| 87 |
results: tuple[ProbeResult, ...] = field(default_factory=tuple) |
| 88 |
|
| 89 |
@property |
| 90 |
def passed(self) -> bool: |
| 91 |
"""All non-skipped probes passed. Skipped probes don't block.""" |
| 92 |
return all(r.passed for r in self.results) |
| 93 |
|
| 94 |
@property |
| 95 |
def failures(self) -> tuple[ProbeResult, ...]: |
| 96 |
return tuple(r for r in self.results if not r.passed) |
| 97 |
|
| 98 |
@property |
| 99 |
def skipped(self) -> tuple[ProbeResult, ...]: |
| 100 |
return tuple(r for r in self.results if r.skipped) |