from __future__ import annotations import sqlite3 import uuid import pytest import yaml from EvoScientist.llm.config_admin import EvoModelConfigAdminService from EvoScientist.llm.contracts import ( CommitModelConfigRequest, EvoRuntimeError, HmacGrantAuthority, ProbeCandidateRouteRequest, ValidateCandidateConfigRequest, ) from EvoScientist.llm.crypto import canonical_json_v1, sha256_id from EvoScientist.llm.model_config import EvoModelConfig, FileEvoModelConfigStore from tests.v3_fixtures import ( RUNTIME_KEY_ID, RUNTIME_SECRET, identity_ring, v3_payload, ) def _grant(authority, *, action, operation_id, payload): return authority.sign_admin( subject_id="admin", action=action, operation_id=operation_id, request_digest=sha256_id(payload), ttl_ms=60_000, ) @pytest.mark.asyncio async def test_validate_probe_commit_bootstraps_v2_config(monkeypatch, tmp_path): monkeypatch.setenv("WEB_RUNTIME_TEST_KEY", "test-secret") authority = HmacGrantAuthority(RUNTIME_SECRET, RUNTIME_KEY_ID) store = FileEvoModelConfigStore( tmp_path / "model_routes.yaml", admin_verifier=authority ) service = EvoModelConfigAdminService( store, grant_authority=authority, identity_key_ring=identity_ring(), probe_runner=lambda *_args: True, ) candidate = v3_payload() candidate.pop("capability_evidence") validate_operation = str(uuid.uuid4()) validate_payload = { "operation_id": validate_operation, "expected_revision": 0, "payload": candidate, } result = service.validate_candidate( ValidateCandidateConfigRequest( validate_operation, 0, candidate, _grant( authority, action="model_config:validate", operation_id=validate_operation, payload=validate_payload, ), ) ) evidence_ids = [] for route in result.concrete_routes: for probe_kind in route.required_probe_kinds: operation_id = str(uuid.uuid4()) payload = { "operation_id": operation_id, "proposal_hash": result.proposal_hash, "route_semantics_hash": route.route_semantics_hash, "probe_kind": probe_kind, } probe = await service.probe_candidate( ProbeCandidateRouteRequest( operation_id, result.proposal_hash, route.route_semantics_hash, probe_kind, _grant( authority, action="model_config:probe", operation_id=operation_id, payload=payload, ), ) ) evidence_ids.append(probe.evidence_id) commit_operation = str(uuid.uuid4()) commit_payload = { "operation_id": commit_operation, "expected_revision": 0, "proposal_hash": result.proposal_hash, "payload": candidate, "evidence_ids": sorted(set(evidence_ids)), } committed = service.commit( CommitModelConfigRequest( commit_operation, 0, result.proposal_hash, candidate, tuple(commit_payload["evidence_ids"]), _grant( authority, action="model_config:commit", operation_id=commit_operation, payload=commit_payload, ), ) ) assert committed.config_revision == 1 assert store.load().config_revision == 1 def test_v2_parser_rejects_old_alias_and_manual_capability_boundary(): payload = v3_payload() payload["providers"]["custom-openai"]["models"][0]["alias"] = "legacy" with pytest.raises(EvoRuntimeError, match="unknown fields"): EvoModelConfig.parse(payload) payload = v3_payload() payload["capability_requirements"] = [] with pytest.raises(EvoRuntimeError, match="unknown fields"): EvoModelConfig.parse(payload) def test_model_output_capability_rejects_oversized_output_token_limit(): payload = v3_payload() payload["providers"]["custom-openai"]["models"][0]["params"] = { "output_token_limit": 4096 } with pytest.raises( EvoRuntimeError, match="output_token_limit exceeds model capability" ): EvoModelConfig.parse(payload) def test_admin_grant_is_bound_to_action_and_payload(monkeypatch, tmp_path): monkeypatch.setenv("WEB_RUNTIME_TEST_KEY", "test-secret") authority = HmacGrantAuthority(RUNTIME_SECRET, RUNTIME_KEY_ID) service = EvoModelConfigAdminService( FileEvoModelConfigStore( tmp_path / "model_routes.yaml", admin_verifier=authority ), grant_authority=authority, identity_key_ring=identity_ring(), ) candidate = v3_payload() candidate.pop("capability_evidence") operation_id = str(uuid.uuid4()) wrong_payload = { "operation_id": operation_id, "expected_revision": 1, "payload": candidate, } with pytest.raises(EvoRuntimeError, match="ADMIN_CONFIG_FORBIDDEN"): service.validate_candidate( ValidateCandidateConfigRequest( operation_id, 0, candidate, _grant( authority, action="model_config:validate", operation_id=operation_id, payload=wrong_payload, ), ) ) def test_validate_operation_replays_and_rejects_changed_payload(monkeypatch, tmp_path): monkeypatch.setenv("WEB_RUNTIME_TEST_KEY", "test-secret") authority = HmacGrantAuthority(RUNTIME_SECRET, RUNTIME_KEY_ID) service = EvoModelConfigAdminService( FileEvoModelConfigStore(tmp_path / "model_routes.yaml"), grant_authority=authority, identity_key_ring=identity_ring(), ) candidate = v3_payload() candidate.pop("capability_evidence") operation_id = str(uuid.uuid4()) payload = { "operation_id": operation_id, "expected_revision": 0, "payload": candidate, } request = ValidateCandidateConfigRequest( operation_id, 0, candidate, _grant( authority, action="model_config:validate", operation_id=operation_id, payload=payload, ), ) first = service.validate_candidate(request) replay = service.validate_candidate(request) assert replay == first changed = {**candidate, "runtime_defaults": {"max_retries": 1}} changed_payload = {**payload, "payload": changed} with pytest.raises(EvoRuntimeError, match="IDEMPOTENCY_CONFLICT"): service.validate_candidate( ValidateCandidateConfigRequest( operation_id, 0, changed, _grant( authority, action="model_config:validate", operation_id=operation_id, payload=changed_payload, ), ) ) def test_provider_defaults_accept_values_up_to_v3_bounds(): from EvoScientist.llm.model_config import _parse_v3_provider_defaults runtime = { "connect_timeout_seconds": 10, "first_event_timeout_seconds": 60, "stream_idle_timeout_seconds": 60, "attempt_timeout_seconds": 600, } defaults = _parse_v3_provider_defaults( {"connect_timeout_seconds": 60, "attempt_timeout_seconds": 600}, runtime ) assert defaults["connect_timeout_seconds"] == 60 assert defaults["attempt_timeout_seconds"] == 600 def test_provider_defaults_reject_values_beyond_v3_bounds(): from EvoScientist.llm.model_config import _parse_v3_provider_defaults runtime = { "connect_timeout_seconds": 10, "first_event_timeout_seconds": 60, "stream_idle_timeout_seconds": 60, "attempt_timeout_seconds": 600, } with pytest.raises(EvoRuntimeError): _parse_v3_provider_defaults({"connect_timeout_seconds": 61}, runtime) def test_store_recovers_renamed_target_from_preparing_journal(tmp_path): path = tmp_path / "model_routes.yaml" ops_path = tmp_path / "model_config_ops.sqlite" store = FileEvoModelConfigStore(path, ops_path=ops_path) store.bootstrap_for_development(v3_payload(), operation_id="bootstrap") target = v3_payload(revision=2) payload_hash = sha256_id(target) path.write_text( yaml.safe_dump(target, allow_unicode=True, sort_keys=False), encoding="utf-8", ) with sqlite3.connect(ops_path) as connection: connection.execute( """INSERT INTO config_operations (subject_id, action, operation_id, request_digest, status, expected_revision, target_revision, payload_hash, canonical_payload, created_at, updated_at) VALUES ('admin', 'model_config:commit', 'recover-op', ?, 'PREPARING', 1, 2, ?, ?, 1, 1)""", ( payload_hash, payload_hash, canonical_json_v1(target).decode("utf-8"), ), ) recovered = FileEvoModelConfigStore(path, ops_path=ops_path) assert recovered.load().config_revision == 2 with sqlite3.connect(ops_path) as connection: status = connection.execute( "SELECT status FROM config_operations WHERE operation_id='recover-op'" ).fetchone()[0] assert status == "COMMITTED" @pytest.mark.asyncio async def test_probe_dispatch_is_durable_and_operation_replay_is_free( monkeypatch, tmp_path ): monkeypatch.setenv("WEB_RUNTIME_TEST_KEY", "test-secret") authority = HmacGrantAuthority(RUNTIME_SECRET, RUNTIME_KEY_ID) sink_events = [] provider_calls = 0 async def sink(payload): sink_events.append(dict(payload)) if payload["outcome"] == "started": return "committed" return f"terminal:{payload['probe_result']}" async def runner(*_args): nonlocal provider_calls provider_calls += 1 return True service = EvoModelConfigAdminService( FileEvoModelConfigStore(tmp_path / "model_routes.yaml"), grant_authority=authority, identity_key_ring=identity_ring(), probe_runner=runner, probe_event_sink=sink, ) candidate = v3_payload() candidate.pop("capability_evidence") validate_id = str(uuid.uuid4()) validate_payload = { "operation_id": validate_id, "expected_revision": 0, "payload": candidate, } validated = service.validate_candidate( ValidateCandidateConfigRequest( validate_id, 0, candidate, _grant( authority, action="model_config:validate", operation_id=validate_id, payload=validate_payload, ), ) ) route = validated.concrete_routes[0] operation_id = str(uuid.uuid4()) payload = { "operation_id": operation_id, "proposal_hash": validated.proposal_hash, "route_semantics_hash": route.route_semantics_hash, "probe_kind": route.required_probe_kinds[0], } request = ProbeCandidateRouteRequest( operation_id, validated.proposal_hash, route.route_semantics_hash, route.required_probe_kinds[0], _grant( authority, action="model_config:probe", operation_id=operation_id, payload=payload, ), ) first = await service.probe_candidate(request) replay = await service.probe_candidate(request) assert replay == first assert provider_calls == 1 assert [event["outcome"] for event in sink_events] == [ "started", "usage_unconfirmed", ] assert all(event["billing_intent"] == "platform_cost" for event in sink_events) @pytest.mark.asyncio async def test_commit_rejects_probe_evidence_after_runtime_secret_changes( monkeypatch, tmp_path ): monkeypatch.setenv("WEB_RUNTIME_TEST_KEY", "secret-before-probe") authority = HmacGrantAuthority(RUNTIME_SECRET, RUNTIME_KEY_ID) store = FileEvoModelConfigStore(tmp_path / "model_routes.yaml") service = EvoModelConfigAdminService( store, grant_authority=authority, identity_key_ring=identity_ring(), probe_runner=lambda *_args: True, ) candidate = v3_payload() candidate.pop("capability_evidence") validate_id = str(uuid.uuid4()) validate_payload = { "operation_id": validate_id, "expected_revision": 0, "payload": candidate, } validated = service.validate_candidate( ValidateCandidateConfigRequest( validate_id, 0, candidate, _grant( authority, action="model_config:validate", operation_id=validate_id, payload=validate_payload, ), ) ) evidence_ids = [] for route in validated.concrete_routes: for probe_kind in route.required_probe_kinds: operation_id = str(uuid.uuid4()) probe_payload = { "operation_id": operation_id, "proposal_hash": validated.proposal_hash, "route_semantics_hash": route.route_semantics_hash, "probe_kind": probe_kind, } result = await service.probe_candidate( ProbeCandidateRouteRequest( operation_id, validated.proposal_hash, route.route_semantics_hash, probe_kind, _grant( authority, action="model_config:probe", operation_id=operation_id, payload=probe_payload, ), ) ) evidence_ids.append(result.evidence_id) monkeypatch.setenv("WEB_RUNTIME_TEST_KEY", "secret-after-probe") commit_id = str(uuid.uuid4()) commit_payload = { "operation_id": commit_id, "expected_revision": 0, "proposal_hash": validated.proposal_hash, "payload": candidate, "evidence_ids": sorted(evidence_ids), } with pytest.raises(EvoRuntimeError, match="CAPABILITY_EVIDENCE_STALE"): service.commit( CommitModelConfigRequest( commit_id, 0, validated.proposal_hash, candidate, tuple(commit_payload["evidence_ids"]), _grant( authority, action="model_config:commit", operation_id=commit_id, payload=commit_payload, ), ) )