erabo/odysseus
1
0
Fork 0
mirror of https://github.com/pewdiepie-archdaemon/odysseus.git synced 2026-08-05 02:45:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Merge 7a7f939461 into 20e7fc0164

Browse source
This commit is contained in:
RaresKeY 2026-08-04 11:38:27 -04:00 • committed by GitHub
commit 11b4db1502
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
22 changed files with 3498 additions and 106 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
app.py
View file

@ -83,9 +83,26 @@ from starlette.responses import RedirectResponse
# ========= LOGGING =========
import logging.handlers
from core.constants import DATA_DIR
from core.log_safety import (
CAPABILITY_DIAGNOSTICS_LOGGER,
ScopedDiagnosticsFilter,
application_log_settings,
configure_uvicorn_log_levels,
uvicorn_log_config,
)
_root_logger = logging.getLogger()
_root_logger.setLevel(logging.INFO)
_log_level_name = os.getenv("LOG_LEVEL", "INFO").strip().upper()
_application_log_level, _capability_debug = application_log_settings(_log_level_name)
_root_logger.setLevel(_application_log_level)
configure_uvicorn_log_levels(_application_log_level)
logging.getLogger(CAPABILITY_DIAGNOSTICS_LOGGER).setLevel(
logging.DEBUG if _capability_debug else logging.NOTSET
)
_diagnostics_filter = ScopedDiagnosticsFilter(
_application_log_level,
capability_debug=_capability_debug,
)
_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# Clear existing handlers to avoid duplicates
@ -93,6 +110,8 @@ for _h in list(_root_logger.handlers):
_root_logger.removeHandler(_h)
_console_h = logging.StreamHandler()
_console_h.setLevel(logging.DEBUG)
_console_h.addFilter(_diagnostics_filter)
_console_h.setFormatter(_formatter)
_root_logger.addHandler(_console_h)
@ -107,6 +126,8 @@ try:
_file_h = logging.handlers.RotatingFileHandler(
_log_file, maxBytes=5 * 1024 * 1024, backupCount=3, encoding="utf-8"
)
_file_h.setLevel(logging.DEBUG)
_file_h.addFilter(_diagnostics_filter)
_file_h.setFormatter(_formatter)
_root_logger.addHandler(_file_h)
except Exception as e:
@ -1278,4 +1299,10 @@ if __name__ == "__main__":
bind_host = os.getenv("APP_BIND", "127.0.0.1")
bind_port = int(os.getenv("APP_PORT", "7000"))
uvicorn.run(app, host=bind_host, port=bind_port, log_level="info")
uvicorn.run(
app,
host=bind_host,
port=bind_port,
log_level=_application_log_level,
log_config=uvicorn_log_config(_application_log_level),
)

83
core/log_safety.py
View file

@ -7,9 +7,92 @@ raw leaks those secrets, so route/diagnostic logs run URLs through
also doubles as a sanitizer barrier for CodeQL's clear-text-logging query.
"""
from __future__ import annotations
from copy import deepcopy
import logging
from urllib.parse import urlparse, urlunparse
CAPABILITY_DIAGNOSTICS_LOGGER = "src.model_capability_readers"
UVICORN_LOGGER_NAMES = (
"uvicorn",
"uvicorn.error",
"uvicorn.access",
"uvicorn.asgi",
)
_LOG_LEVELS = {
"DEBUG": logging.DEBUG,
"INFO": logging.INFO,
"WARN": logging.WARNING,
"WARNING": logging.WARNING,
"ERROR": logging.ERROR,
"FATAL": logging.CRITICAL,
"CRITICAL": logging.CRITICAL,
}
def application_log_settings(value: object) -> tuple[int, bool]:
"""Return the safe app level and whether scoped capability debug is on.
Application-wide DEBUG logging can expose request bodies, provider
responses, or credentials from unrelated libraries. The model capability
catalog has a deliberately bounded DEBUG summary, so a DEBUG request is
translated into INFO for the application and enabled only for that logger.
Unknown values also fail closed to INFO.
"""
requested = _LOG_LEVELS.get(str(value or "INFO").strip().upper(), logging.INFO)
return max(requested, logging.INFO), requested == logging.DEBUG
def configure_uvicorn_log_levels(application_level: int) -> None:
"""Apply the mapped app level to Uvicorn's non-propagating loggers.
External entrypoints configure these loggers before importing ``app`` and
otherwise bypass the root logger's level and scoped diagnostics filter.
"""
for logger_name in UVICORN_LOGGER_NAMES:
logging.getLogger(logger_name).setLevel(application_level)
def uvicorn_log_config(application_level: int) -> dict:
"""Return a Uvicorn config that preserves the mapped level on direct runs."""
from uvicorn.config import LOGGING_CONFIG
config = deepcopy(LOGGING_CONFIG)
loggers = config.setdefault("loggers", {})
for logger_name in UVICORN_LOGGER_NAMES:
loggers.setdefault(logger_name, {})["level"] = application_level
return config
class ScopedDiagnosticsFilter(logging.Filter):
"""Allow normal application records plus one explicitly scoped DEBUG log."""
def __init__(
self,
application_level: int,
*,
capability_debug: bool = False,
) -> None:
super().__init__()
self.application_level = application_level
self.capability_debug = capability_debug
def filter(self, record: logging.LogRecord) -> bool:
if record.levelno >= self.application_level:
return True
return (
self.capability_debug
and record.levelno >= logging.DEBUG
and record.name == CAPABILITY_DIAGNOSTICS_LOGGER
)
def redact_url(url: str) -> str:
"""Return a URL safe for logs by removing userinfo and query/fragment.

12
launcher.py
View file

@ -128,9 +128,13 @@ if __name__ == "__main__":
import uvicorn
# Import the FastAPI app from app.py
from app import app
from core.log_safety import application_log_settings, uvicorn_log_config
bind_host = os.getenv("APP_BIND", "127.0.0.1")
bind_port = int(os.getenv("APP_PORT", "7000"))
application_log_level, _ = application_log_settings(
os.getenv("LOG_LEVEL", "INFO")
)
url = f"http://{bind_host}:{bind_port}"
if getattr(sys, 'frozen', False):
@ -139,4 +143,10 @@ if __name__ == "__main__":
# Start system tray manager thread
threading.Thread(target=setup_system_tray, args=(url,), daemon=True).start()
uvicorn.run(app, host=bind_host, port=bind_port, log_level="info")
uvicorn.run(
app,
host=bind_host,
port=bind_port,
log_level=application_log_level,
log_config=uvicorn_log_config(application_log_level),
)

219
src/model_capability_readers/__init__.py
View file

@ -2,92 +2,287 @@
from __future__ import annotations
from collections.abc import Mapping
import logging
from dataclasses import replace
from typing import Any
from src.model_capability_readers import generic_openai, google, llamacpp, lmstudio, ollama, openai, openrouter
from src import provider_capability_schemas as pcs
from src.model_capability_readers import (
anthropic,
chatgpt_subscription,
cohere,
copilot,
generic_openai,
google,
huggingface,
llamacpp,
lmstudio,
mistral,
ollama,
openai,
openrouter,
sglang,
)
from src.model_capability_readers.base import (
CANONICAL_MODEL_SHAPE_VERSION,
CANONICAL_RUNTIME_CONTEXT_SHAPE_VERSION,
ModelCapabilityRecord,
RuntimeContextAllocationRecord,
VENDOR_ANTHROPIC,
VENDOR_CEREBRAS,
VENDOR_CHATGPT_SUBSCRIPTION,
VENDOR_COHERE,
VENDOR_COPILOT,
VENDOR_DEEPSEEK,
VENDOR_FIREWORKS,
VENDOR_GENERIC_OPENAI,
VENDOR_GOOGLE,
VENDOR_GROQ,
VENDOR_HUGGINGFACE,
VENDOR_LLAMACPP,
VENDOR_LMSTUDIO,
VENDOR_MINIMAX,
VENDOR_MISTRAL,
VENDOR_MOONSHOT,
VENDOR_NVIDIA,
VENDOR_OLLAMA,
VENDOR_OPENAI,
VENDOR_OPENROUTER,
VENDOR_SGLANG,
VENDOR_TOGETHER,
VENDOR_UNKNOWN,
VENDOR_VLLM,
VENDOR_XAI,
VENDOR_ZAI,
detect_vendor,
stable_model_id_for,
)
logger = logging.getLogger(__name__)
READER_MODULES = {
VENDOR_GENERIC_OPENAI: generic_openai,
VENDOR_OPENAI: openai,
VENDOR_OPENROUTER: openrouter,
VENDOR_GOOGLE: google,
VENDOR_ANTHROPIC: anthropic,
VENDOR_LLAMACPP: llamacpp,
VENDOR_OLLAMA: ollama,
VENDOR_LMSTUDIO: lmstudio,
VENDOR_MISTRAL: mistral,
VENDOR_COPILOT: copilot,
VENDOR_CHATGPT_SUBSCRIPTION: chatgpt_subscription,
VENDOR_COHERE: cohere,
VENDOR_SGLANG: sglang,
VENDOR_HUGGINGFACE: huggingface,
}
PLACEHOLDER_VENDOR_IDS = frozenset(
{
VENDOR_ANTHROPIC,
VENDOR_HUGGINGFACE,
VENDOR_SGLANG,
VENDOR_VLLM,
}
)
def reader_for_vendor(vendor: Any):
vendor_id = str(vendor or "").strip().lower().replace("-", "_")
vendor_id = pcs.normalize_provider_id(vendor)
return READER_MODULES.get(vendor_id, generic_openai)
def records_from_payload(
payload: Mapping[str, Any],
payload: Any,
*,
vendor: str | None = None,
base_url: str = "",
endpoint_kind: str = "",
endpoint_id: str = "",
) -> tuple[ModelCapabilityRecord, ...]:
vendor_id = vendor or detect_vendor(base_url, endpoint_kind)
resolution = pcs.resolve_provider(
payload,
provider=vendor,
base_url=base_url,
endpoint_kind=endpoint_kind,
)
vendor_id = resolution.provider_id
if vendor_id == pcs.PROVIDER_UNKNOWN:
vendor_id = detect_vendor(base_url, endpoint_kind)
reader = reader_for_vendor(vendor_id)
if reader is generic_openai:
record_vendor = vendor_id if vendor_id not in {VENDOR_UNKNOWN, ""} else VENDOR_GENERIC_OPENAI
return reader.records_from_payload(
record_vendor = vendor_id if vendor_id else VENDOR_UNKNOWN
def annotate(
records: tuple[ModelCapabilityRecord, ...],
*,
shape_id: str,
fallback: bool,
) -> tuple[ModelCapabilityRecord, ...]:
return tuple(
replace(
record,
provider_source=resolution.provider_source,
catalog_shape_id=shape_id,
fallback=fallback,
)
for record in records
)
shape = pcs.catalog_shape_for_id(resolution.shape_id)
if shape is None:
records = generic_openai.records_from_payload(
payload,
vendor_id=record_vendor,
endpoint_id=endpoint_id,
base_url=base_url,
)
return reader.records_from_payload(payload, endpoint_id=endpoint_id, base_url=base_url)
normalized = annotate(
records,
shape_id=resolution.shape_id,
fallback=resolution.fallback,
)
elif resolution.fallback:
normalized_records: list[ModelCapabilityRecord] = []
for item in shape.items(payload):
fallback_record = generic_openai.record_from_model(
item,
vendor_id=record_vendor,
endpoint_id=endpoint_id,
base_url=base_url,
)
if fallback_record:
normalized_records.extend(
annotate(
(fallback_record,),
shape_id=shape.shape_id,
fallback=True,
)
)
normalized = tuple(normalized_records)
else:
normalized_records: list[ModelCapabilityRecord] = []
catalog_items = shape.items(payload)
select_catalog_items = getattr(reader, "select_catalog_items", None)
if callable(select_catalog_items):
catalog_items = tuple(select_catalog_items(catalog_items))
for item in catalog_items:
item_payload = shape.payload_for_item(payload, item)
if shape.item_matches(item):
if reader is generic_openai:
native_records = reader.records_from_payload(
item_payload,
vendor_id=record_vendor,
endpoint_id=endpoint_id,
base_url=base_url,
)
else:
native_records = reader.records_from_payload(
item_payload,
endpoint_id=endpoint_id,
base_url=base_url,
)
if native_records:
normalized_records.extend(
annotate(native_records, shape_id=shape.shape_id, fallback=False)
)
continue
fallback_record = generic_openai.record_from_model(
item,
vendor_id=record_vendor,
endpoint_id=endpoint_id,
base_url=base_url,
)
if fallback_record:
fallback_shape = pcs.fallback_shape_for_payload(item_payload)
normalized_records.extend(
annotate(
(fallback_record,),
shape_id=fallback_shape.shape_id if fallback_shape else "",
fallback=True,
)
)
normalized = tuple(normalized_records)
if logger.isEnabledFor(logging.DEBUG):
families = sorted({record.capability.family for record in normalized})
features = sorted(
{
feature
for record in normalized
for feature in record.capability.capabilities
}
)
controls = sorted(
{
control.control
for record in normalized
for control in record.deterministic_controls
if control.control
}
)
fallback_count = sum(record.fallback for record in normalized)
diagnostic_provider = (
resolution.provider_id
if resolution.provider_id in pcs.PROVIDER_SCHEMAS
else "unknown"
if resolution.provider_id == pcs.PROVIDER_UNKNOWN
else "unregistered"
)
logger.debug(
"[model-capability] normalized: canonical_version=%s provider=%s "
"provider_source=%s catalog_shape=%s fallback=%s records=%d "
"native_records=%d fallback_records=%d "
"families=%s features=%s controls=%s",
CANONICAL_MODEL_SHAPE_VERSION,
diagnostic_provider,
resolution.provider_source,
resolution.shape_id or "unknown",
bool(fallback_count),
len(normalized),
len(normalized) - fallback_count,
fallback_count,
families,
features,
controls,
)
return normalized
__all__ = [
"ModelCapabilityRecord",
"RuntimeContextAllocationRecord",
"CANONICAL_MODEL_SHAPE_VERSION",
"CANONICAL_RUNTIME_CONTEXT_SHAPE_VERSION",
"PLACEHOLDER_VENDOR_IDS",
"READER_MODULES",
"VENDOR_ANTHROPIC",
"VENDOR_CEREBRAS",
"VENDOR_CHATGPT_SUBSCRIPTION",
"VENDOR_COHERE",
"VENDOR_COPILOT",
"VENDOR_DEEPSEEK",
"VENDOR_FIREWORKS",
"VENDOR_GENERIC_OPENAI",
"VENDOR_GOOGLE",
"VENDOR_GROQ",
"VENDOR_HUGGINGFACE",
"VENDOR_LLAMACPP",
"VENDOR_LMSTUDIO",
"VENDOR_MINIMAX",
"VENDOR_MISTRAL",
"VENDOR_MOONSHOT",
"VENDOR_NVIDIA",
"VENDOR_OLLAMA",
"VENDOR_OPENAI",
"VENDOR_OPENROUTER",
"VENDOR_SGLANG",
"VENDOR_TOGETHER",
"VENDOR_UNKNOWN",
"VENDOR_VLLM",
"VENDOR_XAI",
"VENDOR_ZAI",
"detect_vendor",
"reader_for_vendor",
"records_from_payload",

63
src/model_capability_readers/anthropic.py Normal file
View file

@ -0,0 +1,63 @@
"""Anthropic Models API identity reader.
The current Model resource is availability/identity metadata, not an explicit
per-model capability card, so records stay unknown.
"""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from src import model_capabilities as mc
from src.model_capability_readers.base import (
ModelCapabilityRecord,
VENDOR_ANTHROPIC,
compact_str,
model_id_from,
openai_model_items,
stable_model_id_for,
)
vendor = VENDOR_ANTHROPIC
def record_from_model(
raw: Mapping[str, Any],
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> ModelCapabilityRecord | None:
model_id = model_id_from(raw, "id")
if not model_id:
return None
return ModelCapabilityRecord(
vendor=VENDOR_ANTHROPIC,
model_id=model_id,
stable_model_id=stable_model_id_for(
VENDOR_ANTHROPIC,
model_id,
endpoint_id=endpoint_id,
base_url=base_url,
),
display_name=compact_str(raw.get("display_name")) or model_id,
capability=mc.unknown_capability(
source=mc.SOURCE_PROVIDER_READER,
confidence=mc.CONFIDENCE_UNKNOWN,
),
raw=raw,
)
def records_from_payload(
payload: Any,
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> tuple[ModelCapabilityRecord, ...]:
return tuple(
record
for item in openai_model_items(payload)
if (record := record_from_model(item, endpoint_id=endpoint_id, base_url=base_url))
)

183
src/model_capability_readers/base.py
View file

@ -9,6 +9,7 @@ labels.
from __future__ import annotations
import hashlib
import math
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, field
from typing import Any, Protocol
@ -28,8 +29,25 @@ VENDOR_LLAMACPP = "llamacpp"
VENDOR_VLLM = "vllm"
VENDOR_SGLANG = "sglang"
VENDOR_HUGGINGFACE = "huggingface"
VENDOR_MISTRAL = "mistral"
VENDOR_COPILOT = "copilot"
VENDOR_CHATGPT_SUBSCRIPTION = "chatgpt_subscription"
VENDOR_COHERE = "cohere"
VENDOR_MINIMAX = "minimax"
VENDOR_MOONSHOT = "moonshot"
VENDOR_GROQ = "groq"
VENDOR_NVIDIA = "nvidia"
VENDOR_CEREBRAS = "cerebras"
VENDOR_DEEPSEEK = "deepseek"
VENDOR_TOGETHER = "together"
VENDOR_FIREWORKS = "fireworks"
VENDOR_XAI = "xai"
VENDOR_ZAI = "zai"
VENDOR_UNKNOWN = "unknown"
CANONICAL_MODEL_SHAPE_VERSION = 1
CANONICAL_RUNTIME_CONTEXT_SHAPE_VERSION = 1
@dataclass(frozen=True)
class ModelCapabilityRecord:
@ -40,6 +58,9 @@ class ModelCapabilityRecord:
stable_model_id: str = ""
capability_assertions: tuple[mc.CapabilityAssertion, ...] = ()
deterministic_controls: tuple[mc.DeterministicControl, ...] = ()
provider_source: str = "unknown"
catalog_shape_id: str = ""
fallback: bool = False
raw: Mapping[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
@ -58,14 +79,78 @@ class ModelCapabilityRecord:
)
def to_dict(self, *, include_raw: bool = False) -> dict[str, Any]:
controls = tuple(
dict.fromkeys(
control.control
for control in self.deterministic_controls
if control.control
)
)
data = {
"vendor": self.vendor,
"model_id": self.model_id,
"stable_model_id": self.stable_model_id,
"display_name": self.display_name,
"capability": self.capability.to_dict(),
"capability_assertions": [assertion.to_dict() for assertion in self.capability_assertions],
"deterministic_controls": [control.to_dict() for control in self.deterministic_controls],
"schema_version": CANONICAL_MODEL_SHAPE_VERSION,
"provider": self.vendor,
"model": self.model_id,
"stable_id": self.stable_model_id,
"family": self.capability.family,
"task": self.capability.primary_task,
"modalities": self.capability.modalities.to_dict(),
"features": list(self.capability.capabilities),
"limits": dict(self.capability.limits),
"controls": list(controls),
"evidence": {
"source": self.capability.source,
"confidence": self.capability.confidence,
"provider_source": self.provider_source,
"shape": self.catalog_shape_id,
"fallback": self.fallback,
},
}
if include_raw:
data["raw"] = dict(self.raw)
return data
@dataclass(frozen=True)
class RuntimeContextAllocationRecord:
"""Provider-reported context allocated to one currently loaded model.
This runtime state is deliberately separate from ``ModelCapability.limits``:
a model's maximum, configured ``num_ctx``, and the allocation reported for a
loaded process are different facts with different lifetimes.
"""
vendor: str
model_id: str
allocated_context_tokens: int
stable_model_id: str = ""
source: str = mc.SOURCE_PROVIDER_READER
confidence: str = mc.CONFIDENCE_PROVIDER_REPORTED
runtime_shape_id: str = ""
raw: Mapping[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
context_tokens = int_limit(self.allocated_context_tokens)
if context_tokens is None:
raise ValueError("allocated_context_tokens must be a positive integer")
object.__setattr__(self, "allocated_context_tokens", context_tokens)
if not self.stable_model_id:
object.__setattr__(self, "stable_model_id", stable_model_id_for(self.vendor, self.model_id))
def to_dict(self, *, include_raw: bool = False) -> dict[str, Any]:
data = {
"schema_version": CANONICAL_RUNTIME_CONTEXT_SHAPE_VERSION,
"provider": self.vendor,
"model": self.model_id,
"stable_id": self.stable_model_id,
"runtime": {
"allocated_context_tokens": self.allocated_context_tokens,
},
"evidence": {
"source": self.source,
"confidence": self.confidence,
"shape": self.runtime_shape_id,
"scope": "loaded_model",
},
}
if include_raw:
data["raw"] = dict(self.raw)
@ -77,7 +162,7 @@ class CapabilityReader(Protocol):
def records_from_payload(
self,
payload: Mapping[str, Any],
payload: Any,
*,
endpoint_id: Any = "",
base_url: Any = "",
@ -103,6 +188,12 @@ def compact_str(value: Any) -> str:
return str(value or "").strip()
def identity_str(value: Any) -> str:
"""Return a provider identity only when the payload supplied a string."""
return value.strip() if isinstance(value, str) else ""
def _identity_part(value: Any) -> str:
text = compact_str(value).lower()
out = []
@ -135,16 +226,22 @@ def stable_model_id_for(vendor: Any, model_id: Any, *, endpoint_id: Any = "", ba
def model_id_from(raw: Mapping[str, Any], *keys: str) -> str:
for key in keys:
value = compact_str(raw.get(key))
value = identity_str(raw.get(key))
if value:
return value.removeprefix("models/")
return ""
def int_limit(value: Any) -> int | None:
if isinstance(value, bool):
return None
if isinstance(value, float) and (
not math.isfinite(value) or not value.is_integer()
):
return None
try:
limit = int(value)
except (TypeError, ValueError):
except (OverflowError, TypeError, ValueError):
return None
return limit if limit > 0 else None
@ -168,11 +265,14 @@ def deterministic_controls_from_supported_parameters(values: Any) -> tuple[mc.De
)
def openai_model_items(payload: Mapping[str, Any]) -> tuple[Mapping[str, Any], ...]:
payload = as_mapping(payload)
data = payload.get("data")
if data is None:
data = payload.get("models")
def openai_model_items(payload: Any) -> tuple[Mapping[str, Any], ...]:
if isinstance(payload, (list, tuple)):
data = payload
else:
payload = as_mapping(payload)
data = payload.get("data")
if data is None:
data = payload.get("models")
return tuple(item for item in as_list(data) if isinstance(item, Mapping))
@ -254,6 +354,7 @@ def build_capability(
output_modalities: Iterable[str] = (),
capabilities: Iterable[str] = (),
limits: Mapping[str, Any] | None = None,
source: str = mc.SOURCE_PROVIDER_READER,
confidence: str = mc.CONFIDENCE_PROVIDER_REPORTED,
) -> mc.ModelCapability:
return mc.ModelCapability.build(
@ -263,49 +364,21 @@ def build_capability(
output_modalities=tuple(output_modalities),
capabilities=tuple(capabilities),
limits=limits,
source=mc.SOURCE_PROVIDER_READER,
source=source,
confidence=confidence,
)
def detect_vendor(base_url: Any = "", endpoint_kind: Any = "") -> str:
kind = compact_str(endpoint_kind).lower().replace("-", "_")
kind_map = {
"openai": VENDOR_OPENAI,
"openrouter": VENDOR_OPENROUTER,
"google": VENDOR_GOOGLE,
"gemini": VENDOR_GOOGLE,
"anthropic": VENDOR_ANTHROPIC,
"ollama": VENDOR_OLLAMA,
"lmstudio": VENDOR_LMSTUDIO,
"lm_studio": VENDOR_LMSTUDIO,
"llamacpp": VENDOR_LLAMACPP,
"llama_cpp": VENDOR_LLAMACPP,
"vllm": VENDOR_VLLM,
"sglang": VENDOR_SGLANG,
"huggingface": VENDOR_HUGGINGFACE,
"hf": VENDOR_HUGGINGFACE,
}
if kind in kind_map:
return kind_map[kind]
# Import lazily to keep the reader primitives independent of registry load
# order. Exact endpoint kind and host identity are authoritative enough
# for provider selection; default ports are not.
from src import provider_capability_schemas as pcs
parsed = urlparse(compact_str(base_url))
host = (parsed.hostname or "").lower()
port = parsed.port
if host.endswith("openrouter.ai"):
return VENDOR_OPENROUTER
if host.endswith("openai.com"):
return VENDOR_OPENAI
if host.endswith("anthropic.com"):
return VENDOR_ANTHROPIC
if host.endswith("googleapis.com"):
return VENDOR_GOOGLE
if host.endswith("ollama.com") or port == 11434:
return VENDOR_OLLAMA
if port == 1234:
return VENDOR_LMSTUDIO
if port == 8000:
return VENDOR_VLLM
if port == 30000:
return VENDOR_SGLANG
return VENDOR_GENERIC_OPENAI if host else VENDOR_UNKNOWN
resolution = pcs.resolve_provider(
endpoint_kind=endpoint_kind,
base_url=base_url,
)
if resolution.provider_id != pcs.PROVIDER_UNKNOWN:
return resolution.provider_id
return VENDOR_UNKNOWN

111
src/model_capability_readers/chatgpt_subscription.py Normal file
View file

@ -0,0 +1,111 @@
"""ChatGPT Subscription Codex model-list identity reader."""
from __future__ import annotations
import math
from collections.abc import Mapping
from typing import Any
from src import model_capabilities as mc
from src.model_capability_readers.base import (
ModelCapabilityRecord,
VENDOR_CHATGPT_SUBSCRIPTION,
as_list,
as_mapping,
compact_str,
identity_str,
stable_model_id_for,
)
vendor = VENDOR_CHATGPT_SUBSCRIPTION
_DEFAULT_PRIORITY = 10_000
def _priority_rank(raw: Mapping[str, Any]) -> int | float:
value = raw.get("priority")
if isinstance(value, bool):
return _DEFAULT_PRIORITY
if isinstance(value, int):
return value
if isinstance(value, float) and math.isfinite(value):
return value
return _DEFAULT_PRIORITY
def _is_hidden(raw: Mapping[str, Any]) -> bool:
visibility = raw.get("visibility")
return (
isinstance(visibility, str)
and visibility.strip().lower() in {"hide", "hidden"}
)
def select_catalog_items(
items: tuple[Mapping[str, Any], ...],
) -> tuple[Mapping[str, Any], ...]:
"""Apply the provider's visibility, priority, and slug de-duplication."""
sortable: list[tuple[int | float, str, int, Mapping[str, Any]]] = []
passthrough: list[Mapping[str, Any]] = []
for position, item in enumerate(items):
if _is_hidden(item):
continue
slug = identity_str(item.get("slug"))
if not slug:
passthrough.append(item)
continue
sortable.append((_priority_rank(item), slug, position, item))
sortable.sort(key=lambda entry: (entry[0], entry[1], entry[2]))
selected: list[Mapping[str, Any]] = []
seen: set[str] = set()
for _, slug, _, item in sortable:
if slug not in seen:
selected.append(item)
seen.add(slug)
selected.extend(passthrough)
return tuple(selected)
def record_from_model(
raw: Mapping[str, Any],
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> ModelCapabilityRecord | None:
model_id = identity_str(raw.get("slug"))
if not model_id:
return None
return ModelCapabilityRecord(
vendor=VENDOR_CHATGPT_SUBSCRIPTION,
model_id=model_id,
stable_model_id=stable_model_id_for(
VENDOR_CHATGPT_SUBSCRIPTION,
model_id,
endpoint_id=endpoint_id,
base_url=base_url,
),
display_name=compact_str(raw.get("display_name") or raw.get("title")) or model_id,
capability=mc.unknown_capability(
source=mc.SOURCE_PROVIDER_READER,
confidence=mc.CONFIDENCE_UNKNOWN,
),
raw=raw,
)
def records_from_payload(
payload: Any,
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> tuple[ModelCapabilityRecord, ...]:
values = as_mapping(payload).get("models")
return tuple(
record
for item in select_catalog_items(
tuple(item for item in as_list(values) if isinstance(item, Mapping))
)
if (record := record_from_model(item, endpoint_id=endpoint_id, base_url=base_url))
)

111
src/model_capability_readers/cohere.py Normal file
View file

@ -0,0 +1,111 @@
"""Cohere native model-catalog capability reader.
The `/v1/models` resource reports endpoint compatibility and context size per
model. It does not prove provider-wide chat/tool support for every model, so
the reader maps only those exact model-card fields.
"""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from src import model_capabilities as mc
from src.model_capability_readers.base import (
ModelCapabilityRecord,
VENDOR_COHERE,
as_list,
as_mapping,
build_capability,
compact_str,
deterministic_controls_from_supported_parameters,
identity_str,
int_limit,
stable_model_id_for,
)
vendor = VENDOR_COHERE
_ENDPOINT_FAMILIES = {
"chat": mc.FAMILY_CHAT,
"generate": mc.FAMILY_CHAT,
"embed": mc.FAMILY_EMBEDDING,
"rerank": mc.FAMILY_RERANK,
"classify": mc.FAMILY_CLASSIFICATION,
}
def _family(raw: Mapping[str, Any]) -> str:
families = {
family
for value in as_list(raw.get("endpoints"))
if (family := _ENDPOINT_FAMILIES.get(compact_str(value).lower()))
}
return next(iter(families)) if len(families) == 1 else mc.FAMILY_UNKNOWN
def _modalities(family: str) -> tuple[tuple[str, ...], tuple[str, ...]]:
if family == mc.FAMILY_CHAT:
return (mc.MODALITY_TEXT,), (mc.MODALITY_TEXT,)
if family == mc.FAMILY_EMBEDDING:
return (mc.MODALITY_TEXT,), (mc.MODALITY_EMBEDDING,)
if family in {mc.FAMILY_RERANK, mc.FAMILY_CLASSIFICATION}:
return (mc.MODALITY_TEXT,), (mc.MODALITY_TEXT,)
return (), ()
def record_from_model(
raw: Mapping[str, Any],
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> ModelCapabilityRecord | None:
model_id = identity_str(raw.get("name"))
if not model_id:
return None
family = _family(raw)
inputs, outputs = _modalities(family)
context_tokens = int_limit(raw.get("context_length"))
limits = {"context_tokens": context_tokens} if context_tokens else {}
sampling_defaults = as_mapping(raw.get("sampling_defaults"))
sampling_controls = (
"top_p" if key == "p" else "top_k" if key == "k" else key
for key in sampling_defaults
)
return ModelCapabilityRecord(
vendor=VENDOR_COHERE,
model_id=model_id,
stable_model_id=stable_model_id_for(
VENDOR_COHERE,
model_id,
endpoint_id=endpoint_id,
base_url=base_url,
),
display_name=model_id,
capability=build_capability(
family=family,
input_modalities=inputs,
output_modalities=outputs,
limits=limits,
),
deterministic_controls=deterministic_controls_from_supported_parameters(
sampling_controls
),
raw=raw,
)
def records_from_payload(
payload: Any,
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> tuple[ModelCapabilityRecord, ...]:
values = as_mapping(payload).get("models")
return tuple(
record
for item in as_list(values)
if isinstance(item, Mapping)
if (record := record_from_model(item, endpoint_id=endpoint_id, base_url=base_url))
)

121
src/model_capability_readers/copilot.py Normal file
View file

@ -0,0 +1,121 @@
"""GitHub Copilot model-catalog capability reader."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from src import model_capabilities as mc
from src.model_capability_readers.base import (
ModelCapabilityRecord,
VENDOR_COPILOT,
as_mapping,
build_capability,
compact_str,
int_limit,
merge_unique,
model_id_from,
openai_model_items,
stable_model_id_for,
)
vendor = VENDOR_COPILOT
_SUPPORT_CAPABILITIES = {
"tool_calls": mc.CAP_TOOL_CALL,
"vision": mc.CAP_VISION,
}
def select_catalog_items(
items: tuple[Mapping[str, Any], ...],
) -> tuple[Mapping[str, Any], ...]:
"""Keep picker-enabled models when the catalog advertises any of them."""
if any(item.get("model_picker_enabled") is True for item in items):
return tuple(
item for item in items if item.get("model_picker_enabled") is True
)
return items
def _supports(raw: Mapping[str, Any]) -> Mapping[str, Any]:
return as_mapping(as_mapping(raw.get("capabilities")).get("supports"))
def _limits(raw: Mapping[str, Any]) -> dict[str, int]:
payload = as_mapping(raw.get("limits"))
out: dict[str, int] = {}
for keys, target in (
(("max_prompt_tokens", "input_tokens"), "input_tokens"),
(("max_output_tokens", "output_tokens"), "output_tokens"),
(("max_context_tokens", "context_window"), "context_tokens"),
):
for key in keys:
value = int_limit(payload.get(key)) or int_limit(raw.get(key))
if value:
out[target] = value
break
return out
def record_from_model(
raw: Mapping[str, Any],
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> ModelCapabilityRecord | None:
model_id = model_id_from(raw, "id")
if not model_id:
return None
supports = _supports(raw)
capabilities = merge_unique(
_SUPPORT_CAPABILITIES[key]
for key, enabled in supports.items()
if enabled is True and key in _SUPPORT_CAPABILITIES
)
picker_enabled = raw.get("model_picker_enabled") is True
if picker_enabled or capabilities:
inputs = [mc.MODALITY_TEXT]
if mc.CAP_VISION in capabilities:
inputs.append(mc.MODALITY_IMAGE)
capability = build_capability(
family=mc.FAMILY_CHAT,
input_modalities=inputs,
output_modalities=(mc.MODALITY_TEXT,),
capabilities=capabilities,
limits=_limits(raw),
)
else:
capability = mc.unknown_capability(
source=mc.SOURCE_PROVIDER_READER,
confidence=mc.CONFIDENCE_UNKNOWN,
)
return ModelCapabilityRecord(
vendor=VENDOR_COPILOT,
model_id=model_id,
stable_model_id=stable_model_id_for(
VENDOR_COPILOT,
model_id,
endpoint_id=endpoint_id,
base_url=base_url,
),
display_name=compact_str(raw.get("name")) or model_id,
capability=capability,
raw=raw,
)
def records_from_payload(
payload: Any,
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> tuple[ModelCapabilityRecord, ...]:
records: list[ModelCapabilityRecord] = []
for item in select_catalog_items(openai_model_items(payload)):
record = record_from_model(item, endpoint_id=endpoint_id, base_url=base_url)
if record:
records.append(record)
return tuple(records)

41
src/model_capability_readers/generic_openai.py
View file

@ -1,4 +1,9 @@
"""Reader for bare OpenAI-compatible model-list payloads."""
"""Inventory-only reader for unrecognized model-list envelopes.
Common field names are not a cross-provider capability contract. This reader
therefore recovers model identity and preserves the original record, but never
promotes tasks, modalities, parameters, limits, or capability booleans.
"""
from __future__ import annotations
@ -26,25 +31,34 @@ def record_from_model(
endpoint_id: Any = "",
base_url: Any = "",
) -> ModelCapabilityRecord | None:
model_id = model_id_from(raw, "id", "name", "model")
model_id = model_id_from(raw, "id", "name", "model", "key", "slug")
if not model_id:
return None
capability = mc.unknown_capability(
source=mc.SOURCE_PROVIDER_READER,
confidence=mc.CONFIDENCE_UNKNOWN,
)
return ModelCapabilityRecord(
vendor=vendor_id,
model_id=model_id,
stable_model_id=stable_model_id_for(vendor_id, model_id, endpoint_id=endpoint_id, base_url=base_url),
display_name=compact_str(raw.get("display_name") or raw.get("name")),
capability=capability,
stable_model_id=stable_model_id_for(
vendor_id,
model_id,
endpoint_id=endpoint_id,
base_url=base_url,
),
display_name=compact_str(
raw.get("display_name")
or raw.get("name")
or raw.get("key")
or raw.get("slug")
),
capability=mc.unknown_capability(
source=mc.SOURCE_PROVIDER_READER,
confidence=mc.CONFIDENCE_UNKNOWN,
),
raw=raw,
)
def records_from_payload(
payload: Mapping[str, Any],
payload: Any,
*,
vendor_id: str = VENDOR_GENERIC_OPENAI,
endpoint_id: Any = "",
@ -52,7 +66,12 @@ def records_from_payload(
) -> tuple[ModelCapabilityRecord, ...]:
records: list[ModelCapabilityRecord] = []
for item in openai_model_items(payload):
record = record_from_model(item, vendor_id=vendor_id, endpoint_id=endpoint_id, base_url=base_url)
record = record_from_model(
item,
vendor_id=vendor_id,
endpoint_id=endpoint_id,
base_url=base_url,
)
if record:
records.append(record)
return tuple(records)

6
src/model_capability_readers/google.py
View file

@ -20,7 +20,11 @@ vendor = VENDOR_GOOGLE
def _model_items(payload: Mapping[str, Any]) -> tuple[Mapping[str, Any], ...]:
models = payload.get("models") if isinstance(payload, Mapping) else None
if models is None and isinstance(payload, Mapping) and payload.get("name"):
if (
models is None
and isinstance(payload, Mapping)
and ai_studio.google_model_id(payload)
):
models = [payload]
return tuple(item for item in as_list(models) if isinstance(item, Mapping))

9
src/model_capability_readers/google_ai_studio_mapping.py
View file

@ -13,7 +13,12 @@ from collections.abc import Mapping
from typing import Any
from src import model_capabilities as mc
from src.model_capability_readers.base import as_list, compact_str, int_limit
from src.model_capability_readers.base import (
as_list,
compact_str,
identity_str,
int_limit,
)
METHOD_GENERATE_CONTENT = "generateContent"
@ -55,7 +60,7 @@ MODEL_FIELD_MAP = {
def google_model_id(raw: Mapping[str, Any]) -> str:
value = compact_str(raw.get("baseModelId")) or compact_str(raw.get("name"))
value = identity_str(raw.get("baseModelId")) or identity_str(raw.get("name"))
return value.removeprefix("models/")

151
src/model_capability_readers/huggingface.py Normal file
View file

@ -0,0 +1,151 @@
"""Hugging Face Hub model-info reader using explicit pipeline metadata."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from src import model_capabilities as mc
from src.model_capability_readers.base import (
ModelCapabilityRecord,
VENDOR_HUGGINGFACE,
build_capability,
compact_str,
model_id_from,
stable_model_id_for,
)
vendor = VENDOR_HUGGINGFACE
# Hugging Face publishes ``pipeline_tag`` as a provider-owned task enum. Keep
# its interpretation here, rather than teaching the inventory fallback that a
# similarly named field has the same meaning for every provider.
_PIPELINE_SHAPES = {
"text-generation": (mc.FAMILY_CHAT, (mc.MODALITY_TEXT,), (mc.MODALITY_TEXT,), ()),
"image-text-to-text": (
mc.FAMILY_CHAT,
(mc.MODALITY_TEXT, mc.MODALITY_IMAGE),
(mc.MODALITY_TEXT,),
(mc.CAP_VISION,),
),
"image-question-answering": (
mc.FAMILY_CHAT,
(mc.MODALITY_TEXT, mc.MODALITY_IMAGE),
(mc.MODALITY_TEXT,),
(mc.CAP_VISION,),
),
"feature-extraction": (
mc.FAMILY_EMBEDDING,
(mc.MODALITY_TEXT,),
(mc.MODALITY_EMBEDDING,),
(),
),
"text-to-image": (
mc.FAMILY_IMAGE,
(mc.MODALITY_TEXT,),
(mc.MODALITY_IMAGE,),
(mc.CAP_IMAGE_GENERATION,),
),
"image-to-image": (
mc.FAMILY_IMAGE,
(mc.MODALITY_IMAGE,),
(mc.MODALITY_IMAGE,),
(mc.CAP_IMAGE_GENERATION, mc.CAP_IMAGE_EDITING),
),
"text-to-video": (
mc.FAMILY_VIDEO,
(mc.MODALITY_TEXT,),
(mc.MODALITY_VIDEO,),
(mc.CAP_VIDEO_GENERATION,),
),
"automatic-speech-recognition": (
mc.FAMILY_AUDIO,
(mc.MODALITY_AUDIO,),
(mc.MODALITY_TEXT,),
(mc.CAP_TRANSCRIPTION,),
),
"text-to-speech": (
mc.FAMILY_AUDIO,
(mc.MODALITY_TEXT,),
(mc.MODALITY_AUDIO,),
(mc.CAP_TTS,),
),
"text-classification": (
mc.FAMILY_CLASSIFICATION,
(mc.MODALITY_TEXT,),
(mc.MODALITY_TEXT,),
(),
),
}
def _capability_from_pipeline_tag(value: Any) -> mc.ModelCapability:
shape = _PIPELINE_SHAPES.get(compact_str(value).lower())
if not shape:
return mc.unknown_capability(
source=mc.SOURCE_COOKBOOK_HF,
confidence=mc.CONFIDENCE_UNKNOWN,
)
family, input_modalities, output_modalities, capabilities = shape
return build_capability(
family=family,
input_modalities=input_modalities,
output_modalities=output_modalities,
capabilities=capabilities,
source=mc.SOURCE_COOKBOOK_HF,
confidence=mc.CONFIDENCE_REGISTRY,
)
def record_from_model(
raw: Mapping[str, Any],
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> ModelCapabilityRecord | None:
model_id = model_id_from(raw, "modelId", "id")
if not model_id:
return None
return ModelCapabilityRecord(
vendor=VENDOR_HUGGINGFACE,
model_id=model_id,
stable_model_id=stable_model_id_for(
VENDOR_HUGGINGFACE,
model_id,
endpoint_id=endpoint_id,
base_url=base_url,
),
display_name=(
compact_str(
raw.get("cardData", {}).get("pretty_name")
if isinstance(raw.get("cardData"), Mapping)
else ""
)
or model_id
),
capability=_capability_from_pipeline_tag(raw.get("pipeline_tag")),
raw=raw,
)
def records_from_payload(
payload: Any,
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> tuple[ModelCapabilityRecord, ...]:
if isinstance(payload, Mapping):
record = record_from_model(payload, endpoint_id=endpoint_id, base_url=base_url)
return (record,) if record else ()
if not isinstance(payload, (list, tuple)):
return ()
records: list[ModelCapabilityRecord] = []
for item in payload:
if not isinstance(item, Mapping):
continue
record = record_from_model(item, endpoint_id=endpoint_id, base_url=base_url)
if record:
records.append(record)
return tuple(records)

5
src/model_capability_readers/llamacpp.py
View file

@ -22,6 +22,7 @@ from src.model_capability_readers.base import (
build_capability,
compact_str,
deterministic_controls_from_supported_parameters,
identity_str,
int_limit,
merge_unique,
model_id_from,
@ -53,10 +54,10 @@ def _server_model_entries(payload: Mapping[str, Any]) -> tuple[Mapping[str, Any]
def _model_id_from_props(payload: Mapping[str, Any]) -> str:
payload = as_mapping(payload)
model_alias = compact_str(payload.get("model_alias"))
model_alias = identity_str(payload.get("model_alias"))
if model_alias:
return model_alias
model_path = compact_str(payload.get("model_path"))
model_path = identity_str(payload.get("model_path"))
if model_path:
return PurePosixPath(model_path).name
return ""

111
src/model_capability_readers/mistral.py Normal file
View file

@ -0,0 +1,111 @@
"""Mistral native model-catalog capability reader."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from src import model_capabilities as mc
from src.model_capability_readers.base import (
ModelCapabilityRecord,
VENDOR_MISTRAL,
as_mapping,
build_capability,
compact_str,
int_limit,
merge_unique,
model_id_from,
openai_model_items,
stable_model_id_for,
)
vendor = VENDOR_MISTRAL
def _family(raw: Mapping[str, Any]) -> str:
capabilities = as_mapping(raw.get("capabilities"))
if capabilities.get("classification") is True and not (
capabilities.get("completion_chat") is True
or capabilities.get("completion_fim") is True
):
return mc.FAMILY_CLASSIFICATION
if capabilities.get("completion_chat") is True or capabilities.get("completion_fim") is True:
return mc.FAMILY_CHAT
return mc.FAMILY_UNKNOWN
def _capabilities(raw: Mapping[str, Any]) -> tuple[str, ...]:
payload = as_mapping(raw.get("capabilities"))
values: list[str] = []
for key, capability in (
("vision", mc.CAP_VISION),
("function_calling", mc.CAP_TOOL_CALL),
("reasoning", mc.CAP_REASONING),
("structured_output", mc.CAP_STRUCTURED_OUTPUT),
("structured_outputs", mc.CAP_STRUCTURED_OUTPUT),
):
if payload.get(key) is True:
values.append(capability)
return merge_unique(values)
def record_from_model(
raw: Mapping[str, Any],
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> ModelCapabilityRecord | None:
model_id = model_id_from(raw, "id")
if not model_id:
return None
family = _family(raw)
capabilities = _capabilities(raw)
if family == mc.FAMILY_CHAT:
inputs = [mc.MODALITY_TEXT]
if mc.CAP_VISION in capabilities:
inputs.append(mc.MODALITY_IMAGE)
input_modalities = tuple(inputs)
output_modalities = (mc.MODALITY_TEXT,)
elif family == mc.FAMILY_CLASSIFICATION:
input_modalities = (mc.MODALITY_TEXT,)
output_modalities = (mc.MODALITY_TEXT,)
else:
input_modalities = ()
output_modalities = ()
context_tokens = int_limit(raw.get("max_context_length"))
limits = {"context_tokens": context_tokens} if context_tokens else {}
capability = build_capability(
family=family,
input_modalities=input_modalities,
output_modalities=output_modalities,
capabilities=capabilities,
limits=limits,
)
return ModelCapabilityRecord(
vendor=VENDOR_MISTRAL,
model_id=model_id,
stable_model_id=stable_model_id_for(
VENDOR_MISTRAL,
model_id,
endpoint_id=endpoint_id,
base_url=base_url,
),
display_name=compact_str(raw.get("name")) or model_id,
capability=capability,
raw=raw,
)
def records_from_payload(
payload: Any,
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> tuple[ModelCapabilityRecord, ...]:
records: list[ModelCapabilityRecord] = []
for item in openai_model_items(payload):
record = record_from_model(item, endpoint_id=endpoint_id, base_url=base_url)
if record:
records.append(record)
return tuple(records)

82
src/model_capability_readers/ollama.py
View file

@ -8,11 +8,13 @@ from typing import Any
from src import model_capabilities as mc
from src.model_capability_readers.base import (
ModelCapabilityRecord,
RuntimeContextAllocationRecord,
VENDOR_OLLAMA,
as_list,
as_mapping,
build_capability,
compact_str,
identity_str,
int_limit,
merge_unique,
model_id_from,
@ -23,6 +25,9 @@ from src.model_capability_readers.base import (
vendor = VENDOR_OLLAMA
OLLAMA_PS_SHAPE_ID = "ollama.ps.v1"
_CAPABILITY_MAP = {
"completion": None,
"completions": None,
@ -59,17 +64,11 @@ def _family_from_ollama_capabilities(values: Any) -> str:
def _parameters_mapping(value: Any) -> Mapping[str, Any]:
if isinstance(value, Mapping):
return value
text = compact_str(value)
if not text:
return {}
parsed: dict[str, str] = {}
for line in text.splitlines():
parts = line.strip().split(None, 1)
if len(parts) == 2:
parsed[parts[0]] = parts[1]
return parsed
# `/api/show` currently serializes this field as Modelfile text. Do not
# recover capability truth by reparsing that late text; prefer the native
# structured `model_info.*.context_length` shape. Mapping support remains
# for compatible servers that already return structured parameters.
return value if isinstance(value, Mapping) else {}
def _modalities_for_family(family: str, capabilities: tuple[str, ...]) -> tuple[tuple[str, ...], tuple[str, ...]]:
@ -115,6 +114,65 @@ def _limits_from_show(raw: Mapping[str, Any]) -> dict[str, Any]:
return limits
def _runtime_model_key(value: Any) -> str:
model_id = identity_str(value).casefold()
return model_id.removesuffix(":latest")
def runtime_context_from_ps_payload(
model_id: str,
payload: Mapping[str, Any],
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> RuntimeContextAllocationRecord | None:
"""Normalize one loaded model's allocation from Ollama ``GET /api/ps``.
Exact identity is preferred, with only Ollama's implicit ``:latest`` alias
normalized. Conflicting matching rows fail closed instead of choosing an
arbitrary allocation.
"""
requested_id = identity_str(model_id)
requested_key = _runtime_model_key(requested_id)
if not requested_key:
return None
matches: list[tuple[int, Mapping[str, Any]]] = []
for item in as_list(as_mapping(payload).get("models")):
if not isinstance(item, Mapping):
continue
identities = tuple(
identity
for identity in (identity_str(item.get("model")), identity_str(item.get("name")))
if identity
)
if not any(_runtime_model_key(identity) == requested_key for identity in identities):
continue
context_tokens = int_limit(item.get("context_length"))
if context_tokens is not None:
matches.append((context_tokens, item))
allocations = {context_tokens for context_tokens, _item in matches}
if len(allocations) != 1:
return None
context_tokens, raw = matches[0]
return RuntimeContextAllocationRecord(
vendor=VENDOR_OLLAMA,
model_id=requested_id,
stable_model_id=stable_model_id_for(
VENDOR_OLLAMA,
requested_id,
endpoint_id=endpoint_id,
base_url=base_url,
),
allocated_context_tokens=context_tokens,
runtime_shape_id=OLLAMA_PS_SHAPE_ID,
raw=raw,
)
def record_from_show_payload(
model_id: str,
payload: Mapping[str, Any],
@ -122,7 +180,7 @@ def record_from_show_payload(
endpoint_id: Any = "",
base_url: Any = "",
) -> ModelCapabilityRecord | None:
model_id = compact_str(model_id) or model_id_from(payload, "model", "name")
model_id = identity_str(model_id) or model_id_from(payload, "model", "name")
if not model_id:
return None
capability_values = payload.get("capabilities")

112
src/model_capability_readers/sglang.py Normal file
View file

@ -0,0 +1,112 @@
"""SGLang `/model_info` and OpenAI model-card reader."""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import replace
from pathlib import PurePosixPath
from typing import Any
from src import model_capabilities as mc
from src.model_capability_readers import generic_openai
from src.model_capability_readers.base import (
ModelCapabilityRecord,
VENDOR_SGLANG,
as_mapping,
build_capability,
deterministic_controls_from_supported_parameters,
identity_str,
int_limit,
openai_model_items,
stable_model_id_for,
)
vendor = VENDOR_SGLANG
def _model_id(payload: Mapping[str, Any]) -> str:
value = identity_str(payload.get("served_model_name")) or identity_str(
payload.get("model_path")
)
if not value:
return ""
return PurePosixPath(value).name if value.startswith("/") else value
def record_from_model_info(
payload: Mapping[str, Any],
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> ModelCapabilityRecord | None:
model_id = _model_id(payload)
if not model_id:
return None
capabilities: list[str] = []
inputs: list[str] = []
outputs: list[str] = []
family = mc.FAMILY_UNKNOWN
if payload.get("is_generation") is True:
family = mc.FAMILY_CHAT
inputs.append(mc.MODALITY_TEXT)
outputs.append(mc.MODALITY_TEXT)
if payload.get("has_image_understanding") is True:
inputs.append(mc.MODALITY_IMAGE)
capabilities.append(mc.CAP_VISION)
if payload.get("has_audio_understanding") is True:
inputs.append(mc.MODALITY_AUDIO)
capabilities.append(mc.CAP_AUDIO_INPUT)
capability = build_capability(
family=family,
input_modalities=inputs,
output_modalities=outputs,
capabilities=capabilities,
)
sampling = as_mapping(payload.get("preferred_sampling_params"))
return ModelCapabilityRecord(
vendor=VENDOR_SGLANG,
model_id=model_id,
stable_model_id=stable_model_id_for(
VENDOR_SGLANG,
model_id,
endpoint_id=endpoint_id,
base_url=base_url,
),
display_name=model_id,
capability=capability,
deterministic_controls=deterministic_controls_from_supported_parameters(sampling.keys()),
raw=payload,
)
def records_from_payload(
payload: Any,
*,
endpoint_id: Any = "",
base_url: Any = "",
) -> tuple[ModelCapabilityRecord, ...]:
mapping = as_mapping(payload)
if "is_generation" in mapping and "model_path" in mapping:
record = record_from_model_info(mapping, endpoint_id=endpoint_id, base_url=base_url)
return (record,) if record else ()
records: list[ModelCapabilityRecord] = []
for item in openai_model_items(payload):
record = generic_openai.record_from_model(
item,
vendor_id=VENDOR_SGLANG,
endpoint_id=endpoint_id,
base_url=base_url,
)
if record:
context_tokens = int_limit(item.get("max_model_len"))
if context_tokens:
record = replace(
record,
capability=build_capability(
family=mc.FAMILY_UNKNOWN,
limits={"context_tokens": context_tokens},
),
)
records.append(record)
return tuple(records)

758
src/provider_capability_schemas.py Normal file
View file

@ -0,0 +1,758 @@
"""Provider identity and native model-catalog shape detection.
The registry has one narrow job: identify a configured provider and recognize
tested provider-native catalog envelopes. Generic ``data``/``models``/list
envelopes are marked as fallback inventory only; they never promote model
capabilities.
Request/response transport fields and model-specific behavior belong to their
runtime adapters, not this catalog detector.
"""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any
from urllib.parse import urlparse
PROVIDER_UNKNOWN = "unknown"
PROVIDER_GENERIC_OPENAI = "generic_openai"
PROVIDER_SOURCE_EXPLICIT = "explicit"
PROVIDER_SOURCE_ENDPOINT_KIND = "endpoint_kind"
PROVIDER_SOURCE_HOST = "host"
PROVIDER_SOURCE_PAYLOAD = "payload"
PROVIDER_SOURCE_UNKNOWN = "unknown"
ENVELOPE_DATA = "data"
ENVELOPE_MODELS = "models"
ENVELOPE_BARE_LIST = "bare_list"
ENVELOPE_SINGLE = "single"
_MISSING = object()
def _token(value: Any) -> str:
return str(value or "").strip().lower().replace("-", "_").replace(" ", "_")
def _path_value(value: Any, path: str) -> Any:
current = value
for part in path.split("."):
if not isinstance(current, Mapping) or part not in current:
return _MISSING
current = current[part]
return current
def _path_present(value: Any, path: str) -> bool:
return _path_value(value, path) is not _MISSING
def _items_for_envelope(payload: Any, envelope: str) -> tuple[Mapping[str, Any], ...]:
if envelope == ENVELOPE_BARE_LIST:
values = payload if isinstance(payload, (list, tuple)) else ()
elif envelope == ENVELOPE_SINGLE:
values = (payload,) if isinstance(payload, Mapping) else ()
elif isinstance(payload, Mapping):
values = payload.get(envelope)
values = values if isinstance(values, (list, tuple)) else ()
else:
values = ()
return tuple(item for item in values if isinstance(item, Mapping))
@dataclass(frozen=True)
class ProviderCatalogShape:
"""A tested provider-native shape or an explicit inventory fallback."""
shape_id: str
provider_id: str
envelope: str
identity_paths: tuple[str, ...]
required_root_paths: tuple[str, ...] = ()
required_item_paths: tuple[str, ...] = ()
required_item_any_paths: tuple[str, ...] = ()
item_types: tuple[tuple[str, tuple[Any, ...]], ...] = ()
item_values: tuple[tuple[str, tuple[Any, ...]], ...] = ()
detection_priority: int = 0
fallback: bool = False
def items(self, payload: Any) -> tuple[Mapping[str, Any], ...]:
return _items_for_envelope(payload, self.envelope)
def item_matches(self, item: Mapping[str, Any]) -> bool:
if self.identity_paths and not any(
(value := _path_value(item, path)) is not _MISSING
and isinstance(value, str)
and bool(value.strip())
for path in self.identity_paths
):
return False
if not all(_path_present(item, path) for path in self.required_item_paths):
return False
if self.required_item_any_paths and not any(
_path_present(item, path) for path in self.required_item_any_paths
):
return False
if any(
not isinstance(_path_value(item, path), expected_types)
for path, expected_types in self.item_types
):
return False
if any(_path_value(item, path) not in expected for path, expected in self.item_values):
return False
return True
def matches(self, payload: Any) -> bool:
if self.required_root_paths:
if not isinstance(payload, Mapping):
return False
if not all(_path_present(payload, path) for path in self.required_root_paths):
return False
return any(self.item_matches(item) for item in self.items(payload))
def payload_for_item(self, payload: Any, item: Mapping[str, Any]) -> Any:
"""Return a one-item payload in the same provider-native envelope."""
if self.envelope == ENVELOPE_BARE_LIST:
return [item]
if self.envelope == ENVELOPE_SINGLE:
return {
key: value
for key, value in item.items()
if key not in {ENVELOPE_DATA, ENVELOPE_MODELS}
}
if isinstance(payload, Mapping):
narrowed = {
key: value
for key, value in payload.items()
if key not in {ENVELOPE_DATA, ENVELOPE_MODELS}
}
narrowed[self.envelope] = [item]
return narrowed
return {self.envelope: [item]}
@dataclass(frozen=True)
class ProviderCapabilitySchema:
provider_id: str
aliases: tuple[str, ...] = ()
host_suffixes: tuple[str, ...] = ()
catalog_shapes: tuple[ProviderCatalogShape, ...] = ()
@dataclass(frozen=True)
class ProviderResolution:
provider_id: str = PROVIDER_UNKNOWN
provider_source: str = PROVIDER_SOURCE_UNKNOWN
shape_id: str = ""
fallback: bool = False
def to_dict(self) -> dict[str, Any]:
return {
"provider": self.provider_id,
"provider_source": self.provider_source,
"shape": self.shape_id,
"fallback": self.fallback,
}
# Generic envelopes are inventory fallbacks only. Their field names are not a
# portable capability contract, so readers may recover identity but nothing
# else from them.
GENERAL_DATA_SHAPE = ProviderCatalogShape(
shape_id="fallback.models.data.v1",
provider_id=PROVIDER_UNKNOWN,
envelope=ENVELOPE_DATA,
identity_paths=("id", "name", "model", "key", "slug"),
fallback=True,
)
GENERAL_MODELS_SHAPE = ProviderCatalogShape(
shape_id="fallback.models.envelope.v1",
provider_id=PROVIDER_UNKNOWN,
envelope=ENVELOPE_MODELS,
identity_paths=("id", "name", "model", "key", "slug"),
fallback=True,
)
GENERAL_BARE_SHAPE = ProviderCatalogShape(
shape_id="fallback.models.list.v1",
provider_id=PROVIDER_UNKNOWN,
envelope=ENVELOPE_BARE_LIST,
identity_paths=("id", "name", "model", "key", "slug"),
fallback=True,
)
FALLBACK_CATALOG_SHAPES = (
GENERAL_DATA_SHAPE,
GENERAL_MODELS_SHAPE,
GENERAL_BARE_SHAPE,
)
OPENAI_MODELS_SHAPE = ProviderCatalogShape(
shape_id="openai.models.identity.v1",
provider_id="openai",
envelope=ENVELOPE_DATA,
identity_paths=("id",),
required_item_paths=("object", "created", "owned_by"),
item_values=(("object", ("model",)),),
)
OPENROUTER_MODELS_SHAPE = ProviderCatalogShape(
shape_id="openrouter.models.rich.v1",
provider_id="openrouter",
envelope=ENVELOPE_DATA,
identity_paths=("id",),
required_item_paths=(
"architecture",
"canonical_slug",
"pricing",
"supported_parameters",
"top_provider",
),
item_types=(
("architecture", (Mapping,)),
("canonical_slug", (str,)),
("pricing", (Mapping,)),
("supported_parameters", (list, tuple)),
("top_provider", (Mapping,)),
),
detection_priority=90,
)
GOOGLE_MODELS_SHAPE = ProviderCatalogShape(
shape_id="google.generative-language.models.v1beta",
provider_id="google",
envelope=ENVELOPE_MODELS,
identity_paths=("baseModelId", "name"),
required_item_paths=("supportedGenerationMethods",),
item_types=(("supportedGenerationMethods", (list, tuple)),),
detection_priority=100,
)
GOOGLE_MODEL_SHAPE = ProviderCatalogShape(
shape_id="google.generative-language.model.v1beta",
provider_id="google",
envelope=ENVELOPE_SINGLE,
identity_paths=("baseModelId", "name"),
required_item_paths=("supportedGenerationMethods",),
item_types=(("supportedGenerationMethods", (list, tuple)),),
detection_priority=100,
)
OLLAMA_TAGS_SHAPE = ProviderCatalogShape(
shape_id="ollama.tags.v1",
provider_id="ollama",
envelope=ENVELOPE_MODELS,
identity_paths=("model", "name"),
required_item_any_paths=("digest", "details.family", "details.families"),
# `name` plus a digest/details field is not globally provider-specific.
# Configured provider context remains authoritative for local Ollama
# inventories; payload-only detection would create false provider identity.
detection_priority=0,
)
OLLAMA_SHOW_SHAPE = ProviderCatalogShape(
shape_id="ollama.show.v1",
provider_id="ollama",
envelope=ENVELOPE_SINGLE,
identity_paths=(),
required_item_paths=("capabilities",),
required_item_any_paths=("model_info", "details", "template", "parameters"),
item_types=(("capabilities", (list, tuple)),),
# `/api/show` capability and parameter fields are not sufficiently unique
# to identify an otherwise unknown provider. Local/default ports are also
# deliberately non-authoritative, so require configured provider context
# before interpreting this singleton response as Ollama-native metadata.
detection_priority=0,
)
LMSTUDIO_MODELS_V1_SHAPE = ProviderCatalogShape(
shape_id="lmstudio.models.native.v1",
provider_id="lmstudio",
envelope=ENVELOPE_MODELS,
identity_paths=("key",),
required_item_paths=("type",),
required_item_any_paths=(
"capabilities",
"loaded_instances",
"max_context_length",
"architecture",
"quantization",
),
item_types=(("type", (str,)),),
detection_priority=0,
)
LMSTUDIO_MODELS_V0_SHAPE = ProviderCatalogShape(
shape_id="lmstudio.models.native.v0",
provider_id="lmstudio",
envelope=ENVELOPE_DATA,
identity_paths=("id",),
required_item_paths=("type",),
required_item_any_paths=("arch", "compatibility_type", "state", "max_context_length"),
item_types=(("type", (str,)),),
detection_priority=0,
)
LLAMACPP_PROPS_SHAPE = ProviderCatalogShape(
shape_id="llamacpp.props.v1",
provider_id="llamacpp",
envelope=ENVELOPE_SINGLE,
identity_paths=("model_alias", "model_path"),
required_item_paths=("default_generation_settings",),
required_item_any_paths=("chat_template_caps", "modalities", "total_slots"),
item_types=(("default_generation_settings", (Mapping,)),),
detection_priority=100,
)
LLAMACPP_MODELS_SHAPE = ProviderCatalogShape(
shape_id="llamacpp.models.native.v1",
provider_id="llamacpp",
envelope=ENVELOPE_MODELS,
identity_paths=("id", "name", "model"),
required_item_paths=("capabilities",),
item_types=(("capabilities", (list, tuple)),),
# Model/capability fields are not globally provider-specific. Interpret
# them only after explicit llama.cpp endpoint/provider selection.
detection_priority=0,
)
MISTRAL_MODELS_SHAPE = ProviderCatalogShape(
shape_id="mistral.models.rich.v1",
provider_id="mistral",
envelope=ENVELOPE_DATA,
identity_paths=("id",),
required_item_paths=("capabilities",),
required_item_any_paths=(
"capabilities.completion_chat",
"capabilities.completion_fim",
"capabilities.function_calling",
"capabilities.vision",
"capabilities.classification",
),
item_types=(("capabilities", (Mapping,)),),
detection_priority=0,
)
COPILOT_MODELS_SHAPE = ProviderCatalogShape(
shape_id="github-copilot.models.v1",
provider_id="copilot",
envelope=ENVELOPE_DATA,
identity_paths=("id",),
required_item_paths=("model_picker_enabled", "capabilities.supports"),
item_types=(
("model_picker_enabled", (bool,)),
("capabilities.supports", (Mapping,)),
),
detection_priority=100,
)
ANTHROPIC_MODELS_SHAPE = ProviderCatalogShape(
shape_id="anthropic.models.identity.v1",
provider_id="anthropic",
envelope=ENVELOPE_DATA,
identity_paths=("id",),
required_item_paths=("type", "display_name", "created_at"),
item_values=(("type", ("model",)),),
# These model-resource fields are not globally provider-specific. Require
# explicit Anthropic endpoint/provider context before assigning identity.
detection_priority=0,
)
CHATGPT_MODELS_SHAPE = ProviderCatalogShape(
shape_id="chatgpt-subscription.codex-models.v1",
provider_id="chatgpt_subscription",
envelope=ENVELOPE_MODELS,
identity_paths=("slug",),
required_item_any_paths=("visibility", "priority"),
detection_priority=0,
)
SGLANG_MODEL_INFO_SHAPE = ProviderCatalogShape(
shape_id="sglang.model-info.v2",
provider_id="sglang",
envelope=ENVELOPE_SINGLE,
identity_paths=("model_path",),
required_item_paths=("is_generation",),
required_item_any_paths=(
"tokenizer_path",
"has_image_understanding",
"has_audio_understanding",
),
item_types=(("is_generation", (bool,)),),
detection_priority=100,
)
SGLANG_MODELS_SHAPE = ProviderCatalogShape(
shape_id="sglang.models.openai.v1",
provider_id="sglang",
envelope=ENVELOPE_DATA,
identity_paths=("id",),
required_item_paths=("root", "max_model_len"),
item_values=(("owned_by", ("sglang",)),),
detection_priority=80,
)
VLLM_MODELS_SHAPE = ProviderCatalogShape(
shape_id="vllm.models.openai.v1",
provider_id="vllm",
envelope=ENVELOPE_DATA,
identity_paths=("id",),
required_item_paths=("root", "max_model_len", "permission"),
item_values=(("owned_by", ("vllm",)),),
detection_priority=80,
)
HUGGINGFACE_MODEL_SHAPE = ProviderCatalogShape(
shape_id="huggingface.hub.model-info.v1",
provider_id="huggingface",
envelope=ENVELOPE_SINGLE,
identity_paths=("modelId", "id"),
# Hub ModelInfo exposes pipeline_tag as optional metadata. Provider/host
# context is still required because this shape has priority zero, so an
# identity-only card can stay native without making generic ``id`` payloads
# look like Hugging Face catalogs.
detection_priority=0,
)
HUGGINGFACE_MODELS_LIST_SHAPE = ProviderCatalogShape(
shape_id="huggingface.hub.model-info-list.v1",
provider_id="huggingface",
envelope=ENVELOPE_BARE_LIST,
identity_paths=("modelId", "id"),
detection_priority=0,
)
COHERE_MODELS_SHAPE = ProviderCatalogShape(
shape_id="cohere.models.rich.v1",
provider_id="cohere",
envelope=ENVELOPE_MODELS,
identity_paths=("name",),
required_item_paths=("endpoints",),
required_item_any_paths=(
"context_length",
"default_endpoints",
"features",
"sampling_defaults",
),
item_types=(("endpoints", (list, tuple)),),
detection_priority=0,
)
MINIMAX_MODELS_SHAPE = ProviderCatalogShape(
shape_id="minimax.models.identity.v1",
provider_id="minimax",
envelope=ENVELOPE_DATA,
identity_paths=("id",),
required_item_paths=("object", "owned_by"),
item_values=(("object", ("model",)), ("owned_by", ("minimax",))),
detection_priority=90,
)
def _provider(
provider_id: str,
*,
aliases: tuple[str, ...] = (),
hosts: tuple[str, ...] = (),
shapes: tuple[ProviderCatalogShape, ...] = (),
) -> ProviderCapabilitySchema:
return ProviderCapabilitySchema(
provider_id=provider_id,
aliases=aliases,
host_suffixes=hosts,
catalog_shapes=shapes,
)
PROVIDER_SCHEMAS = {
PROVIDER_GENERIC_OPENAI: _provider(
PROVIDER_GENERIC_OPENAI,
aliases=("openai_compatible", "openai_compat"),
),
"openai": _provider("openai", hosts=("openai.com",), shapes=(OPENAI_MODELS_SHAPE,)),
"openrouter": _provider(
"openrouter",
hosts=("openrouter.ai",),
shapes=(OPENROUTER_MODELS_SHAPE,),
),
"google": _provider(
"google",
aliases=("gemini", "google_ai_studio"),
hosts=("generativelanguage.googleapis.com",),
shapes=(GOOGLE_MODELS_SHAPE, GOOGLE_MODEL_SHAPE),
),
"anthropic": _provider(
"anthropic",
hosts=("anthropic.com",),
shapes=(ANTHROPIC_MODELS_SHAPE,),
),
"ollama": _provider(
"ollama",
hosts=("ollama.com",),
shapes=(OLLAMA_SHOW_SHAPE, OLLAMA_TAGS_SHAPE),
),
"lmstudio": _provider(
"lmstudio",
aliases=("lm_studio",),
shapes=(LMSTUDIO_MODELS_V1_SHAPE, LMSTUDIO_MODELS_V0_SHAPE),
),
"llamacpp": _provider(
"llamacpp",
aliases=("llama.cpp", "llama_cpp", "llama_server"),
shapes=(LLAMACPP_PROPS_SHAPE, LLAMACPP_MODELS_SHAPE),
),
"mistral": _provider(
"mistral",
hosts=("mistral.ai",),
shapes=(MISTRAL_MODELS_SHAPE,),
),
"copilot": _provider(
"copilot",
aliases=("github_copilot",),
hosts=("api.githubcopilot.com",),
shapes=(COPILOT_MODELS_SHAPE,),
),
"chatgpt_subscription": _provider(
"chatgpt_subscription",
aliases=("chatgpt-subscription", "chatgpt", "codex_subscription"),
hosts=("chatgpt.com",),
shapes=(CHATGPT_MODELS_SHAPE,),
),
"sglang": _provider(
"sglang",
shapes=(SGLANG_MODEL_INFO_SHAPE, SGLANG_MODELS_SHAPE),
),
"vllm": _provider("vllm", shapes=(VLLM_MODELS_SHAPE,)),
"huggingface": _provider(
"huggingface",
aliases=("hf", "hugging_face"),
hosts=("huggingface.co",),
shapes=(HUGGINGFACE_MODEL_SHAPE, HUGGINGFACE_MODELS_LIST_SHAPE),
),
"cohere": _provider(
"cohere",
hosts=("cohere.ai", "cohere.com"),
shapes=(COHERE_MODELS_SHAPE,),
),
"minimax": _provider(
"minimax",
hosts=("minimax.io", "minimaxi.com"),
shapes=(MINIMAX_MODELS_SHAPE,),
),
}
_GENERAL_PROVIDER_ALIASES = {
"moonshot": ("moonshot_ai",),
"nvidia": ("nvidia_nim", "nim"),
"xai": ("x_ai",),
"zai": ("z.ai", "z_ai"),
"opencode": ("opencode_go", "opencode_zen"),
"together": ("together_ai",),
"fireworks": ("fireworks_ai",),
"atlas_cloud": ("atlas",),
"azure_openai": ("azure",),
"bedrock": ("aws_bedrock",),
"cloudflare_workers_ai": ("workers_ai",),
"mlx_lm": ("mlx",),
"text_generation_inference": ("tgi", "huggingface_tgi", "hugging_face_tgi"),
}
for _provider_id, _hosts in (
("moonshot", ("moonshot.ai", "moonshot.cn")),
("groq", ("groq.com",)),
("nvidia", ("nvidia.com",)),
("cerebras", ("cerebras.ai",)),
("deepseek", ("deepseek.com",)),
("together", ("together.xyz", "together.ai")),
("fireworks", ("fireworks.ai",)),
("xai", ("x.ai",)),
("zai", ("z.ai",)),
("opencode", ("opencode.ai",)),
("perplexity", ("perplexity.ai",)),
("github_models", ("models.inference.ai.azure.com",)),
("atlas_cloud", ("atlascloud.ai",)),
("siliconflow", ("siliconflow.cn", "siliconflow.com")),
("kimi_code", ("kimi.com",)),
("venice", ("venice.ai",)),
("azure_openai", ("openai.azure.com",)),
("bedrock", ()),
("cloudflare_workers_ai", ()),
("mlx_lm", ()),
("text_generation_inference", ()),
("lmdeploy", ()),
("litellm", ()),
):
PROVIDER_SCHEMAS[_provider_id] = _provider(
_provider_id,
aliases=_GENERAL_PROVIDER_ALIASES.get(_provider_id, ()),
hosts=_hosts,
)
UNKNOWN_SCHEMA = ProviderCapabilitySchema(provider_id=PROVIDER_UNKNOWN)
_ALIASES = {
_token(alias): provider_id
for provider_id, schema in PROVIDER_SCHEMAS.items()
for alias in (provider_id, *schema.aliases)
}
def normalize_provider_id(value: Any) -> str:
token = _token(value)
if not token or token == PROVIDER_UNKNOWN:
return PROVIDER_UNKNOWN
# An explicit, previously unseen provider id is still useful identity. It
# selects the inventory-only reader until a native schema is added; it does
# not acquire capabilities merely by being preserved here.
return _ALIASES.get(token, token)
def schema_for_provider(value: Any) -> ProviderCapabilitySchema:
return PROVIDER_SCHEMAS.get(normalize_provider_id(value), UNKNOWN_SCHEMA)
def provider_from_endpoint_kind(value: Any) -> str:
"""Return a provider only for registered provider-valued endpoint kinds.
Endpoint configuration normally stores transport categories such as
``auto``, ``local``, ``api``, and ``proxy``. Those categories and unknown
values must not preempt provider identity from a host or native payload.
"""
provider_id = normalize_provider_id(value)
return provider_id if provider_id in PROVIDER_SCHEMAS else PROVIDER_UNKNOWN
def _host_matches(host: str, suffix: str) -> bool:
return host == suffix or host.endswith("." + suffix)
def provider_from_host(base_url: Any) -> str:
try:
host = (urlparse(str(base_url or "")).hostname or "").lower().rstrip(".")
except Exception:
return PROVIDER_UNKNOWN
if not host:
return PROVIDER_UNKNOWN
if host.startswith("copilot-api.") and host.endswith(".ghe.com"):
return "copilot"
matches = {
schema.provider_id
for schema in PROVIDER_SCHEMAS.values()
if any(_host_matches(host, suffix) for suffix in schema.host_suffixes)
}
return next(iter(matches)) if len(matches) == 1 else PROVIDER_UNKNOWN
def native_shape_for_payload(
payload: Any,
*,
provider_id: Any = None,
) -> ProviderCatalogShape | None:
normalized = normalize_provider_id(provider_id)
if normalized == PROVIDER_UNKNOWN:
shapes = tuple(
shape
for schema in PROVIDER_SCHEMAS.values()
for shape in schema.catalog_shapes
if shape.detection_priority > 0
)
else:
schema = PROVIDER_SCHEMAS.get(normalized)
shapes = schema.catalog_shapes if schema else ()
matches = [shape for shape in shapes if shape.matches(payload)]
if not matches:
return None
providers = {shape.provider_id for shape in matches}
if len(providers) != 1:
return None
priority = max(shape.detection_priority for shape in matches)
best = [shape for shape in matches if shape.detection_priority == priority]
# Registry declaration order expresses preference between revisions of the
# same provider shape (for example LM Studio v1 before v0). Alphabetical
# shape ids invert that version preference for otherwise equal evidence.
return best[0]
def catalog_shape_for_id(shape_id: Any) -> ProviderCatalogShape | None:
return next(
(
shape
for shape in (
*FALLBACK_CATALOG_SHAPES,
*(
provider_shape
for schema in PROVIDER_SCHEMAS.values()
for provider_shape in schema.catalog_shapes
),
)
if shape.shape_id == shape_id
),
None,
)
def fallback_shape_for_payload(payload: Any) -> ProviderCatalogShape | None:
return next((shape for shape in FALLBACK_CATALOG_SHAPES if shape.matches(payload)), None)
def resolve_provider(
payload: Any = None,
*,
provider: Any = None,
endpoint_kind: Any = None,
base_url: Any = None,
) -> ProviderResolution:
provider_id = normalize_provider_id(provider)
provider_source = PROVIDER_SOURCE_EXPLICIT
if provider_id == PROVIDER_UNKNOWN:
provider_id = provider_from_endpoint_kind(endpoint_kind)
provider_source = PROVIDER_SOURCE_ENDPOINT_KIND
if provider_id == PROVIDER_UNKNOWN:
provider_id = provider_from_host(base_url)
provider_source = PROVIDER_SOURCE_HOST
if provider_id != PROVIDER_UNKNOWN:
native = (
native_shape_for_payload(payload, provider_id=provider_id)
if payload is not None
else None
)
if native:
return ProviderResolution(provider_id, provider_source, native.shape_id, False)
fallback = fallback_shape_for_payload(payload) if payload is not None else None
return ProviderResolution(
provider_id,
provider_source,
fallback.shape_id if fallback else "",
bool(fallback),
)
native = native_shape_for_payload(payload) if payload is not None else None
if native:
return ProviderResolution(
native.provider_id,
PROVIDER_SOURCE_PAYLOAD,
native.shape_id,
False,
)
fallback = fallback_shape_for_payload(payload) if payload is not None else None
return ProviderResolution(
PROVIDER_UNKNOWN,
PROVIDER_SOURCE_UNKNOWN,
fallback.shape_id if fallback else "",
bool(fallback),
)
__all__ = [
"FALLBACK_CATALOG_SHAPES",
"GENERAL_BARE_SHAPE",
"GENERAL_DATA_SHAPE",
"GENERAL_MODELS_SHAPE",
"PROVIDER_GENERIC_OPENAI",
"PROVIDER_SCHEMAS",
"PROVIDER_SOURCE_ENDPOINT_KIND",
"PROVIDER_SOURCE_EXPLICIT",
"PROVIDER_SOURCE_HOST",
"PROVIDER_SOURCE_PAYLOAD",
"PROVIDER_SOURCE_UNKNOWN",
"PROVIDER_UNKNOWN",
"ProviderCapabilitySchema",
"ProviderCatalogShape",
"ProviderResolution",
"catalog_shape_for_id",
"fallback_shape_for_payload",
"native_shape_for_payload",
"normalize_provider_id",
"provider_from_endpoint_kind",
"provider_from_host",
"resolve_provider",
"schema_for_provider",
]

136
tests/test_log_safety.py
View file

@ -1,4 +1,19 @@
from core.log_safety import redact_url
import logging
import subprocess
import sys
from textwrap import dedent
import pytest
from core.log_safety import (
CAPABILITY_DIAGNOSTICS_LOGGER,
ScopedDiagnosticsFilter,
UVICORN_LOGGER_NAMES,
application_log_settings,
configure_uvicorn_log_levels,
redact_url,
uvicorn_log_config,
)
def test_strips_userinfo():
@ -30,3 +45,122 @@ def test_empty_and_none():
def test_garbage_does_not_raise():
# urlparse is lenient; just assert no credential-looking userinfo survives.
assert "@" not in redact_url("::::not a url::::")
@pytest.mark.parametrize(
("configured", "expected_level", "expected_capability_debug"),
(
("DEBUG", logging.INFO, True),
("debug", logging.INFO, True),
("INFO", logging.INFO, False),
("WARNING", logging.WARNING, False),
("ERROR", logging.ERROR, False),
("CRITICAL", logging.CRITICAL, False),
("not-a-level", logging.INFO, False),
(None, logging.INFO, False),
),
)
def test_application_log_settings_scope_debug_and_fail_closed(
configured,
expected_level,
expected_capability_debug,
):
assert application_log_settings(configured) == (
expected_level,
expected_capability_debug,
)
def test_configure_uvicorn_log_levels_clamps_non_propagating_loggers():
logger_names = UVICORN_LOGGER_NAMES
previous_levels = {
name: logging.getLogger(name).level for name in logger_names
}
try:
for name in logger_names:
logging.getLogger(name).setLevel(logging.DEBUG)
configure_uvicorn_log_levels(logging.ERROR)
assert all(
logging.getLogger(name).level == logging.ERROR for name in logger_names
)
finally:
for name, level in previous_levels.items():
logging.getLogger(name).setLevel(level)
def test_uvicorn_log_config_sets_all_named_loggers_without_mutating_default():
from uvicorn.config import LOGGING_CONFIG
configured = uvicorn_log_config(logging.ERROR)
assert all(
configured["loggers"][name]["level"] == logging.ERROR
for name in UVICORN_LOGGER_NAMES
)
assert LOGGING_CONFIG["loggers"]["uvicorn"]["level"] == "INFO"
def test_uvicorn_levels_hold_across_external_and_direct_config_order():
result = subprocess.run(
[
sys.executable,
"-c",
dedent(
"""
import logging
import uvicorn
from core.log_safety import (
UVICORN_LOGGER_NAMES,
configure_uvicorn_log_levels,
uvicorn_log_config,
)
uvicorn.Config("app:app", log_level="debug")
configure_uvicorn_log_levels(logging.INFO)
assert all(
logging.getLogger(name).level == logging.INFO
for name in UVICORN_LOGGER_NAMES
)
uvicorn.Config(
"app:app",
log_level=logging.ERROR,
log_config=uvicorn_log_config(logging.ERROR),
)
assert all(
logging.getLogger(name).level == logging.ERROR
for name in UVICORN_LOGGER_NAMES
)
"""
),
],
capture_output=True,
text=True,
check=False,
)
assert result.returncode == 0, result.stderr
def _record(name: str, level: int) -> logging.LogRecord:
return logging.LogRecord(name, level, __file__, 1, "message", (), None)
def test_scoped_diagnostics_filter_allows_only_bounded_debug_logger():
log_filter = ScopedDiagnosticsFilter(logging.INFO, capability_debug=True)
assert log_filter.filter(_record(CAPABILITY_DIAGNOSTICS_LOGGER, logging.DEBUG))
assert log_filter.filter(_record("unrelated.library", logging.INFO))
assert not log_filter.filter(_record("unrelated.library", logging.DEBUG))
assert not log_filter.filter(_record(f"{CAPABILITY_DIAGNOSTICS_LOGGER}.raw", logging.DEBUG))
def test_scoped_diagnostics_filter_respects_higher_application_level():
log_filter = ScopedDiagnosticsFilter(logging.WARNING, capability_debug=False)
assert log_filter.filter(_record("application", logging.WARNING))
assert not log_filter.filter(_record("application", logging.INFO))
assert not log_filter.filter(_record(CAPABILITY_DIAGNOSTICS_LOGGER, logging.DEBUG))

74
tests/test_model_capability_diagnostics.py Normal file
View file

@ -0,0 +1,74 @@
import logging
from pathlib import Path
from src.model_capability_readers import records_from_payload
def test_normalization_debug_log_reports_shape_without_payload_identity(caplog):
payload = {
"data": [
{
"id": "sensitive-model-id",
"architecture": {"modality": "text+image->text"},
"canonical_slug": "provider/sensitive-model-id",
"pricing": {"prompt": "0.1", "completion": "0.2"},
"supported_parameters": ["tools", "temperature"],
"top_provider": {"context_length": 32768},
"private_field": "secret-value",
}
]
}
with caplog.at_level(logging.DEBUG, logger="src.model_capability_readers"):
records = records_from_payload(payload, vendor="openrouter")
assert len(records) == 1
message = caplog.messages[-1]
assert "[model-capability] normalized:" in message
assert "canonical_version=1" in message
assert "provider=openrouter" in message
assert "provider_source=explicit" in message
assert "catalog_shape=openrouter.models.rich.v1" in message
assert "fallback=False" in message
assert "records=1" in message
assert "native_records=1" in message
assert "fallback_records=0" in message
assert "families=['chat']" in message
assert "features=['tool_call', 'vision']" in message
assert "controls=['temperature']" in message
assert "sensitive-model-id" not in message
assert "secret-value" not in message
def test_fallback_debug_log_is_explicit_and_has_no_capability_claims(caplog):
payload = [{"id": "future-model", "capabilities": {"tools": True}}]
with caplog.at_level(logging.DEBUG, logger="src.model_capability_readers"):
records = records_from_payload(payload, vendor="future-provider")
assert records[0].capability.capabilities == ()
message = caplog.messages[-1]
assert "provider=unregistered" in message
assert "catalog_shape=fallback.models.list.v1" in message
assert "fallback=True" in message
assert "native_records=0" in message
assert "fallback_records=1" in message
assert "features=[]" in message
def test_web_app_logging_uses_existing_log_level_environment_toggle():
root = Path(__file__).resolve().parents[1]
source = (root / "app.py").read_text(encoding="utf-8")
launcher_source = (root / "launcher.py").read_text(encoding="utf-8")
assert 'os.getenv("LOG_LEVEL", "INFO")' in source
assert "application_log_settings(_log_level_name)" in source
assert "_root_logger.setLevel(_application_log_level)" in source
assert "configure_uvicorn_log_levels(_application_log_level)" in source
assert "_console_h.addFilter(_diagnostics_filter)" in source
assert "_file_h.addFilter(_diagnostics_filter)" in source
assert "log_level=_application_log_level" in source
assert "log_config=uvicorn_log_config(_application_log_level)" in source
assert "application_log_settings(" in launcher_source
assert "log_level=application_log_level" in launcher_source
assert "log_config=uvicorn_log_config(application_log_level)" in launcher_source

81
tests/test_model_capability_readers.py
View file

@ -9,6 +9,7 @@ from src.model_capability_readers.base import (
VENDOR_OLLAMA,
VENDOR_OPENAI,
VENDOR_OPENROUTER,
VENDOR_UNKNOWN,
detect_vendor,
stable_model_id_for,
)
@ -18,16 +19,16 @@ def surfaces(record):
return set(mc.display_surfaces_for(record.capability))
def test_detect_vendor_uses_endpoint_kind_then_host_and_common_local_ports():
def test_detect_vendor_uses_endpoint_kind_and_host_but_not_ambiguous_local_ports():
assert detect_vendor("https://example.test/v1", endpoint_kind="ollama") == VENDOR_OLLAMA
assert detect_vendor("http://127.0.0.1:8080", endpoint_kind="llama_cpp") == VENDOR_LLAMACPP
assert detect_vendor("https://openrouter.ai/api/v1") == VENDOR_OPENROUTER
assert detect_vendor("https://api.openai.com/v1") == VENDOR_OPENAI
assert detect_vendor("https://generativelanguage.googleapis.com/v1beta/openai") == VENDOR_GOOGLE
assert detect_vendor("http://127.0.0.1:11434") == VENDOR_OLLAMA
assert detect_vendor("http://127.0.0.1:1234") == VENDOR_LMSTUDIO
assert detect_vendor("http://127.0.0.1:8080") == VENDOR_GENERIC_OPENAI
assert detect_vendor("http://localhost:7000/v1") == VENDOR_GENERIC_OPENAI
assert detect_vendor("http://127.0.0.1:11434") == VENDOR_UNKNOWN
assert detect_vendor("http://127.0.0.1:1234") == VENDOR_UNKNOWN
assert detect_vendor("http://127.0.0.1:8080") == VENDOR_UNKNOWN
assert detect_vendor("http://localhost:7000/v1") == VENDOR_UNKNOWN
def test_generic_openai_reader_keeps_basic_model_payload_unknown():
@ -345,7 +346,9 @@ def test_ollama_reader_maps_show_capabilities_and_tags_are_unknown():
"nomic-embed-text:latest",
{"capabilities": ["embedding"]},
)
tags = ollama.records_from_tags_payload({"models": [{"name": "qwen3:latest"}]})
tags = ollama.records_from_tags_payload(
{"models": [{"name": "qwen3:latest", "details": {"family": "qwen3"}}]}
)
assert vision is not None
assert vision.capability.family == mc.FAMILY_CHAT
@ -381,7 +384,9 @@ def test_ollama_reader_uses_show_shape_without_architecture_name_matching():
assert record.capability.modalities.input == (mc.MODALITY_TEXT,)
assert record.capability.modalities.output == (mc.MODALITY_TEXT,)
assert record.capability.capabilities == (mc.CAP_REASONING, mc.CAP_TOOL_CALL)
assert dict(record.capability.limits) == {"context_tokens": 8192}
# Serialized Modelfile text is not reparsed for capability truth. The
# structured native `model_info.*.context_length` field wins.
assert dict(record.capability.limits) == {"context_tokens": 32768}
assert surfaces(record) == {"chat"}
@ -399,6 +404,68 @@ def test_ollama_reader_uses_generic_model_info_context_length_when_no_num_ctx():
assert dict(record.capability.limits) == {"context_tokens": 32768}
def test_ollama_ps_reader_keeps_loaded_allocation_in_runtime_shape():
record = ollama.runtime_context_from_ps_payload(
"hf.co/example/Qwen3:Q6_K",
{
"models": [
{
"name": "hf.co/example/Qwen3:Q6_K",
"model": "hf.co/example/Qwen3:Q6_K",
"context_length": 65536,
"details": {"family": "qwen3"},
}
]
},
endpoint_id="7",
)
assert record is not None
assert record.allocated_context_tokens == 65536
assert record.stable_model_id == "ollama|endpoint:7|hf.co/example/qwen3:q6_k"
serialized = record.to_dict()
assert serialized["runtime"] == {"allocated_context_tokens": 65536}
assert serialized["evidence"] == {
"source": mc.SOURCE_PROVIDER_READER,
"confidence": mc.CONFIDENCE_PROVIDER_REPORTED,
"shape": "ollama.ps.v1",
"scope": "loaded_model",
}
assert "limits" not in serialized
assert "raw" not in serialized
def test_ollama_ps_reader_matches_only_exact_or_latest_identity():
payload = {
"models": [
{"model": "qwen3:latest", "context_length": 32768},
{"model": "qwen3:14b", "context_length": 65536},
]
}
record = ollama.runtime_context_from_ps_payload("qwen3", payload)
assert record is not None
assert record.allocated_context_tokens == 32768
assert ollama.runtime_context_from_ps_payload("qwen3:8b", payload) is None
def test_ollama_ps_reader_fails_closed_for_invalid_or_conflicting_allocation():
assert ollama.runtime_context_from_ps_payload(
"qwen3",
{"models": [{"model": "qwen3", "context_length": True}]},
) is None
assert ollama.runtime_context_from_ps_payload(
"qwen3",
{
"models": [
{"model": "qwen3", "context_length": 32768},
{"name": "qwen3:latest", "context_length": 65536},
]
},
) is None
def test_lmstudio_reader_uses_native_v1_capabilities_when_present():
records = lmstudio.records_from_payload(
{

1104
tests/test_provider_capability_schemas.py Normal file
View file

File diff suppressed because it is too large Load diff