ouyangbo/EvoScientist
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 3 Wiki Activity

feat(model-registry): add ModelRegistryResolver and run snapshot service

Browse Source 
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.
This commit is contained in:
m4
2026-07-21 08:33:12 +08:00
parent b2e28249fd
commit b1233d42dc
8 changed files with 2121 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
EvoScientist/model_registry/__init__.py
+24
View File
@@ -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",
]
EvoScientist/model_registry/errors.py
+2
View File
@@ -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,
EvoScientist/model_registry/resolver.py
+438
View File
@@ -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")
EvoScientist/model_registry/snapshots.py
+413
View File
@@ -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."
)
EvoScientist/model_registry/store.py
+156 -12
View File
@@ -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()
tests/test_model_registry_schemas.py
+1
View File
@@ -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,
tests/test_resolver.py
+520
View File
@@ -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}
tests/test_snapshots.py
+567
View File
@@ -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