"""Runtime loader for versioned Glas harness profiles and rein descriptors.""" from __future__ import annotations import importlib import inspect import os import re from pathlib import Path from typing import Any import yaml from pydantic import ValidationError from glas_harness.contract import ( CONTRACT_VERSION, HarnessProfile, Rein, ReinDescriptor, ResolvedExecutionContext, ) class ProfileError(ValueError): pass class UnknownProfileError(ProfileError): pass class AmbiguousProfileError(ProfileError): pass class IncompatibleProfileError(ProfileError): pass class OperationallyBlockedProfileError(ProfileError): pass _SENSITIVE_KEY = re.compile( r"(^|_)(api_key|password|passwd|secret|secret_value|token_value|private_key)$", re.IGNORECASE, ) _SENSITIVE_VALUE = re.compile(r"^(sk-[A-Za-z0-9_-]{12,}|hvs\.[A-Za-z0-9_-]{12,})$") def _source_root() -> Path: return Path(__file__).resolve().parents[2] def _default_data_dir(kind: str) -> Path: env_name = "GLAS_PROFILE_DIR" if kind == "profiles" else "GLAS_REIN_REGISTRY_DIR" if configured := os.environ.get(env_name): return Path(configured).expanduser().resolve() packaged = Path(__file__).resolve().parent / "data" / kind if packaged.is_dir(): return packaged return _source_root() / ("profiles" if kind == "profiles" else "registry/reins") def _read_yaml(path: Path) -> dict[str, Any]: try: data = yaml.safe_load(path.read_text()) except (OSError, yaml.YAMLError) as exc: raise ProfileError(f"cannot read {path}: {exc}") from exc if not isinstance(data, dict): raise ProfileError(f"{path}: expected a YAML object") _reject_inline_secrets(data, path=path) return data def _reject_inline_secrets(value: Any, *, path: Path, key_path: str = "") -> None: if isinstance(value, dict): for key, child in value.items(): key_str = str(key) child_path = f"{key_path}.{key_str}" if key_path else key_str if _SENSITIVE_KEY.search(key_str) and child not in (None, "", [], {}): raise ProfileError(f"{path}: inline secret material forbidden at {child_path}") _reject_inline_secrets(child, path=path, key_path=child_path) elif isinstance(value, list): for index, child in enumerate(value): _reject_inline_secrets(child, path=path, key_path=f"{key_path}[{index}]") elif isinstance(value, str) and _SENSITIVE_VALUE.match(value): raise ProfileError(f"{path}: token-looking inline value forbidden at {key_path}") def _capability_satisfies(actual: Any, required: Any) -> bool: if isinstance(required, list): if not isinstance(actual, list): return False return all(item in actual for item in required) return actual == required class ProfileCatalog: def __init__( self, profile_dir: str | Path | None = None, rein_dir: str | Path | None = None, ) -> None: self.profile_dir = Path(profile_dir) if profile_dir else _default_data_dir("profiles") self.rein_dir = Path(rein_dir) if rein_dir else _default_data_dir("reins") self._profiles: dict[tuple[str, str], HarnessProfile] | None = None self._reins: dict[str, ReinDescriptor] | None = None def profiles(self) -> dict[tuple[str, str], HarnessProfile]: if self._profiles is None: loaded: dict[tuple[str, str], HarnessProfile] = {} for path in sorted(self.profile_dir.glob("*.yaml")): try: profile = HarnessProfile.model_validate(_read_yaml(path)) except ValidationError as exc: raise ProfileError(f"{path}: invalid harness profile: {exc}") from exc key = (profile.id, profile.version) if key in loaded: raise ProfileError(f"duplicate harness profile {profile.id}@{profile.version}") loaded[key] = profile self._profiles = loaded return self._profiles def reins(self) -> dict[str, ReinDescriptor]: if self._reins is None: loaded: dict[str, ReinDescriptor] = {} for path in sorted(self.rein_dir.glob("*.yaml")): try: descriptor = ReinDescriptor.model_validate(_read_yaml(path)) except ValidationError as exc: raise ProfileError(f"{path}: invalid rein descriptor: {exc}") from exc if descriptor.id in loaded: raise ProfileError(f"duplicate rein descriptor {descriptor.id}") loaded[descriptor.id] = descriptor self._reins = loaded return self._reins def resolve(self, reference: str) -> tuple[HarnessProfile, ReinDescriptor]: profile_id, separator, version = reference.partition("@") candidates = [ profile for (candidate_id, candidate_version), profile in self.profiles().items() if candidate_id == profile_id and (not separator or candidate_version == version) ] if not candidates: raise UnknownProfileError(f"unknown harness profile: {reference}") if len(candidates) != 1: refs = ", ".join(sorted(str(candidate.ref) for candidate in candidates)) raise AmbiguousProfileError( f"ambiguous harness profile {reference}; pin one of: {refs}" ) profile = candidates[0] if profile.status != "enabled": raise IncompatibleProfileError(f"harness profile disabled: {profile.ref}") if profile.contract_version != CONTRACT_VERSION: raise IncompatibleProfileError( f"profile {profile.ref} requires contract {profile.contract_version}; " f"gateway supports {CONTRACT_VERSION}" ) descriptor = self.reins().get(profile.rein.id) if descriptor is None: raise IncompatibleProfileError( f"profile {profile.ref} references unknown rein {profile.rein.id}" ) if descriptor.status != "implemented": raise IncompatibleProfileError( f"rein {descriptor.id} is not enabled for governed execution " f"(status={descriptor.status})" ) if profile.contract_version not in descriptor.contract_versions: raise IncompatibleProfileError( f"rein {descriptor.id}@{descriptor.version} does not implement " f"contract {profile.contract_version}" ) for name, required in profile.rein.required_capabilities.items(): actual = descriptor.capabilities.get(name) if not _capability_satisfies(actual, required): raise IncompatibleProfileError( f"rein {descriptor.id} capability {name!r} is {actual!r}; " f"profile requires {required!r}" ) return profile, descriptor def resolve_context(self, reference: str) -> ResolvedExecutionContext: profile, descriptor = self.resolve(reference) return ResolvedExecutionContext( contract_version=CONTRACT_VERSION, profile=profile.ref, rein_id=descriptor.id, rein_version=descriptor.version, operational_readiness=profile.operational_readiness, sandbox_profile=profile.sandbox_profile, tool_profile=profile.tool_profile, model=profile.model, limits=profile.limits, ) def require_operational(self, profile: HarnessProfile) -> None: """Refuse known runtime blockers while allowing labeled proof attempts.""" readiness = profile.operational_readiness if readiness.status != "blocked": return raise OperationallyBlockedProfileError( f"profile {profile.ref} operational readiness is blocked: " f"{readiness.reason} (owner={readiness.owner}; " f"evidence={readiness.evidence_ref})" ) def build_rein(self, profile: HarnessProfile, descriptor: ReinDescriptor) -> Rein: module_name, class_name = descriptor.handler.split(":", 1) module = importlib.import_module(module_name) rein_type = getattr(module, class_name) if not inspect.isclass(rein_type) or not issubclass(rein_type, Rein): raise IncompatibleProfileError( f"handler {descriptor.handler} does not implement Rein" ) candidate_kwargs = { "model": profile.model.model, "tool_profile": profile.tool_profile, "max_turns": profile.limits.max_turns, "budget_tokens": profile.limits.budget_tokens, "stream_tool_events": bool(profile.metadata.get("stream_tool_events", False)), } parameters = inspect.signature(rein_type).parameters kwargs = { name: value for name, value in candidate_kwargs.items() if name in parameters and value is not None } return rein_type(**kwargs) def validate_all(self) -> list[ResolvedExecutionContext]: return [ self.resolve_context(str(profile.ref)) for profile in sorted(self.profiles().values(), key=lambda item: (item.id, item.version)) ]