diff --git a/packages/sie_gateway/src/handlers/decisions.rs b/packages/sie_gateway/src/handlers/decisions.rs index 19de6388b..6e60429a2 100644 --- a/packages/sie_gateway/src/handlers/decisions.rs +++ b/packages/sie_gateway/src/handlers/decisions.rs @@ -309,7 +309,7 @@ fn normalize_request(parsed: &Value) -> Result<(Value, Vec), Invalid> { native.insert(format!("question_{index}"), normalized); } let selectors = - super::lora::selectors(body).map_err(|message| invalid(message, "lora_adapter"))?; + super::lora::selectors(body).map_err(|error| invalid(error.message, error.field))?; let mut request = json!({"model": model, "state": body["input"], "questions": native}); request.as_object_mut().unwrap().extend(selectors); Ok((request, questions.clone())) @@ -664,6 +664,103 @@ mod tests { assert!(normalize_request(&conflict).is_err()); } + async fn assert_lora_error_fields(path: &str, base: Value, status: StatusCode) { + let gateway = TestGateway::new(&[LOCAL_DECISION_MODEL]).await; + let router = create_router( + Arc::clone(&gateway.state), + Arc::clone(&gateway.state.config), + ); + let mut cases = Vec::new(); + for field in ["lora", "lora_id", "lora_adapter"] { + for value in [ + json!(false), + json!(1), + json!([]), + json!({}), + json!(""), + json!(" \t"), + ] { + cases.push(( + json!({field: value}), + field, + format!("'{field}' must be a non-empty string or null"), + )); + } + } + for (first, second) in [ + ("lora_adapter", "lora"), + ("lora_adapter", "lora_id"), + ("lora", "lora_id"), + ] { + for (left, right) in [ + (Value::Null, json!("private-selection-a")), + (json!("private-selection-a"), Value::Null), + (json!("private-selection-a"), json!("private-selection-b")), + ] { + cases.push(( + json!({first: left, second: right}), + second, + "LoRA selector fields must agree, including explicit null values".into(), + )); + } + } + for (selectors, field, message) in cases { + let mut payload = base.clone(); + payload + .as_object_mut() + .unwrap() + .extend(selectors.as_object().unwrap().clone()); + let request = Request::builder() + .method(Method::POST) + .uri(path) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(serde_json::to_vec(&payload).unwrap())) + .unwrap(); + let response = router.clone().oneshot(request).await.unwrap(); + assert_eq!(response.status(), status, "{path}: {selectors}"); + let body = body_json(response).await; + if path == "/v1/decisions" { + assert_eq!( + body["error"], + json!({ + "message": message, "param": field, + "type": "invalid_request_error", "code": "invalid_request", + }), + "{selectors}" + ); + } else { + assert_eq!( + body["detail"][0]["loc"], + json!(["body", field]), + "{selectors}" + ); + assert_eq!(body["detail"][0]["msg"], message, "{selectors}"); + } + assert!(!body.to_string().contains("private-selection")); + assert!(gateway.dispatcher.dispatched().is_empty()); + } + } + + #[tokio::test] + async fn registered_decisions_route_identifies_invalid_lora_fields() { + assert_lora_error_fields( + "/v1/decisions", + predicate_request(), + StatusCode::BAD_REQUEST, + ) + .await; + } + + #[tokio::test] + async fn registered_systemone_route_identifies_invalid_lora_fields() { + assert_lora_error_fields( + "/v1/systemone", + json!({"model": "acme/decide", "state": "Continue?", "questions": {"q": {"type": "noul"}}}), + StatusCode::UNPROCESSABLE_ENTITY, + ) + .await; + } + #[test] fn shared_fixtures_pin_request_and_response_parity() { for case in fixture()["requests"].as_array().unwrap() { diff --git a/packages/sie_gateway/src/handlers/lora.rs b/packages/sie_gateway/src/handlers/lora.rs index ba1f4fa51..82c0cf253 100644 --- a/packages/sie_gateway/src/handlers/lora.rs +++ b/packages/sie_gateway/src/handlers/lora.rs @@ -4,8 +4,14 @@ use serde_json::{Map, Value}; pub(crate) const FIELDS: [&str; 3] = ["lora_adapter", "lora", "lora_id"]; +#[derive(Debug)] +pub(crate) struct SelectorError { + pub field: &'static str, + pub message: String, +} + /// Retain the supplied spellings so legacy source-only admission cannot be lost. -pub(crate) fn selectors(body: &Map) -> Result, String> { +pub(crate) fn selectors(body: &Map) -> Result, SelectorError> { let mut selected = None; let mut fields = Map::new(); for field in FIELDS { @@ -13,10 +19,16 @@ pub(crate) fn selectors(body: &Map) -> Result, continue; }; if !value.is_null() && !value.as_str().is_some_and(|s| !s.trim().is_empty()) { - return Err(format!("'{field}' must be a non-empty string or null")); + return Err(SelectorError { + field, + message: format!("'{field}' must be a non-empty string or null"), + }); } if selected.is_some_and(|prior| prior != value) { - return Err("LoRA selector fields must agree, including explicit null values".into()); + return Err(SelectorError { + field, + message: "LoRA selector fields must agree, including explicit null values".into(), + }); } selected = Some(value); fields.insert(field.into(), value.clone()); @@ -27,7 +39,9 @@ pub(crate) fn selectors(body: &Map) -> Result, pub(crate) fn validate_options(options: Option<&Value>) -> Result<(), String> { match options { None | Some(Value::Null) => Ok(()), - Some(Value::Object(options)) => selectors(options).map(|_| ()), + Some(Value::Object(options)) => selectors(options) + .map(|_| ()) + .map_err(|error| error.message), _ => Err("'options' must be an object or null".into()), } } @@ -61,7 +75,9 @@ pub(crate) fn validate_options_rmpv(options: Option<&rmpv::Value>) -> Result<(), return Err(format!("duplicate LoRA selector '{key}'")); } } - selectors(&selected).map(|_| ()) + selectors(&selected) + .map(|_| ()) + .map_err(|error| error.message) } #[cfg(test)] diff --git a/packages/sie_gateway/src/handlers/systemone.rs b/packages/sie_gateway/src/handlers/systemone.rs index 6b22f5126..b201d6cc9 100644 --- a/packages/sie_gateway/src/handlers/systemone.rs +++ b/packages/sie_gateway/src/handlers/systemone.rs @@ -169,7 +169,7 @@ pub(crate) fn normalize_request(parsed: &Value) -> Result dict[str, str]: + paths = resolved.loadtime.get("lora_paths") + return {name: lora_entry_ref(ref)[0] for name, ref in paths.items() if ref} if isinstance(paths, Mapping) else {} + + def _check_declared_lora( config: ModelConfig, resolved: ResolvedProfile, profile_name: str, request_options: dict[str, Any] ) -> None: @@ -92,21 +103,59 @@ def _check_declared_lora( Raises: UndeclaredLoraError: Either option is not null and not one of those names. """ + ids = resolved.lora_ids() + names = ids | _profile_lora_aliases(resolved).keys() for key, declared in ( - ("lora_id", resolved.lora_ids), - ("lora", resolved.lora_names), - ("lora_adapter", resolved.lora_names), + ("lora_id", ids), + ("lora", names), + ("lora_adapter", names), ): value = request_options.get(key) if value is None: continue - if not isinstance(value, str) or value not in declared(): + if not isinstance(value, str) or value not in declared: raise UndeclaredLoraError( f"'options.{key}' must name a LoRA adapter that profile '{profile_name}' of model " f"'{config.sie_id}' declares; {value!r} is not one of them." ) +def _profile_lora_selection(resolved: ResolvedProfile) -> LoraSelection: + """Compare profile aliases by source while preserving the served selector. + + Unlike request aliases, a profile's legacy ``lora_id`` also declares a + source to load. ``lora_adapter`` / ``lora`` select that source by served + name, or explicitly select the base model with null. + """ + selections = [ + split_lora_selection({field: resolved.runtime[field]})[0] + for field in LORA_SELECTION_FIELDS + if field in resolved.runtime + ] + if not selections: + return LoraSelection(provided=False, value=None, fields=()) + + aliases = _profile_lora_aliases(resolved) + declared = resolved.lora_ids() | aliases.keys() + sources: list[str | None] = [] + for selection in selections: + value = selection.value + field = selection.fields[0] + if value is not None and field != "lora_id" and value not in declared: + raise InvalidLoraSelectionError(f"Profile '{field}' must select a declared LoRA adapter") + sources.append(aliases.get(value, value) if value is not None and field != "lora_id" else value) + + selected = selections[0] + for selection, source in zip(selections[1:], sources[1:], strict=True): + # A source declaration is not a competing base selection; likewise, + # a declared source must not override an explicit base default. + if selection.fields == ("lora_id",) and (selected.value is None or selection.value is None): + continue + if source != sources[0]: + raise InvalidLoraSelectionError("Profile LoRA defaults must resolve to the same declared adapter") + return selected + + def merge_runtime_options_with_profile( config: ModelConfig, request_options: dict[str, Any] | None, @@ -123,11 +172,14 @@ def merge_runtime_options_with_profile( selection, request = split_lora_selection(request_options or {}) except InvalidLoraSelectionError as exc: raise InvalidLoraOptionError(str(exc)) from exc - defaults, merged = split_lora_selection(resolved.runtime) + merged = {key: value for key, value in resolved.runtime.items() if key not in LORA_SELECTION_FIELDS} if selection.provided: _check_declared_lora(config, resolved, request.get("profile") or "default", request_options or {}) else: - selection = defaults + try: + selection = _profile_lora_selection(resolved) + except InvalidLoraSelectionError as exc: + raise InvalidLoraOptionError(str(exc)) from exc merged |= {key: value for key, value in request.items() if key != "profile"} if selection.provided: merged["lora"] = selection.value diff --git a/packages/sie_server/tests/api/test_lora_aliases.py b/packages/sie_server/tests/api/test_lora_aliases.py index 80b697b35..fc42e52a9 100644 --- a/packages/sie_server/tests/api/test_lora_aliases.py +++ b/packages/sie_server/tests/api/test_lora_aliases.py @@ -1,13 +1,17 @@ -"""Native HTTP/MessagePack ingress rejects selectors before routing or loading.""" +"""Native ingress preserves profile defaults and rejects invalid request selectors.""" +from concurrent.futures import ThreadPoolExecutor from unittest.mock import AsyncMock, MagicMock +import numpy as np import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from sie_sdk._msgpack import packb from sie_server.api import encode, extract, score from sie_server.config.model import ModelConfig +from sie_server.core.inference_output import EncodeOutput +from sie_server.core.postprocessor_registry import PostprocessorRegistry from sie_server.core.registry import ModelRegistry @@ -18,6 +22,7 @@ [ {"lora_adapter": None, "lora": "org/legal"}, {"lora_adapter": "org/legal", "lora_id": "other"}, + {"lora": "legal", "lora_id": "org/legal"}, {"lora_adapter": []}, {"lora_adapter": " "}, {"lora_id": "legal"}, @@ -65,3 +70,115 @@ def test_bad_lora_options_stop_before_routing_loading_or_worker(endpoint, wire, route.assert_not_awaited() registry.ensure_lora_loaded_async.assert_not_called() registry.start_worker.assert_not_called() + + +@pytest.mark.parametrize("wire", ["json", "msgpack"]) +@pytest.mark.parametrize( + ("selection", "expected"), + [ + ({}, "legal"), + ({"lora_adapter": None}, None), + ({"lora": None}, None), + ({"lora_id": None}, None), + ({"lora_adapter": "org/legal"}, "org/legal"), + ], +) +def test_http_profile_alias_and_source_default_reaches_adapter(wire, selection, expected): + config = ModelConfig.model_validate( + { + "sie_id": "m", + "hf_id": "unloaded", + "tasks": {"encode": {"dense": {"dim": 8}}}, + "profiles": { + "default": {"max_batch_tokens": 8192, "adapter_path": "test:TestAdapter"}, + "legal": { + "extends": "default", + "adapter_options": { + "loadtime": {"lora_paths": {"legal": {"id": "org/legal", "revision": "a" * 40}}}, + "runtime": {"lora": "legal", "lora_id": "org/legal"}, + }, + }, + }, + } + ) + registry = MagicMock(spec=ModelRegistry) + registry.has_model.return_value = registry.is_loaded.return_value = True + registry.is_loading.return_value = registry.is_unloading.return_value = registry.is_failed.return_value = False + registry.get_config.return_value = config + registry.device = "cpu" + registry.preprocessor_registry = MagicMock() + registry.preprocessor_registry.has_tokenizer.return_value = False + registry.preprocessor_registry.has_preprocessor.return_value = False + registry.ensure_lora_loaded_async = AsyncMock(return_value=(True, False)) + adapter = MagicMock() + adapter.encode.return_value = EncodeOutput(dense=np.ones((1, 8), dtype=np.float32), batch_size=1, dense_dim=8) + registry.get.return_value = adapter + app = FastAPI() + app.include_router(encode.router) + app.state.registry = registry + body = {"items": [{"text": "x"}], "params": {"options": {"profile": "legal", **selection}}} + with ThreadPoolExecutor(max_workers=1) as pool: + registry.postprocessor_registry = PostprocessorRegistry(pool) + with TestClient(app) as client: + if wire == "json": + response = client.post("/v1/encode/m", json=body, headers={"accept": "application/json"}) + else: + response = client.post( + "/v1/encode/m", + content=packb(body), + headers={"content-type": "application/msgpack", "accept": "application/json"}, + ) + + assert response.status_code == 200, response.text + assert response.json()["items"][0]["dense"]["values"] == [1.0] * 8 + adapter.encode.assert_called_once() + assert adapter.encode.call_args.kwargs["options"] == {"lora": expected, "output_dtype": "float32"} + if expected is None: + registry.ensure_lora_loaded_async.assert_not_awaited() + else: + registry.ensure_lora_loaded_async.assert_awaited_once_with("m", expected) + + +@pytest.mark.parametrize("endpoint", ["encode", "extract", "score"]) +@pytest.mark.parametrize("field", [None, "lora_adapter", "lora", "lora_id"]) +def test_profile_selection_does_not_authorize_sibling_source_before_http_dispatch(endpoint, field, monkeypatch): + config = ModelConfig.model_validate( + { + "sie_id": "m", + "hf_id": "unloaded", + "tasks": {"encode": {"dense": {"dim": 8}}, "extract": {}, "score": {}}, + "profiles": { + "default": { + "max_batch_tokens": 8192, + "adapter_path": "test:TestAdapter", + "adapter_options": {"runtime": {"lora": "org/legal"}}, + }, + "legal": { + "extends": "default", + "adapter_options": {"loadtime": {"lora_paths": {"legal": "org/legal"}}}, + }, + }, + } + ) + registry = MagicMock(spec=ModelRegistry) + registry.has_model.return_value = registry.is_loaded.return_value = True + registry.is_loading.return_value = registry.is_unloading.return_value = registry.is_failed.return_value = False + registry.get_config.return_value = config + registry.device = "cpu" + app = FastAPI() + module = {"encode": encode, "extract": extract, "score": score}[endpoint] + app.include_router(module.router) + app.state.registry = registry + route = AsyncMock(side_effect=AssertionError("undeclared source reached route")) + monkeypatch.setattr(module, "route_request", route) + options = {} if field is None else {field: "org/legal"} + body = {"items": [{"text": "x"}], "params": {"labels": ["label"], "options": options}} + if endpoint == "score": + body = {"query": {"text": "q"}, "items": [{"text": "x"}], "options": options} + with TestClient(app) as client: + response = client.post(f"/v1/{endpoint}/m", json=body) + assert response.status_code == 400, response.text + assert "declar" in response.text + route.assert_not_awaited() + registry.ensure_lora_loaded_async.assert_not_called() + registry.start_worker.assert_not_called() diff --git a/packages/sie_server/tests/core/test_runtime_options.py b/packages/sie_server/tests/core/test_runtime_options.py index ac4738178..adfcf6f65 100644 --- a/packages/sie_server/tests/core/test_runtime_options.py +++ b/packages/sie_server/tests/core/test_runtime_options.py @@ -136,6 +136,108 @@ def test_explicit_null_overrides_legacy_profile_default(field: str) -> None: assert merge_runtime_options(_lora_config(), {"profile": "banking"}) == {"lora": "org/banking-lora"} +def _served_lora_config(runtime: dict[str, Any], *, pinned: bool = False) -> ModelConfig: + data = _lora_config().model_dump() + options = data["profiles"]["served"]["adapter_options"] + options["runtime"] = {"threshold": 0.5, **runtime} + if pinned: + options["loadtime"]["lora_paths"]["legal"] = {"id": "org/legal-lora", "revision": "a" * 40} + return ModelConfig.model_validate(data) + + +@pytest.mark.parametrize("pinned", [False, True]) +@pytest.mark.parametrize( + ("runtime", "expected"), + [ + ({"lora": "legal", "lora_id": "org/legal-lora"}, "legal"), + ({"lora_adapter": "legal", "lora_id": "org/legal-lora"}, "legal"), + ({"lora_adapter": "legal", "lora": "org/legal-lora", "lora_id": "org/legal-lora"}, "legal"), + ({"lora": None, "lora_id": "org/legal-lora"}, None), + ({"lora": "legal", "lora_id": None}, "legal"), + ({"lora_adapter": None, "lora": None, "lora_id": "org/legal-lora"}, None), + ({"lora_id": "org/legal-lora"}, "org/legal-lora"), + ], +) +def test_profile_lora_defaults_preserve_served_selection(runtime: dict[str, Any], expected, pinned: bool) -> None: + config = _served_lora_config(runtime, pinned=pinned) + before = config.model_dump() + + merged = merge_runtime_options(config, {"profile": "served"}) + + assert merged == {"threshold": 0.5, "lora": expected} + assert config.model_dump() == before + assert merge_runtime_options(config, None) == {} + + +@pytest.mark.parametrize("field", ["lora_adapter", "lora", "lora_id"]) +@pytest.mark.parametrize("value", [None, "org/legal-lora"]) +def test_request_selector_overrides_profile_alias_and_source(field: str, value: str | None) -> None: + config = _served_lora_config({"lora": "legal", "lora_id": "org/legal-lora"}, pinned=True) + options = {"profile": "served", field: value, "threshold": 0.75} + + assert merge_runtime_options(config, options) == {"threshold": 0.75, "lora": value} + assert options == {"profile": "served", field: value, "threshold": 0.75} + + +@pytest.mark.parametrize( + "runtime", + [ + {"lora": "unknown-alias", "lora_id": "org/legal-lora"}, + {"lora": "unknown-alias"}, + {"lora": "org/banking-lora"}, + {"lora": "legal", "lora_id": "org/other"}, + {"lora_adapter": "undeclared"}, + {"lora_adapter": None, "lora": "legal"}, + {"lora_adapter": "legal", "lora": None}, + {"lora": False}, + {"lora_id": []}, + ], +) +def test_invalid_profile_lora_defaults_fail_only_when_selected(runtime: dict[str, Any]) -> None: + config = _served_lora_config(runtime) + with pytest.raises(InvalidLoraOptionError): + merge_runtime_options(config, {"profile": "served"}) + assert merge_runtime_options(config, {"profile": "served", "lora_adapter": None}) == { + "threshold": 0.5, + "lora": None, + } + assert merge_runtime_options(config, None) == {} + + +@pytest.mark.parametrize("ref", [None, ""]) +def test_profile_lora_default_cannot_resolve_through_an_empty_declaration(ref) -> None: + data = _served_lora_config({"lora": "legal", "lora_id": "org/legal-lora"}).model_dump() + data["profiles"]["served"]["adapter_options"]["loadtime"]["lora_paths"]["legal"] = ref + config = ModelConfig.model_validate(data) + with pytest.raises(InvalidLoraOptionError, match="declared"): + merge_runtime_options(config, {"profile": "served"}) + + +@pytest.mark.parametrize("field", [None, "lora_adapter", "lora", "lora_id"]) +def test_runtime_selection_cannot_authorize_another_profiles_source(field: str | None) -> None: + data = _served_lora_config({"lora": "org/banking-lora"}).model_dump() + data["profiles"]["served"]["adapter_options"]["loadtime"] = {} + config = ModelConfig.model_validate(data) + options = {"profile": "served"} + if field is not None: + options[field] = "org/banking-lora" + with pytest.raises(InvalidInputError, match="declar"): + merge_runtime_options(config, options) + assert merge_runtime_options(config, {"profile": "served", "lora_adapter": None}) == { + "threshold": 0.5, + "lora": None, + } + assert merge_runtime_options(config, {"profile": "banking"}) == {"lora": "org/banking-lora"} + + +def test_request_aliases_still_require_raw_agreement_for_same_declared_source() -> None: + config = _served_lora_config({"lora": "legal", "lora_id": "org/legal-lora"}, pinned=True) + with pytest.raises(InvalidLoraOptionError, match="agree"): + merge_runtime_options(config, {"profile": "served", "lora": "legal", "lora_id": "org/legal-lora"}) + with pytest.raises(runtime_options.UndeclaredLoraError): + merge_runtime_options(config, {"lora_adapter": "legal"}) + + @pytest.mark.parametrize( "options", [ diff --git a/packages/sie_server/tests/test_queue_executor.py b/packages/sie_server/tests/test_queue_executor.py index dedf01f4f..bdf5ce1dd 100644 --- a/packages/sie_server/tests/test_queue_executor.py +++ b/packages/sie_server/tests/test_queue_executor.py @@ -259,7 +259,7 @@ def _runtime_profile_config() -> ModelConfig: "max_batch_tokens": 8192, "adapter_path": "sie_server.adapters.fake.adapter:FakeAdapter", "adapter_options": { - "loadtime": {"lora_paths": ["request-lora"]}, + "loadtime": {"lora_paths": ["request-lora", "profile-lora"]}, "runtime": { "threshold": 0.5, "lora": "profile-lora", @@ -1313,15 +1313,22 @@ class TestProcessScoreBatch: @pytest.mark.asyncio @pytest.mark.parametrize("key", ["lora_adapter", "lora", "lora_id"]) @pytest.mark.parametrize("supports_lora", [True, False]) - @pytest.mark.parametrize("legacy_default", [False, True]) + @pytest.mark.parametrize("default_style", ["served", "source", "alias_and_source"]) async def test_real_worker_preserves_authorized_lora_and_isolates_rejections( - self, supports_lora: bool, legacy_default: bool, key: str + self, supports_lora: bool, default_style: str, key: str ) -> None: reg = _make_registry() config_data = _runtime_profile_config().model_dump() - if legacy_default: - runtime = config_data["profiles"]["tuned"]["adapter_options"]["runtime"] + options = config_data["profiles"]["tuned"]["adapter_options"] + runtime = options["runtime"] + if default_style == "source": runtime["lora_id"] = runtime.pop("lora") + elif default_style == "alias_and_source": + runtime["lora_id"] = "org/profile-lora" + options["loadtime"]["lora_paths"] = { + "profile-lora": {"id": "org/profile-lora", "revision": "a" * 40}, + "request-lora": "request-lora", + } reg.get_config.return_value = ModelConfig.model_validate(config_data) adapter = MagicMock() adapter.supports_lora.return_value = supports_lora @@ -1352,6 +1359,10 @@ def score(queries, items, **kwargs) -> ScoreOutput: _score_item(wiid="reset.0", options={"profile": "tuned", key: None}), _score_item(wiid="profile-again.0", options={"profile": "tuned"}), _score_item(wiid="undeclared.0", options={"profile": "tuned", "lora": "untrusted"}), + _score_item( + wiid="conflicting.0", + options={"profile": "tuned", "lora": "profile-lora", "lora_id": "org/profile-lora"}, + ), ], ) ) @@ -1371,6 +1382,8 @@ def score(queries, items, **kwargs) -> ScoreOutput: assert by_id[wiid].raw_output.score.scores == [0.0] assert by_id["undeclared.0"].error_code == "INVALID_INPUT" assert "declares" in by_id["undeclared.0"].error + assert by_id["conflicting.0"].error_code == "INVALID_INPUT" + assert "agree" in by_id["conflicting.0"].error selections = [call.args[0] for call in adapter.set_active_lora.call_args_list] assert "untrusted" not in selections assert selections == (["profile-lora", "request-lora", None] if supports_lora else []) @@ -2276,3 +2289,33 @@ async def test_invalid_lora_aliases_are_caller_errors_before_queue_forward(kind: worker.submit_score_preformed_batch.assert_not_called() worker.submit_extract_preformed_batch.assert_not_called() reg.ensure_lora_loaded_async.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("kind", ["encode", "score", "extract"]) +@pytest.mark.parametrize("field", [None, "lora_adapter", "lora", "lora_id"]) +async def test_profile_selection_does_not_authorize_sibling_source_before_queue_forward(kind: str, field) -> None: + reg = _make_registry() + data = _runtime_profile_config().model_dump() + data["profiles"]["default"]["adapter_options"]["runtime"]["lora"] = "org/other" + data["profiles"]["tuned"]["adapter_options"] = {"loadtime": {"lora_paths": ["org/other"]}} + reg.get_config.return_value = ModelConfig.model_validate(data) + worker = AsyncMock() + reg.start_worker = AsyncMock(return_value=worker) + request_type, item = { + "encode": (ProcessEncodeBatchRequest, _encode_item), + "score": (ProcessScoreBatchRequest, _score_item), + "extract": (ProcessExtractBatchRequest, _extract_item), + }[kind] + options = {} if field is None else {field: "org/other"} + with patch("sie_server.queue_executor.EncodePipeline.run_encode", new_callable=AsyncMock) as encode: + outcome = await getattr(QueueExecutor(reg), f"process_{kind}_batch")( + request_type(model_id="test/model", items=[item(options=options)]) + ) + assert len(outcome.outcomes) == 1 + assert outcome.outcomes[0].error_code == "INVALID_INPUT" + assert "declar" in outcome.outcomes[0].error + encode.assert_not_called() + worker.submit_score_preformed_batch.assert_not_called() + worker.submit_extract_preformed_batch.assert_not_called() + reg.ensure_lora_loaded_async.assert_not_called()