From b1233d42dc81cda5131eb8cd2ffd3742d00cacb8 Mon Sep 17 00:00:00 2001 From: m4 Date: Tue, 21 Jul 2026 08:33:12 +0800 Subject: [PATCH] feat(model-registry): add ModelRegistryResolver and run snapshot service Resolver (8.1): validates provider/model/credential/capability/limits and the 6.5 four-mode input budget, freezes ResolvedModelConfig; resolve_for_test relaxes only the enabled-visibility check (9.4); compute_availability is the single 4.3 six-state judgement (stale beats configured, selectable only when enabled). SnapshotService (8.2, shared by the Task 5 HTTP API and Task 7 local entry): freezes both roles' full ResolvedModelConfig with adapter spec revision, fixed reserves, capabilities, and credential revisions; selection-hash idempotency with pre-resolution semantics; prepared(15min)/bound(+24h)/ expired/aborted lifecycle with atomic bind; binding-checked reads that revalidate frozen spec revisions; per-call credential resolution against the frozen revision with no in-process secret cache (5.2); public diagnostic view limited to the 8.2 safe subset. Store gains additive helpers (credential pointer lookup, verification listing, active-triplet lookup, conditional bind, due-expiry sweep) and the taxonomy gains SNAPSHOT_NOT_FOUND (404) for missing snapshots. --- EvoScientist/model_registry/__init__.py | 24 + EvoScientist/model_registry/errors.py | 2 + EvoScientist/model_registry/resolver.py | 438 +++++++++++++++++ EvoScientist/model_registry/snapshots.py | 413 +++++++++++++++++ EvoScientist/model_registry/store.py | 168 ++++++- tests/test_model_registry_schemas.py | 1 + tests/test_resolver.py | 520 +++++++++++++++++++++ tests/test_snapshots.py | 567 +++++++++++++++++++++++ 8 files changed, 2121 insertions(+), 12 deletions(-) create mode 100644 EvoScientist/model_registry/resolver.py create mode 100644 EvoScientist/model_registry/snapshots.py create mode 100644 tests/test_resolver.py create mode 100644 tests/test_snapshots.py diff --git a/EvoScientist/model_registry/__init__.py b/EvoScientist/model_registry/__init__.py index 18305de..5d52173 100644 --- a/EvoScientist/model_registry/__init__.py +++ b/EvoScientist/model_registry/__init__.py @@ -23,6 +23,7 @@ from .endpoint_policy import EndpointPolicy from .errors import ERROR_HTTP_STATUS, ErrorDetail, ErrorPayload, ModelRegistryError from .factory import build_chat_model from .hashing import configuration_hash +from .resolver import ModelRegistryResolver from .safe_transport import ( AsyncSafeHttpTransport, AsyncSafeNetworkBackend, @@ -52,10 +53,24 @@ from .schemas import ( ResolvedModelConfig, VerificationInfo, ) +from .snapshots import ( + BOUND_RETENTION_SECONDS, + PREPARED_TTL_SECONDS, + RuntimeSnapshot, + SnapshotCreateRequest, + SnapshotCreation, + SnapshotPayload, + SnapshotService, + compute_selection_hash, + config_for_role, + public_snapshot_view, +) from .store import ModelRuntimeStore, SharedStorageError __all__ = [ + "BOUND_RETENTION_SECONDS", "ERROR_HTTP_STATUS", + "PREPARED_TTL_SECONDS", "Adapter", "AdapterParameterSpec", "AsyncSafeHttpTransport", @@ -76,6 +91,7 @@ __all__ = [ "ModelConfig", "ModelRef", "ModelRegistryError", + "ModelRegistryResolver", "ModelRuntimeConfig", "ModelRuntimeStore", "ParameterRule", @@ -84,17 +100,25 @@ __all__ = [ "RegistryV4", "ResolvedModelConfig", "ResolvedParameters", + "RuntimeSnapshot", "SafeHttpTransport", "SafeNetworkBackend", "SharedStorageError", + "SnapshotCreateRequest", + "SnapshotCreation", + "SnapshotPayload", + "SnapshotService", "VerificationInfo", "adapter_specs", "build_chat_model", "build_safe_async_http_client", "build_safe_http_client", "compute_effective_capabilities", + "compute_selection_hash", + "config_for_role", "configuration_hash", "find_adapter_spec", "get_adapter", + "public_snapshot_view", "resolve_parameters", ] diff --git a/EvoScientist/model_registry/errors.py b/EvoScientist/model_registry/errors.py index fac815e..5043512 100644 --- a/EvoScientist/model_registry/errors.py +++ b/EvoScientist/model_registry/errors.py @@ -28,6 +28,7 @@ DELEGATION_REPLAYED = "DELEGATION_REPLAYED" # 404 — missing resource. MODEL_NOT_FOUND = "MODEL_NOT_FOUND" +SNAPSHOT_NOT_FOUND = "SNAPSHOT_NOT_FOUND" # 422 — configuration and runtime precondition failures. MODEL_REGISTRY_NOT_READY = "MODEL_REGISTRY_NOT_READY" @@ -56,6 +57,7 @@ ERROR_HTTP_STATUS: dict[str, int] = { MODEL_CONFIGURATION_CHANGED: 409, DELEGATION_REPLAYED: 401, MODEL_NOT_FOUND: 404, + SNAPSHOT_NOT_FOUND: 404, MODEL_REGISTRY_NOT_READY: 422, MODEL_DISABLED: 422, MODEL_NOT_AVAILABLE: 422, diff --git a/EvoScientist/model_registry/resolver.py b/EvoScientist/model_registry/resolver.py new file mode 100644 index 0000000..0bd63c4 --- /dev/null +++ b/EvoScientist/model_registry/resolver.py @@ -0,0 +1,438 @@ +"""ModelRegistryResolver: the single registry resolution path (section 8.1). + +``resolve(ModelRef, role)`` validates the provider, model, credentials, +capabilities, limits, and input budget, then freezes a complete +``ResolvedModelConfig`` (section 6.4). ``resolve_for_test`` relaxes only the +"model must be enabled" user-visibility check (section 9.4); every other +validation still runs. ``compute_availability`` implements the section 4.3 +six-state judgement order and is the only availability computation — callers +must not derive state from ``enabled`` flags or test timestamps on their own. + +The resolver never reads ``secret_value``: a credential check resolves the +pointer's ``current_revision`` and freezes it into ``auth_ref``. Secrets are +resolved per provider call by the snapshot layer against that frozen +revision (section 5.2). +""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping +from datetime import UTC, datetime +from typing import Any, get_args + +from .adapters import ( + adapter_specs, + compute_effective_capabilities, + find_adapter_spec, + resolve_parameters, +) +from .errors import ( + CONTEXT_BUDGET_UNSATISFIABLE, + CREDENTIAL_NOT_CONFIGURED, + MODEL_DISABLED, + MODEL_LIMITS_UNCONFIRMED, + MODEL_NOT_AVAILABLE, + MODEL_NOT_FOUND, + ModelRegistryError, +) +from .hashing import configuration_hash +from .schemas import ( + AdapterParameterSpec, + AuthRef, + Capabilities, + ClientOptions, + FixedReserves, + InputBudget, + ModelAvailability, + ModelConfig, + ModelRef, + ModelRole, + ProviderConfig, + RegistryV4, + RequestOptions, + ResolvedModelConfig, + VerificationInfo, +) +from .store import ModelRuntimeStore + +_CAPABILITY_NAMES = ("tools", "vision", "structured_output") +_MODEL_ROLES = get_args(ModelRole) + +# Availability reason codes (section 4.3 `reason_code` strings). +REASON_PROVIDER_DISABLED = "PROVIDER_DISABLED" +REASON_NO_ADAPTER_CONTRACT = "NO_ADAPTER_CONTRACT" +REASON_MODEL_DISABLED = "MODEL_DISABLED" +REASON_VERIFICATION_FAILED = "VERIFICATION_FAILED" +REASON_VERIFICATION_STALE = "VERIFICATION_STALE" + + +def _rfc3339(epoch_seconds: int) -> str: + return datetime.fromtimestamp(epoch_seconds, UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _check_role(role: str) -> None: + if role not in _MODEL_ROLES: + raise ValueError( + f"Unknown model role {role!r}; expected one of {list(_MODEL_ROLES)}." + ) + + +class ModelRegistryResolver: + """Validates ModelRefs and freezes per-run resolved configurations.""" + + def __init__( + self, + store: ModelRuntimeStore, + *, + specs: Iterable[AdapterParameterSpec] | None = None, + ) -> None: + self._store = store + self._specs = tuple(adapter_specs() if specs is None else specs) + + @property + def specs(self) -> tuple[AdapterParameterSpec, ...]: + """The adapter contracts this resolver matches against.""" + return self._specs + + def resolve( + self, + model_ref: ModelRef, + role: ModelRole = "primary", + *, + registry: RegistryV4 | None = None, + ) -> ResolvedModelConfig: + """Resolve an enabled, verified model into its frozen run config.""" + return self._resolve(model_ref, role, registry=registry, require_enabled=True) + + def resolve_for_test(self, model_ref: ModelRef) -> ResolvedModelConfig: + """Resolve for a provider test (section 9.4). + + Only the "model must be enabled" visibility check is relaxed; auth, + parameter, capability, limits, and budget validation all still run. + """ + return self._resolve(model_ref, "primary", registry=None, require_enabled=False) + + def _resolve( + self, + model_ref: ModelRef, + role: ModelRole, + *, + registry: RegistryV4 | None, + require_enabled: bool, + ) -> ResolvedModelConfig: + _check_role(role) + if registry is None: + registry = self._store.load_registry() + provider = registry.find_provider(model_ref.provider_id) + if provider is None: + raise ModelRegistryError( + MODEL_NOT_FOUND, + f"Provider {model_ref.provider_id!r} does not exist.", + details=[{"path": "provider_id", "code": MODEL_NOT_FOUND}], + ) + model = provider.find_model(model_ref.model_key) + if model is None: + raise ModelRegistryError( + MODEL_NOT_FOUND, + f"Model {model_ref.model_key!r} does not exist on provider " + f"{model_ref.provider_id!r}.", + details=[{"path": "model_key", "code": MODEL_NOT_FOUND}], + ) + if not provider.enabled: + raise ModelRegistryError( + MODEL_NOT_AVAILABLE, + f"Provider {provider.id!r} is disabled.", + details=[{"path": "provider.enabled", "code": MODEL_NOT_AVAILABLE}], + ) + if require_enabled and not model.enabled: + raise ModelRegistryError( + MODEL_DISABLED, + f"Model {model_ref.model_key!r} is disabled.", + details=[{"path": "model.enabled", "code": MODEL_DISABLED}], + ) + + spec = find_adapter_spec( + provider.adapter, model.upstream_model_id, specs=self._specs + ) + if spec is None: + raise ModelRegistryError( + MODEL_NOT_AVAILABLE, + "No adapter contract matches " + f"{provider.adapter!r}/{model.upstream_model_id!r}; the model " + "may only remain 'configured'.", + details=[{"path": "model.adapter", "code": MODEL_NOT_AVAILABLE}], + ) + + # Save-time contract checks re-applied at resolve time: auth mode, + # credential reference, declared capabilities, and parameters. + parameters = resolve_parameters(provider, model, spec) + + auth_spec = spec.auth_specs[provider.auth.mode] + credential_revision: int | None = None + if auth_spec.credential_required and provider.auth.credential_id is not None: + credential_revision = self._store.current_credential_revision( + provider.auth.credential_id + ) + if credential_revision is None: + raise ModelRegistryError( + CREDENTIAL_NOT_CONFIGURED, + "The credential referenced by this provider is not configured.", + details=[ + { + "path": "auth.credential_id", + "code": CREDENTIAL_NOT_CONFIGURED, + } + ], + ) + auth_ref = AuthRef( + mode=provider.auth.mode, + credential_id=provider.auth.credential_id, + credential_revision=credential_revision, + ) + + # The current verification five-tuple must exist and have passed; + # any configuration or credential change invalidates it (4.3). + record = self._store.get_model_verification( + provider_id=provider.id, + model_key=model.key, + configuration_hash=configuration_hash(provider, model), + credential_revision=credential_revision or 0, + adapter_spec_revision=spec.spec_revision, + ) + if record is None or record["result"] != "passed": + raise ModelRegistryError( + MODEL_NOT_AVAILABLE, + "The model has no passing verification for its current configuration.", + details=[{"path": "model.verification", "code": MODEL_NOT_AVAILABLE}], + ) + + if model.runtime.limits_status != "confirmed": + raise ModelRegistryError( + MODEL_LIMITS_UNCONFIRMED, + "The model's context limits are not confirmed.", + details=[ + {"path": "runtime.limits_status", "code": MODEL_LIMITS_UNCONFIRMED} + ], + ) + + budget = self._resolve_budget(model, parameters.max_output_tokens) + verified = record["verified_capabilities"] + verified_capabilities = Capabilities( + **{name: bool(verified.get(name, False)) for name in _CAPABILITY_NAMES} + ) + effective_capabilities = compute_effective_capabilities( + spec.protocol_capabilities, + model.runtime.declared_capabilities, + verified_capabilities, + ) + return ResolvedModelConfig( + model_ref=model_ref, + role=role, + adapter_id=spec.adapter_id, + adapter_spec_revision=spec.spec_revision, + upstream_model_id=model.upstream_model_id, + base_url=provider.base_url, + auth_ref=auth_ref, + client_options=ClientOptions( + timeout_seconds=parameters.timeout_seconds, + max_retries=parameters.max_retries, + ), + request_options=RequestOptions( + max_output_tokens=parameters.max_output_tokens, + temperature=parameters.temperature, + top_p=parameters.top_p, + reasoning_effort=parameters.reasoning_effort, + ), + budget=budget, + effective_capabilities=effective_capabilities, + ) + + @staticmethod + def _resolve_budget(model: ModelConfig, max_output_tokens: int) -> InputBudget: + """Apply the section 6.5 formulas and the four-mode invariant guard. + + The frozen ``message_budget`` is the base budget (system reserve + deducted, no tools/attachments); ``MessageBudgetMiddleware`` + recomputes the per-call budget from the frozen reserves. All four + tool/attachment combinations must stay above + ``min_effective_input_tokens`` or the run cannot be created. + """ + runtime = model.runtime + if runtime.limit_mode == "combined": + assert runtime.context_window_tokens is not None # schema invariant + resolved_input_limit = runtime.context_window_tokens - max_output_tokens + else: + assert runtime.max_input_tokens is not None # schema invariant + resolved_input_limit = runtime.max_input_tokens + reserves = FixedReserves( + fixed_system_reserve_tokens=runtime.fixed_system_reserve_tokens, + fixed_tools_reserve_tokens=runtime.fixed_tools_reserve_tokens, + fixed_attachments_reserve_tokens=runtime.fixed_attachments_reserve_tokens, + ) + base = resolved_input_limit - reserves.fixed_system_reserve_tokens + worst_case = ( + base + - reserves.fixed_tools_reserve_tokens + - reserves.fixed_attachments_reserve_tokens + ) + if resolved_input_limit <= 0 or worst_case < runtime.min_effective_input_tokens: + raise ModelRegistryError( + CONTEXT_BUDGET_UNSATISFIABLE, + "The input budget cannot satisfy min_effective_input_tokens " + "for every tool/attachment combination.", + details=[ + { + "path": "runtime.min_effective_input_tokens", + "code": CONTEXT_BUDGET_UNSATISFIABLE, + } + ], + ) + return InputBudget( + resolved_input_limit=resolved_input_limit, + fixed_reserves=reserves, + message_budget=base, + ) + + def compute_availability( + self, + registry: RegistryV4, + verifications: Iterable[dict[str, Any]], + *, + credential_revisions: Mapping[str, int] | None = None, + ) -> list[ModelAvailability]: + """The section 4.3 six-state judgement for every configured model. + + ``verifications`` carries full records (five-tuple, result, verified + capabilities, timestamp, error code) as returned by + ``ModelRuntimeStore.list_model_verifications``; + ``credential_revisions`` maps credential IDs to their current pointer + revision for the five-tuple comparison. + """ + revisions = dict(credential_revisions or {}) + records_by_model: dict[tuple[str, str], list[dict[str, Any]]] = {} + for record in verifications: + key = (str(record["provider_id"]), str(record["model_key"])) + records_by_model.setdefault(key, []).append(record) + return [ + self._judge_model( + provider, + model, + records_by_model.get((provider.id, model.key), []), + revisions, + ) + for provider in registry.providers + for model in provider.models + ] + + def _judge_model( + self, + provider: ProviderConfig, + model: ModelConfig, + records: list[dict[str, Any]], + credential_revisions: Mapping[str, int], + ) -> ModelAvailability: + model_ref = ModelRef(provider_id=provider.id, model_key=model.key) + try: + spec = find_adapter_spec( + provider.adapter, model.upstream_model_id, specs=self._specs + ) + except ModelRegistryError: + spec = None + + current = self._current_record( + provider, model, spec, records, credential_revisions + ) + verification = self._verification_info(current, records) + effective = Capabilities() + if spec is not None and current is not None and current["result"] == "passed": + verified_caps = Capabilities( + **{ + name: bool(current["verified_capabilities"].get(name, False)) + for name in _CAPABILITY_NAMES + } + ) + effective = compute_effective_capabilities( + spec.protocol_capabilities, + model.runtime.declared_capabilities, + verified_caps, + ) + + # Section 4.3 judgement order; the first match wins. + if not provider.enabled or spec is None: + state = "unavailable" + reason = ( + REASON_PROVIDER_DISABLED + if not provider.enabled + else REASON_NO_ADAPTER_CONTRACT + ) + elif current is not None and current["result"] == "passed" and model.enabled: + state, reason = "enabled", None + elif current is not None and current["result"] == "passed": + state, reason = "verified", REASON_MODEL_DISABLED + elif current is not None: + state = "verification_failed" + reason = current["error_code"] or REASON_VERIFICATION_FAILED + elif records: + # Records exist, but none matches the current five-tuple; stale + # takes priority over configured. + state, reason = "verification_stale", REASON_VERIFICATION_STALE + else: + state, reason = "configured", None + return ModelAvailability( + model_ref=model_ref, + state=state, + selectable=state == "enabled", + reason_code=reason, + verification=verification, + effective_capabilities=effective, + ) + + @staticmethod + def _current_record( + provider: ProviderConfig, + model: ModelConfig, + spec: AdapterParameterSpec | None, + records: list[dict[str, Any]], + credential_revisions: Mapping[str, int], + ) -> dict[str, Any] | None: + if spec is None: + return None + auth_spec = spec.auth_specs.get(provider.auth.mode) + credential_revision = 0 + if ( + auth_spec is not None + and auth_spec.credential_required + and provider.auth.credential_id is not None + ): + credential_revision = credential_revisions.get( + provider.auth.credential_id, 0 + ) + config_hash = configuration_hash(provider, model) + for record in records: + if ( + record["configuration_hash"] == config_hash + and record["credential_revision"] == credential_revision + and record["adapter_spec_revision"] == spec.spec_revision + ): + return record + return None + + @staticmethod + def _verification_info( + current: dict[str, Any] | None, + records: list[dict[str, Any]], + ) -> VerificationInfo: + if current is not None: + return VerificationInfo( + status=current["result"], + verified_at=_rfc3339(int(current["verified_at"])), + adapter_spec_revision=int(current["adapter_spec_revision"]), + ) + if records: + latest = max(records, key=lambda record: int(record["verified_at"])) + return VerificationInfo( + status="stale", + verified_at=_rfc3339(int(latest["verified_at"])), + adapter_spec_revision=int(latest["adapter_spec_revision"]), + ) + return VerificationInfo(status="none") diff --git a/EvoScientist/model_registry/snapshots.py b/EvoScientist/model_registry/snapshots.py new file mode 100644 index 0000000..6e1f003 --- /dev/null +++ b/EvoScientist/model_registry/snapshots.py @@ -0,0 +1,413 @@ +"""Run runtime snapshot service (design doc 5.2, 8.1, 8.2). + +``SnapshotService`` is the single implementation shared by the HTTP snapshot +API (Task 5) and the local CLI/channel/scheduler entry points (Task 7): it +freezes both roles' complete ``ResolvedModelConfig`` — adapter spec +revision, fixed budget reserves, capabilities, and credential revisions — +into ``run_runtime_snapshots.payload_json``. Payloads and logs never carry +``secret_value``. + +- Idempotency: one ``{deployment_id, thread_id, run_request_id}`` triplet + maps to at most one non-terminal snapshot. A repeated request with the + same ``selection_hash`` returns the original snapshot; a different hash + raises ``RUN_REQUEST_CONFLICT``. ``selection_hash`` covers the + pre-resolution ``{primary, auxiliary}`` selection (inherit participates as + ``null``); ``model_selection_revision`` is an audit field and never part + of the hash. +- Lifecycle: ``prepared`` (TTL 15 minutes) → ``bound`` (retained 24 hours + after binding) → ``expired`` (terminal, via ``cleanup_expired``); + ``prepared`` → ``aborted`` on run creation failure. Terminal snapshots + free their triplet for recreation and reject reads/binds with + ``SNAPSHOT_EXPIRED``. +- Reads revalidate the frozen ``adapter_spec_revision`` against the current + contracts and verify the deployment/thread binding. +- Credentials are resolved per call from the frozen ``credential_revision``; + there is no in-process secret cache (section 5.2 rule 4). +""" + +from __future__ import annotations + +import hashlib +import json +import sqlite3 +import time +import uuid +from typing import Any, Literal, get_args + +from pydantic import BaseModel, NonNegativeInt, PositiveInt + +from .adapters import find_adapter_spec +from .errors import ( + ADAPTER_NOT_SUPPORTED, + MODEL_REGISTRY_NOT_READY, + RUN_REQUEST_CONFLICT, + SNAPSHOT_ALREADY_BOUND, + SNAPSHOT_EXPIRED, + SNAPSHOT_NOT_FOUND, + ModelRegistryError, +) +from .resolver import ModelRegistryResolver +from .schemas import ( + ModelRef, + ModelRole, + NonEmptyString, + ResolvedModelConfig, +) +from .store import ModelRuntimeStore + +PREPARED_TTL_SECONDS = 15 * 60 +BOUND_RETENTION_SECONDS = 24 * 60 * 60 + +_MODEL_ROLES = get_args(ModelRole) +_TERMINAL_STATUSES = ("expired", "aborted") + + +class SnapshotCreateRequest(BaseModel): + """The section 8.2 snapshot creation request body.""" + + run_request_id: NonEmptyString + thread_id: NonEmptyString + deployment_id: NonEmptyString + model_selection_revision: NonNegativeInt = 0 + # ``None`` means inherit: the registry defaults are resolved at creation. + primary: ModelRef | None = None + auxiliary: ModelRef | None = None + + +class SnapshotPayload(BaseModel): + """The frozen private payload; never contains secrets.""" + + registry_revision: PositiveInt + model_selection_revision: NonNegativeInt + primary: ResolvedModelConfig + auxiliary: ResolvedModelConfig | None = None + + +class RuntimeSnapshot(BaseModel): + """One ``run_runtime_snapshots`` row with a typed payload.""" + + snapshot_id: str + deployment_id: str + thread_id: str + run_request_id: str + selection_hash: str + status: Literal["prepared", "bound", "expired", "aborted"] + langgraph_run_id: str | None + payload: SnapshotPayload + created_at: int + expires_at: int + + @classmethod + def from_row(cls, row: dict[str, Any]) -> RuntimeSnapshot: + return cls( + snapshot_id=row["snapshot_id"], + deployment_id=row["deployment_id"], + thread_id=row["thread_id"], + run_request_id=row["run_request_id"], + selection_hash=row["selection_hash"], + status=row["status"], + langgraph_run_id=row["langgraph_run_id"], + payload=SnapshotPayload.model_validate(row["payload"]), + created_at=row["created_at"], + expires_at=row["expires_at"], + ) + + +class SnapshotCreation(BaseModel): + """``create`` result; ``created`` distinguishes 201 from 200 semantics.""" + + snapshot: RuntimeSnapshot + created: bool + + +def compute_selection_hash(primary: ModelRef | None, auxiliary: ModelRef | None) -> str: + """Hash the pre-resolution selection; inherit participates as ``null``.""" + + def entry(ref: ModelRef | None) -> dict[str, str] | None: + if ref is None: + return None + return {"provider_id": ref.provider_id, "model_key": ref.model_key} + + encoded = json.dumps( + {"primary": entry(primary), "auxiliary": entry(auxiliary)}, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def config_for_role(snapshot: RuntimeSnapshot, role: ModelRole) -> ResolvedModelConfig: + """The section 6.1 role mapping; no other mapping may be guessed. + + ``primary`` → ``snapshot.primary``; ``auxiliary``, ``summary``, and + ``tool_selector`` → ``snapshot.auxiliary ?? snapshot.primary``. + """ + if role not in _MODEL_ROLES: + raise ValueError( + f"Unknown model role {role!r}; expected one of {list(_MODEL_ROLES)}." + ) + if role == "primary" or snapshot.payload.auxiliary is None: + return snapshot.payload.primary + return snapshot.payload.auxiliary + + +def public_snapshot_view(snapshot: RuntimeSnapshot) -> dict[str, Any]: + """The section 8.2 public diagnostic subset; never contains secrets.""" + + def public_config(config: ResolvedModelConfig) -> dict[str, Any]: + return { + "provider_id": config.model_ref.provider_id, + "model_key": config.model_ref.model_key, + "adapter_spec_revision": config.adapter_spec_revision, + "runtime": { + "max_output_tokens": config.request_options.max_output_tokens, + "temperature": config.request_options.temperature, + "top_p": config.request_options.top_p, + "timeout_seconds": config.client_options.timeout_seconds, + "max_retries": config.client_options.max_retries, + }, + } + + auxiliary = snapshot.payload.auxiliary + return { + "snapshot_id": snapshot.snapshot_id, + "registry_revision": snapshot.payload.registry_revision, + "primary": public_config(snapshot.payload.primary), + "auxiliary": None if auxiliary is None else public_config(auxiliary), + } + + +class SnapshotService: + """Creates, binds, reads, and cleans up run runtime snapshots.""" + + def __init__( + self, store: ModelRuntimeStore, resolver: ModelRegistryResolver + ) -> None: + self._store = store + self._resolver = resolver + + # --- creation --------------------------------------------------------- + + def create(self, request: SnapshotCreateRequest) -> SnapshotCreation: + """Freeze the requested (or inherited) selection into a snapshot.""" + registry = self._store.load_registry() + if registry.state != "active": + raise ModelRegistryError( + MODEL_REGISTRY_NOT_READY, + "The model registry is in bootstrap; configure and enable a " + "primary model first.", + ) + selection_hash = compute_selection_hash(request.primary, request.auxiliary) + existing = self._store.find_active_run_snapshot( + deployment_id=request.deployment_id, + thread_id=request.thread_id, + run_request_id=request.run_request_id, + ) + if existing is not None: + if existing["selection_hash"] == selection_hash: + return SnapshotCreation( + snapshot=RuntimeSnapshot.from_row(existing), created=False + ) + raise ModelRegistryError( + RUN_REQUEST_CONFLICT, + "run_request_id already exists with a different model selection.", + details=[{"path": "run_request_id", "code": RUN_REQUEST_CONFLICT}], + ) + + primary_ref = ( + request.primary + if request.primary is not None + else registry.defaults.primary + ) + auxiliary_ref = ( + request.auxiliary + if request.auxiliary is not None + else registry.defaults.auxiliary + ) + if primary_ref is None: # pragma: no cover - active registries set it + raise ModelRegistryError( + MODEL_REGISTRY_NOT_READY, + "The model registry has no default primary model.", + ) + primary_config = self._resolver.resolve( + primary_ref, "primary", registry=registry + ) + auxiliary_config = ( + None + if auxiliary_ref is None + else self._resolver.resolve(auxiliary_ref, "auxiliary", registry=registry) + ) + payload = { + "registry_revision": registry.revision, + "model_selection_revision": request.model_selection_revision, + "primary": primary_config.model_dump(mode="json"), + "auxiliary": ( + None + if auxiliary_config is None + else auxiliary_config.model_dump(mode="json") + ), + } + snapshot_id = f"snap-{uuid.uuid4().hex}" + try: + self._store.insert_run_snapshot( + snapshot_id=snapshot_id, + deployment_id=request.deployment_id, + thread_id=request.thread_id, + run_request_id=request.run_request_id, + selection_hash=selection_hash, + payload=payload, + expires_at=int(time.time()) + PREPARED_TTL_SECONDS, + ) + except sqlite3.IntegrityError: + # Lost a creation race: the winner now occupies the triplet, so + # the standard idempotency comparison applies. + winner = self._store.find_active_run_snapshot( + deployment_id=request.deployment_id, + thread_id=request.thread_id, + run_request_id=request.run_request_id, + ) + if winner is not None and winner["selection_hash"] == selection_hash: + return SnapshotCreation( + snapshot=RuntimeSnapshot.from_row(winner), created=False + ) + raise ModelRegistryError( + RUN_REQUEST_CONFLICT, + "run_request_id already exists with a different model selection.", + details=[{"path": "run_request_id", "code": RUN_REQUEST_CONFLICT}], + ) from None + row = self._store.get_run_snapshot(snapshot_id) + assert row is not None # pragma: no cover - inserted above + return SnapshotCreation(snapshot=RuntimeSnapshot.from_row(row), created=True) + + # --- lifecycle ---------------------------------------------------------- + + def bind(self, snapshot_id: str, langgraph_run_id: str) -> RuntimeSnapshot: + """Transition ``prepared`` → ``bound`` once and extend the retention. + + Re-binding with the same ``langgraph_run_id`` is idempotent; a + different value conflicts with ``SNAPSHOT_ALREADY_BOUND``. + """ + while True: + row = self._store.get_run_snapshot(snapshot_id) + if row is None: + raise _snapshot_not_found() + if row["status"] == "bound": + if row["langgraph_run_id"] == langgraph_run_id: + return RuntimeSnapshot.from_row(row) + raise ModelRegistryError( + SNAPSHOT_ALREADY_BOUND, + "The snapshot is already bound to a LangGraph run.", + ) + if row["status"] in _TERMINAL_STATUSES: + raise _snapshot_expired() + expires_at = int(time.time()) + BOUND_RETENTION_SECONDS + if self._store.bind_run_snapshot( + snapshot_id, + langgraph_run_id=langgraph_run_id, + expires_at=expires_at, + ): + bound = self._store.get_run_snapshot(snapshot_id) + assert bound is not None # pragma: no cover - just updated + return RuntimeSnapshot.from_row(bound) + # Lost a state-transition race; re-read and apply the rules. + + def abort(self, snapshot_id: str) -> None: + """Mark a ``prepared`` snapshot ``aborted`` (run creation failed).""" + row = self._store.get_run_snapshot(snapshot_id) + if row is None: + raise _snapshot_not_found() + if row["status"] == "aborted": + return + if row["status"] == "bound": + raise ModelRegistryError( + SNAPSHOT_ALREADY_BOUND, + "A bound snapshot cannot be aborted.", + ) + if row["status"] == "expired": + raise _snapshot_expired() + self._store.set_run_snapshot_status(snapshot_id, "aborted") + + def get( + self, + snapshot_id: str, + *, + deployment_id: str, + thread_id: str, + ) -> RuntimeSnapshot: + """Read a snapshot after verifying its deployment/thread binding. + + The frozen ``adapter_spec_revision`` is revalidated against the + current contracts: a removed revision fails loudly with + ``ADAPTER_NOT_SUPPORTED`` instead of being silently substituted. + """ + row = self._store.get_run_snapshot(snapshot_id) + if ( + row is None + or row["deployment_id"] != deployment_id + or row["thread_id"] != thread_id + ): + raise _snapshot_not_found() + if row["status"] in _TERMINAL_STATUSES: + raise _snapshot_expired() + snapshot = RuntimeSnapshot.from_row(row) + self._ensure_frozen_specs_available(snapshot) + return snapshot + + def cleanup_expired(self, now: int | None = None) -> list[str]: + """Mark due snapshots ``expired``; returns the transitioned IDs.""" + return self._store.expire_due_run_snapshots( + int(time.time()) if now is None else now + ) + + # --- credentials -------------------------------------------------------- + + def resolve_snapshot_credential( + self, snapshot: RuntimeSnapshot, role: ModelRole + ) -> str: + """Resolve the secret for the role's frozen credential revision. + + Reads the credential store on every call — secrets are never cached + in process memory (section 5.2). A destroyed revision raises + ``RUN_CREDENTIAL_REVISION_UNAVAILABLE``; ``mode=none`` models have + no credential and yield an empty string. + """ + config = config_for_role(snapshot, role) + auth_ref = config.auth_ref + if auth_ref.mode == "none": + return "" + assert auth_ref.credential_id is not None # AuthSpec validation + assert auth_ref.credential_revision is not None + return self._store.resolve_credential( + auth_ref.credential_id, auth_ref.credential_revision + ) + + # --- internals ------------------------------------------------------------ + + def _ensure_frozen_specs_available(self, snapshot: RuntimeSnapshot) -> None: + configs = [snapshot.payload.primary] + if snapshot.payload.auxiliary is not None: + configs.append(snapshot.payload.auxiliary) + for config in configs: + spec = find_adapter_spec( + config.adapter_id, + config.upstream_model_id, + spec_revision=config.adapter_spec_revision, + specs=self._resolver.specs, + ) + if spec is None: + raise ModelRegistryError( + ADAPTER_NOT_SUPPORTED, + f"The adapter contract {config.adapter_id!r} at " + f"spec_revision {config.adapter_spec_revision} frozen by " + "this snapshot no longer exists.", + ) + + +def _snapshot_not_found() -> ModelRegistryError: + return ModelRegistryError(SNAPSHOT_NOT_FOUND, "The snapshot does not exist.") + + +def _snapshot_expired() -> ModelRegistryError: + return ModelRegistryError( + SNAPSHOT_EXPIRED, "The snapshot has reached a terminal state." + ) diff --git a/EvoScientist/model_registry/store.py b/EvoScientist/model_registry/store.py index c0b720f..39db28a 100644 --- a/EvoScientist/model_registry/store.py +++ b/EvoScientist/model_registry/store.py @@ -107,6 +107,21 @@ class SharedStorageError(RuntimeError): """Raised at startup when the database cannot take a write lock.""" +def _run_snapshot_row_to_dict(row: tuple) -> dict[str, Any]: + return { + "snapshot_id": str(row[0]), + "deployment_id": str(row[1]), + "thread_id": str(row[2]), + "run_request_id": str(row[3]), + "selection_hash": str(row[4]), + "status": str(row[5]), + "langgraph_run_id": row[6] if row[6] is None else str(row[6]), + "payload": json.loads(str(row[7])), + "created_at": int(row[8]), + "expires_at": int(row[9]), + } + + def _rfc3339(epoch_seconds: int) -> str: return datetime.fromtimestamp(epoch_seconds, UTC).strftime("%Y-%m-%dT%H:%M:%SZ") @@ -482,6 +497,17 @@ class ModelRuntimeStore: finally: connection.close() + def current_credential_revision(self, credential_id: str) -> int | None: + """Return the pointer's current revision, or ``None`` when unconfigured.""" + _validate_credential_id(credential_id) + with self._lock, self._connect() as connection: + row = connection.execute( + "SELECT current_revision FROM credential_pointers " + "WHERE credential_id = ?", + (credential_id,), + ).fetchone() + return None if row is None else int(row[0]) + def credential_status(self, credential_id: str) -> CredentialStatus: """Return the redacted browser-safe status; never the plaintext.""" _validate_credential_id(credential_id) @@ -603,6 +629,51 @@ class ModelRuntimeStore: "error_code": row[3] if row[3] is None else str(row[3]), } + def list_model_verifications( + self, + *, + provider_id: str | None = None, + model_key: str | None = None, + ) -> list[dict[str, Any]]: + """Return verification records, optionally narrowed to one model. + + Each record carries the full five-tuple plus the result, verified + capabilities, timestamp, and redacted error code — the input the + section 4.3 availability judgement consumes. + """ + query = ( + "SELECT provider_id, model_key, configuration_hash, " + "credential_revision, adapter_spec_revision, result, " + "verified_capabilities_json, verified_at, error_code " + "FROM model_verifications" + ) + clauses: list[str] = [] + params: list[str] = [] + if provider_id is not None: + clauses.append("provider_id = ?") + params.append(provider_id) + if model_key is not None: + clauses.append("model_key = ?") + params.append(model_key) + if clauses: + query += " WHERE " + " AND ".join(clauses) + with self._lock, self._connect() as connection: + rows = connection.execute(query, params).fetchall() + return [ + { + "provider_id": str(row[0]), + "model_key": str(row[1]), + "configuration_hash": str(row[2]), + "credential_revision": int(row[3]), + "adapter_spec_revision": int(row[4]), + "result": str(row[5]), + "verified_capabilities": json.loads(str(row[6])), + "verified_at": int(row[7]), + "error_code": row[8] if row[8] is None else str(row[8]), + } + for row in rows + ] + # --- run runtime snapshots --------------------------------------------- def insert_run_snapshot( @@ -698,15 +769,88 @@ class ModelRuntimeStore: ).fetchone() if row is None: return None - return { - "snapshot_id": str(row[0]), - "deployment_id": str(row[1]), - "thread_id": str(row[2]), - "run_request_id": str(row[3]), - "selection_hash": str(row[4]), - "status": str(row[5]), - "langgraph_run_id": row[6] if row[6] is None else str(row[6]), - "payload": json.loads(str(row[7])), - "created_at": int(row[8]), - "expires_at": int(row[9]), - } + return _run_snapshot_row_to_dict(row) + + def find_active_run_snapshot( + self, + *, + deployment_id: str, + thread_id: str, + run_request_id: str, + ) -> dict[str, Any] | None: + """Return the non-terminal snapshot for one triplet, if any. + + Only ``prepared``/``bound`` rows occupy the triplet (the partial + unique index), so at most one row matches. + """ + with self._lock, self._connect() as connection: + row = connection.execute( + "SELECT snapshot_id, deployment_id, thread_id, run_request_id, " + "selection_hash, status, langgraph_run_id, payload_json, " + "created_at, expires_at FROM run_runtime_snapshots " + "WHERE deployment_id = ? AND thread_id = ? " + "AND run_request_id = ? AND status IN ('prepared', 'bound')", + (deployment_id, thread_id, run_request_id), + ).fetchone() + if row is None: + return None + return _run_snapshot_row_to_dict(row) + + def bind_run_snapshot( + self, + snapshot_id: str, + *, + langgraph_run_id: str, + expires_at: int, + ) -> bool: + """Bind a ``prepared`` snapshot; return False when it left that state. + + The conditional update keeps the prepared→bound transition atomic so + a concurrent bind/abort/expire cannot produce a double bind. + """ + with self._lock: + connection = self._connect() + try: + connection.execute("BEGIN IMMEDIATE") + cursor = connection.execute( + "UPDATE run_runtime_snapshots " + "SET status = 'bound', langgraph_run_id = ?, expires_at = ? " + "WHERE snapshot_id = ? AND status = 'prepared'", + (langgraph_run_id, expires_at, snapshot_id), + ) + connection.commit() + return cursor.rowcount == 1 + except BaseException: + connection.rollback() + raise + finally: + connection.close() + + def expire_due_run_snapshots(self, now: int) -> list[str]: + """Mark every due prepared/bound snapshot ``expired`` (terminal). + + Returns the IDs of the snapshots transitioned by this call. + """ + with self._lock: + connection = self._connect() + try: + connection.execute("BEGIN IMMEDIATE") + rows = connection.execute( + "SELECT snapshot_id FROM run_runtime_snapshots " + "WHERE status IN ('prepared', 'bound') AND expires_at <= ?", + (now,), + ).fetchall() + snapshot_ids = [str(row[0]) for row in rows] + if snapshot_ids: + connection.execute( + "UPDATE run_runtime_snapshots SET status = 'expired' " + "WHERE status IN ('prepared', 'bound') AND expires_at <= ?", + (now,), + ) + connection.commit() + return snapshot_ids + except BaseException: + connection.rollback() + raise + finally: + connection.close() diff --git a/tests/test_model_registry_schemas.py b/tests/test_model_registry_schemas.py index fe67eff..f9a429a 100644 --- a/tests/test_model_registry_schemas.py +++ b/tests/test_model_registry_schemas.py @@ -50,6 +50,7 @@ ALL_ERROR_CODES = { "MODEL_CONFIGURATION_CHANGED": 409, "DELEGATION_REPLAYED": 401, "MODEL_NOT_FOUND": 404, + "SNAPSHOT_NOT_FOUND": 404, "MODEL_REGISTRY_NOT_READY": 422, "MODEL_DISABLED": 422, "MODEL_NOT_AVAILABLE": 422, diff --git a/tests/test_resolver.py b/tests/test_resolver.py new file mode 100644 index 0000000..da12764 --- /dev/null +++ b/tests/test_resolver.py @@ -0,0 +1,520 @@ +"""Tests for ModelRegistryResolver (design doc 4.3, 6.1, 6.4, 6.5, 8.1). + +Covers ``resolve``/``resolve_for_test`` (provider, model, contract, +verification five-tuple, limits, and budget validation producing a frozen +``ResolvedModelConfig``) and ``compute_availability`` (the section 4.3 +six-state judgement order, the single availability computation). +""" + +from __future__ import annotations + +import pytest + +from EvoScientist.model_registry.adapters import adapter_specs, find_adapter_spec +from EvoScientist.model_registry.errors import ( + ADAPTER_NOT_SUPPORTED, + AUTH_MODE_UNSUPPORTED, + CONTEXT_BUDGET_UNSATISFIABLE, + CREDENTIAL_NOT_CONFIGURED, + MODEL_DISABLED, + MODEL_LIMITS_UNCONFIRMED, + MODEL_NOT_AVAILABLE, + MODEL_NOT_FOUND, + ModelRegistryError, +) +from EvoScientist.model_registry.hashing import configuration_hash +from EvoScientist.model_registry.resolver import ModelRegistryResolver +from EvoScientist.model_registry.schemas import ( + CredentialWrite, + ModelRef, + RegistryV4, +) +from EvoScientist.model_registry.store import ModelRuntimeStore + +ZHIPU_REF = ModelRef(provider_id="zhipu-glm", model_key="glm-5.2") +OLLAMA_REF = ModelRef(provider_id="local-ollama", model_key="qwen3") +SECRET = "sk-live-9876abcd" +VERIFIED_AT = 1_750_000_000 # 2025-06-15T15:06:40Z + + +def _model_runtime(**overrides): + payload = { + "limit_mode": "combined", + "context_window_tokens": 1048576, + "max_input_tokens": None, + "max_output_tokens": 32768, + "min_effective_input_tokens": 8192, + "fixed_system_reserve_tokens": 4096, + "fixed_tools_reserve_tokens": 8192, + "fixed_attachments_reserve_tokens": 4096, + "limits_status": "confirmed", + "limits_source": "provider", + "temperature": None, + "top_p": None, + "reasoning_effort": "auto", + "declared_capabilities": { + "tools": True, + "vision": False, + "structured_output": True, + }, + } + payload.update(overrides) + return payload + + +def _zhipu_provider(**overrides): + payload = { + "id": "zhipu-glm", + "name": "Zhipu GLM", + "adapter": "openai-compatible", + "base_url": "https://open.bigmodel.cn/api/paas/v4", + "auth": {"mode": "api_key", "credential_id": "zhipu-primary"}, + "enabled": True, + "runtime": { + "timeout_seconds": 120, + "max_retries": 2, + "default_temperature": 0.7, + "default_top_p": 0.95, + "default_reasoning_effort": "auto", + }, + "models": [ + { + "key": "glm-5.2", + "name": "GLM-5.2", + "upstream_model_id": "glm-5.2", + "enabled": True, + "runtime": _model_runtime(), + } + ], + } + payload.update(overrides) + return payload + + +def _ollama_provider(**overrides): + payload = { + "id": "local-ollama", + "name": "Local Ollama", + "adapter": "ollama", + "base_url": "http://localhost:11434", + "auth": {"mode": "none", "credential_id": None}, + "enabled": True, + "runtime": {"timeout_seconds": 120, "max_retries": 2}, + "models": [ + { + "key": "qwen3", + "name": "Qwen3", + "upstream_model_id": "qwen3", + "enabled": True, + "runtime": _model_runtime(), + } + ], + } + payload.update(overrides) + return payload + + +def _registry(providers=None, **overrides) -> RegistryV4: + payload = { + "version": 4, + "revision": 1, + "state": "bootstrap", + "defaults": {"primary": None, "auxiliary": None}, + "providers": [_zhipu_provider()] if providers is None else providers, + } + payload.update(overrides) + return RegistryV4.model_validate(payload) + + +@pytest.fixture +def store(tmp_path): + return ModelRuntimeStore(config_dir=tmp_path) + + +def _activate(store, registry, *, credential=True): + writes = ( + [CredentialWrite(credential_id="zhipu-primary", secret_value=SECRET)] + if credential + else None + ) + return store.save_registry( + expected_revision=1, registry=registry, credential_writes=writes + ) + + +def _verify( + store, + registry, + provider_id="zhipu-glm", + model_key="glm-5.2", + *, + result="passed", + config_hash=None, + credential_revision=None, + adapter_spec_revision=None, + caps=None, +): + provider = registry.find_provider(provider_id) + model = provider.find_model(model_key) + spec = find_adapter_spec(provider.adapter, model.upstream_model_id) + if credential_revision is None: + credential_revision = 1 if provider.auth.credential_id else 0 + store.record_model_verification( + provider_id=provider_id, + model_key=model_key, + configuration_hash=config_hash or configuration_hash(provider, model), + credential_revision=credential_revision, + adapter_spec_revision=adapter_spec_revision or spec.spec_revision, + result=result, + verified_capabilities=caps + or {"tools": True, "vision": False, "structured_output": True}, + ) + + +@pytest.fixture +def active_store(store): + registry = _activate( + store, _registry(providers=[_zhipu_provider(), _ollama_provider()]) + ) + _verify(store, registry) + return store + + +@pytest.fixture +def resolver(active_store): + return ModelRegistryResolver(active_store) + + +class TestResolve: + def test_freezes_complete_resolved_config(self, resolver): + config = resolver.resolve(ZHIPU_REF, "primary") + assert config.model_ref == ZHIPU_REF + assert config.role == "primary" + assert config.adapter_id == "openai-compatible" + # The glm-5.2 model-specific contract wins over the generic one. + assert config.adapter_spec_revision == 1 + assert config.upstream_model_id == "glm-5.2" + assert config.base_url == "https://open.bigmodel.cn/api/paas/v4" + assert config.auth_ref.mode == "api_key" + assert config.auth_ref.credential_id == "zhipu-primary" + assert config.auth_ref.credential_revision == 1 + assert config.client_options.timeout_seconds == 120 + assert config.client_options.max_retries == 2 + assert config.request_options.max_output_tokens == 32768 + assert config.request_options.temperature == 0.7 + assert config.request_options.top_p == 0.95 + assert config.request_options.reasoning_effort == "auto" + # Section 6.5: combined mode subtracts the output budget. + assert config.budget.resolved_input_limit == 1048576 - 32768 + assert config.budget.fixed_reserves.fixed_system_reserve_tokens == 4096 + assert config.budget.fixed_reserves.fixed_tools_reserve_tokens == 8192 + assert config.budget.fixed_reserves.fixed_attachments_reserve_tokens == 4096 + assert config.budget.message_budget == 1048576 - 32768 - 4096 + # protocol AND declared AND verified (glm-5.2 has no vision). + assert config.effective_capabilities.tools is True + assert config.effective_capabilities.vision is False + assert config.effective_capabilities.structured_output is True + + def test_resolved_config_never_carries_secret(self, resolver): + config = resolver.resolve(ZHIPU_REF, "primary") + assert SECRET not in config.model_dump_json() + + def test_all_roles_are_stamped(self, resolver): + for role in ("primary", "auxiliary", "summary", "tool_selector"): + assert resolver.resolve(ZHIPU_REF, role).role == role + + def test_unknown_role_rejected(self, resolver): + with pytest.raises(ValueError, match="role"): + resolver.resolve(ZHIPU_REF, "planner") + + def test_unknown_provider_is_not_found(self, resolver): + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(ModelRef(provider_id="missing", model_key="glm-5.2")) + assert excinfo.value.code == MODEL_NOT_FOUND + + def test_unknown_model_is_not_found(self, resolver): + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(ModelRef(provider_id="zhipu-glm", model_key="missing")) + assert excinfo.value.code == MODEL_NOT_FOUND + + def test_disabled_provider_is_unavailable(self, store): + registry = _activate( + store, _registry(providers=[_zhipu_provider(enabled=False)]) + ) + _verify(store, registry) + resolver = ModelRegistryResolver(store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(ZHIPU_REF) + assert excinfo.value.code == MODEL_NOT_AVAILABLE + + def test_disabled_model_is_disabled(self, store): + provider = _zhipu_provider() + provider["models"][0]["enabled"] = False + registry = _activate(store, _registry(providers=[provider])) + _verify(store, registry) + resolver = ModelRegistryResolver(store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(ZHIPU_REF) + assert excinfo.value.code == MODEL_DISABLED + + def test_unopened_adapter_rejected(self, store): + provider = _zhipu_provider(adapter="google-genai") + _activate(store, _registry(providers=[provider])) + resolver = ModelRegistryResolver(store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(ZHIPU_REF) + assert excinfo.value.code == ADAPTER_NOT_SUPPORTED + + def test_no_matching_contract_is_unavailable(self, store): + registry = _activate(store, _registry(providers=[_ollama_provider()])) + _verify(store, registry, "local-ollama", "qwen3") + # Only the glm-5.2 model-specific contract: qwen3 matches nothing. + specs = tuple(spec for spec in adapter_specs() if spec.model_selector != "*") + resolver = ModelRegistryResolver(store, specs=specs) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(OLLAMA_REF) + assert excinfo.value.code == MODEL_NOT_AVAILABLE + + def test_missing_verification_record_is_unavailable(self, active_store): + # The active_store fixture recorded no verification for qwen3. + resolver = ModelRegistryResolver(active_store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(OLLAMA_REF) + assert excinfo.value.code == MODEL_NOT_AVAILABLE + + def test_failed_verification_is_unavailable(self, store): + registry = _activate(store, _registry()) + _verify(store, registry, result="failed") + resolver = ModelRegistryResolver(store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(ZHIPU_REF) + assert excinfo.value.code == MODEL_NOT_AVAILABLE + + def test_stale_configuration_hash_is_unavailable(self, store): + registry = _activate(store, _registry()) + _verify(store, registry, config_hash="0" * 64) + resolver = ModelRegistryResolver(store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(ZHIPU_REF) + assert excinfo.value.code == MODEL_NOT_AVAILABLE + + def test_credential_rotation_invalidates_verification(self, active_store): + # Verifed at revision 1; rotating the key moves the pointer to 2. + active_store.write_credential_version("zhipu-primary", "sk-rotated-2222") + resolver = ModelRegistryResolver(active_store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(ZHIPU_REF) + assert excinfo.value.code == MODEL_NOT_AVAILABLE + + def test_unconfirmed_limits_rejected(self, store): + provider = _zhipu_provider() + provider["models"][0]["runtime"]["limits_status"] = "needs_confirmation" + registry = _activate(store, _registry(providers=[provider])) + _verify(store, registry) + resolver = ModelRegistryResolver(store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(ZHIPU_REF) + assert excinfo.value.code == MODEL_LIMITS_UNCONFIRMED + + def test_budget_unsatisfiable_rejected(self, store): + provider = _zhipu_provider() + provider["models"][0]["runtime"] = _model_runtime( + context_window_tokens=16384, + max_output_tokens=8192, + min_effective_input_tokens=8192, + ) + registry = _activate(store, _registry(providers=[provider])) + _verify(store, registry) + resolver = ModelRegistryResolver(store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(ZHIPU_REF) + assert excinfo.value.code == CONTEXT_BUDGET_UNSATISFIABLE + + def test_missing_credential_is_rejected(self, store): + _activate(store, _registry(), credential=False) + resolver = ModelRegistryResolver(store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(ZHIPU_REF) + assert excinfo.value.code == CREDENTIAL_NOT_CONFIGURED + + def test_auth_mode_not_in_contract_rejected(self, store): + provider = _zhipu_provider( + auth={"mode": "bearer", "credential_id": "zhipu-primary"} + ) + _activate(store, _registry(providers=[provider])) + resolver = ModelRegistryResolver(store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve(ZHIPU_REF) + assert excinfo.value.code == AUTH_MODE_UNSUPPORTED + + def test_mode_none_resolves_without_credential(self, store): + registry = _activate(store, _registry(providers=[_ollama_provider()])) + _verify(store, registry, "local-ollama", "qwen3") + resolver = ModelRegistryResolver(store) + config = resolver.resolve(OLLAMA_REF, "auxiliary") + assert config.auth_ref.mode == "none" + assert config.auth_ref.credential_id is None + assert config.auth_ref.credential_revision is None + + +class TestResolveForTest: + def test_allows_disabled_model(self, store): + provider = _zhipu_provider() + provider["models"][0]["enabled"] = False + registry = _activate(store, _registry(providers=[provider])) + _verify(store, registry) + resolver = ModelRegistryResolver(store) + config = resolver.resolve_for_test(ZHIPU_REF) + assert config.model_ref == ZHIPU_REF + + def test_still_enforces_credential(self, store): + _activate(store, _registry(), credential=False) + resolver = ModelRegistryResolver(store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve_for_test(ZHIPU_REF) + assert excinfo.value.code == CREDENTIAL_NOT_CONFIGURED + + def test_still_enforces_disabled_provider(self, store): + registry = _activate( + store, _registry(providers=[_zhipu_provider(enabled=False)]) + ) + _verify(store, registry) + resolver = ModelRegistryResolver(store) + with pytest.raises(ModelRegistryError) as excinfo: + resolver.resolve_for_test(ZHIPU_REF) + assert excinfo.value.code == MODEL_NOT_AVAILABLE + + +def _record( + registry, + provider_id="zhipu-glm", + model_key="glm-5.2", + *, + result="passed", + config_hash=None, + credential_revision=1, + adapter_spec_revision=1, + verified_at=VERIFIED_AT, + caps=None, + error_code=None, +): + provider = registry.find_provider(provider_id) + model = provider.find_model(model_key) + return { + "provider_id": provider_id, + "model_key": model_key, + "configuration_hash": config_hash or configuration_hash(provider, model), + "credential_revision": credential_revision, + "adapter_spec_revision": adapter_spec_revision, + "result": result, + "verified_capabilities": caps + or {"tools": True, "vision": False, "structured_output": True}, + "verified_at": verified_at, + "error_code": error_code, + } + + +class TestComputeAvailability: + def test_enabled_state(self, store): + registry = _registry() + resolver = ModelRegistryResolver(store) + records = [_record(registry)] + (availability,) = resolver.compute_availability( + registry, records, credential_revisions={"zhipu-primary": 1} + ) + assert availability.model_ref == ZHIPU_REF + assert availability.state == "enabled" + assert availability.selectable is True + assert availability.reason_code is None + assert availability.verification.status == "passed" + assert availability.verification.verified_at == "2025-06-15T15:06:40Z" + assert availability.verification.adapter_spec_revision == 1 + assert availability.effective_capabilities.tools is True + assert availability.effective_capabilities.vision is False + assert availability.effective_capabilities.structured_output is True + + def test_verified_when_model_disabled(self, store): + provider = _zhipu_provider() + provider["models"][0]["enabled"] = False + registry = _registry(providers=[provider]) + resolver = ModelRegistryResolver(store) + (availability,) = resolver.compute_availability( + registry, [_record(registry)], credential_revisions={"zhipu-primary": 1} + ) + assert availability.state == "verified" + assert availability.selectable is False + assert availability.verification.status == "passed" + + def test_unavailable_when_provider_disabled(self, store): + registry = _registry(providers=[_zhipu_provider(enabled=False)]) + resolver = ModelRegistryResolver(store) + (availability,) = resolver.compute_availability( + registry, [_record(registry)], credential_revisions={"zhipu-primary": 1} + ) + assert availability.state == "unavailable" + assert availability.selectable is False + assert availability.reason_code == "PROVIDER_DISABLED" + + def test_unavailable_without_matching_contract(self, store): + registry = _registry(providers=[_ollama_provider()]) + specs = tuple(spec for spec in adapter_specs() if spec.model_selector != "*") + resolver = ModelRegistryResolver(store, specs=specs) + (availability,) = resolver.compute_availability( + registry, + [_record(registry, "local-ollama", "qwen3", credential_revision=0)], + ) + assert availability.state == "unavailable" + assert availability.reason_code == "NO_ADAPTER_CONTRACT" + + def test_verification_failed(self, store): + registry = _registry() + resolver = ModelRegistryResolver(store) + (availability,) = resolver.compute_availability( + registry, + [_record(registry, result="failed", error_code="PROVIDER_UNREACHABLE")], + credential_revisions={"zhipu-primary": 1}, + ) + assert availability.state == "verification_failed" + assert availability.selectable is False + assert availability.reason_code == "PROVIDER_UNREACHABLE" + assert availability.verification.status == "failed" + assert availability.effective_capabilities.tools is False + + def test_stale_when_only_old_records_exist(self, store): + registry = _registry() + resolver = ModelRegistryResolver(store) + records = [_record(registry, config_hash="0" * 64)] + (availability,) = resolver.compute_availability( + registry, records, credential_revisions={"zhipu-primary": 1} + ) + # Stale wins over configured when records exist for other five-tuples. + assert availability.state == "verification_stale" + assert availability.selectable is False + assert availability.reason_code == "VERIFICATION_STALE" + assert availability.verification.status == "stale" + assert availability.verification.verified_at == "2025-06-15T15:06:40Z" + + def test_stale_after_credential_rotation(self, store): + registry = _registry() + resolver = ModelRegistryResolver(store) + records = [_record(registry, credential_revision=1)] + (availability,) = resolver.compute_availability( + registry, records, credential_revisions={"zhipu-primary": 2} + ) + assert availability.state == "verification_stale" + + def test_configured_without_any_records(self, store): + registry = _registry() + resolver = ModelRegistryResolver(store) + (availability,) = resolver.compute_availability(registry, []) + assert availability.state == "configured" + assert availability.selectable is False + assert availability.verification.status == "none" + assert availability.effective_capabilities.tools is False + + def test_covers_every_model_of_every_provider(self, store): + registry = _registry(providers=[_zhipu_provider(), _ollama_provider()]) + resolver = ModelRegistryResolver(store) + availabilities = resolver.compute_availability(registry, []) + refs = {availability.model_ref for availability in availabilities} + assert refs == {ZHIPU_REF, OLLAMA_REF} diff --git a/tests/test_snapshots.py b/tests/test_snapshots.py new file mode 100644 index 0000000..70a6111 --- /dev/null +++ b/tests/test_snapshots.py @@ -0,0 +1,567 @@ +"""Tests for the run runtime snapshot service (design doc 5.2, 8.1, 8.2). + +Covers creation with inherit semantics, the selection-hash idempotency +rules, bind/abort state transitions, binding validation on reads, TTLs, +adapter-spec revalidation, and credential resolution against frozen +revisions — with no in-process secret caching. +""" + +from __future__ import annotations + +import hashlib +import json +import time + +import pytest + +from EvoScientist.model_registry.adapters import adapter_specs, find_adapter_spec +from EvoScientist.model_registry.errors import ( + ADAPTER_NOT_SUPPORTED, + MODEL_REGISTRY_NOT_READY, + RUN_CREDENTIAL_REVISION_UNAVAILABLE, + RUN_REQUEST_CONFLICT, + SNAPSHOT_ALREADY_BOUND, + SNAPSHOT_EXPIRED, + SNAPSHOT_NOT_FOUND, + ModelRegistryError, +) +from EvoScientist.model_registry.hashing import configuration_hash +from EvoScientist.model_registry.resolver import ModelRegistryResolver +from EvoScientist.model_registry.schemas import ( + CredentialWrite, + ModelRef, + RegistryV4, +) +from EvoScientist.model_registry.snapshots import ( + BOUND_RETENTION_SECONDS, + PREPARED_TTL_SECONDS, + SnapshotCreateRequest, + SnapshotService, + compute_selection_hash, + config_for_role, + public_snapshot_view, +) +from EvoScientist.model_registry.store import ModelRuntimeStore + +ZHIPU_REF = ModelRef(provider_id="zhipu-glm", model_key="glm-5.2") +OLLAMA_REF = ModelRef(provider_id="local-ollama", model_key="qwen3") +SECRET = "sk-live-9876abcd" + + +def _model_runtime(**overrides): + payload = { + "limit_mode": "combined", + "context_window_tokens": 1048576, + "max_input_tokens": None, + "max_output_tokens": 32768, + "min_effective_input_tokens": 8192, + "fixed_system_reserve_tokens": 4096, + "fixed_tools_reserve_tokens": 8192, + "fixed_attachments_reserve_tokens": 4096, + "limits_status": "confirmed", + "limits_source": "provider", + "temperature": None, + "top_p": None, + "reasoning_effort": "auto", + "declared_capabilities": { + "tools": True, + "vision": False, + "structured_output": True, + }, + } + payload.update(overrides) + return payload + + +def _zhipu_provider(): + return { + "id": "zhipu-glm", + "name": "Zhipu GLM", + "adapter": "openai-compatible", + "base_url": "https://open.bigmodel.cn/api/paas/v4", + "auth": {"mode": "api_key", "credential_id": "zhipu-primary"}, + "enabled": True, + "runtime": { + "timeout_seconds": 120, + "max_retries": 2, + "default_temperature": 0.7, + "default_top_p": 0.95, + "default_reasoning_effort": "auto", + }, + "models": [ + { + "key": "glm-5.2", + "name": "GLM-5.2", + "upstream_model_id": "glm-5.2", + "enabled": True, + "runtime": _model_runtime(), + } + ], + } + + +def _ollama_provider(): + return { + "id": "local-ollama", + "name": "Local Ollama", + "adapter": "ollama", + "base_url": "http://localhost:11434", + "auth": {"mode": "none", "credential_id": None}, + "enabled": True, + "runtime": {"timeout_seconds": 120, "max_retries": 2}, + "models": [ + { + "key": "qwen3", + "name": "Qwen3", + "upstream_model_id": "qwen3", + "enabled": True, + "runtime": _model_runtime(), + } + ], + } + + +def _registry(*, auxiliary_default=True) -> RegistryV4: + return RegistryV4.model_validate( + { + "version": 4, + "revision": 1, + "state": "bootstrap", + "defaults": { + "primary": {"provider_id": "zhipu-glm", "model_key": "glm-5.2"}, + "auxiliary": ( + {"provider_id": "local-ollama", "model_key": "qwen3"} + if auxiliary_default + else None + ), + }, + "providers": [_zhipu_provider(), _ollama_provider()], + } + ) + + +def _verify(store, registry, provider_id, model_key): + provider = registry.find_provider(provider_id) + model = provider.find_model(model_key) + spec = find_adapter_spec(provider.adapter, model.upstream_model_id) + store.record_model_verification( + provider_id=provider_id, + model_key=model_key, + configuration_hash=configuration_hash(provider, model), + credential_revision=1 if provider.auth.credential_id else 0, + adapter_spec_revision=spec.spec_revision, + result="passed", + verified_capabilities={ + "tools": True, + "vision": False, + "structured_output": True, + }, + ) + + +@pytest.fixture +def store(tmp_path): + return ModelRuntimeStore(config_dir=tmp_path) + + +@pytest.fixture +def active_store(store): + registry = store.save_registry( + expected_revision=1, + registry=_registry(), + credential_writes=[ + CredentialWrite(credential_id="zhipu-primary", secret_value=SECRET) + ], + ) + assert registry.state == "active" + _verify(store, registry, "zhipu-glm", "glm-5.2") + _verify(store, registry, "local-ollama", "qwen3") + return store + + +@pytest.fixture +def service(active_store): + return SnapshotService(active_store, ModelRegistryResolver(active_store)) + + +def _request(**overrides): + payload = { + "run_request_id": "req-1", + "thread_id": "thread-1", + "deployment_id": "local", + "model_selection_revision": 0, + "primary": None, + "auxiliary": None, + } + payload.update(overrides) + return SnapshotCreateRequest.model_validate(payload) + + +def _expected_selection_hash(primary, auxiliary): + def entry(ref): + if ref is None: + return None + return {"provider_id": ref.provider_id, "model_key": ref.model_key} + + encoded = json.dumps( + {"primary": entry(primary), "auxiliary": entry(auxiliary)}, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +class TestCreate: + def test_inherit_resolves_registry_defaults(self, service, active_store): + creation = service.create(_request(model_selection_revision=4)) + assert creation.created is True + snapshot = creation.snapshot + assert snapshot.status == "prepared" + assert snapshot.langgraph_run_id is None + payload = snapshot.payload + assert payload.registry_revision == 2 + assert payload.model_selection_revision == 4 + assert payload.primary.model_ref == ZHIPU_REF + assert payload.primary.role == "primary" + assert payload.auxiliary is not None + assert payload.auxiliary.model_ref == OLLAMA_REF + assert payload.auxiliary.role == "auxiliary" + + def test_inherit_without_auxiliary_default(self, active_store): + store = active_store + registry = store.load_registry() + store.save_registry( + expected_revision=registry.revision, + registry=_registry(auxiliary_default=False), + ) + service = SnapshotService(store, ModelRegistryResolver(store)) + creation = service.create(_request()) + assert creation.snapshot.payload.auxiliary is None + + def test_explicit_selection_freezes_both_roles(self, service): + creation = service.create(_request(primary=ZHIPU_REF, auxiliary=OLLAMA_REF)) + payload = creation.snapshot.payload + assert payload.primary.model_ref == ZHIPU_REF + assert payload.auxiliary is not None + assert payload.auxiliary.model_ref == OLLAMA_REF + + def test_bootstrap_registry_is_not_ready(self, store): + service = SnapshotService(store, ModelRegistryResolver(store)) + with pytest.raises(ModelRegistryError) as excinfo: + service.create(_request()) + assert excinfo.value.code == MODEL_REGISTRY_NOT_READY + assert excinfo.value.http_status == 422 + + def test_selection_hash_uses_pre_resolution_semantics(self, service): + creation = service.create(_request()) + snapshot = creation.snapshot + assert snapshot.selection_hash == _expected_selection_hash(None, None) + assert compute_selection_hash(None, None) == _expected_selection_hash( + None, None + ) + # An explicit selection equal to the defaults still hashes differently. + assert compute_selection_hash(ZHIPU_REF, OLLAMA_REF) != snapshot.selection_hash + + def test_selection_hash_ignores_selection_revision(self, service): + first = service.create(_request(model_selection_revision=4)) + second = service.create(_request(model_selection_revision=9)) + assert second.created is False + assert second.snapshot.snapshot_id == first.snapshot.snapshot_id + + def test_idempotent_same_selection_returns_original(self, service): + first = service.create(_request()) + second = service.create(_request()) + assert first.created is True + assert second.created is False + assert second.snapshot.snapshot_id == first.snapshot.snapshot_id + + def test_conflicting_selection_same_triplet(self, service): + service.create(_request()) + with pytest.raises(ModelRegistryError) as excinfo: + service.create(_request(primary=ZHIPU_REF)) + assert excinfo.value.code == RUN_REQUEST_CONFLICT + assert excinfo.value.http_status == 409 + + def test_recreate_after_expired(self, service, active_store): + first = service.create(_request()) + active_store.set_run_snapshot_status(first.snapshot.snapshot_id, "expired") + second = service.create(_request()) + assert second.created is True + assert second.snapshot.snapshot_id != first.snapshot.snapshot_id + + def test_recreate_after_aborted(self, service): + first = service.create(_request()) + service.abort(first.snapshot.snapshot_id) + second = service.create(_request()) + assert second.created is True + assert second.snapshot.snapshot_id != first.snapshot.snapshot_id + + def test_payload_freezes_credential_revision_without_secret(self, service): + creation = service.create(_request()) + payload = creation.snapshot.payload + assert payload.primary.auth_ref.credential_id == "zhipu-primary" + assert payload.primary.auth_ref.credential_revision == 1 + encoded = json.dumps(payload.model_dump(mode="json")) + assert SECRET not in encoded + + def test_prepared_ttl_is_fifteen_minutes(self, service): + creation = service.create(_request()) + snapshot = creation.snapshot + assert PREPARED_TTL_SECONDS == 15 * 60 + delta = snapshot.expires_at - snapshot.created_at + assert PREPARED_TTL_SECONDS - 1 <= delta <= PREPARED_TTL_SECONDS + 1 + + +class TestBind: + def test_bind_once(self, service): + creation = service.create(_request()) + before = int(time.time()) + bound = service.bind(creation.snapshot.snapshot_id, "lg-run-1") + assert bound.status == "bound" + assert bound.langgraph_run_id == "lg-run-1" + assert BOUND_RETENTION_SECONDS == 24 * 60 * 60 + delta = bound.expires_at - before + assert BOUND_RETENTION_SECONDS - 1 <= delta <= BOUND_RETENTION_SECONDS + 1 + + def test_bind_same_run_id_is_idempotent(self, service): + creation = service.create(_request()) + service.bind(creation.snapshot.snapshot_id, "lg-run-1") + again = service.bind(creation.snapshot.snapshot_id, "lg-run-1") + assert again.status == "bound" + assert again.langgraph_run_id == "lg-run-1" + + def test_bind_different_run_id_conflicts(self, service): + creation = service.create(_request()) + service.bind(creation.snapshot.snapshot_id, "lg-run-1") + with pytest.raises(ModelRegistryError) as excinfo: + service.bind(creation.snapshot.snapshot_id, "lg-run-2") + assert excinfo.value.code == SNAPSHOT_ALREADY_BOUND + + def test_bind_expired_snapshot(self, service, active_store): + creation = service.create(_request()) + active_store.set_run_snapshot_status(creation.snapshot.snapshot_id, "expired") + with pytest.raises(ModelRegistryError) as excinfo: + service.bind(creation.snapshot.snapshot_id, "lg-run-1") + assert excinfo.value.code == SNAPSHOT_EXPIRED + + def test_bind_aborted_snapshot(self, service): + creation = service.create(_request()) + service.abort(creation.snapshot.snapshot_id) + with pytest.raises(ModelRegistryError) as excinfo: + service.bind(creation.snapshot.snapshot_id, "lg-run-1") + assert excinfo.value.code == SNAPSHOT_EXPIRED + + def test_bind_unknown_snapshot(self, service): + with pytest.raises(ModelRegistryError) as excinfo: + service.bind("snap-missing", "lg-run-1") + assert excinfo.value.code == SNAPSHOT_NOT_FOUND + assert excinfo.value.http_status == 404 + + +class TestAbort: + def test_abort_prepared(self, service, active_store): + creation = service.create(_request()) + service.abort(creation.snapshot.snapshot_id) + row = active_store.get_run_snapshot(creation.snapshot.snapshot_id) + assert row["status"] == "aborted" + + def test_abort_is_idempotent(self, service): + creation = service.create(_request()) + service.abort(creation.snapshot.snapshot_id) + service.abort(creation.snapshot.snapshot_id) + + def test_abort_bound_conflicts(self, service): + creation = service.create(_request()) + service.bind(creation.snapshot.snapshot_id, "lg-run-1") + with pytest.raises(ModelRegistryError) as excinfo: + service.abort(creation.snapshot.snapshot_id) + assert excinfo.value.code == SNAPSHOT_ALREADY_BOUND + + def test_abort_expired(self, service, active_store): + creation = service.create(_request()) + active_store.set_run_snapshot_status(creation.snapshot.snapshot_id, "expired") + with pytest.raises(ModelRegistryError) as excinfo: + service.abort(creation.snapshot.snapshot_id) + assert excinfo.value.code == SNAPSHOT_EXPIRED + + def test_abort_unknown(self, service): + with pytest.raises(ModelRegistryError) as excinfo: + service.abort("snap-missing") + assert excinfo.value.code == SNAPSHOT_NOT_FOUND + + +class TestGet: + def test_get_returns_snapshot(self, service): + creation = service.create(_request()) + snapshot = service.get( + creation.snapshot.snapshot_id, + deployment_id="local", + thread_id="thread-1", + ) + assert snapshot.snapshot_id == creation.snapshot.snapshot_id + assert snapshot.payload.primary.model_ref == ZHIPU_REF + + def test_get_rejects_foreign_thread(self, service): + creation = service.create(_request()) + with pytest.raises(ModelRegistryError) as excinfo: + service.get( + creation.snapshot.snapshot_id, + deployment_id="local", + thread_id="thread-2", + ) + assert excinfo.value.code == SNAPSHOT_NOT_FOUND + + def test_get_rejects_foreign_deployment(self, service): + creation = service.create(_request()) + with pytest.raises(ModelRegistryError) as excinfo: + service.get( + creation.snapshot.snapshot_id, + deployment_id="other", + thread_id="thread-1", + ) + assert excinfo.value.code == SNAPSHOT_NOT_FOUND + + def test_get_expired_fails(self, service, active_store): + creation = service.create(_request()) + active_store.set_run_snapshot_status(creation.snapshot.snapshot_id, "expired") + with pytest.raises(ModelRegistryError) as excinfo: + service.get( + creation.snapshot.snapshot_id, + deployment_id="local", + thread_id="thread-1", + ) + assert excinfo.value.code == SNAPSHOT_EXPIRED + + def test_get_fails_when_frozen_spec_revision_removed(self, active_store): + service = SnapshotService(active_store, ModelRegistryResolver(active_store)) + creation = service.create(_request()) + # Simulate a contract upgrade that removed spec_revision 1 entirely. + bumped = tuple( + spec.model_copy(update={"spec_revision": 2}) for spec in adapter_specs() + ) + stale_service = SnapshotService( + active_store, ModelRegistryResolver(active_store, specs=bumped) + ) + with pytest.raises(ModelRegistryError) as excinfo: + stale_service.get( + creation.snapshot.snapshot_id, + deployment_id="local", + thread_id="thread-1", + ) + assert excinfo.value.code == ADAPTER_NOT_SUPPORTED + + +class TestCleanupExpired: + def _insert(self, store, snapshot_id, *, status, expires_at): + store.insert_run_snapshot( + snapshot_id=snapshot_id, + deployment_id="local", + thread_id="thread-1", + run_request_id=f"req-{snapshot_id}", + selection_hash="b" * 64, + payload={}, + expires_at=expires_at, + status=status, + langgraph_run_id="lg-run" if status == "bound" else None, + ) + + def test_cleanup_marks_due_snapshots_expired(self, store): + service = SnapshotService(store, ModelRegistryResolver(store)) + now = 1_000_000 + self._insert(store, "snap-p-old", status="prepared", expires_at=now - 1) + self._insert(store, "snap-b-old", status="bound", expires_at=now - 1) + self._insert(store, "snap-p-new", status="prepared", expires_at=now + 900) + expired = service.cleanup_expired(now) + assert sorted(expired) == ["snap-b-old", "snap-p-old"] + assert store.get_run_snapshot("snap-p-old")["status"] == "expired" + assert store.get_run_snapshot("snap-b-old")["status"] == "expired" + assert store.get_run_snapshot("snap-p-new")["status"] == "prepared" + + +class TestRoleMapping: + def test_auxiliary_roles_fall_back_to_primary(self, active_store): + store = active_store + registry = store.load_registry() + store.save_registry( + expected_revision=registry.revision, + registry=_registry(auxiliary_default=False), + ) + service = SnapshotService(store, ModelRegistryResolver(store)) + snapshot = service.create(_request()).snapshot + for role in ("primary", "auxiliary", "summary", "tool_selector"): + assert config_for_role(snapshot, role).model_ref == ZHIPU_REF + + def test_auxiliary_roles_use_frozen_auxiliary(self, service): + snapshot = service.create(_request()).snapshot + assert config_for_role(snapshot, "primary").model_ref == ZHIPU_REF + for role in ("auxiliary", "summary", "tool_selector"): + config = config_for_role(snapshot, role) + assert config.model_ref == OLLAMA_REF + assert config.role == "auxiliary" + + def test_unknown_role_rejected(self, service): + snapshot = service.create(_request()).snapshot + with pytest.raises(ValueError, match="role"): + config_for_role(snapshot, "planner") + + +class TestSnapshotCredentials: + def test_resolves_frozen_revision(self, service): + snapshot = service.create(_request()).snapshot + assert service.resolve_snapshot_credential(snapshot, "primary") == SECRET + + def test_rotation_keeps_frozen_revision(self, service, active_store): + snapshot = service.create(_request()).snapshot + active_store.write_credential_version("zhipu-primary", "sk-rotated-2222") + assert service.resolve_snapshot_credential(snapshot, "primary") == SECRET + + def test_destroyed_revision_fails_loudly(self, service, active_store): + snapshot = service.create(_request()).snapshot + active_store.retire_credential_version("zhipu-primary", 1) + with pytest.raises(ModelRegistryError) as excinfo: + service.resolve_snapshot_credential(snapshot, "primary") + assert excinfo.value.code == RUN_CREDENTIAL_REVISION_UNAVAILABLE + + def test_no_in_process_secret_cache(self, service, active_store): + snapshot = service.create(_request()).snapshot + # A brand-new store/service pair on the same database still resolves. + fresh_store = ModelRuntimeStore(config_dir=active_store.config_dir) + fresh_service = SnapshotService(fresh_store, ModelRegistryResolver(fresh_store)) + assert fresh_service.resolve_snapshot_credential(snapshot, "primary") == SECRET + + def test_auxiliary_role_resolves_auxiliary_credential(self, service): + snapshot = service.create(_request()).snapshot + # The frozen auxiliary is the mode=none ollama model: no secret. + assert service.resolve_snapshot_credential(snapshot, "summary") == "" + assert service.resolve_snapshot_credential(snapshot, "tool_selector") == "" + + def test_mode_none_returns_empty_string(self, service): + snapshot = service.create(_request(primary=OLLAMA_REF)).snapshot + assert service.resolve_snapshot_credential(snapshot, "primary") == "" + + +class TestPublicView: + def test_matches_section_8_2_shape(self, service): + snapshot = service.create(_request()).snapshot + view = public_snapshot_view(snapshot) + assert view["snapshot_id"] == snapshot.snapshot_id + assert view["registry_revision"] == 2 + primary = view["primary"] + assert primary["provider_id"] == "zhipu-glm" + assert primary["model_key"] == "glm-5.2" + assert primary["adapter_spec_revision"] == 1 + assert primary["runtime"] == { + "max_output_tokens": 32768, + "temperature": 0.7, + "top_p": 0.95, + "timeout_seconds": 120, + "max_retries": 2, + } + assert view["auxiliary"]["provider_id"] == "local-ollama" + + def test_never_contains_secret_or_base_url(self, service): + snapshot = service.create(_request()).snapshot + encoded = json.dumps(public_snapshot_view(snapshot)) + assert SECRET not in encoded + assert "open.bigmodel.cn" not in encoded + assert "localhost" not in encoded