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>
549 lines
20 KiB
Python
549 lines
20 KiB
Python
"""Tests for ``EvoScientist.middleware.configurable_model`` (design doc 8.3).
|
|
|
|
Verifies the snapshot-only contract: the middleware resolves the frozen
|
|
``ResolvedModelConfig`` for its role from ``configurable["runtime_snapshot_id"]``
|
|
and builds the model via ``build_chat_model``; ``model``/``model_provider``
|
|
in ``configurable`` are rejected with ``MODEL_CONFIG_OUTSIDE_SNAPSHOT``; a
|
|
run without a snapshot ID gets a lazily created local snapshot bound to its
|
|
thread (section 8.1), or fails closed when no thread ID exists either.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import contextmanager
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from langchain_ollama import ChatOllama
|
|
from langchain_openai import ChatOpenAI
|
|
|
|
from EvoScientist.middleware.configurable_model import (
|
|
ConfigurableModelMiddleware,
|
|
check_no_outside_snapshot_model_config,
|
|
ensure_snapshot_binding,
|
|
read_snapshot_binding,
|
|
)
|
|
from EvoScientist.model_registry.errors import (
|
|
MODEL_CONFIG_OUTSIDE_SNAPSHOT,
|
|
MODEL_REGISTRY_NOT_READY,
|
|
SNAPSHOT_EXPIRED,
|
|
SNAPSHOT_NOT_FOUND,
|
|
ModelRegistryError,
|
|
)
|
|
from EvoScientist.model_registry.runtime import SnapshotRuntime
|
|
from EvoScientist.model_registry.store import ModelRuntimeStore
|
|
from tests.registry_fixtures import (
|
|
ZHIPU_SECRET,
|
|
make_active_store,
|
|
make_snapshot,
|
|
)
|
|
|
|
|
|
@contextmanager
|
|
def _patched_config(configurable: dict | object | None):
|
|
"""Patch ``langgraph.config.get_config`` to return a controlled value.
|
|
|
|
Pass ``None`` to simulate "outside a runnable context" (raises
|
|
``RuntimeError`` like the real ``get_config()`` does).
|
|
Pass a dict for ``configurable`` to expose just that key.
|
|
Pass any other object to simulate a malformed config.
|
|
"""
|
|
import langgraph.config as _lg_cfg
|
|
|
|
if configurable is None:
|
|
# Simulate get_config raising outside a runnable context.
|
|
with patch.object(
|
|
_lg_cfg,
|
|
"get_config",
|
|
side_effect=RuntimeError("Called get_config outside of a runnable context"),
|
|
):
|
|
yield
|
|
elif isinstance(configurable, dict):
|
|
with patch.object(
|
|
_lg_cfg,
|
|
"get_config",
|
|
return_value={"configurable": configurable},
|
|
):
|
|
yield
|
|
else:
|
|
with patch.object(
|
|
_lg_cfg,
|
|
"get_config",
|
|
return_value=configurable,
|
|
):
|
|
yield
|
|
|
|
|
|
def _make_request():
|
|
"""Build a minimal ``ModelRequest`` stub.
|
|
|
|
``request.override(model=...)`` returns a new request whose ``model``
|
|
field reflects the override.
|
|
"""
|
|
req = MagicMock()
|
|
|
|
def _override(**kwargs):
|
|
new = MagicMock()
|
|
new.model = kwargs.get("model", req.model)
|
|
return new
|
|
|
|
req.override = MagicMock(side_effect=_override)
|
|
return req
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path):
|
|
return make_active_store(tmp_path / "model-runtime")
|
|
|
|
|
|
@pytest.fixture
|
|
def runtime(store):
|
|
return SnapshotRuntime(store)
|
|
|
|
|
|
def _configurable_for(snapshot, **overrides):
|
|
configurable = {
|
|
"runtime_snapshot_id": snapshot.snapshot_id,
|
|
"thread_id": snapshot.thread_id,
|
|
}
|
|
configurable.update(overrides)
|
|
return configurable
|
|
|
|
|
|
# =============================================================================
|
|
# 1. Outside-snapshot model configuration is rejected (section 8.2)
|
|
# =============================================================================
|
|
|
|
|
|
class TestOutsideSnapshotConfig:
|
|
@pytest.mark.parametrize("key", ["model", "model_provider"])
|
|
def test_model_keys_rejected(self, key):
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
check_no_outside_snapshot_model_config({key: "gpt-5"})
|
|
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
|
|
assert excinfo.value.http_status == 422
|
|
assert excinfo.value.details[0].path == key
|
|
|
|
def test_rejected_alongside_snapshot_id(self, store, runtime):
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
snapshot = make_snapshot(store)
|
|
req = _make_request()
|
|
with (
|
|
_patched_config(_configurable_for(snapshot, model="gpt-5")),
|
|
pytest.raises(ModelRegistryError) as excinfo,
|
|
):
|
|
mw.wrap_model_call(req, MagicMock())
|
|
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
|
|
|
|
def test_rejected_without_snapshot_id(self, runtime):
|
|
"""The check applies even on the pass-through path (验收 13)."""
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
req = _make_request()
|
|
handler = MagicMock()
|
|
with (
|
|
_patched_config({"model_provider": "openai"}),
|
|
pytest.raises(ModelRegistryError) as excinfo,
|
|
):
|
|
mw.wrap_model_call(req, handler)
|
|
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
|
|
handler.assert_not_called()
|
|
|
|
def test_none_values_treated_as_absent(self):
|
|
check_no_outside_snapshot_model_config({"model": None, "thread_id": "t"})
|
|
|
|
|
|
# =============================================================================
|
|
# 2. Snapshot binding parsing
|
|
# =============================================================================
|
|
|
|
|
|
class TestReadSnapshotBinding:
|
|
def test_full_binding(self):
|
|
# ``workspace_deployment_id`` names the workspace-isolation scope and
|
|
# must NOT be picked up as the snapshot issuer (section 8.2).
|
|
binding = read_snapshot_binding(
|
|
{
|
|
"runtime_snapshot_id": "snap-1",
|
|
"workspace_deployment_id": "deploy-1",
|
|
"thread_id": "thread-1",
|
|
}
|
|
)
|
|
assert binding == ("snap-1", "thread-1")
|
|
|
|
def test_missing_snapshot_id_returns_none(self):
|
|
assert read_snapshot_binding({}) is None
|
|
assert read_snapshot_binding({"runtime_snapshot_id": ""}) is None
|
|
assert read_snapshot_binding({"runtime_snapshot_id": 42}) is None
|
|
|
|
def test_missing_thread_falls_back(self):
|
|
assert read_snapshot_binding({"runtime_snapshot_id": "snap-1"}) == (
|
|
"snap-1",
|
|
None,
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# 3. Fail-closed behavior (no snapshot in configurable, section 8.1)
|
|
# =============================================================================
|
|
|
|
|
|
class TestFailClosed:
|
|
def test_sync_no_snapshot_no_thread_rejected(self, runtime):
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
req = _make_request()
|
|
handler = MagicMock()
|
|
with (
|
|
_patched_config({}),
|
|
pytest.raises(ModelRegistryError) as excinfo,
|
|
):
|
|
mw.wrap_model_call(req, handler)
|
|
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
|
|
handler.assert_not_called()
|
|
req.override.assert_not_called()
|
|
|
|
async def test_async_no_snapshot_no_thread_rejected(self, runtime):
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
req = _make_request()
|
|
|
|
async def handler(r):
|
|
raise AssertionError("handler must not run") # pragma: no cover
|
|
|
|
with (
|
|
_patched_config({}),
|
|
pytest.raises(ModelRegistryError) as excinfo,
|
|
):
|
|
await mw.awrap_model_call(req, handler)
|
|
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
|
|
req.override.assert_not_called()
|
|
|
|
def test_outside_runnable_context_fails_closed(self, runtime):
|
|
"""No runnable context means no thread binding — never pass through."""
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
req = _make_request()
|
|
handler = MagicMock()
|
|
with (
|
|
_patched_config(None),
|
|
pytest.raises(ModelRegistryError) as excinfo,
|
|
):
|
|
mw.wrap_model_call(req, handler)
|
|
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
|
|
handler.assert_not_called()
|
|
|
|
def test_bootstrap_registry_fails_closed_not_ready(self, tmp_path):
|
|
bootstrap_runtime = SnapshotRuntime(
|
|
ModelRuntimeStore(config_dir=tmp_path / "model-runtime")
|
|
)
|
|
mw = ConfigurableModelMiddleware(runtime=bootstrap_runtime)
|
|
req = _make_request()
|
|
with (
|
|
_patched_config({"thread_id": "cli-thread"}),
|
|
pytest.raises(ModelRegistryError) as excinfo,
|
|
):
|
|
mw.wrap_model_call(req, MagicMock())
|
|
assert excinfo.value.code == MODEL_REGISTRY_NOT_READY
|
|
|
|
|
|
class TestLazyLocalSnapshot:
|
|
def test_thread_without_snapshot_lazily_creates_one(self, store, runtime):
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
req = _make_request()
|
|
captured = {}
|
|
|
|
def handler(r):
|
|
captured["model"] = r.model
|
|
return "ok"
|
|
|
|
with _patched_config({"thread_id": "cron-thread-1"}):
|
|
assert mw.wrap_model_call(req, handler) == "ok"
|
|
# The override carries the registry-default primary model.
|
|
assert captured["model"] is not req.model
|
|
# A snapshot was created and bound to the run's own thread.
|
|
found = store.find_active_run_snapshot(
|
|
deployment_id=runtime.local_deployment_id,
|
|
thread_id="cron-thread-1",
|
|
run_request_id="auto:cron-thread-1",
|
|
)
|
|
assert found is not None
|
|
|
|
def test_lazy_creation_converges_within_a_thread(self, store, runtime):
|
|
"""Every middleware in the run lands on the same lazily created snapshot."""
|
|
first = ensure_snapshot_binding({"thread_id": "t-1"}, runtime)
|
|
second = ensure_snapshot_binding({"thread_id": "t-1"}, runtime)
|
|
assert first == second
|
|
assert first[1] == "t-1"
|
|
|
|
def test_explicit_snapshot_wins_over_lazy_creation(self, store, runtime):
|
|
snapshot = make_snapshot(store)
|
|
binding = ensure_snapshot_binding(
|
|
{"runtime_snapshot_id": snapshot.snapshot_id, "thread_id": "other"},
|
|
runtime,
|
|
)
|
|
assert binding[0] == snapshot.snapshot_id
|
|
|
|
def test_webui_snapshot_survives_scope_deployment_mismatch(self, store):
|
|
"""BFF regression: the run's ``workspace_deployment_id`` names the
|
|
workspace-isolation scope, not the snapshot issuer. A snapshot issued
|
|
by a registered WebUI deployment must resolve even when the two ids
|
|
differ (the mismatch previously failed with SNAPSHOT_NOT_FOUND)."""
|
|
runtime = SnapshotRuntime(
|
|
store, allowed_deployment_ids=("local", "webui-local")
|
|
)
|
|
snapshot = make_snapshot(store, deployment_id="webui-local")
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
req = _make_request()
|
|
handler = MagicMock(return_value="response")
|
|
configurable = _configurable_for(
|
|
snapshot, workspace_deployment_id="61d1b61b-078b-4bda-84c3-6c68a4afcd38"
|
|
)
|
|
with _patched_config(configurable):
|
|
assert mw.wrap_model_call(req, handler) == "response"
|
|
assert handler.call_args[0][0].model is not req.model
|
|
|
|
def test_unregistered_deployment_snapshot_rejected(self, store, runtime):
|
|
"""A snapshot from an unregistered deployment fails closed."""
|
|
snapshot = make_snapshot(store, deployment_id="rogue-webui")
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
req = _make_request()
|
|
with (
|
|
_patched_config(_configurable_for(snapshot)),
|
|
pytest.raises(ModelRegistryError) as excinfo,
|
|
):
|
|
mw.wrap_model_call(req, MagicMock())
|
|
assert excinfo.value.code == SNAPSHOT_NOT_FOUND
|
|
|
|
|
|
# =============================================================================
|
|
# 4. Snapshot-driven model construction (full chain)
|
|
# =============================================================================
|
|
|
|
|
|
class TestSnapshotDrivenConstruction:
|
|
def test_sync_builds_snapshot_primary_model(self, store, runtime):
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
snapshot = make_snapshot(store)
|
|
req = _make_request()
|
|
handler = MagicMock(return_value="response")
|
|
|
|
with _patched_config(_configurable_for(snapshot)):
|
|
mw.wrap_model_call(req, handler)
|
|
|
|
called_with = handler.call_args[0][0]
|
|
assert called_with is not req
|
|
model = called_with.model
|
|
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.openai_api_key.get_secret_value() == ZHIPU_SECRET
|
|
assert model.max_retries == 2
|
|
|
|
async def test_async_path_parity(self, store, runtime):
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
snapshot = make_snapshot(store)
|
|
req = _make_request()
|
|
|
|
async def handler(r):
|
|
return r.model
|
|
|
|
with _patched_config(_configurable_for(snapshot)):
|
|
model = await mw.awrap_model_call(req, handler)
|
|
assert isinstance(model, ChatOpenAI)
|
|
assert model.model_name == "glm-5.2"
|
|
|
|
def test_unknown_role_rejected(self, runtime):
|
|
with pytest.raises(ValueError, match="role"):
|
|
ConfigurableModelMiddleware(role="auxiliary", runtime=runtime)
|
|
|
|
def test_deployment_id_from_configurable_is_verified(self, store, runtime):
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
snapshot = make_snapshot(store)
|
|
req = _make_request()
|
|
handler = MagicMock(return_value="ok")
|
|
|
|
with _patched_config(
|
|
_configurable_for(snapshot, workspace_deployment_id="local")
|
|
):
|
|
mw.wrap_model_call(req, handler)
|
|
handler.assert_called_once()
|
|
|
|
def test_wrong_thread_binding_fails_closed(self, store, runtime):
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
snapshot = make_snapshot(store)
|
|
req = _make_request()
|
|
|
|
with (
|
|
_patched_config(_configurable_for(snapshot, thread_id="other-thread")),
|
|
pytest.raises(ModelRegistryError) as excinfo,
|
|
):
|
|
mw.wrap_model_call(req, MagicMock())
|
|
assert excinfo.value.code == SNAPSHOT_NOT_FOUND
|
|
|
|
def test_scope_deployment_id_is_not_the_snapshot_issuer(self, store, runtime):
|
|
"""``workspace_deployment_id`` names the workspace-isolation scope, so
|
|
any value — matching or not — is irrelevant to the snapshot binding;
|
|
the issuer check is the registered-deployment membership (see
|
|
test_unregistered_deployment_snapshot_rejected)."""
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
snapshot = make_snapshot(store)
|
|
req = _make_request()
|
|
handler = MagicMock(return_value="response")
|
|
|
|
with _patched_config(
|
|
_configurable_for(snapshot, workspace_deployment_id="other-deploy")
|
|
):
|
|
assert mw.wrap_model_call(req, handler) == "response"
|
|
|
|
def test_expired_snapshot_fails_loudly(self, store, runtime):
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
snapshot = make_snapshot(store)
|
|
store.abort_run_snapshot(snapshot.snapshot_id)
|
|
req = _make_request()
|
|
|
|
with (
|
|
_patched_config(_configurable_for(snapshot)),
|
|
pytest.raises(ModelRegistryError) as excinfo,
|
|
):
|
|
mw.wrap_model_call(req, MagicMock())
|
|
assert excinfo.value.code == SNAPSHOT_EXPIRED
|
|
|
|
|
|
# =============================================================================
|
|
# 5. Frozen credential revision resolution
|
|
# =============================================================================
|
|
|
|
|
|
class TestCredentialRevision:
|
|
def test_snapshot_keeps_its_frozen_credential_revision(self, store, runtime):
|
|
"""Rotating the credential after freezing must not affect the run."""
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
snapshot = make_snapshot(store)
|
|
store.write_credential_version("zhipu-primary", "sk-rotated-0000")
|
|
req = _make_request()
|
|
handler = MagicMock(return_value="ok")
|
|
|
|
with _patched_config(_configurable_for(snapshot)):
|
|
mw.wrap_model_call(req, handler)
|
|
|
|
model = handler.call_args[0][0].model
|
|
assert model.openai_api_key.get_secret_value() == ZHIPU_SECRET
|
|
|
|
|
|
# =============================================================================
|
|
# 6. Per-instance caching
|
|
# =============================================================================
|
|
|
|
|
|
class TestCache:
|
|
def test_second_call_hits_cache(self, store, runtime):
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
snapshot = make_snapshot(store)
|
|
req = _make_request()
|
|
handler = MagicMock(return_value="ok")
|
|
|
|
with (
|
|
_patched_config(_configurable_for(snapshot)),
|
|
patch.object(
|
|
SnapshotRuntime, "build_role_model", wraps=runtime.build_role_model
|
|
) as build,
|
|
):
|
|
mw.wrap_model_call(req, handler)
|
|
mw.wrap_model_call(req, handler)
|
|
|
|
assert build.call_count == 1
|
|
|
|
def test_independent_instances_have_independent_caches(self, store, runtime):
|
|
snapshot = make_snapshot(store)
|
|
req = _make_request()
|
|
handler = MagicMock(return_value="ok")
|
|
|
|
with (
|
|
_patched_config(_configurable_for(snapshot)),
|
|
patch.object(
|
|
SnapshotRuntime, "build_role_model", wraps=runtime.build_role_model
|
|
) as build,
|
|
):
|
|
ConfigurableModelMiddleware(runtime=runtime).wrap_model_call(req, handler)
|
|
ConfigurableModelMiddleware(runtime=runtime).wrap_model_call(req, handler)
|
|
|
|
assert build.call_count == 2
|
|
|
|
|
|
# =============================================================================
|
|
# 7. Constructor validation
|
|
# =============================================================================
|
|
|
|
|
|
class TestRoleValidation:
|
|
def test_unknown_role_rejected(self):
|
|
with pytest.raises(ValueError, match="Unknown model role"):
|
|
ConfigurableModelMiddleware(role="bogus")
|
|
|
|
|
|
# =============================================================================
|
|
# 8. Fallback removal startup assertion (design doc 8.3)
|
|
# =============================================================================
|
|
|
|
|
|
class TestFallbackChainRemoved:
|
|
def test_fallback_middleware_gone_from_package(self):
|
|
import importlib
|
|
|
|
import EvoScientist.middleware as mw
|
|
|
|
assert not hasattr(mw, "ModelFallbackMiddleware")
|
|
assert not hasattr(mw, "load_fallback_chain")
|
|
with pytest.raises(ModuleNotFoundError):
|
|
importlib.import_module("EvoScientist.middleware.model_fallback")
|
|
|
|
|
|
# =============================================================================
|
|
# 9. Integration — real langgraph contextvar (no get_config mock)
|
|
# =============================================================================
|
|
|
|
|
|
class TestRunnableContextVarIntegration:
|
|
"""Set the actual ``var_child_runnable_config`` contextvar that LangGraph
|
|
populates per node, then verify the middleware reads through to it.
|
|
|
|
This catches breakage of the ``langgraph.config.get_config()`` contract
|
|
that pure-mock tests would miss (e.g. if get_config is moved to a
|
|
different module, or the contextvar mechanism changes).
|
|
"""
|
|
|
|
def test_real_contextvar_drives_snapshot_resolution(self, store, runtime):
|
|
from langchain_core.runnables.config import var_child_runnable_config
|
|
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
snapshot = make_snapshot(store)
|
|
req = _make_request()
|
|
handler = MagicMock(return_value="ok")
|
|
|
|
token = var_child_runnable_config.set(
|
|
{"configurable": _configurable_for(snapshot)}
|
|
)
|
|
try:
|
|
mw.wrap_model_call(req, handler)
|
|
finally:
|
|
var_child_runnable_config.reset(token)
|
|
|
|
model = handler.call_args[0][0].model
|
|
assert isinstance(model, ChatOpenAI)
|
|
assert model.model_name == "glm-5.2"
|
|
|
|
def test_real_contextvar_unset_fails_closed(self, runtime):
|
|
from langchain_core.runnables.config import var_child_runnable_config
|
|
|
|
# Defensive: ensure no leftover contextvar from another test.
|
|
token = var_child_runnable_config.set(None)
|
|
try:
|
|
mw = ConfigurableModelMiddleware(runtime=runtime)
|
|
req = _make_request()
|
|
handler = MagicMock(return_value="ok")
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
mw.wrap_model_call(req, handler)
|
|
finally:
|
|
var_child_runnable_config.reset(token)
|
|
|
|
assert excinfo.value.code == MODEL_CONFIG_OUTSIDE_SNAPSHOT
|
|
handler.assert_not_called()
|
|
req.override.assert_not_called()
|