m4
bb9bed82e1
ModelRole collapses to "primary": every role (main, tool selector, memory
agents, subagents, summarizer) resolves to the snapshot's frozen primary
model, per design 6.1/8.3 — users typically configure a single usable LLM,
so compile-time auxiliary bindings were bypassing run snapshots and
mis-attributing usage. Legacy auxiliary keys in stored snapshots, registry
JSON, and thread metadata are tolerated on read and dropped.
BREAKING CHANGE: ThreadModelSelection no longer carries an auxiliary ref;
snapshot selection_hash is computed over {primary, reasoning_effort} only;
ConfigurableModelMiddleware(role="auxiliary") is rejected.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
218 lines
8.2 KiB
Python
218 lines
8.2 KiB
Python
"""Tests for ``EvoScientist.model_registry.runtime`` (design doc 8.3).
|
|
|
|
Covers the runtime-link glue: registry-default primary resolution, snapshot
|
|
role construction with per-call credential resolution, safe-client wiring
|
|
(ollama carries retries on the transports), the shared default-runtime
|
|
accessor, and the tolerant local platform field reader.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from langchain_ollama import ChatOllama
|
|
from langchain_openai import ChatOpenAI
|
|
|
|
from EvoScientist.model_registry.endpoint_policy import EndpointPolicy
|
|
from EvoScientist.model_registry.errors import (
|
|
MODEL_REGISTRY_NOT_READY,
|
|
ModelRegistryError,
|
|
)
|
|
from EvoScientist.model_registry.platform import (
|
|
DEFAULT_LOCAL_DEPLOYMENT_ID,
|
|
PlatformConfigError,
|
|
)
|
|
from EvoScientist.model_registry.runtime import (
|
|
SnapshotRuntime,
|
|
_read_local_platform_fields,
|
|
get_snapshot_runtime,
|
|
set_snapshot_runtime_for_tests,
|
|
)
|
|
from EvoScientist.model_registry.schemas import DevelopmentEndpoint
|
|
from EvoScientist.model_registry.store import ModelRuntimeStore
|
|
from tests.registry_fixtures import (
|
|
OLLAMA_REF,
|
|
ZHIPU_REF,
|
|
ZHIPU_SECRET,
|
|
make_active_store,
|
|
make_snapshot,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path):
|
|
return make_active_store(tmp_path / "model-runtime")
|
|
|
|
|
|
@pytest.fixture
|
|
def runtime(store):
|
|
return SnapshotRuntime(store)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_default_runtime():
|
|
set_snapshot_runtime_for_tests(None)
|
|
yield
|
|
set_snapshot_runtime_for_tests(None)
|
|
|
|
|
|
class TestRegistryDefaults:
|
|
def test_bootstrap_registry_is_not_ready(self, tmp_path):
|
|
runtime = SnapshotRuntime(ModelRuntimeStore(config_dir=tmp_path / "db"))
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
runtime.registry_default()
|
|
assert excinfo.value.code == MODEL_REGISTRY_NOT_READY
|
|
assert excinfo.value.http_status == 422
|
|
|
|
def test_active_registry_returns_default(self, runtime):
|
|
primary, revision = runtime.registry_default()
|
|
assert primary == ZHIPU_REF
|
|
assert revision == 2
|
|
|
|
|
|
class TestResolveDefaultConfig:
|
|
def test_resolves_primary_default(self, runtime):
|
|
config = runtime.resolve_default_config()
|
|
assert config.model_ref == ZHIPU_REF
|
|
assert config.role == "primary"
|
|
|
|
|
|
class TestBuildDefaultRoleModel:
|
|
def test_primary_builds_chat_openai_with_frozen_options(self, runtime):
|
|
model = runtime.build_default_role_model("primary")
|
|
assert isinstance(model, ChatOpenAI)
|
|
assert model.model_name == "glm-5.2"
|
|
assert model.openai_api_base == "https://open.bigmodel.cn/api/paas/v4"
|
|
assert model.max_retries == 2
|
|
assert model.temperature == 0.7
|
|
assert model.top_p == 0.95
|
|
|
|
def test_openai_compatible_never_reads_env_api_key(self, runtime, monkeypatch):
|
|
"""The credential comes from the frozen auth_ref, never OPENAI_API_KEY."""
|
|
monkeypatch.setenv("OPENAI_API_KEY", "sk-env-must-not-leak")
|
|
model = runtime.build_default_role_model("primary")
|
|
assert model.openai_api_key.get_secret_value() == ZHIPU_SECRET
|
|
|
|
|
|
class TestBuildRoleModelFromSnapshot:
|
|
def test_primary_role_uses_snapshot_primary(self, store, runtime):
|
|
snapshot = make_snapshot(store)
|
|
model = runtime.build_role_model(snapshot, "primary")
|
|
assert isinstance(model, ChatOpenAI)
|
|
assert model.model_name == "glm-5.2"
|
|
assert model.openai_api_key.get_secret_value() == ZHIPU_SECRET
|
|
|
|
|
|
|
|
class TestSharedDefaultRuntime:
|
|
def test_set_and_get_runtime(self, runtime):
|
|
set_snapshot_runtime_for_tests(runtime)
|
|
assert get_snapshot_runtime() is runtime
|
|
|
|
def test_builds_from_missing_config_yaml(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(
|
|
"EvoScientist.model_registry.runtime.get_config_path",
|
|
lambda: tmp_path / "config.yaml",
|
|
)
|
|
monkeypatch.setattr(
|
|
"EvoScientist.model_registry.store.DEFAULT_CONFIG_DIR",
|
|
tmp_path / "config-dir",
|
|
)
|
|
runtime = get_snapshot_runtime()
|
|
assert runtime.local_deployment_id == DEFAULT_LOCAL_DEPLOYMENT_ID
|
|
assert get_snapshot_runtime() is runtime
|
|
|
|
|
|
class TestLocalPlatformFields:
|
|
def _read(self, tmp_path, monkeypatch, text: str | None):
|
|
config_path = tmp_path / "config.yaml"
|
|
if text is not None:
|
|
config_path.write_text(text, encoding="utf-8")
|
|
monkeypatch.setattr(
|
|
"EvoScientist.model_registry.runtime.get_config_path", lambda: config_path
|
|
)
|
|
return _read_local_platform_fields()
|
|
|
|
def test_missing_config_uses_defaults(self, tmp_path, monkeypatch):
|
|
deployment_id, db_path, endpoints, webui_ids = self._read(
|
|
tmp_path, monkeypatch, None
|
|
)
|
|
assert deployment_id == DEFAULT_LOCAL_DEPLOYMENT_ID
|
|
assert db_path is None
|
|
assert endpoints == ()
|
|
assert webui_ids == ()
|
|
|
|
def test_reads_runtime_fields(self, tmp_path, monkeypatch):
|
|
deployment_id, db_path, endpoints, webui_ids = self._read(
|
|
tmp_path,
|
|
monkeypatch,
|
|
"local_deployment_id: dev-deploy\n"
|
|
"model_runtime_db: /tmp/mr.sqlite3\n"
|
|
"development_endpoints:\n"
|
|
" - {id: ollama, url: 'http://localhost:11434', label: Local}\n"
|
|
"webui_delegation_public_keys:\n"
|
|
" - {deployment_id: webui-local, public_key: '-----BEGIN PUBLIC KEY-----\\nMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE\\n-----END PUBLIC KEY-----\\n'}\n",
|
|
)
|
|
assert deployment_id == "dev-deploy"
|
|
assert db_path == __import__("pathlib").Path("/tmp/mr.sqlite3")
|
|
assert endpoints == (
|
|
DevelopmentEndpoint(
|
|
id="ollama", url="http://localhost:11434", label="Local"
|
|
),
|
|
)
|
|
assert webui_ids == ("webui-local",)
|
|
|
|
def test_invalid_yaml_raises(self, tmp_path, monkeypatch):
|
|
with pytest.raises(PlatformConfigError):
|
|
self._read(tmp_path, monkeypatch, "local_deployment_id: [unclosed")
|
|
|
|
def test_invalid_field_type_raises(self, tmp_path, monkeypatch):
|
|
with pytest.raises(PlatformConfigError):
|
|
self._read(tmp_path, monkeypatch, "local_deployment_id: 42")
|
|
|
|
|
|
class TestEndpointPolicyWiring:
|
|
def test_development_endpoints_allow_loopback_ollama(self, store):
|
|
policy = EndpointPolicy(
|
|
(
|
|
DevelopmentEndpoint(
|
|
id="ollama", url="http://localhost:11434", label="Local"
|
|
),
|
|
)
|
|
)
|
|
runtime = SnapshotRuntime(store, endpoint_policy=policy)
|
|
# The loopback ollama provider builds only because its base URL is a
|
|
# registered development endpoint.
|
|
config = runtime.resolver.resolve(OLLAMA_REF, "primary")
|
|
model = runtime._build(config, "")
|
|
assert isinstance(model, ChatOllama)
|
|
|
|
|
|
class TestCreateLocalSnapshot:
|
|
"""Section 8.1 local entry: fixed binding convention through SnapshotService."""
|
|
|
|
def test_bootstrap_registry_fails_closed(self, tmp_path):
|
|
runtime = SnapshotRuntime(ModelRuntimeStore(config_dir=tmp_path / "db"))
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
runtime.create_local_snapshot("thread-1")
|
|
assert excinfo.value.code == MODEL_REGISTRY_NOT_READY
|
|
|
|
def test_binding_convention_is_fixed(self, store, runtime):
|
|
from EvoScientist.model_registry.snapshots import config_for_role
|
|
|
|
snapshot = runtime.create_local_snapshot("cli-thread-1")
|
|
assert snapshot.thread_id == "cli-thread-1"
|
|
assert snapshot.deployment_id == runtime.local_deployment_id
|
|
assert snapshot.payload.model_selection_revision == 0
|
|
# primary=None inherits — the registry default is frozen at creation.
|
|
assert config_for_role(snapshot, "primary").model_ref == ZHIPU_REF
|
|
|
|
def test_same_run_request_id_reuses_snapshot(self, store, runtime):
|
|
first = runtime.create_local_snapshot("t-1", run_request_id="req-1")
|
|
second = runtime.create_local_snapshot("t-1", run_request_id="req-1")
|
|
assert second.snapshot_id == first.snapshot_id
|
|
|
|
def test_default_run_request_id_freezes_per_call(self, store, runtime):
|
|
first = runtime.create_local_snapshot("t-1")
|
|
second = runtime.create_local_snapshot("t-1")
|
|
assert second.snapshot_id != first.snapshot_id
|