diff --git a/.github/workflows/auto-unit-test.yml b/.github/workflows/auto-unit-test.yml index b7b5c4a6be..20893a1b05 100644 --- a/.github/workflows/auto-unit-test.yml +++ b/.github/workflows/auto-unit-test.yml @@ -82,6 +82,9 @@ jobs: uv pip install -e "../sdk[dev]" cd .. + - name: Check packaged model profiles before submission + run: backend/.venv/bin/python scripts/check_model_profiles.py + - name: Run all tests and collect coverage env: NEXENT_PYTEST_WORKERS: ${{ github.event.inputs.pytest_workers || 'auto' }} diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000000..0d65fed91f --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,9 @@ +repos: + - repo: local + hooks: + - id: model-profile-check + name: Check packaged model profiles + entry: python scripts/check_model_profiles.py + language: system + pass_filenames: false + files: ^(sdk/nexent/core/model_service/|scripts/check_model_profiles\.py) diff --git a/scripts/check_model_profiles.py b/scripts/check_model_profiles.py new file mode 100644 index 0000000000..69f00a796c --- /dev/null +++ b/scripts/check_model_profiles.py @@ -0,0 +1,26 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Check packaged model profiles before submission, without starting application services.""" + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "sdk")) + +from nexent.core.model_service.profiles.submission import check_profile_catalog + + +def main(): + try: + catalog = check_profile_catalog() + except ValueError as error: + # SourceReadError contains identifiers only; never print source values. + category = getattr(error, "category", "PROFILE_CONFIGURATION_INVALID") + print(f"{category}: {getattr(error, 'file_path', '')}: {getattr(error, 'source_path', '')}", file=sys.stderr) + return 1 + print(f"Model profile submission checks passed: {catalog.revision}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/sdk/nexent/core/model_service/adapters/openai_chat.py b/sdk/nexent/core/model_service/adapters/openai_chat.py new file mode 100644 index 0000000000..33c59015af --- /dev/null +++ b/sdk/nexent/core/model_service/adapters/openai_chat.py @@ -0,0 +1,309 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""OpenAI-compatible chat framing; runtime execution and retries stay outside.""" + +import json +import math + +from pydantic import ValidationError + +from nexent.core.model_service.contracts import AssistantMessage, ModelRequest, TokenUsage, ToolCall +from nexent.core.model_service.http_contracts import PreparedHttpRequest, RawHttpResponse +from nexent.core.model_service.parameter_transforms import parameter_transform_references, transform_parameter +from nexent.core.model_service.profiles.contracts import VersionedReference +from nexent.core.model_service.profiles.instruction_mapping import validate_instruction_mapping +from nexent.core.model_service.profiles.parameter_bindings import validate_parameter_bindings +from nexent.core.model_service.profiles.parameter_rules import compile_parameter_rules, validate_saved_parameter_values +from nexent.core.model_service.profiles.parameter_schema import ParameterValidationError +from nexent.core.model_service.profiles.request_parameters import _set, _tokens, apply_extra_api_parameters +from nexent.core.model_service.profiles.resolved_profile import ResolvedModelProfile + + +class ModelAdapterError(ValueError): + """Fixed protocol category without model text, tool arguments or raw body.""" + + def __init__(self, category: str, *, response: RawHttpResponse | None = None): + self.category = category + self.response = response + super().__init__(category) + + +def _unique_object(pairs): + result = {} + for key, value in pairs: + if key in result: + raise ValueError("Duplicate JSON member") + result[key] = value + return result + + +def _invalid_constant(value): + raise ValueError("Invalid JSON constant") + + +def _finite_float(value): + number = float(value) + if not math.isfinite(number): + raise ValueError("Invalid JSON number") + return number + + +def parse_json(text: str): + return json.loads( + text, object_pairs_hook=_unique_object, parse_constant=_invalid_constant, parse_float=_finite_float + ) + + +def _content_parts(value): + if value is None: + return [] + if isinstance(value, str): + return [{"type": "text", "text": value}] + if isinstance(value, list) and all(isinstance(part, dict) and isinstance(part.get("type"), str) for part in value): + return value + raise ModelAdapterError("MODEL_RESPONSE_INVALID") + + +def _usage(raw, semantics_id): + if raw is None: + return TokenUsage(semantics_id=semantics_id) + if not isinstance(raw, dict): + raise ModelAdapterError("MODEL_USAGE_INVALID") + details = [] + for key in ("prompt_tokens_details", "completion_tokens_details"): + value = raw.get(key) + if value is None: + value = {} + if not isinstance(value, dict): + raise ModelAdapterError("MODEL_USAGE_INVALID") + details.append(value) + try: + return TokenUsage( + input_tokens=raw.get("prompt_tokens"), + output_tokens=raw.get("completion_tokens"), + total_tokens=raw.get("total_tokens"), + cached_input_tokens=details[0].get("cached_tokens"), + reasoning_tokens=details[1].get("reasoning_tokens"), + origin="provider", + semantics_id=semantics_id, + ) + except ValidationError: + raise ModelAdapterError("MODEL_USAGE_INVALID") from None + + +def _tool_call(raw, index): + if not isinstance(raw, dict) or raw.get("type") != "function" or not isinstance(raw.get("function"), dict): + raise ModelAdapterError("MODEL_TOOL_CALL_INVALID") + function = raw["function"] + arguments = function.get("arguments") + if not isinstance(arguments, str): + raise ModelAdapterError("MODEL_TOOL_CALL_INVALID") + parsed, error = None, None + try: + parsed = parse_json(arguments) + if not isinstance(parsed, dict): + raise ValueError("Tool arguments must be an object") + except (ValueError, RecursionError): + parsed, error = None, "INVALID_TOOL_ARGUMENTS" + try: + return ToolCall( + provider_call_id=raw.get("id"), + index=index, + name=function.get("name"), + raw_arguments=arguments, + parsed_arguments=parsed, + argument_error=error, + ) + except ValidationError: + raise ModelAdapterError("MODEL_TOOL_CALL_INVALID") from None + + +class OpenAIChatAdapter: + """Single-choice messages with declared parameter bindings and wire metadata.""" + + reference = VersionedReference(ref="openai_chat_v1", version=1) + protocol_reference = VersionedReference(ref="openai_chat_completions", version=1) + + def __init__(self, *, reasoning_replay: bool = False): + if type(reasoning_replay) is not bool: + raise ValueError("Reasoning replay declaration must be boolean") + self._reasoning_replay = reasoning_replay + + def prepare_http_request(self, request: ModelRequest, profile: ResolvedModelProfile) -> PreparedHttpRequest: + if request.outlet_id != profile.outlet_id or request.config_revision != profile.config_revision: + raise ModelAdapterError("MODEL_SNAPSHOT_MISMATCH") + if profile.bindings.protocol != self.protocol_reference: + raise ModelAdapterError("MODEL_PROTOCOL_UNSUPPORTED") + validate_instruction_mapping(profile.instruction_mapping) + if request.tools and profile.capabilities.tools.status != "supported": + raise ModelAdapterError("NATIVE_TOOL_PROTOCOL_UNAVAILABLE") + mappings = profile.instruction_mapping.mappings.model_dump() + messages = [] + for message in request.messages: + role = mappings.get(message.semantic_role, message.semantic_role) + if role not in profile.instruction_mapping.supported_roles: + raise ModelAdapterError("MODEL_ROLE_UNSUPPORTED") + if message.tool_calls and role != "assistant" or message.tool_call_id is not None and role != "tool": + raise ModelAdapterError("MODEL_MESSAGE_INVALID") + if role == "tool" and message.tool_call_id is None: + raise ModelAdapterError("MODEL_MESSAGE_INVALID") + if ( + message.tool_calls or message.tool_call_id is not None or role == "tool" + ) and profile.capabilities.tools.status != "supported": + raise ModelAdapterError("NATIVE_TOOL_PROTOCOL_UNAVAILABLE") + parts = message.content_parts + if any( + part.get("type") + not in {"text", "image_url", "input_audio", "audio_url", "video_url", "input_image", "input_video"} + for part in parts + ): + raise ModelAdapterError("MODEL_CONTENT_UNSUPPORTED") + if any(part.get("type") == "text" and not isinstance(part.get("text"), str) for part in parts): + raise ModelAdapterError("MODEL_CONTENT_UNSUPPORTED") + simple_text = len(parts) == 1 and set(parts[0]) == {"type", "text"} + wire = {"role": role, "content": parts[0]["text"] if simple_text else parts} + if not parts: + wire["content"] = None if role == "assistant" and message.tool_calls else "" + if message.tool_call_id is not None: + wire["tool_call_id"] = message.tool_call_id + if message.tool_calls: + wire["tool_calls"] = [ + { + "id": call.provider_call_id, + "type": "function", + "function": {"name": call.name, "arguments": call.raw_arguments}, + } + for call in message.tool_calls + ] + if message.reasoning_parts: + if ( + not self._reasoning_replay + or role != "assistant" + or any( + part.get("type") != "text" or not isinstance(part.get("text"), str) + for part in message.reasoning_parts + ) + ): + raise ModelAdapterError("MODEL_REASONING_FORMAT_UNSUPPORTED") + wire["reasoning_content"] = "".join(part["text"] for part in message.reasoning_parts) + messages.append(wire) + body = {"model": profile.identity.request_model_id, "messages": messages, "stream": request.stream} + if request.tools: + body["tools"] = [ + { + "type": "function", + "function": {"name": tool.name, "description": tool.description, "parameters": tool.input_schema}, + } + for tool in request.tools + ] + if request.stream: + body["stream_options"] = {"include_usage": True} + try: + rules = compile_parameter_rules(profile.parameter_rules) + validate_saved_parameter_values(rules, profile.parameter_values) + bindings = validate_parameter_bindings(rules, parameter_transform_references()) + for key, binding in bindings.items(): + if key in {"temperature", "top_p"} and getattr(profile.capabilities.sampling, key).status == "unsupported": + continue + if key in profile.parameter_values: + _set(body, _tokens(binding["path"]), transform_parameter(binding, profile.parameter_values[key]), + binding["path"]) + body = apply_extra_api_parameters( + body, profile.extra_api_parameters, parameter_paths=[binding["path"] for binding in bindings.values()] + ) + return PreparedHttpRequest( + request_id=request.request_id, + method=profile.connection_rules.method, + base_url=profile.connection.base_url, + relative_path=profile.connection_rules.relative_path, + headers=profile.connection_rules.headers, + credential_ref=profile.connection.credential_ref, + json_body=body, + timeout_seconds=profile.connection.timeout_seconds, + ssl_verify=profile.connection.ssl_verify, + stream_protocol="sse" if request.stream else None, + ) + except (ParameterValidationError, ValueError, TypeError): + raise ModelAdapterError("MODEL_REQUEST_CONFIGURATION_INVALID") from None + + def decode_response( + self, raw: RawHttpResponse, request: ModelRequest, profile: ResolvedModelProfile + ) -> AssistantMessage: + try: + body = parse_json(raw.body.decode("utf-8")) + if isinstance(body, dict) and body.get("error") is not None: + raise ModelAdapterError("MODEL_PROVIDER_ERROR", response=raw) + if not isinstance(body, dict) or not isinstance(body.get("choices"), list) or len(body["choices"]) != 1: + raise ModelAdapterError("MODEL_RESPONSE_INVALID") + choice = body["choices"][0] + if not isinstance(choice, dict) or type(choice.get("index")) is not int or choice["index"] != 0: + raise ModelAdapterError("MODEL_RESPONSE_INVALID") + message = choice.get("message") + if not isinstance(message, dict) or message.get("role") != "assistant": + raise ModelAdapterError("MODEL_RESPONSE_INVALID") + finish = choice.get("finish_reason") + if not isinstance(finish, str) or not finish.strip(): + raise ModelAdapterError("MODEL_RESPONSE_INCOMPLETE") + calls = message.get("tool_calls") + if calls is None: + calls = [] + if not isinstance(calls, list): + raise ModelAdapterError("MODEL_TOOL_CALL_INVALID") + parts = _content_parts(message.get("content")) + if message.get("audio") is not None: + parts.append({"type": "audio", "audio": message["audio"]}) + return AssistantMessage( + content_parts=parts, + reasoning_parts=_content_parts(message.get("reasoning_content")), + refusal=message.get("refusal"), + tool_calls=[_tool_call(call, index) for index, call in enumerate(calls)], + finish_reason=finish, + usage=_usage(body.get("usage"), profile.capabilities.usage.semantics_id), + provider_metadata={ + "response_id": body.get("id"), + "model": body.get("model"), + "request_id": request.request_id, + "unmapped_message_fields": { + key: value + for key, value in message.items() + if key not in {"role", "content", "reasoning_content", "refusal", "tool_calls", "audio"} + }, + }, + ) + except ModelAdapterError: + raise + except (ValueError, TypeError, UnicodeError, RecursionError): + raise ModelAdapterError("MODEL_RESPONSE_INVALID") from None + + def new_stream_decoder(self, request, profile): + """Import the stream state lazily to keep complete-message helpers reusable.""" + from nexent.core.model_service.adapters.openai_stream import OpenAIStreamDecoder + + return OpenAIStreamDecoder(request, profile) + + def normalize_error(self, error, *, request_id, attempt=1, max_attempts=1, will_retry=False): + from nexent.core.model_service.errors import normalize_model_error + + return normalize_model_error( + error, request_id=request_id, attempt=attempt, max_attempts=max_attempts, will_retry=will_retry + ) + + def decode_stream(self, raw_events, request, profile): + """Stop at the verified completion marker; HTTP ownership is external.""" + decoder = self.new_stream_decoder(request, profile) + for frame in raw_events: + for event in decoder.push(frame): + yield event + if event.type == "message_completed": + return + yield from decoder.finish() + + async def adecode_stream(self, raw_events, request, profile): + decoder = self.new_stream_decoder(request, profile) + async for frame in raw_events: + for event in decoder.push(frame): + yield event + if event.type == "message_completed": + return + for event in decoder.finish(): + yield event diff --git a/sdk/nexent/core/model_service/adapters/openai_stream.py b/sdk/nexent/core/model_service/adapters/openai_stream.py new file mode 100644 index 0000000000..369c3e5fb9 --- /dev/null +++ b/sdk/nexent/core/model_service/adapters/openai_stream.py @@ -0,0 +1,235 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""One-message OpenAI-compatible stream state, independent of HTTP execution.""" + +import json +from uuid import uuid4 + +from nexent.core.model_service.adapters.openai_chat import ModelAdapterError, OpenAIChatAdapter, _usage, parse_json +from nexent.core.model_service.contracts import ModelEvent, ModelRequest, ToolCallDelta +from nexent.core.model_service.http_contracts import RawHttpResponse +from nexent.core.model_service.profiles.resolved_profile import ResolvedModelProfile +from nexent.core.model_service.sse import SSEEvent + + +class OpenAIStreamDecoder: + """Accumulate complete calls only after a declared provider completion boundary.""" + + def __init__( + self, + request: ModelRequest, + profile: ResolvedModelProfile, + *, + require_done: bool = True, + max_message_bytes: int = 32 * 1024 * 1024, + max_tool_calls: int = 128, + ): + if type(require_done) is not bool or any( + type(value) is not int or value <= 0 for value in (max_message_bytes, max_tool_calls) + ): + raise ValueError("Invalid model stream limits or completion declaration") + if not request.stream: + raise ModelAdapterError("MODEL_STREAM_REQUEST_REQUIRED") + self._request, self._profile = request, profile + self._require_done = require_done + self._limit, self._tool_limit, self._size = max_message_bytes, max_tool_calls, 0 + self._message_id = uuid4() + self._sequence = 0 + self._closed = False + self._finish = None + self._identity = {} + self._content, self._reasoning, self._refusal = [], [], [] + self._calls = {} + self._usage_raw = None + self._unknown = [] + + def _fail(self, category): + self._closed = True + raise ModelAdapterError(category) + + def _event(self, kind, **payload): + self._sequence += 1 + return ModelEvent( + request_id=self._request.request_id, + message_id=self._message_id, + sequence=self._sequence, + type=kind, + **payload, + ) + + def _complete(self): + if self._finish is None: + self._fail("MODEL_RESPONSE_INCOMPLETE") + ordered = sorted(self._calls.items()) + message = { + "role": "assistant", + "content": "".join(self._content) if self._content else None, + "reasoning_content": "".join(self._reasoning) if self._reasoning else None, + "refusal": "".join(self._refusal) if self._refusal else None, + "tool_calls": [ + { + "id": call["id"], + "type": "function", + "function": {"name": "".join(call["name"]), "arguments": "".join(call["arguments"])}, + } + for _, call in ordered + ], + } + if self._finish == "tool_calls" and not ordered: + self._fail("MODEL_TOOL_CALL_INVALID") + body = { + **self._identity, + "choices": [{"index": 0, "message": message, "finish_reason": self._finish}], + "usage": self._usage_raw, + } + raw = RawHttpResponse(200, {}, json.dumps(body, ensure_ascii=False).encode()) + complete = OpenAIChatAdapter().decode_response(raw, self._request, self._profile) + calls = [ + call.model_copy(update={"index": index, "local_call_id": state["local_id"]}) + for call, (index, state) in zip(complete.tool_calls, ordered) + ] + complete = complete.model_copy( + update={ + "message_id": self._message_id, + "tool_calls": calls, + "provider_metadata": {**complete.provider_metadata, "unmapped_delta_fields": self._unknown}, + } + ) + self._closed = True + return [self._event("message_completed", message=complete, finish_reason=complete.finish_reason)] + + def push(self, frame: SSEEvent) -> list[ModelEvent]: + if self._closed: + raise ModelAdapterError("MODEL_STREAM_CLOSED") + try: + self._size += len(frame.data.encode("utf-8")) + if self._size > self._limit: + self._fail("MODEL_RESPONSE_TOO_LARGE") + if frame.data == "[DONE]": + return self._complete() + body = parse_json(frame.data) + if isinstance(body, dict) and body.get("error") is not None: + raise ModelAdapterError( + "MODEL_PROVIDER_ERROR", response=RawHttpResponse(200, {}, frame.data.encode("utf-8")) + ) + if not isinstance(body, dict) or not isinstance(body.get("choices"), list): + self._fail("MODEL_RESPONSE_INVALID") + for key in ("id", "model"): + if key in body: + value = body[key] + if ( + not isinstance(value, str) + or not value.strip() + or key in self._identity + and self._identity[key] != value + ): + self._fail("MODEL_STREAM_IDENTITY_CONFLICT") + self._identity[key] = value + events = [] + if body.get("usage") is not None: + usage = _usage(body["usage"], self._profile.capabilities.usage.semantics_id) + if self._usage_raw is not None: + self._fail("MODEL_USAGE_CONFLICT") + self._usage_raw = body["usage"] + events.append(self._event("usage", usage=usage)) + choices = body["choices"] + if not choices: + if body.get("usage") is None: + self._fail("MODEL_RESPONSE_INVALID") + return events + if len(choices) != 1: + self._fail("MODEL_RESPONSE_INVALID") + choice = choices[0] + if not isinstance(choice, dict) or type(choice.get("index")) is not int or choice["index"] != 0: + self._fail("MODEL_RESPONSE_INVALID") + delta = choice.get("delta") + if not isinstance(delta, dict) or delta.get("role", "assistant") != "assistant": + self._fail("MODEL_RESPONSE_INVALID") + if self._finish is not None: + self._fail("MODEL_STREAM_AFTER_FINISH") + for key, target, kind in ( + ("reasoning_content", self._reasoning, "reasoning_delta"), + ("content", self._content, "text_delta"), + ("refusal", self._refusal, "text_delta"), + ): + value = delta.get(key) + if value is None: + continue + if not isinstance(value, str): + self._fail("MODEL_RESPONSE_INVALID") + if value: + target.append(value) + payload = {"delta": value} + if kind == "text_delta": + payload["content_kind"] = "refusal" if key == "refusal" else "text" + events.append(self._event(kind, **payload)) + raw_calls = delta.get("tool_calls") + if raw_calls is not None: + if not isinstance(raw_calls, list): + self._fail("MODEL_TOOL_CALL_INVALID") + for raw_call in raw_calls: + events.append(self._call_delta(raw_call)) + unknown = { + key: value + for key, value in delta.items() + if key not in {"role", "content", "reasoning_content", "refusal", "tool_calls"} + } + if unknown: + self._unknown.append(unknown) + finish = choice.get("finish_reason") + if finish is not None: + if not isinstance(finish, str) or not finish.strip(): + self._fail("MODEL_RESPONSE_INVALID") + self._finish = finish + return events + except ModelAdapterError: + self._closed = True + raise + except (ValueError, TypeError, UnicodeError, RecursionError): + self._fail("MODEL_RESPONSE_INVALID") + + def _call_delta(self, raw): + if not isinstance(raw, dict) or type(raw.get("index")) is not int or raw["index"] < 0: + self._fail("MODEL_TOOL_CALL_INVALID") + index = raw["index"] + if index not in self._calls: + if len(self._calls) >= self._tool_limit: + self._fail("MODEL_TOOL_CALL_LIMIT") + self._calls[index] = {"local_id": uuid4(), "id": None, "name": [], "arguments": []} + state = self._calls[index] + if "type" in raw and raw["type"] != "function": + self._fail("MODEL_TOOL_CALL_INVALID") + if raw.get("id") is not None: + value = raw["id"] + if not isinstance(value, str) or not value.strip() or state["id"] is not None and state["id"] != value: + self._fail("MODEL_STREAM_IDENTITY_CONFLICT") + state["id"] = value + function = raw.get("function", {}) + if not isinstance(function, dict): + self._fail("MODEL_TOOL_CALL_INVALID") + for key in ("name", "arguments"): + if key in function: + if not isinstance(function[key], str): + self._fail("MODEL_TOOL_CALL_INVALID") + state[key].append(function[key]) + return self._event( + "tool_call_delta", + tool_call_delta=ToolCallDelta( + local_call_id=state["local_id"], + index=index, + provider_call_id=state["id"], + name_delta=function.get("name"), + arguments_delta=function.get("arguments"), + ), + ) + + def finish(self) -> list[ModelEvent]: + if self._closed: + return [] + if self._require_done: + self._fail("MODEL_RESPONSE_INCOMPLETE") + try: + return self._complete() + except ModelAdapterError: + self._closed = True + raise diff --git a/sdk/nexent/core/model_service/errors.py b/sdk/nexent/core/model_service/errors.py new file mode 100644 index 0000000000..114046a95b --- /dev/null +++ b/sdk/nexent/core/model_service/errors.py @@ -0,0 +1,95 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Fixed error classification; provider text remains private to the adapter.""" + +from nexent.core.model_service.adapters.openai_chat import ModelAdapterError, parse_json +from nexent.core.model_service.contracts import ModelError +from nexent.core.model_service.http_contracts import ModelTransportError + + +_CATEGORIES = { + "MODEL_AUTHENTICATION_ERROR": (False, "Model service authentication failed."), + "MODEL_NOT_FOUND": (False, "The requested model service was not found."), + "MODEL_RATE_LIMITED": (True, "The model service is rate limited."), + "MODEL_SERVICE_UNAVAILABLE": (True, "The model service is temporarily unavailable."), + "MODEL_REQUEST_INVALID": (False, "The model service rejected the request."), + "MODEL_CONTEXT_EXCEEDED": (False, "The model service context limit was exceeded."), + "MODEL_QUOTA_EXHAUSTED": (False, "The model service quota is exhausted."), + "MODEL_TIMEOUT": (True, "The model request timed out."), + "MODEL_NETWORK_ERROR": (True, "The model service connection failed."), + "MODEL_TLS_ERROR": (False, "The model service TLS connection failed."), + "MODEL_HTTP_ERROR": (False, "The model service returned an unsupported HTTP status."), + "MODEL_PROVIDER_ERROR": (False, "The model service returned a provider error."), + "MODEL_CREDENTIAL_ERROR": (False, "Model service credentials could not be resolved."), + "MODEL_PROTOCOL_ERROR": (False, "The model response could not be processed."), + "MODEL_CONCURRENCY_TIMEOUT": (False, "Model concurrency admission timed out."), + "MODEL_RUN_DEADLINE": (False, "The model run time budget was exhausted."), +} +_PROVIDER_CODES = { + "context_length_exceeded": "MODEL_CONTEXT_EXCEEDED", + "invalid_api_key": "MODEL_AUTHENTICATION_ERROR", + "insufficient_quota": "MODEL_QUOTA_EXHAUSTED", + "model_not_found": "MODEL_NOT_FOUND", + "rate_limit_exceeded": "MODEL_RATE_LIMITED", +} + + +def normalize_model_error( + error, *, request_id: str, attempt: int = 1, max_attempts: int = 1, will_retry: bool = False +) -> ModelError: + """Classify trusted error types and known codes, never arbitrary error messages.""" + category, scope, provider_code = "MODEL_PROTOCOL_ERROR", "protocol", None + if isinstance(error, ModelTransportError): + scope = "transport" + category = { + "HTTP_TIMEOUT": "MODEL_TIMEOUT", + "HTTP_NETWORK_ERROR": "MODEL_NETWORK_ERROR", + "HTTP_TLS_ERROR": "MODEL_TLS_ERROR", + "CREDENTIAL_RESOLUTION_FAILED": "MODEL_CREDENTIAL_ERROR", + "CREDENTIAL_HEADER_CONFLICT": "MODEL_CREDENTIAL_ERROR", + "HTTP_CONCURRENCY_TIMEOUT": "MODEL_CONCURRENCY_TIMEOUT", + "HTTP_RUN_DEADLINE": "MODEL_RUN_DEADLINE", + }.get(error.category, "MODEL_PROTOCOL_ERROR") + response = ( + getattr(error, "response", None) if isinstance(error, (ModelTransportError, ModelAdapterError)) else None + ) + if response is not None: + scope = "provider" + status = response.status_code + category = "MODEL_PROVIDER_ERROR" if 200 <= status < 300 else "MODEL_HTTP_ERROR" + if status in (401, 403): + category = "MODEL_AUTHENTICATION_ERROR" + elif status == 404: + category = "MODEL_NOT_FOUND" + elif status == 429: + category = "MODEL_RATE_LIMITED" + elif 500 <= status < 600: + category = "MODEL_SERVICE_UNAVAILABLE" + elif status in (400, 422): + category = "MODEL_REQUEST_INVALID" + try: + body = parse_json(response.body.decode("utf-8")) + code = ( + body.get("error", {}).get("code") + if isinstance(body, dict) and isinstance(body.get("error"), dict) + else None + ) + if isinstance(code, str) and code in _PROVIDER_CODES: + provider_code = code + mapped = _PROVIDER_CODES[code] + if mapped != "MODEL_RATE_LIMITED" or status == 429 or 500 <= status < 600 or 200 <= status < 300: + category = mapped + except (ValueError, TypeError, UnicodeError, RecursionError): + pass + retryable, message = _CATEGORIES[category] + return ModelError( + category=category, + scope=scope, + retryable=retryable, + will_retry=will_retry and retryable and attempt < max_attempts, + attempt=attempt, + max_attempts=max_attempts, + provider_code=provider_code, + safe_message=message, + request_id=request_id, + ) diff --git a/sdk/nexent/core/model_service/profiles/_models_dev/inheritance.py b/sdk/nexent/core/model_service/profiles/_models_dev/inheritance.py new file mode 100644 index 0000000000..76391c5938 --- /dev/null +++ b/sdk/nexent/core/model_service/profiles/_models_dev/inheritance.py @@ -0,0 +1,138 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Pinned authored base_model inheritance with per-field source ownership.""" + +from dataclasses import dataclass +from typing import Mapping + +from nexent.core.model_service.profiles._models_dev.reader import SourceDocument, SourceReadError, _freeze + + +_EXCLUDED_BASE_FIELDS = frozenset({"id", "license", "links", "weights", "benchmarks"}) + + +@dataclass(frozen=True) +class FieldOrigin: + """Source ownership before the versioned semantic mapping is applied.""" + + source_kind: str + file_path: str + source_path: str + content_sha256: str + + +@dataclass(frozen=True) +class EffectiveSourceModel: + """Read-only source result, distinct from a resolved Nexent outlet.""" + + document: SourceDocument + data: Mapping[str, object] + field_origins: Mapping[str, FieldOrigin] + dependencies: tuple[SourceDocument, ...] + omitted_paths: tuple[str, ...] + + +def _error(category: str, document: SourceDocument, source_path: str) -> SourceReadError: + error = SourceReadError(category, document.file_path, document.kind) + error.source_path = source_path + error.rule_id = "source_inheritance_v1" + return error + + +def _thaw(value: object) -> object: + if isinstance(value, Mapping): + return {key: _thaw(item) for key, item in value.items()} + if isinstance(value, (tuple, list)): + return [_thaw(item) for item in value] + return value + + +def _collect_origins(value: object, path: str, document: SourceDocument, origins: dict) -> None: + if isinstance(value, Mapping) and value: + for key, item in value.items(): + _collect_origins(item, f"{path}.{key}" if path else key, document, origins) + elif isinstance(value, (tuple, list)) and value: + for index, item in enumerate(value): + _collect_origins(item, f"{path}[{index}]", document, origins) + elif path: + origins[path] = FieldOrigin(document.kind, document.file_path, path, document.content_sha256) + + +def _clear_origins(origins: dict, path: str) -> None: + for key in tuple(origins): + if key == path or key.startswith(path + ".") or key.startswith(path + "["): + del origins[key] + + +def _overlay(target: dict, updates: Mapping, path: str, document: SourceDocument, origins: dict) -> None: + for key, value in updates.items(): + field_path = f"{path}.{key}" if path else key + if isinstance(target.get(key), dict) and isinstance(value, Mapping): + if value: + origins.pop(field_path, None) + _overlay(target[key], value, field_path, document, origins) + else: + _clear_origins(origins, field_path) + target[key] = _thaw(value) + _collect_origins(value, field_path, document, origins) + + +def _omit(target: dict, path: str, origins: dict) -> None: + parts = path.split(".") + current = target + parents = [] + for part in parts[:-1]: + child = current.get(part) + if not isinstance(child, dict): + return + parents.append((current, part)) + current = child + if parts[-1] not in current: + return + del current[parts[-1]] + _clear_origins(origins, path) + for parent, key in reversed(parents): + if parent[key]: + break + del parent[key] + + +def resolve_source_model(document: SourceDocument, sources: tuple[SourceDocument, ...]) -> EffectiveSourceModel: + """Use only explicit base references; never infer a model from its name.""" + if document.kind not in ("model", "provider_model"): + raise _error("SOURCE_MODEL_KIND_INVALID", document, "") + models = {} + for source in sources: + if source.kind == "model": + if source.key in models: + raise _error("SOURCE_DUPLICATE", source, "") + models[source.key] = source + + origins = {} + dependencies = (document,) + data = {} + if "base_model" in document.data: + base_key = document.data["base_model"] + if document.kind != "provider_model" or not isinstance(base_key, str) or not base_key.strip(): + raise _error("SOURCE_BASE_REFERENCE_INVALID", document, "base_model") + base = models.get(base_key) + if base is None: + raise _error("SOURCE_BASE_MODEL_MISSING", document, "base_model") + if "base_model" in base.data: + raise _error("SOURCE_BASE_REFERENCE_INVALID", base, "base_model") + data = _thaw({key: value for key, value in base.data.items() if key not in _EXCLUDED_BASE_FIELDS}) + _collect_origins(data, "", base, origins) + dependencies = (base, document) + elif "base_model_omit" in document.data: + raise _error("SOURCE_OMIT_WITHOUT_BASE", document, "base_model_omit") + + omissions = document.data.get("base_model_omit", ()) + if not isinstance(omissions, (tuple, list)) or any( + not isinstance(path, str) or any(not part for part in path.split(".")) for path in omissions + ): + raise _error("SOURCE_OMIT_INVALID", document, "base_model_omit") + updates = {key: value for key, value in document.data.items() if key not in ("base_model", "base_model_omit")} + _overlay(data, updates, "", document, origins) + for path in omissions: + _omit(data, path, origins) + return EffectiveSourceModel(document, _freeze(data), _freeze(origins), dependencies, tuple(omissions)) diff --git a/sdk/nexent/core/model_service/profiles/_models_dev/mapping.py b/sdk/nexent/core/model_service/profiles/_models_dev/mapping.py new file mode 100644 index 0000000000..2149319f4d --- /dev/null +++ b/sdk/nexent/core/model_service/profiles/_models_dev/mapping.py @@ -0,0 +1,143 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Code-owned source conversion; submission checks live outside the runtime path.""" + +import hashlib +import json +import re +from dataclasses import asdict, dataclass +from typing import Mapping + +from nexent.core.model_service.profiles._models_dev.inheritance import EffectiveSourceModel, FieldOrigin +from nexent.core.model_service.profiles._models_dev.reader import SourceDocument, _freeze + + +@dataclass(frozen=True) +class SourceMappingRule: + source_kind: str + source_path: str + target_path: str + transform_ref: str = "identity_v1" + missing_policy: str = "omit" + + @property + def rule_id(self): + return self.source_kind + "." + self.source_path + + rule_version = 1 + + +@dataclass(frozen=True) +class SourceMappingRegistry: + rules: tuple[SourceMappingRule, ...] + mapping_id: str = "models_dev_source_v1" + mapping_version: int = 1 + + +# Common model facts are shared by base and hosted model definitions. +_COMMON = ( + ('name', '/identity/name'), + ('description', '/identity/description'), + ('family', '/identity/family'), + ('attachment', '/reference_metadata/attachment'), + ('reasoning', '/reasoning/status', 'boolean_capability_v1', 'unknown'), + ('tool_call', '/tools/status', 'boolean_capability_v1', 'unknown'), + ('structured_output', '/structured_output/status', 'boolean_capability_v1', 'unknown'), + ('temperature', '/sampling/temperature/status', 'boolean_capability_v1', 'unknown'), + ('open_weights', '/reference_metadata/open_weights'), + ('knowledge', '/reference_metadata/knowledge'), + ('release_date', '/reference_metadata/release_date'), + ('last_updated', '/reference_metadata/last_updated'), + ('modalities', '/modalities'), + ('modalities.input', '/modalities/input'), + ('modalities.input[]', '/modalities/input[]'), + ('modalities.output', '/modalities/output'), + ('modalities.output[]', '/modalities/output[]'), + ('limit', '/reference_metadata/limit'), + ('limit.context', '/context/combined_window', 'zero_unknown_tokens_v1', 'unknown'), + ('limit.input', '/context/input_limit', 'zero_unknown_tokens_v1', 'unknown'), + ('limit.output', '/context/output_limit', 'zero_unknown_tokens_v1', 'unknown'), +) + + +def _rules(): + for kind in ("model", "provider_model"): + for fields in _COMMON: + yield SourceMappingRule(kind, *fields) + yield SourceMappingRule('provider', 'name', '/identity/name') + yield SourceMappingRule('provider', 'api', '/connection_suggestions/base_url') + yield SourceMappingRule('provider_model', 'reasoning_options', '/reasoning/control_candidates') + yield SourceMappingRule('provider_model', 'reasoning_options[]', '/reasoning/control_candidates[]') + yield SourceMappingRule('provider_model', 'reasoning_options[].type', '/reasoning/control_candidates[]/type') + yield SourceMappingRule('provider_model', 'reasoning_options[].values', '/reasoning/control_candidates[]/values') + yield SourceMappingRule('provider_model', 'reasoning_options[].values[]', '/reasoning/control_candidates[]/values[]') + yield SourceMappingRule('provider_model', 'reasoning_options[].min', '/reasoning/control_candidates[]/min') + yield SourceMappingRule('provider_model', 'reasoning_options[].max', '/reasoning/control_candidates[]/max') + + +SOURCE_MAPPING = SourceMappingRegistry(tuple(_rules())) + + +def mapping_fingerprint(registry): + return hashlib.sha256(json.dumps(asdict(registry), sort_keys=True).encode()).hexdigest() + + +@dataclass(frozen=True) +class MappedFieldOrigin: + source: FieldOrigin + rule_id: str + rule_version: int + target_path: str + + +@dataclass(frozen=True) +class MappedSourceFields: + """Canonical fields before supplements and runtime profile composition.""" + + source: EffectiveSourceModel | SourceDocument + mapping_id: str + mapping_version: int + values: Mapping[str, object] + field_origins: Mapping[str, MappedFieldOrigin] + + + +def _leaves(value: object, path: str = ""): + if isinstance(value, Mapping) and value: + for key, item in value.items(): + yield from _leaves(item, f"{path}.{key}" if path else key) + elif isinstance(value, (tuple, list)) and value: + for index, item in enumerate(value): + yield from _leaves(item, f"{path}[{index}]") + elif path: + yield path, value + + + +def map_source_fields(source, registry=SOURCE_MAPPING) -> MappedSourceFields: + """Convert checked source facts; unconsumed fields stay isolated as reference data.""" + document = source.document if isinstance(source, EffectiveSourceModel) else source + rules = {rule.source_path: rule for rule in registry.rules if rule.source_kind == document.kind} + values, origins = {}, {} + for path, value in _leaves(source.data): + pattern = re.sub(r"\[\d+\]", "[]", path) + rule = rules.get(pattern) + if rule is None: + target = "/reference_metadata/" + re.sub(r"\[(\d+)\]", r"/\1", path).replace(".", "/") + rule = SourceMappingRule(document.kind, pattern, target) + indexes = iter(re.findall(r"\[(\d+)\]", path)) + target = re.sub(r"\[\]", lambda _: "/" + next(indexes), rule.target_path) + if rule.transform_ref == "boolean_capability_v1": + value = "supported" if value else "unsupported" + elif rule.transform_ref == "zero_unknown_tokens_v1" and value == 0: + value = None + values[target] = value + origins[target] = MappedFieldOrigin( + source.field_origins[path] if isinstance(source, EffectiveSourceModel) + else FieldOrigin(document.kind, document.file_path, path, document.content_sha256), + rule.rule_id, rule.rule_version, target, + ) + for rule in rules.values(): + if "[]" not in rule.source_path and rule.missing_policy == "unknown" and rule.target_path not in values: + values[rule.target_path] = "unknown" if rule.transform_ref == "boolean_capability_v1" else None + return MappedSourceFields(source, registry.mapping_id, registry.mapping_version, _freeze(values), _freeze(origins)) diff --git a/sdk/nexent/core/model_service/profiles/composition.py b/sdk/nexent/core/model_service/profiles/composition.py new file mode 100644 index 0000000000..be5a8c066a --- /dev/null +++ b/sdk/nexent/core/model_service/profiles/composition.py @@ -0,0 +1,344 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Immutable atomic views and exact provider-model policy composition.""" + +import json +from dataclasses import dataclass +from typing import Mapping, Sequence + +from pydantic import ValidationError + +from nexent.core.model_service.profiles._models_dev.reader import _freeze +from nexent.core.model_service.profiles.contracts import ProfileCapabilities, VersionedReference +from nexent.core.model_service.profiles.instruction_mapping import resolve_instruction_mapping +from nexent.core.model_service.profiles.parameter_bindings import validate_parameter_bindings +from nexent.core.model_service.profiles.parameter_rules import ( + CompiledParameterRules, + apply_parameter_values, + compile_parameter_rules, +) +from nexent.core.model_service.profiles.parameter_schema import ( + ParameterValidationError, + _json_copy, + compile_parameter_schema, + json_values_equal, + validate_parameter_value, +) +from nexent.core.model_service.profiles.resolver import SourceCatalog, SourceCatalogEntry, _revision +from nexent.core.model_service.profiles.supplements import SupplementCatalog + + +COMPOSER_REF = "profile_policy_composer_v6" +_CAPABILITY_PATHS = frozenset( + f"/{group}/{field}" + for group, fields in { + "context": ("combined_window", "input_limit", "output_limit"), + "reasoning": ("status", "controls", "response_paths"), + "tools": ("status", "schema_constraints", "result_format"), + "sampling": ("temperature", "top_p"), + "structured_output": ("status",), + "modalities": ("input", "output"), + "response": ("supported_finish_reasons", "stream_format"), + "usage": ("semantics_id", "input_includes_cached", "output_includes_reasoning"), + }.items() + for field in fields +) + + +class ProfileCompositionError(ValueError): + """A conflict diagnostic never includes parameter or credential values.""" + + def __init__(self, category: str, path: str = ""): + self.category, self.path = category, path + super().__init__(f"{category}: {path}") + + +@dataclass(frozen=True) +class AtomicProfileDefinition: + kind: str + key: str + source: SourceCatalogEntry + capabilities: Mapping[str, object] + parameters: tuple[Mapping[str, object], ...] + bindings: Mapping[str, object] + instruction_reference: VersionedReference | None + model_profile_key: str | None + aliases: tuple[str, ...] + endpoint_hints: tuple[str, ...] + overrides: tuple[Mapping[str, object], ...] + revision: str + + +@dataclass(frozen=True) +class ProfileCatalog: + revision: str + models: Mapping[str, AtomicProfileDefinition] + providers: Mapping[str, AtomicProfileDefinition] + provider_models: Mapping[str, AtomicProfileDefinition] + + +@dataclass(frozen=True) +class ProviderModelPolicy: + """A validated policy draft, not a saved outlet or a transport implementation.""" + + request_model_id: str + model_profile_key: str | None + provider_key: str | None + capabilities: Mapping[str, object] + instruction_mapping: Mapping[str, object] + bindings: Mapping[str, object] + parameter_rules: CompiledParameterRules + parameter_bindings: Mapping[str, Mapping[str, object]] + parameter_values: Mapping[str, object] + overrides: tuple[Mapping[str, object], ...] + catalog_revision: str + revision: str + + +def _check_profile_size(document: Mapping) -> None: + try: + encoded = json.dumps(_json_copy(document), ensure_ascii=False, allow_nan=False, separators=(",", ":")).encode( + "utf-8" + ) + except (ValueError, TypeError, RecursionError, UnicodeError): + raise ProfileCompositionError("PROFILE_JSON_INVALID") from None + if len(encoded) > 1024 * 1024: + raise ProfileCompositionError("PROFILE_SIZE_LIMIT") + + +def _capabilities(value: Mapping) -> dict: + try: + return ProfileCapabilities.model_validate(_json_copy(value)).model_dump(mode="json") + except ValidationError: + raise ProfileCompositionError("PROFILE_CAPABILITIES_INVALID") from None + + +def _apply_override(capabilities: dict, override: Mapping, source: str) -> dict: + path = override["path"] + if path not in _CAPABILITY_PATHS: + raise ProfileCompositionError("PROFILE_OVERRIDE_PATH_INVALID", path) + group, key = path[1:].split("/") + previous = capabilities[group][key] + capabilities[group][key] = _json_copy(override["value"]) + return { + "path": path, + "source": source, + "original_value": previous, + "reason": override["reason"], + "evidence": override["evidence"], + } + + +def _merge_capability_facts(target: dict, facts: Mapping) -> None: + """Overlay authored source facts only; absent fields preserve the base.""" + for key, value in facts.items(): + if isinstance(value, Mapping) and isinstance(target.get(key), dict): + _merge_capability_facts(target[key], value) + else: + target[key] = _json_copy(value) + + +def _compose_declarations(base: Mapping, *layers: Mapping) -> dict: + """Overlay complete declarations in explicit model/provider/combination order.""" + result = _json_copy(base) + for declarations in layers: + result.update(_json_copy(declarations)) + return result + + +def build_profile_catalog(sources: SourceCatalog, supplements: SupplementCatalog) -> ProfileCatalog: + """Publish all atomic views together; unsupported bindings are checked at selection.""" + if supplements.source_revision != sources.revision: + raise ProfileCompositionError("PROFILE_SOURCE_REVISION_MISMATCH") + rules = {(entry.rule["target_kind"], entry.rule["target_key"]): entry for entry in supplements.entries.values()} + collections = {"model": {}, "provider": {}, "provider_model": {}} + for kind, entries in ( + ("model", sources.models), + ("provider", sources.providers), + ("provider_model", sources.provider_models), + ): + for key, source in entries.items(): + supplement = rules.get((kind, key)) + rule = supplement.rule if supplement else {} + capabilities = _capabilities(source.capabilities) + overrides, seen = [], {} + for override in rule.get("capability_overrides", ()): + path = override["path"] + if path in seen: + if not json_values_equal(seen[path], override["value"]): + raise ProfileCompositionError("PROFILE_CONFLICT", path) + continue + seen[path] = override["value"] + overrides.append(_apply_override(capabilities, override, supplement.file_path)) + capabilities = _capabilities(capabilities) + reference = rule.get("instruction_mapping") + reference = VersionedReference.model_validate(_json_copy(reference)) if reference else None + parts = { + "composer": COMPOSER_REF, + "source": source.profile_revision, + "supplement_schema": supplements.schema_ref, + } + if supplement: + parts[supplement.file_path] = supplement.content_sha256 + _check_profile_size( + { + "identity": source.identity, + "capabilities": capabilities, + "reference_metadata": source.reference_metadata, + "connection_suggestions": source.connection_suggestions, + "supplement": rule, + } + ) + collections[kind][key] = AtomicProfileDefinition( + kind, + key, + source, + _freeze(capabilities), + _freeze(_json_copy(rule.get("parameters", ()))), + _freeze(_json_copy(rule.get("bindings", {}))), + reference, + rule.get("model_profile_key"), + tuple(rule.get("aliases", ())), + tuple(rule.get("endpoint_hints", ())), + _freeze(overrides), + _revision(parts), + ) + return ProfileCatalog( + _revision({"source": sources.revision, "supplements": supplements.revision, "composer": COMPOSER_REF}), + _freeze(collections["model"]), + _freeze(collections["provider"]), + _freeze(collections["provider_model"]), + ) + + +def compose_provider_model_policy( + catalog: ProfileCatalog, + *, + provider_key: str | None, + request_model_id: str, + model_profile_key: str | None, + base_bindings: Mapping[str, object], + base_instruction: VersionedReference, + registered_references: Mapping[str, Sequence[VersionedReference]], + submitted_values: Mapping[str, object], + protected_api_paths: Sequence[str] = (), + protocol_parameter_rules: Sequence[Mapping[str, object]] = (), +) -> ProviderModelPolicy: + """Compose exact selections against a trusted implementation-availability snapshot.""" + if not isinstance(request_model_id, str) or not request_model_id.strip() or len(request_model_id) > 100: + raise ProfileCompositionError("REQUEST_MODEL_ID_INVALID") + provider = catalog.providers.get(provider_key) if provider_key is not None else None + if provider_key is not None and provider is None: + raise ProfileCompositionError("PROVIDER_PROFILE_NOT_FOUND") + combo = catalog.provider_models.get(f"{provider_key}/{request_model_id}") if provider else None + linked_model = combo.model_profile_key if combo else None + if model_profile_key is not None and linked_model is not None and model_profile_key != linked_model: + raise ProfileCompositionError("PROFILE_CONFLICT", "/selection/model_profile_key") + model_profile_key = model_profile_key if model_profile_key is not None else linked_model + model = catalog.models.get(model_profile_key) if model_profile_key is not None else None + if model_profile_key is not None and model is None: + raise ProfileCompositionError("MODEL_PROFILE_NOT_FOUND") + # Complete the upstream baseline before applying any Nexent experience rule. + capabilities = _capabilities({}) + for atom in (model, provider, combo): + if atom is not None: + _merge_capability_facts(capabilities, atom.source.capabilities) + overrides = [] + parameters = {} + for atom in (model, provider, combo): + if atom is None: + continue + for override in atom.overrides: + group, field = override["path"][1:].split("/") + overrides.append( + _apply_override( + capabilities, + {**override, "value": atom.capabilities[group][field]}, + override["source"], + ) + ) + for parameter in atom.parameters: + parameters[parameter["key"]] = _json_copy(parameter) + capabilities = _capabilities(capabilities) + for rule in protocol_parameter_rules: + declaration = _json_copy(rule) + parameters.setdefault(rule["key"], declaration) + compiled = compile_parameter_rules(list(parameters.values())) + values = apply_parameter_values(compiled, submitted_values, scope="outlet", saved_values={}) + atomic_bindings = [atom.bindings if atom else {} for atom in (model, provider)] + explicit_bindings = combo.bindings if combo else {} + bindings = _compose_declarations(base_bindings, *atomic_bindings, explicit_bindings) + instruction_declarations = [ + {"strategy": atom.instruction_reference.model_dump(mode="json")} if atom and atom.instruction_reference else {} + for atom in (model, provider, combo) + ] + instruction = VersionedReference.model_validate( + _compose_declarations( + {"strategy": base_instruction.model_dump(mode="json")}, + *instruction_declarations, + )["strategy"] + ) + for name in ("protocol", "auth", "adapter", "code_converter", "native_converter"): + if name not in bindings: + raise ProfileCompositionError("PROFILE_BINDING_MISSING", "/bindings/" + name) + try: + reference = VersionedReference.model_validate(bindings[name]) + except ValidationError: + raise ProfileCompositionError("PROFILE_BINDING_INVALID", "/bindings/" + name) from None + if reference not in registered_references.get(name, ()): + raise ProfileCompositionError("PROFILE_BINDING_UNREGISTERED", "/bindings/" + name) + parameter_bindings = validate_parameter_bindings( + compiled, + registered_references.get("parameter_transform", ()), + protected_paths=protected_api_paths, + ) + parameter_schemas = {rule["key"]: compile_parameter_schema(rule["schema"]) for rule in compiled.declarations} + for control in capabilities["reasoning"]["controls"] or (): + if control["parameter_key"] not in parameter_bindings: + raise ProfileCompositionError("REASONING_PARAMETER_BINDING_MISSING", "/capabilities/reasoning/controls") + candidates = list(control.get("values", ())) + list(control.get("special_values", ())) + for bound in ("min", "max"): + if control.get(bound) is not None: + candidates.append(control[bound]) + try: + for candidate in candidates: + validate_parameter_value(parameter_schemas[control["parameter_key"]], candidate) + except ParameterValidationError: + raise ProfileCompositionError( + "REASONING_PARAMETER_SCHEMA_CONFLICT", "/capabilities/reasoning/controls" + ) from None + mapping = resolve_instruction_mapping(instruction).model_dump(mode="json") + _check_profile_size( + { + "request_model_id": request_model_id, + "capabilities": capabilities, + "instruction_mapping": mapping, + "bindings": bindings, + "parameter_rules": compiled.declarations, + "parameter_values": values, + "overrides": overrides, + } + ) + parts = { + "composer": COMPOSER_REF, + "request_model_id": request_model_id, + "bindings": json.dumps(bindings, sort_keys=True), + "instruction": instruction.model_dump_json(), + } + for atom in (model, provider, combo): + if atom: + parts[f"{atom.kind}:{atom.key}"] = atom.revision + return ProviderModelPolicy( + request_model_id, + model_profile_key, + provider_key, + _freeze(capabilities), + _freeze(mapping), + _freeze(bindings), + compiled, + parameter_bindings, + _freeze(values), + _freeze(overrides), + catalog.revision, + _revision(parts), + ) diff --git a/sdk/nexent/core/model_service/profiles/resolver.py b/sdk/nexent/core/model_service/profiles/resolver.py new file mode 100644 index 0000000000..5c639d359a --- /dev/null +++ b/sdk/nexent/core/model_service/profiles/resolver.py @@ -0,0 +1,286 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Atomic loading of pinned source profiles before Nexent supplements and API bindings.""" + +import hashlib +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Mapping + +from pydantic import ValidationError + +from nexent.core.model_service.profiles._models_dev.inheritance import EffectiveSourceModel, resolve_source_model +from nexent.core.model_service.profiles._models_dev.mapping import ( + MappedFieldOrigin, + map_source_fields, + SOURCE_MAPPING, + mapping_fingerprint, +) +from nexent.core.model_service.profiles._models_dev.reader import ( + SourceDocument, + SourceReadError, + _freeze, + read_inventory, +) +from nexent.core.model_service.profiles.contracts import ( + AtomicProfileReference, + ContextCapabilities, + ReasoningCapabilities, +) + + +PARSER_REF = "models_dev_source_catalog_v3" +SUPPORTED_UPSTREAM_REVISION = "e69e1844210e5cfe7c7e7f68d57328bc7451a51c" + + +@dataclass(frozen=True) +class CatalogFieldOrigin: + mapping: MappedFieldOrigin + target_path: str + normalization_ref: str = PARSER_REF + + +@dataclass(frozen=True) +class SourceCatalogEntry: + """Source-only declaration, not a complete ModelProfile or usable outlet.""" + + kind: str + key: str + identity: Mapping[str, object] + capabilities: Mapping[str, object] + reference_metadata: Mapping[str, object] + connection_suggestions: Mapping[str, object] + field_origins: Mapping[str, CatalogFieldOrigin] + source_refs: tuple[AtomicProfileReference, ...] + profile_revision: str + + +@dataclass(frozen=True) +class SourceCatalog: + revision: str + upstream_revision: str + parser_ref: str + mapping_id: str + mapping_version: int + models: Mapping[str, SourceCatalogEntry] + providers: Mapping[str, SourceCatalogEntry] + provider_models: Mapping[str, SourceCatalogEntry] + + +def _error(category: str, document: SourceDocument, path: str = "") -> SourceReadError: + error = SourceReadError(category, document.file_path, document.kind) + error.source_path = path + error.rule_id = PARSER_REF + return error + + +def _revision(parts: Mapping[str, str]) -> str: + digest = hashlib.sha256() + for key, value in sorted(parts.items()): + for item in (key, value): + data = item.encode("utf-8") + digest.update(len(data).to_bytes(8, "big")) + digest.update(data) + return digest.hexdigest() + + +def _tree(values: Mapping[str, object]) -> dict: + """Materialize canonical pointer fields, preserving explicit arrays and nulls.""" + root = {} + for pointer, value in values.items(): + parts = pointer.split("/")[1:] + current = root + for index, part in enumerate(parts): + last = index == len(parts) - 1 + key = int(part) if isinstance(current, list) else part + if isinstance(current, list): + while len(current) <= key: + current.append(None) + if last: + current[key] = value + else: + next_value = [] if parts[index + 1].isdigit() else {} + if isinstance(current, list): + if current[key] is None: + current[key] = next_value + elif key not in current: + current[key] = next_value + current = current[key] + return root + + +def _controls(reasoning: dict) -> list | None: + if "control_candidates" not in reasoning: + return None + controls = [] + for option in reasoning["control_candidates"]: + mode = option["type"] + if mode == "toggle": + controls.append({"mode": "toggle", "values": [True, False]}) + elif mode == "effort": + controls.append( + {"mode": "effort", "values": [None if value == "null" else value for value in option["values"]]} + ) + else: + lower = option.get("min") + controls.append( + { + "mode": "budget_tokens", + "min": None if lower == -1 else lower, + "max": option.get("max"), + "special_values": [-1] if lower == -1 else [], + } + ) + return controls + + +def _normalized_origins(mapped, controls) -> dict: + origins = {} + for path, origin in mapped.field_origins.items(): + target = path + if path.startswith("/reasoning/control_candidates"): + target = path.replace("/reasoning/control_candidates", "/capabilities/reasoning/controls", 1) + target = re.sub(r"/type$", "/mode", target) + if path.endswith("/min") and mapped.values[path] == -1: + target = target[: -len("min")] + "special_values/0" + elif path.split("/")[1] in {"context", "reasoning", "tools", "sampling", "structured_output", "modalities"}: + target = "/capabilities" + path + origins[target] = CatalogFieldOrigin(origin, target) + for index, control in enumerate(controls or ()): + prefix = f"/capabilities/reasoning/controls/{index}" + type_origin = origins[prefix + "/mode"].mapping + origins[prefix + "/parameter_key"] = CatalogFieldOrigin(type_origin, prefix + "/parameter_key") + if control["mode"] == "toggle": + for value_index in range(2): + path = prefix + f"/values/{value_index}" + origins[path] = CatalogFieldOrigin(type_origin, path) + return origins + + +def _entry( + document: SourceDocument, source, registry, providers: Mapping, revision: str, version_parts: dict +) -> SourceCatalogEntry: + mapped = map_source_fields(source, registry) + tree = _tree(mapped.values) + unexpected = set(tree) - { + "identity", + "context", + "reasoning", + "tools", + "sampling", + "structured_output", + "modalities", + "reference_metadata", + "connection_suggestions", + } + if unexpected: + raise _error("SOURCE_TARGET_UNHANDLED", document) + reasoning = tree.get("reasoning", {}) + controls = _controls(reasoning) + try: + context = ContextCapabilities.model_validate(tree.get("context", {})) + except ValidationError: + raise _error("SOURCE_NORMALIZATION_INVALID", document, "limit") from None + try: + thinking = ReasoningCapabilities(status=reasoning.get("status", "unknown"), controls=controls) + except ValidationError: + raise _error("SOURCE_NORMALIZATION_INVALID", document, "reasoning_options") from None + capabilities = { + "context": context.model_dump(), + "reasoning": thinking.model_dump(), + "tools": tree.get("tools", {"status": "unknown"}), + "sampling": tree.get("sampling", {}), + "structured_output": tree.get("structured_output", {"status": "unknown"}), + "modalities": tree.get("modalities", {"input": None, "output": None}), + } + if document.kind == "provider": + capabilities = {} + dependencies = source.dependencies if isinstance(source, EffectiveSourceModel) else (document,) + if document.kind == "provider_model": + provider_key = document.key.split("/", 1)[0] + if provider_key not in providers: + raise _error("SOURCE_PROVIDER_MISSING", document) + dependencies = (*dependencies, providers[provider_key]) + profile_parts = dict(version_parts) + refs = [] + for dependency in dependencies: + profile_parts[dependency.file_path] = dependency.content_sha256 + refs.append( + AtomicProfileReference( + key=dependency.key, + catalog_revision=dependency.source_revision or revision, + file_path=dependency.file_path, + content_sha256=dependency.content_sha256, + ) + ) + metadata = dict(tree.get("reference_metadata", {})) + if document.origin == "nexent": + origin = {"kind": document.origin, "revision": document.source_revision, "url": document.source_url} + metadata["source_origin"] = origin + profile_parts.update({"origin_" + key: value for key, value in origin.items()}) + return SourceCatalogEntry( + document.kind, + document.key, + _freeze(tree.get("identity", {})), + _freeze(capabilities), + _freeze(metadata), + _freeze(tree.get("connection_suggestions", {})), + _freeze(_normalized_origins(mapped, controls)), + tuple(refs), + _revision(profile_parts), + ) + + +def load_source_catalog(root: Path | None = None) -> SourceCatalog: + """Convert the submission-checked immutable source catalog without repeating source validation.""" + root = (root if root is not None else Path(__file__).parent).resolve() + inventory = read_inventory(root) + if inventory.upstream_revision != SUPPORTED_UPSTREAM_REVISION: + error = SourceReadError("SOURCE_SCHEMA_UNSUPPORTED", "manifest.toml", "manifest") + error.source_path = "upstream_revision" + error.rule_id = PARSER_REF + raise error + registry = SOURCE_MAPPING + version_parts = { + "parser_ref": PARSER_REF, + "source_schema_ref": inventory.source_schema_ref, + "mapping_ref": registry.mapping_id, + "mapping_version": str(registry.mapping_version), + "mapping_sha256": mapping_fingerprint(registry), + "upstream_revision": inventory.upstream_revision, + } + catalog_parts = { + **version_parts, + "manifest_sha256": inventory.manifest_sha256, + **{document.file_path: document.content_sha256 for document in inventory.documents}, + } + revision = _revision(catalog_parts) + providers = {document.key: document for document in inventory.documents if document.kind == "provider"} + entries = {"model": {}, "provider": {}, "provider_model": {}} + for document in inventory.documents: + source = document + if document.kind != "provider": + source = resolve_source_model(document, inventory.documents) + entries[document.kind][document.key] = _entry(document, source, registry, providers, revision, version_parts) + return SourceCatalog( + revision, + inventory.upstream_revision, + PARSER_REF, + registry.mapping_id, + registry.mapping_version, + _freeze(entries["model"]), + _freeze(entries["provider"]), + _freeze(entries["provider_model"]), + ) + + +def load_profile_catalog(root: Path | None = None): + """Load source facts and explicit Nexent policy through one atomic entry point.""" + from nexent.core.model_service.profiles.composition import build_profile_catalog + from nexent.core.model_service.profiles.supplements import load_supplements + + root = (root if root is not None else Path(__file__).parent).resolve() + sources = load_source_catalog(root) + supplements = load_supplements(root, sources) + return build_profile_catalog(sources, supplements) diff --git a/sdk/nexent/core/model_service/profiles/submission.py b/sdk/nexent/core/model_service/profiles/submission.py new file mode 100644 index 0000000000..8556c362e3 --- /dev/null +++ b/sdk/nexent/core/model_service/profiles/submission.py @@ -0,0 +1,31 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Submission checks for trusted packaged profiles; never called by the application.""" + +from pathlib import Path + +from nexent.core.model_service.profiles._models_dev.inheritance import resolve_source_model +from nexent.core.model_service.profiles._models_dev.reader import SourceDocument, read_inventory +from nexent.core.model_service.profiles._models_dev.validation import validate_source_document +from nexent.core.model_service.profiles.resolver import load_profile_catalog, load_source_catalog + + +def check_source_catalog(root: Path | None = None): + """Check consumed authored/inherited facts, then exercise the actual catalog composition.""" + root = (root or Path(__file__).parent).resolve() + inventory = read_inventory(root) + for document in inventory.documents: + validate_source_document(document) + if document.kind != "provider": + source = resolve_source_model(document, inventory.documents) + validate_source_document(SourceDocument( + document.kind, document.key, document.file_path, document.content_sha256, source.data, + ), effective=True) + return load_source_catalog(root) + + + +def check_profile_catalog(root: Path | None = None): + """Check the complete packaged source and experience-profile composition.""" + check_source_catalog(root) + return load_profile_catalog(root) diff --git a/sdk/nexent/core/model_service/profiles/supplements.py b/sdk/nexent/core/model_service/profiles/supplements.py new file mode 100644 index 0000000000..89cb327018 --- /dev/null +++ b/sdk/nexent/core/model_service/profiles/supplements.py @@ -0,0 +1,248 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Read explicit Nexent supplement declarations separately from upstream TOML.""" + +import hashlib +import json +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Literal, Mapping + +from pydantic import Field, JsonValue, ValidationError, model_validator + +from nexent.core.model_service.profiles._models_dev.reader import SourceReadError, _freeze, _parse, _read +from nexent.core.model_service.profiles.contracts import ( + AtomicProfileReference, + NonBlankString, + ProfileContract, + VersionedReference, +) +from nexent.core.model_service.profiles.instruction_mapping import InstructionMappingError, resolve_instruction_mapping +from nexent.core.model_service.profiles.parameter_rules import ParameterRule, compile_parameter_rules +from nexent.core.model_service.profiles.parameter_schema import ( + ParameterValidationError, + compile_parameter_schema, + validate_parameter_value, +) +from nexent.core.model_service.profiles.resolver import SourceCatalog, _revision + + +MANIFEST_PATH = "rules/manifest.toml" +SUPPLEMENT_SCHEMA_REF = "nexent_supplements_v1" + + +class CapabilityOverride(ProfileContract): + """An explicit changed fact, accompanied by human-readable evidence.""" + + path: NonBlankString + value: JsonValue + reason: NonBlankString + evidence: NonBlankString + + +class SupplementBindings(ProfileContract): + """Declared versioned dependencies; availability is checked during composition.""" + + protocol: VersionedReference | None = None + auth: VersionedReference | None = None + adapter: VersionedReference | None = None + code_converter: VersionedReference | None = None + native_converter: VersionedReference | None = None + + +class SupplementRule(ProfileContract): + """One exact source target; no regex, dynamic imports or implicit name matching.""" + + schema_version: int = Field(ge=1, le=1) + rule_id: NonBlankString + version: int = Field(gt=0) + target_kind: Literal["model", "provider", "provider_model"] + target_key: NonBlankString + reason: NonBlankString + evidence: NonBlankString + model_profile_key: NonBlankString | None = None + aliases: list[NonBlankString] = Field(default_factory=list) + endpoint_hints: list[NonBlankString] = Field(default_factory=list) + instruction_mapping: VersionedReference | None = None + bindings: SupplementBindings | None = None + parameters: list[ParameterRule] = Field(default_factory=list) + capability_overrides: list[CapabilityOverride] = Field(default_factory=list) + + @model_validator(mode="after") + def validate_ownership(self): + if len(set(self.aliases)) != len(self.aliases) or len(set(self.endpoint_hints)) != len(self.endpoint_hints): + raise ValueError("Supplement hints must be unique") + if self.model_profile_key is not None and self.target_kind != "provider_model": + raise ValueError("Only provider-model rules can link a model profile") + if self.aliases and self.target_kind != "model": + raise ValueError("Model aliases require model ownership") + if self.endpoint_hints and self.target_kind != "provider": + raise ValueError("Endpoint hints require provider ownership") + return self + + +class SupplementFile(ProfileContract): + path: NonBlankString + content_sha256: NonBlankString + + +class SupplementManifest(ProfileContract): + schema_version: int = Field(ge=1, le=1) + rules: list[SupplementFile] + + +class SupplementReadError(ValueError): + """Do not expose file contents, fixed values or operating system details.""" + + def __init__(self, category: str, file_path: str, field_path: str = ""): + self.category, self.file_path, self.field_path = category, file_path, field_path + super().__init__(f"{category}: {file_path}: {field_path}") + + +@dataclass(frozen=True) +class SupplementEntry: + file_path: str + content_sha256: str + rule: Mapping[str, object] + + +@dataclass(frozen=True) +class SupplementCatalog: + revision: str + schema_ref: str + entries: Mapping[str, SupplementEntry] + source_revision: str + + +def _document(root: Path, path: str) -> tuple[dict, bytes]: + try: + data = _read(root, path, "supplement") + return _parse(data, path, "supplement"), data + except SourceReadError: + raise SupplementReadError("SUPPLEMENT_FILE_INVALID", path) from None + + +def _decode_json_fields(authored: dict, file_path: str) -> dict: + """Explicitly encode JSON-only values; ordinary TOML strings keep their meaning.""" + copied = dict(authored) + directives = copied.pop("json_fields", {}) + if not isinstance(directives, dict): + raise SupplementReadError("SUPPLEMENT_JSON_FIELD_INVALID", file_path, "json_fields") + + def unique_object(pairs): + result = {} + for key, value in pairs: + if key in result: + raise ValueError("Duplicate JSON field") + result[key] = value + return result + + def invalid_constant(value): + raise ValueError("Nonfinite JSON value") + + for pointer, encoded in directives.items(): + match = re.fullmatch( + r"/(parameters|capability_overrides)/(0|[1-9][0-9]*)/(schema|default_value|fixed_value|condition|value)", + pointer, + ) + if match is None or not isinstance(encoded, str): + raise SupplementReadError("SUPPLEMENT_JSON_FIELD_INVALID", file_path, "json_fields") + group, index_text, field = match.groups() + allowed = { + "parameters": {"schema", "default_value", "fixed_value", "condition"}, + "capability_overrides": {"value"}, + } + items = copied.get(group) + if ( + field not in allowed[group] + or not isinstance(items, list) + or len(index_text) > len(str(len(items))) + or int(index_text) >= len(items) + or not isinstance(items[int(index_text)], dict) + ): + raise SupplementReadError("SUPPLEMENT_JSON_FIELD_INVALID", file_path, pointer) + item = items[int(index_text)] + if field in item: + raise SupplementReadError("SUPPLEMENT_JSON_FIELD_CONFLICT", file_path, pointer) + try: + value = json.loads(encoded, object_pairs_hook=unique_object, parse_constant=invalid_constant) + validate_parameter_value(compile_parameter_schema({}), value) + except (ValueError, RecursionError): + raise SupplementReadError("SUPPLEMENT_JSON_FIELD_INVALID", file_path, pointer) from None + # Copy the touched container rather than mutating the authored declaration. + items = [dict(entry) if isinstance(entry, dict) else entry for entry in items] + items[int(index_text)][field] = value + copied[group] = items + return copied + + +def load_supplements(root: Path, sources: SourceCatalog) -> SupplementCatalog: + """Read a complete immutable declaration catalog; final policy composition follows.""" + root = root.resolve() + authored, manifest_bytes = _document(root, MANIFEST_PATH) + try: + manifest = SupplementManifest.model_validate(authored) + except ValidationError as error: + location = "/".join(str(part) for part in error.errors(include_input=False)[0]["loc"]) + raise SupplementReadError("SUPPLEMENT_MANIFEST_INVALID", MANIFEST_PATH, location) from None + entries, paths, targets = {}, set(), set() + parts = { + "schema_ref": SUPPLEMENT_SCHEMA_REF, + "source_catalog": sources.revision, + "manifest": hashlib.sha256(manifest_bytes).hexdigest(), + } + available = {"model": sources.models, "provider": sources.providers, "provider_model": sources.provider_models} + for declared in manifest.rules: + path = declared.path + try: + AtomicProfileReference( + key="supplement", + catalog_revision=sources.revision, + file_path=path, + content_sha256=declared.content_sha256, + ) + except ValidationError: + raise SupplementReadError("SUPPLEMENT_PATH_INVALID", path) from None + if not path.startswith(("rules/models/", "rules/providers/")) or not path.endswith(".toml") or path in paths: + raise SupplementReadError("SUPPLEMENT_PATH_INVALID", path) + paths.add(path) + authored, data = _document(root, path) + digest = hashlib.sha256(data).hexdigest() + if digest != declared.content_sha256: + raise SupplementReadError("SUPPLEMENT_HASH_MISMATCH", path) + try: + validate_parameter_value(compile_parameter_schema({}), authored) + except ParameterValidationError as error: + raise SupplementReadError("SUPPLEMENT_RULE_INVALID", path, error.path) from None + authored = _decode_json_fields(authored, path) + try: + rule = SupplementRule.model_validate(authored) + # Conditions can refer to parameters in another participating atom. + # Compile the combined dependency graph only after profile composition. + declarations = [parameter.model_dump(by_alias=True, exclude_unset=True) for parameter in rule.parameters] + compile_parameter_rules([{**declaration, "condition": None} for declaration in declarations]) + if rule.instruction_mapping is not None: + resolve_instruction_mapping(rule.instruction_mapping) + except ValidationError as error: + location = "/".join(str(part) for part in error.errors(include_input=False)[0]["loc"]) + raise SupplementReadError("SUPPLEMENT_RULE_INVALID", path, location) from None + except ParameterValidationError as error: + raise SupplementReadError("SUPPLEMENT_RULE_INVALID", path, "parameters" + error.path) from None + except InstructionMappingError: + raise SupplementReadError("SUPPLEMENT_RULE_INVALID", path, "instruction_mapping") from None + target = (rule.target_kind, rule.target_key) + expected_directory = "rules/models/" if rule.target_kind == "model" else "rules/providers/" + if not path.startswith(expected_directory): + raise SupplementReadError("SUPPLEMENT_PATH_OWNERSHIP", path) + if target in targets or rule.rule_id in entries: + raise SupplementReadError("SUPPLEMENT_DUPLICATE", path) + if rule.target_key not in available[rule.target_kind]: + raise SupplementReadError("SUPPLEMENT_TARGET_MISSING", path) + if rule.model_profile_key is not None and rule.model_profile_key not in sources.models: + raise SupplementReadError("SUPPLEMENT_MODEL_MISSING", path) + targets.add(target) + normalized = rule.model_dump(mode="json", by_alias=True, exclude_unset=True) + entries[rule.rule_id] = SupplementEntry(path, digest, _freeze(normalized)) + parts[path] = digest + return SupplementCatalog(_revision(parts), SUPPLEMENT_SCHEMA_REF, _freeze(entries), sources.revision) diff --git a/sdk/nexent/core/model_service/sse.py b/sdk/nexent/core/model_service/sse.py new file mode 100644 index 0000000000..b72a017711 --- /dev/null +++ b/sdk/nexent/core/model_service/sse.py @@ -0,0 +1,140 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Bounded incremental SSE framing shared by sync and async HTTP senders.""" + +import codecs +from collections.abc import Iterator +from dataclasses import dataclass + + +class SSEDecodeError(ValueError): + """Safe framing failure with no provider payload in diagnostics.""" + + def __init__(self, category: str): + self.category = category + super().__init__(category) + + +@dataclass(frozen=True) +class SSEEvent: + """A complete SSE frame; provider JSON interpretation belongs to adapters.""" + + data: str + event: str = "message" + event_id: str = "" + + +class SSEDecoder: + """Decode UTF-8 text fragments without reconnecting or interpreting retry fields. + + Only blank-line-terminated data blocks become events. Incomplete data at EOF + is an explicit Nexent failure rather than a successful model completion. + """ + + def __init__(self, *, max_event_bytes: int = 1024 * 1024): + if type(max_event_bytes) is not int or max_event_bytes <= 0: + raise ValueError("SSE event byte limit must be a positive integer") + self._limit = max_event_bytes + self._size = 0 + self._line: list[str] = [] + self._data: list[str] = [] + self._event = "" + self._id = "" + self._first = True + self._after_cr = False + self._closed = False + + def _fail(self, category: str) -> None: + self._closed = True + self._line.clear() + self._data.clear() + raise SSEDecodeError(category) from None + + def _consume_line(self) -> SSEEvent | None: + line = "".join(self._line) + self._line.clear() + if not line: + result = SSEEvent("\n".join(self._data), self._event or "message", self._id) if self._data else None + self._data.clear() + self._event = "" + self._size = 0 + return result + if line.startswith(":"): + return None + field, separator, value = line.partition(":") + if separator and value.startswith(" "): + value = value[1:] + if field == "data": + self._data.append(value) + elif field == "event": + self._event = value + elif field == "id" and "\x00" not in value: + self._id = value + return None + + def feed(self, fragment: str) -> Iterator[SSEEvent]: + """Yield completed frames immediately, even before a later frame fails. + + Exhaust this iterator before feeding another fragment. HTTP iteration + supplies already incrementally decoded UTF-8 text, never raw bytes. + """ + if self._closed: + raise SSEDecodeError("SSE_DECODER_CLOSED") + if not isinstance(fragment, str): + self._fail("SSE_INVALID_TEXT") + for character in fragment: + if self._first: + self._first = False + if character == "\ufeff": + continue + if self._after_cr: + self._after_cr = False + if character == "\n": + continue + try: + self._size += len(character.encode("utf-8")) + except UnicodeEncodeError: + self._fail("SSE_INVALID_TEXT") + if self._size > self._limit: + self._fail("SSE_EVENT_TOO_LARGE") + if character in ("\r", "\n"): + self._after_cr = character == "\r" + event = self._consume_line() + if event is not None: + yield event + else: + self._line.append(character) + + def finish(self) -> None: + """Close framing; provider finish markers still require adapter validation.""" + if self._closed: + raise SSEDecodeError("SSE_DECODER_CLOSED") + self._closed = True + if self._line: + self._consume_line() + if self._data: + self._fail("SSE_INCOMPLETE_EVENT") + + +class SSEByteDecoder: + """Strict UTF-8 decoding keeps byte splits transparent and corruption explicit.""" + + def __init__(self, *, max_event_bytes: int = 1024 * 1024): + self._text = codecs.getincrementaldecoder("utf-8")() + self._frames = SSEDecoder(max_event_bytes=max_event_bytes) + + def feed(self, fragment: bytes) -> Iterator[SSEEvent]: + if not isinstance(fragment, bytes): + self._frames._fail("SSE_INVALID_TEXT") + try: + text = self._text.decode(fragment) + except UnicodeError: + self._frames._fail("SSE_INVALID_TEXT") + yield from self._frames.feed(text) + + def finish(self) -> None: + try: + self._text.decode(b"", final=True) + except UnicodeError: + self._frames._fail("SSE_INVALID_TEXT") + self._frames.finish() diff --git a/test/sdk/core/model_service/test_catalog_normalization.py b/test/sdk/core/model_service/test_catalog_normalization.py new file mode 100644 index 0000000000..568b94107e --- /dev/null +++ b/test/sdk/core/model_service/test_catalog_normalization.py @@ -0,0 +1,155 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Isolated materialization, normalization and revision boundary rules.""" + +import pytest +from nexent.core.model_service.profiles._models_dev.mapping import SOURCE_MAPPING +from nexent.core.model_service.profiles._models_dev.reader import ( + SourceDocument, + SourceReadError, + _freeze, +) +from nexent.core.model_service.profiles.resolver import ( + _controls, + _entry, + _revision, + _tree, +) + + + +def test_canonical_tree_preserves_arrays_empty_objects_and_explicit_null(): + tree = _tree( + { + "/reference_metadata/items/0/name": "first", + "/reference_metadata/items/1/name": "second", + "/reference_metadata/empty": {}, + "/modalities/output": (), + "/reasoning/control_candidates/0/values/0": None, + } + ) + assert tree["reference_metadata"]["items"] == [ + {"name": "first"}, + {"name": "second"}, + ] + assert tree["reference_metadata"]["empty"] == {} + assert tree["modalities"]["output"] == () + assert tree["reasoning"]["control_candidates"][0]["values"] == [None] + + +def test_declared_control_domains_preserve_sentinels_and_unknown_bounds(): + assert _controls({}) is None + assert _controls({"control_candidates": []}) == [] + assert _controls( + { + "control_candidates": [ + {"type": "toggle"}, + {"type": "effort", "values": ["none", "null", "default"]}, + {"type": "budget_tokens", "min": -1, "max": 0}, + ] + } + ) == [ + {"mode": "toggle", "values": [True, False]}, + {"mode": "effort", "values": ["none", None, "default"]}, + {"mode": "budget_tokens", "min": None, "max": 0, "special_values": [-1]}, + ] + assert _controls({"control_candidates": [{"type": "budget_tokens", "min": 0}]}) == [ + {"mode": "budget_tokens", "min": 0, "max": None, "special_values": []} + ] + + +def test_revision_is_order_independent_and_length_delimited(): + assert _revision({"parser": "v1", "mapping": "v1"}) == _revision( + {"mapping": "v1", "parser": "v1"} + ) + assert _revision({"ab": "c"}) != _revision({"a": "bc"}) + assert _revision({"parser": "v1"}) != _revision({"parser": "v2"}) + + +def test_normalized_source_snapshot_is_deeply_immutable_and_not_an_outlet(): + source = SourceDocument( + "provider_model", + "vendor/model", + "providers/vendor/models/model.toml", + "a" * 64, + _freeze( + { + "reasoning": True, + "reasoning_options": [ + {"type": "toggle"}, + {"type": "budget_tokens", "min": -1, "max": 10}, + ], + } + ), + ) + provider = SourceDocument( + "provider", + "vendor", + "providers/vendor/provider.toml", + "b" * 64, + _freeze({"name": "Vendor"}), + ) + entry = _entry( + source, + source, + SOURCE_MAPPING, + {"vendor": provider}, + "catalog-v1", + {"parser": "v1"}, + ) + assert entry.capabilities["reasoning"]["controls"][1]["special_values"] == (-1,) + assert entry.capabilities["reasoning"]["controls"][1]["min"] is None + assert ( + entry.field_origins[ + "/capabilities/reasoning/controls/1/special_values/0" + ].mapping.source.source_path + == "reasoning_options[1].min" + ) + assert ( + entry.field_origins[ + "/capabilities/reasoning/controls/0/values/0" + ].mapping.source.source_path + == "reasoning_options[0].type" + ) + with pytest.raises(TypeError): + entry.capabilities["reasoning"]["controls"][0]["values"][0] = False + assert [ref.key for ref in entry.source_refs] == ["vendor/model", "vendor"] + assert not hasattr(entry, "outlet_id") + + +@pytest.mark.parametrize( + "data,path", + [ + ({"limit": {"context": 10, "output": 20}}, "limit"), + ( + { + "reasoning": True, + "reasoning_options": [{"type": "effort", "values": []}], + }, + "reasoning_options", + ), + ], +) +def test_invalid_normalized_domains_fail_with_sanitized_diagnostics(data, path): + source = SourceDocument( + "model" if path == "limit" else "provider_model", + "lab/model", + "models/lab/model.toml", + "a" * 64, + _freeze(data), + ) + with pytest.raises(SourceReadError) as error: + _entry(source, source, SOURCE_MAPPING, {}, "catalog-v1", {}) + assert error.value.category == "SOURCE_NORMALIZATION_INVALID" + assert error.value.source_path == path + + +def test_local_provider_reference_and_documentation_are_not_upstream_attribution(): + from nexent.core.model_service.profiles.resolver import load_source_catalog + + provider = load_source_catalog().providers["tokenpony"] + assert provider.source_refs[0].catalog_revision == "nexent_tokenpony_provider_v1" + assert provider.reference_metadata["source_origin"] == { + "kind": "nexent", "revision": "nexent_tokenpony_provider_v1", "url": "https://docs.tokenpony.cn/", + } + assert provider.connection_suggestions["base_url"] == "https://api.tokenpony.cn/v1" diff --git a/test/sdk/core/model_service/test_experience_overlay.py b/test/sdk/core/model_service/test_experience_overlay.py new file mode 100644 index 0000000000..049166421c --- /dev/null +++ b/test/sdk/core/model_service/test_experience_overlay.py @@ -0,0 +1,128 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Experience precedence is explicit and separate from upstream source composition.""" + +import pytest + +from nexent.core.model_service.profiles.composition import build_profile_catalog, compose_provider_model_policy +from nexent.core.model_service.profiles.contracts import VersionedReference +from nexent.core.model_service.profiles.parameter_schema import ParameterValidationError +from nexent.core.model_service.profiles.resolver import SourceCatalog, SourceCatalogEntry +from nexent.core.model_service.profiles.supplements import SupplementCatalog, SupplementEntry + + +REF = VersionedReference(ref="fixture_v1", version=1) +BINDINGS = {name: REF.model_dump() for name in ("protocol", "auth", "adapter", "code_converter", "native_converter")} + + +def definitions(*, model=None, provider=None, combination=None): + def source(kind, key, facts): + return SourceCatalogEntry(kind, key, {}, facts, {}, {}, {}, (), kind + key) + + sources = SourceCatalog( + "source", + "upstream", + "parser", + "mapping", + 1, + { + "family/model": source( + "model", "family/model", {"context": {"combined_window": 200000, "output_limit": 16000}} + ) + }, + {"provider": source("provider", "provider", {})}, + {"provider/exact": source("provider_model", "provider/exact", {"context": {"output_limit": 12000}})}, + ) + entries = {} + for kind, key, rule in ( + ("model", "family/model", model), + ("provider", "provider", provider), + ("provider_model", "provider/exact", combination), + ): + entries[kind] = SupplementEntry( + "rules/" + kind + ".toml", + kind, + { + "target_kind": kind, + "target_key": key, + **(rule or {}), + **({"model_profile_key": "family/model"} if kind == "provider_model" else {}), + }, + ) + return build_profile_catalog(sources, SupplementCatalog("supplements", "schema", entries, sources.revision)) + + +def resolve(catalog, values=None): + return compose_provider_model_policy( + catalog, + provider_key="provider", + request_model_id="exact", + model_profile_key=None, + base_bindings=BINDINGS, + base_instruction=VersionedReference(ref="system_only_v1", version=1), + registered_references={name: [REF] for name in (*BINDINGS, "parameter_transform")}, + submitted_values=values or {}, + ) + + +def parameter(value, **changes): + return { + "key": "temperature", + "schema": {"type": "number", "minimum": 0, "maximum": 2}, + "visible": True, + "editable": True, + "required": False, + "allowed_scopes": ["outlet"], + "ui_group": "sampling", + "owner": "model", + "source": {"ref": "fixture"}, + "default_value": value, + "api_binding": {"path": "/temperature", "transform_ref": REF.ref, "transform_version": 1}, + **changes, + } + + +@pytest.mark.parametrize("layer,expected", [("model", 0.2), ("provider", 0.4), ("combination", 0.6)]) +def test_defaults_are_layered_before_permitted_user_values(layer, expected): + rules = {"model": {"parameters": [parameter(0.2)]}} + if layer != "model": + rules["provider"] = {"parameters": [parameter(0.4, owner="provider")]} + if layer == "combination": + rules["combination"] = {"parameters": [parameter(0.6, owner="combination")]} + catalog = definitions(**rules) + before = resolve(catalog) + assert before.parameter_values["temperature"] == expected + assert resolve(catalog, {"temperature": 0.9}).parameter_values["temperature"] == 0.9 + assert before.parameter_values["temperature"] == expected + assert before.parameter_rules.declarations[0]["source"]["ref"] == "fixture" + + +def test_model_experience_is_applied_after_exact_upstream_facts(): + rule = { + "capability_overrides": [ + { + "path": "/context/output_limit", + "value": 8000, + "reason": "Reviewed service behavior", + "evidence": "controlled fixture", + } + ] + } + policy = resolve(definitions(model=rule)) + assert policy.capabilities["context"]["output_limit"] == 8000 + assert policy.overrides[0]["original_value"] == 12000 + assert policy.overrides[0]["source"] == "rules/model.toml" + assert policy.capabilities["context"]["combined_window"] == 200000 + + +def test_effective_fixed_value_and_range_still_reject_user_changes(): + catalog = definitions( + model={"parameters": [parameter(0.2)]}, + provider={"parameters": [parameter(0.4)]}, + combination={"parameters": [parameter(0.6, fixed_value=0.6, editable=False)]}, + ) + assert resolve(catalog).parameter_values["temperature"] == 0.6 + with pytest.raises(ParameterValidationError, match="FIXED_PARAMETER"): + resolve(catalog, {"temperature": 0.9}) + with pytest.raises(ParameterValidationError, match="PARAMETER_VALUE_INVALID"): + resolve(definitions(model={"parameters": [parameter(0.2)]}), {"temperature": 3}) diff --git a/test/sdk/core/model_service/test_source_inheritance.py b/test/sdk/core/model_service/test_source_inheritance.py new file mode 100644 index 0000000000..c411421086 --- /dev/null +++ b/test/sdk/core/model_service/test_source_inheritance.py @@ -0,0 +1,233 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Pinned inheritance rules and source ownership, independent of semantic mapping.""" + +import pytest +from nexent.core.model_service.profiles._models_dev.inheritance import ( + resolve_source_model, +) +from nexent.core.model_service.profiles._models_dev.reader import ( + SourceDocument, + SourceReadError, + _freeze, +) + + +def source(kind, key, data): + path = f"models/{key}.toml" if kind == "model" else f"providers/{key}.toml" + return SourceDocument(kind, key, path, "a" * 64, _freeze(data)) + + +def test_inheritance_merges_objects_replaces_arrays_and_preserves_field_ownership(): + base = source( + "model", + "lab/model", + { + "name": "Base", + "reasoning": True, + "tool_call": True, + "limit": {"context": 100, "input": 90, "output": 20}, + "modalities": {"input": ["text", "image"], "output": ["text"]}, + }, + ) + outlet = source( + "provider_model", + "vendor/model", + { + "base_model": base.key, + "name": "Hosted", + "reasoning": False, + "limit": {"output": 10}, + "modalities": {"input": ["text"]}, + }, + ) + effective = resolve_source_model(outlet, (base, outlet)) + assert effective.data["name"] == "Hosted" + assert effective.data["reasoning"] is False + assert dict(effective.data["limit"]) == {"context": 100, "input": 90, "output": 10} + assert effective.data["modalities"]["input"] == ("text",) + assert "base_model" not in effective.data + assert effective.dependencies == (base, outlet) + assert effective.document.key == "vendor/model" + assert effective.field_origins["limit.context"].file_path == base.file_path + assert effective.field_origins["limit.output"].file_path == outlet.file_path + assert ( + effective.field_origins["modalities.input[0]"].source_path + == "modalities.input[0]" + ) + assert "modalities.input[1]" not in effective.field_origins + assert base.data["limit"]["output"] == 20 + assert base.data["modalities"]["input"] == ("text", "image") + with pytest.raises(TypeError): + effective.data["limit"]["output"] = 1 + + +def test_provider_inheritance_excludes_reference_only_fields(): + base = source( + "model", + "lab/model", + { + "name": "Base", + "id": "base-id", + "license": "MIT", + "links": ["docs"], + "weights": ["weights"], + "benchmarks": [{"name": "score", "score": 1}], + }, + ) + outlet = source("provider_model", "vendor/model", {"base_model": base.key}) + result = resolve_source_model(outlet, (base, outlet)) + assert dict(result.data) == {"name": "Base"} + assert set(result.field_origins) == {"name"} + assert base.data["license"] == "MIT" + atomic = resolve_source_model(base, (base, outlet)) + assert atomic.data["license"] == "MIT" + + +def test_omit_deletes_nested_paths_and_prunes_empty_parents(): + base = source( + "model", + "lab/model", + { + "name": "Base", + "nested": {"only": {"field": 1}}, + "limit": {"context": 100, "output": 10}, + }, + ) + outlet = source( + "provider_model", + "vendor/model", + { + "base_model": base.key, + "base_model_omit": ["nested.only.field", "limit.output", "missing.path"], + }, + ) + result = resolve_source_model(outlet, (base, outlet)) + assert "nested" not in result.data + assert dict(result.data["limit"]) == {"context": 100} + assert "limit.output" not in result.field_origins + assert result.omitted_paths == ("nested.only.field", "limit.output", "missing.path") + assert "base_model_omit" not in result.data + assert base.data["nested"]["only"]["field"] == 1 + + +def test_omit_does_not_traverse_arrays_or_scalars(): + base = source("model", "lab/model", {"array": ["item"], "scalar": 1}) + outlet = source( + "provider_model", + "vendor/model", + { + "base_model": base.key, + "base_model_omit": ["array.0", "scalar.child"], + }, + ) + assert dict(resolve_source_model(outlet, (base, outlet)).data) == { + "array": ("item",), + "scalar": 1, + } + + +def test_empty_array_replacement_preserves_explicit_empty_declaration(): + base = source("model", "lab/model", {"values": ["low", "high"]}) + outlet = source( + "provider_model", "vendor/model", {"base_model": base.key, "values": []} + ) + result = resolve_source_model(outlet, (base, outlet)) + assert result.data["values"] == () + assert set(result.field_origins) == {"values"} + assert result.field_origins["values"].file_path == outlet.file_path + + +def test_without_base_never_infers_identity_or_capabilities_from_name(): + base = source("model", "lab/model", {"name": "Same", "reasoning": True}) + outlet = source("provider_model", "vendor/model", {"name": "Same"}) + result = resolve_source_model(outlet, (base, outlet)) + assert dict(result.data) == {"name": "Same"} + assert result.dependencies == (outlet,) + + +@pytest.mark.parametrize( + "data,category", + [ + ({"base_model": "missing"}, "SOURCE_BASE_MODEL_MISSING"), + ({"base_model": False}, "SOURCE_BASE_REFERENCE_INVALID"), + ({"base_model": " "}, "SOURCE_BASE_REFERENCE_INVALID"), + ({"base_model": "lab/model", "base_model_omit": "name"}, "SOURCE_OMIT_INVALID"), + ( + {"base_model": "lab/model", "base_model_omit": [False]}, + "SOURCE_OMIT_INVALID", + ), + ( + {"base_model": "lab/model", "base_model_omit": ["limit..output"]}, + "SOURCE_OMIT_INVALID", + ), + ({"base_model_omit": ["name"]}, "SOURCE_OMIT_WITHOUT_BASE"), + ], +) +def test_invalid_references_and_omissions_fail_explicitly(data, category): + base = source("model", "lab/model", {"name": "Base"}) + outlet = source("provider_model", "vendor/model", data) + with pytest.raises(SourceReadError) as caught: + resolve_source_model(outlet, (base, outlet)) + assert caught.value.category == category + assert caught.value.file_path == outlet.file_path + assert caught.value.rule_id == "source_inheritance_v1" + + +def test_duplicate_model_identity_is_rejected(): + base = source("model", "lab/model", {"name": "Base"}) + outlet = source("provider_model", "vendor/model", {"base_model": base.key}) + with pytest.raises(SourceReadError, match="SOURCE_DUPLICATE"): + resolve_source_model(outlet, (base, base, outlet)) + + +def test_atomic_model_cannot_declare_nested_or_cyclic_base_in_this_source_schema(): + base = source("model", "lab/model", {"base_model": "lab/model"}) + outlet = source("provider_model", "vendor/model", {"base_model": base.key}) + for document in (base, outlet): + with pytest.raises(SourceReadError, match="SOURCE_BASE_REFERENCE_INVALID"): + resolve_source_model(document, (base, outlet)) + + +def test_provider_metadata_is_not_a_model_document(): + provider = source("provider", "vendor", {"name": "Vendor"}) + with pytest.raises(SourceReadError, match="SOURCE_MODEL_KIND_INVALID"): + resolve_source_model(provider, (provider,)) + + +def test_nonempty_overlay_replaces_empty_container_origin_with_leaf_ownership(): + base = source("model", "lab/model", {"limit": {}}) + outlet = source( + "provider_model", + "vendor/model", + {"base_model": base.key, "limit": {"output": 10}}, + ) + result = resolve_source_model(outlet, (base, outlet)) + assert dict(result.data["limit"]) == {"output": 10} + assert set(result.field_origins) == {"limit.output"} + assert result.field_origins["limit.output"].file_path == outlet.file_path + assert dict(base.data["limit"]) == {} + + +def test_empty_object_overlay_preserves_inherited_leaf_values_and_origins(): + base = source("model", "lab/model", {"limit": {"context": 100}}) + outlet = source( + "provider_model", "vendor/model", {"base_model": base.key, "limit": {}} + ) + result = resolve_source_model(outlet, (base, outlet)) + assert dict(result.data["limit"]) == {"context": 100} + assert set(result.field_origins) == {"limit.context"} + assert result.field_origins["limit.context"].file_path == base.file_path + + +def test_omit_prunes_only_empty_parents_and_keeps_sibling_origins(): + base = source("model", "lab/model", {"nested": {"remove": {"leaf": 1}, "keep": 2}}) + outlet = source( + "provider_model", + "vendor/model", + {"base_model": base.key, "base_model_omit": ["nested.remove.leaf"]}, + ) + result = resolve_source_model(outlet, (base, outlet)) + assert dict(result.data["nested"]) == {"keep": 2} + assert set(result.field_origins) == {"nested.keep"} + assert result.field_origins["nested.keep"].file_path == base.file_path diff --git a/test/sdk/core/model_service/test_source_mapping.py b/test/sdk/core/model_service/test_source_mapping.py new file mode 100644 index 0000000000..fa13dbfbff --- /dev/null +++ b/test/sdk/core/model_service/test_source_mapping.py @@ -0,0 +1,122 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Versioned source dispositions without API bindings or catalog execution.""" + +import pytest +from nexent.core.model_service.profiles._models_dev.inheritance import ( + resolve_source_model, +) +from nexent.core.model_service.profiles._models_dev.mapping import ( + SOURCE_MAPPING, + mapping_fingerprint, + map_source_fields, +) +from nexent.core.model_service.profiles._models_dev.reader import ( + SourceDocument, + SourceReadError, + _freeze, +) + + + +def doc(kind, key, data): + return SourceDocument(kind, key, f"{kind}/{key}.toml", "a" * 64, _freeze(data)) + + +def test_mapping_keeps_unknown_false_and_zero_separate_and_records_rule_origin(): + source = doc( + "model", + "lab/model", + {"name": "Model", "reasoning": False, "limit": {"context": 0, "output": 10}}, + ) + result = map_source_fields(source, SOURCE_MAPPING) + assert result.values["/reasoning/status"] == "unsupported" + assert result.values["/tools/status"] == "unknown" + assert result.values["/context/combined_window"] is None + assert result.values["/context/input_limit"] is None + assert result.values["/context/output_limit"] == 10 + origin = result.field_origins["/context/combined_window"] + assert origin.rule_id == "model.limit.context" + assert origin.rule_version == 1 + assert origin.source.source_path == "limit.context" + assert "/tools/status" not in result.field_origins + assert source.data["limit"]["context"] == 0 + with pytest.raises(TypeError): + result.values["/reasoning/status"] = "supported" + + +def test_inherited_field_ownership_is_preserved_through_mapping(): + base = doc("model", "lab/model", {"tool_call": True, "limit": {"context": 100}}) + hosted = doc( + "provider_model", + "vendor/model", + {"base_model": base.key, "limit": {"output": 10}}, + ) + result = map_source_fields( + resolve_source_model(hosted, (base, hosted)), SOURCE_MAPPING + ) + assert result.field_origins["/tools/status"].source.file_path == base.file_path + assert ( + result.field_origins["/context/output_limit"].source.file_path + == hosted.file_path + ) + assert result.field_origins["/tools/status"].rule_id == "provider_model.tool_call" + + +def test_array_reference_fields_and_control_candidates_preserve_exact_values(): + source = doc( + "provider_model", + "vendor/model", + { + "reasoning_options": [ + {"type": "effort", "values": ["none", "null"]}, + {"type": "budget_tokens", "min": -1}, + ], + "cost": {"input": 0.1, "output": 0.2}, + "modalities": {"input": ["text", "image"], "output": []}, + }, + ) + result = map_source_fields(source, SOURCE_MAPPING) + assert result.values["/reasoning/control_candidates/0/values/0"] == "none" + assert result.values["/reasoning/control_candidates/0/values/1"] == "null" + assert result.values["/reasoning/control_candidates/1/min"] == -1 + assert result.values["/reference_metadata/cost/input"] == 0.1 + assert result.values["/modalities/output"] == () + assert not any("api" in key or "body" in key for key in result.values) + + +def test_provider_metadata_remains_a_suggestion_without_api_selection(): + source = doc( + "provider", + "vendor", + {"api": "https://example.com", "env": ["PRIVATE_API_KEY"], "npm": "sdk"}, + ) + result = map_source_fields(source, SOURCE_MAPPING) + assert result.values["/connection_suggestions/base_url"] == "https://example.com" + assert result.values["/reference_metadata/env/0"] == "PRIVATE_API_KEY" + assert not any("protocol" in key or "credential" in key for key in result.values) + + +def test_empty_reference_arrays_and_text_benchmark_scores_remain_reference_data(): + source = doc( + "model", + "lab/model", + {"links": [], "benchmarks": [{"name": "Score", "score": "pass"}]}, + ) + result = map_source_fields(source, SOURCE_MAPPING) + assert result.values["/reference_metadata/links"] == () + assert result.values["/reference_metadata/benchmarks/0/score"] == "pass" + assert result.values["/reasoning/status"] == "unknown" + + + +def test_policy_fingerprint_is_deterministic_and_versioned(): + from dataclasses import replace + assert mapping_fingerprint(SOURCE_MAPPING) == mapping_fingerprint(SOURCE_MAPPING) + assert mapping_fingerprint(replace(SOURCE_MAPPING, mapping_version=2)) != mapping_fingerprint(SOURCE_MAPPING) + + +def test_unconsumed_fields_are_isolated_reference_data(): + result = map_source_fields(doc("model", "lab/model", {"benchmarks": [{"new_field": "value"}]})) + assert result.values["/reference_metadata/benchmarks/0/new_field"] == "value" + assert result.values["/reasoning/status"] == "unknown" diff --git a/test/sdk/core/model_service/test_sse.py b/test/sdk/core/model_service/test_sse.py new file mode 100644 index 0000000000..8b6a0553f5 --- /dev/null +++ b/test/sdk/core/model_service/test_sse.py @@ -0,0 +1,152 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Isolated SSE framing proof; actual HTTP integration belongs to MST-D3-001.""" + +import pytest +from nexent.core.model_service.sse import ( + SSEByteDecoder, + SSEDecodeError, + SSEDecoder, + SSEEvent, +) + + +def decode(fragments, **options): + decoder = SSEDecoder(**options) + events = [event for fragment in fragments for event in decoder.feed(fragment)] + decoder.finish() + return events + + +@pytest.mark.parametrize("newline", ["\n", "\r", "\r\n"]) +def test_every_fragment_boundary_preserves_unicode_and_multiline_events(newline): + text = newline.join( + [ + "\ufeff: heartbeat", + "id: call-1", + "event: delta", + "data: 中文", + "data: second", + "", + "data", + "", + "", + ] + ) + expected = [ + SSEEvent("中文\n second", "delta", "call-1"), + SSEEvent("", "message", "call-1"), + ] + for split in range(len(text) + 1): + assert decode([text[:split], text[split:]]) == expected + assert decode(list(text)) == expected + + +def test_comment_unknown_retry_and_null_id_fields_do_not_create_events(): + text = "id: retained\n\ndata: one\n\nid: ignored\x00\nevent: custom\nretry: 0\nunknown: hidden\n: data: false\ndata: two\n\nid:\ndata: three\n\n" + assert decode([text]) == [ + SSEEvent("one", event_id="retained"), + SSEEvent("two", "custom", "retained"), + SSEEvent("three"), + ] + + +def test_only_leading_bom_and_one_separator_space_are_removed(): + assert decode(["", "\ufeff", "data: \ufeffx \n\n"]) == [SSEEvent("\ufeffx ")] + + +@pytest.mark.parametrize( + "suffix", + ["data: private", "data: private\n", "data\n", "event: delta\ndata: private\r"], +) +def test_eof_never_promotes_incomplete_data_to_completed_event(suffix): + decoder = SSEDecoder() + assert list(decoder.feed(suffix)) == [] + with pytest.raises(SSEDecodeError, match="^SSE_INCOMPLETE_EVENT$"): + decoder.finish() + with pytest.raises(SSEDecodeError, match="^SSE_DECODER_CLOSED$"): + list(decoder.feed("\n")) + + +@pytest.mark.parametrize( + "text", ["", ": heartbeat", "event: delta\n", "id: final\n\n", "retry: 100"] +) +def test_eof_without_pending_data_is_safe_but_not_a_model_completion(text): + assert decode([text]) == [] + + +def test_byte_limit_is_per_event_and_counts_utf8_not_characters(): + text = "data: 中\n\n" + size = len(text.encode("utf-8")) + assert decode([text] * 100, max_event_bytes=size) == [SSEEvent("中")] * 100 + decoder = SSEDecoder(max_event_bytes=size - 1) + with pytest.raises(SSEDecodeError, match="^SSE_EVENT_TOO_LARGE$"): + list(decoder.feed(text)) + with pytest.raises(SSEDecodeError, match="^SSE_DECODER_CLOSED$"): + decoder.finish() + + +@pytest.mark.parametrize( + "text", [":" * 20, "event: " + "x" * 20, "unknown: " + "x" * 20] +) +def test_non_data_fields_cannot_bypass_buffer_limit(text): + with pytest.raises(SSEDecodeError, match="^SSE_EVENT_TOO_LARGE$"): + decode(list(text), max_event_bytes=10) + + +@pytest.mark.parametrize("limit", [0, -1, True, 1.0, None]) +def test_invalid_byte_limits_are_rejected(limit): + with pytest.raises(ValueError, match="positive integer"): + SSEDecoder(max_event_bytes=limit) + + +@pytest.mark.parametrize("fragment", [b"data: bytes\n\n", "data: \ud800"]) +def test_invalid_text_is_sanitized_and_decoder_becomes_terminal(fragment): + decoder = SSEDecoder() + with pytest.raises(SSEDecodeError, match="^SSE_INVALID_TEXT$") as caught: + list(decoder.feed(fragment)) + assert caught.value.__cause__ is None + with pytest.raises(SSEDecodeError, match="^SSE_DECODER_CLOSED$"): + list(decoder.feed("data: ignored\n\n")) + + +def test_valid_event_remains_visible_before_later_failure_in_same_fragment(): + decoder = SSEDecoder(max_event_bytes=16) + events = decoder.feed("data: first\n\ndata: " + "private" * 20) + assert next(events) == SSEEvent("first") + with pytest.raises(SSEDecodeError, match="^SSE_EVENT_TOO_LARGE$"): + next(events) + + +def test_finished_decoder_cannot_accept_data_or_finish_again(): + decoder = SSEDecoder() + decoder.finish() + with pytest.raises(SSEDecodeError, match="^SSE_DECODER_CLOSED$"): + list(decoder.feed("")) + with pytest.raises(SSEDecodeError, match="^SSE_DECODER_CLOSED$"): + decoder.finish() + + +def test_utf8_byte_splits_do_not_corrupt_multibyte_content(): + text = "\ufeffdata: 中文😀\r\n\r\n".encode() + for split in range(len(text) + 1): + decoder = SSEByteDecoder() + events = list(decoder.feed(text[:split])) + list(decoder.feed(text[split:])) + decoder.finish() + assert events == [SSEEvent("中文😀")] + + +@pytest.mark.parametrize("fragment", [b"data: \xff\n\n", "not bytes"]) +def test_invalid_utf8_is_an_error_and_not_replacement_characters(fragment): + decoder = SSEByteDecoder() + with pytest.raises(SSEDecodeError, match="^SSE_INVALID_TEXT$"): + list(decoder.feed(fragment)) + with pytest.raises(SSEDecodeError, match="^SSE_DECODER_CLOSED$"): + list(decoder.feed(b"")) + + +def test_incomplete_utf8_at_eof_is_explicit_failure(): + decoder = SSEByteDecoder() + assert list(decoder.feed(b"data: \xe4")) == [] + with pytest.raises(SSEDecodeError, match="^SSE_INVALID_TEXT$"): + decoder.finish() diff --git a/test/sdk/core/model_service/test_supplements.py b/test/sdk/core/model_service/test_supplements.py new file mode 100644 index 0000000000..2be33f1573 --- /dev/null +++ b/test/sdk/core/model_service/test_supplements.py @@ -0,0 +1,370 @@ +# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved. + +"""Supplement declarations have exact ownership, digests and immutable snapshots.""" + +import hashlib +from pathlib import Path +from types import MappingProxyType + +import pytest +from nexent.core.model_service.profiles.resolver import SourceCatalog +from nexent.core.model_service.profiles.supplements import ( + SupplementReadError, + load_supplements, +) + +MODEL_RULE = """schema_version = 1 +rule_id = "test_model_v1" +version = 1 +target_kind = "model" +target_key = "alibaba/qwen3.5-plus" +reason = "Declare traditional mapping separately" +evidence = "test-rule-v1" +aliases = ["qwen3.5-plus"] +[instruction_mapping] +ref = "system_only_v1" +version = 1 +""" + + +@pytest.fixture +def sources(): + # Source normalization is a collaborator, covered separately by its owning UT/D1. + return SourceCatalog( + "a" * 64, + "upstream", + "parser", + "mapping", + 1, + MappingProxyType({"alibaba/qwen3.5-plus": None}), + MappingProxyType({"alibaba-cn": None}), + MappingProxyType({"alibaba-cn/qwen3.5-plus": None}), + ) + + +def write_manifest(root, entries): + (root / "rules").mkdir(exist_ok=True) + content = "schema_version = 1\n" + for path, digest in entries: + content += f'[[rules]]\npath = "{path}"\ncontent_sha256 = "{digest}"\n' + (root / "rules/manifest.toml").write_text(content) + + +def write_rule(root, content=MODEL_RULE, path="rules/models/test.toml"): + file = root / path + file.parent.mkdir(parents=True, exist_ok=True) + file.write_text(content) + digest = hashlib.sha256(file.read_bytes()).hexdigest() + write_manifest(root, [(path, digest)]) + return file, digest + + +def test_valid_rule_is_frozen_and_revisions_include_source_and_rule_bytes( + tmp_path, sources +): + file, digest = write_rule(tmp_path) + catalog = load_supplements(tmp_path, sources) + assert catalog.source_revision == sources.revision + entry = catalog.entries["test_model_v1"] + assert entry.content_sha256 == digest + assert entry.rule["aliases"] == ("qwen3.5-plus",) + assert entry.rule["instruction_mapping"]["ref"] == "system_only_v1" + assert load_supplements(tmp_path, sources).revision == catalog.revision + with pytest.raises(TypeError): + entry.rule["aliases"] = () + write_rule(tmp_path, MODEL_RULE.replace('"test-rule-v1"', '"changed-evidence"')) + assert load_supplements(tmp_path, sources).revision != catalog.revision + assert catalog.entries["test_model_v1"].rule["evidence"] == "test-rule-v1" + assert file.exists() + + +def test_empty_explicit_manifest_is_valid(tmp_path, sources): + (tmp_path / "rules").mkdir() + (tmp_path / "rules/manifest.toml").write_text("schema_version = 1\nrules = []\n") + assert dict(load_supplements(tmp_path, sources).entries) == {} + + +@pytest.mark.parametrize( + "replacement", + [ + ("schema_version = 1", "schema_version = true"), + ("schema_version = 1", "schema_version = 2"), + ("version = 1\ntarget_kind", "version = true\ntarget_kind"), + ("version = 1\ntarget_kind", "version = 0\ntarget_kind"), + ('target_kind = "model"', 'target_kind = "unknown"'), + ('aliases = ["qwen3.5-plus"]', 'aliases = ["same", "same"]'), + ('aliases = ["qwen3.5-plus"]', 'endpoint_hints = ["url"]'), + ('aliases = ["qwen3.5-plus"]', 'model_profile_key = "alibaba/qwen3.5-plus"'), + ('ref = "system_only_v1"', 'ref = "unregistered"'), + ('reason = "Declare traditional mapping separately"', 'reason = " "'), + ], +) +def test_invalid_rule_contract_is_rejected_without_raw_contents( + tmp_path, sources, replacement +): + write_rule(tmp_path, MODEL_RULE.replace(*replacement)) + with pytest.raises( + SupplementReadError, match="SUPPLEMENT_RULE_INVALID" + ) as captured: + load_supplements(tmp_path, sources) + assert captured.value.file_path == "rules/models/test.toml" + assert "Declare traditional" not in str(captured.value) + + +@pytest.mark.parametrize( + "content,category", + [ + ( + MODEL_RULE.replace('"alibaba/qwen3.5-plus"', '"missing"'), + "SUPPLEMENT_TARGET_MISSING", + ), + ( + MODEL_RULE.replace( + 'target_kind = "model"', 'target_kind = "provider_model"' + ) + .replace( + 'target_key = "alibaba/qwen3.5-plus"', + 'target_key = "alibaba-cn/qwen3.5-plus"', + ) + .replace('aliases = ["qwen3.5-plus"]', 'model_profile_key = "missing"'), + "SUPPLEMENT_PATH_OWNERSHIP", + ), + ], +) +def test_missing_target_and_wrong_directory_are_explicit( + tmp_path, sources, content, category +): + write_rule(tmp_path, content) + with pytest.raises(SupplementReadError, match=category): + load_supplements(tmp_path, sources) + + +def test_provider_model_link_requires_known_atom(tmp_path, sources): + content = MODEL_RULE.replace( + 'target_kind = "model"', 'target_kind = "provider_model"' + ) + content = content.replace( + 'target_key = "alibaba/qwen3.5-plus"', 'target_key = "alibaba-cn/qwen3.5-plus"' + ) + content = content.replace( + 'aliases = ["qwen3.5-plus"]', 'model_profile_key = "missing"' + ) + write_rule(tmp_path, content, "rules/providers/test.toml") + with pytest.raises(SupplementReadError, match="SUPPLEMENT_MODEL_MISSING"): + load_supplements(tmp_path, sources) + + +@pytest.mark.parametrize( + "path", + [ + "../outside.toml", + "rules/models/../test.toml", + "models/test.toml", + "rules/models/test.json", + ], +) +def test_manifest_paths_are_canonical_and_controlled(tmp_path, sources, path): + write_manifest(tmp_path, [(path, "a" * 64)]) + with pytest.raises(SupplementReadError, match="SUPPLEMENT_PATH_INVALID"): + load_supplements(tmp_path, sources) + + +def test_content_digest_mismatch_and_missing_file(tmp_path, sources): + file, _ = write_rule(tmp_path) + file.write_text(file.read_text() + "\n# different bytes\n") + with pytest.raises(SupplementReadError, match="SUPPLEMENT_HASH_MISMATCH"): + load_supplements(tmp_path, sources) + file.unlink() + with pytest.raises(SupplementReadError, match="SUPPLEMENT_FILE_INVALID"): + load_supplements(tmp_path, sources) + + +def test_invalid_utf8_toml_and_external_symlink_are_sanitized(tmp_path, sources): + path = "rules/models/test.toml" + file, _ = write_rule(tmp_path) + for data in [b"\xff", b"not toml = ["]: + file.write_bytes(data) + write_manifest(tmp_path, [(path, hashlib.sha256(data).hexdigest())]) + with pytest.raises(SupplementReadError, match="SUPPLEMENT_FILE_INVALID"): + load_supplements(tmp_path, sources) + file.unlink() + file.symlink_to(Path(__file__)) + with pytest.raises(SupplementReadError, match="SUPPLEMENT_FILE_INVALID"): + load_supplements(tmp_path, sources) + + +def test_duplicate_files_and_rules_never_publish_partial_catalog(tmp_path, sources): + file, digest = write_rule(tmp_path) + catalog = load_supplements(tmp_path, sources) + write_manifest( + tmp_path, + [("rules/models/test.toml", digest), ("rules/models/test.toml", digest)], + ) + with pytest.raises(SupplementReadError, match="SUPPLEMENT_PATH_INVALID"): + load_supplements(tmp_path, sources) + other = file.parent / "other.toml" + other.write_text(file.read_text()) + write_manifest( + tmp_path, + [("rules/models/test.toml", digest), ("rules/models/other.toml", digest)], + ) + with pytest.raises(SupplementReadError, match="SUPPLEMENT_DUPLICATE"): + load_supplements(tmp_path, sources) + assert catalog.entries["test_model_v1"].rule["target_key"] == "alibaba/qwen3.5-plus" + + +def test_parameter_contracts_checked_but_cross_atom_conditions_wait_for_composition( + tmp_path, sources +): + parameters = """\n[[parameters]] +key = "temperature" +visible = true +editable = true +required = false +allowed_scopes = ["outlet"] +ui_group = "common" +owner = "model" +default_value = 0 +[parameters.source] +rule_id = "test" +[parameters.schema] +type = "number" +minimum = 0 +[parameters.condition] +key = "provider_switch" +op = "eq" +value = true +""" + write_rule(tmp_path, MODEL_RULE + parameters) + loaded = load_supplements(tmp_path, sources) + assert loaded.entries["test_model_v1"].rule["parameters"][0]["default_value"] == 0 + write_rule( + tmp_path, + MODEL_RULE + parameters.replace("default_value = 0", "default_value = false"), + ) + with pytest.raises(SupplementReadError, match="SUPPLEMENT_RULE_INVALID"): + load_supplements(tmp_path, sources) + + +@pytest.mark.parametrize( + "manifest", + [ + "schema_version = 2\nrules = []", + "schema_version = true\nrules = []", + 'schema_version = 1\nunknown = "secret"\nrules = []', + ], +) +def test_manifest_schema_version_and_unknown_fields_fail(tmp_path, sources, manifest): + (tmp_path / "rules").mkdir() + (tmp_path / "rules/manifest.toml").write_text(manifest) + with pytest.raises(SupplementReadError, match="SUPPLEMENT_MANIFEST_INVALID"): + load_supplements(tmp_path, sources) + + +def test_diagnostics_identify_fields_without_echoing_values(tmp_path, sources): + write_rule( + tmp_path, + MODEL_RULE.replace( + "version = 1\ntarget_kind", 'version = "secret"\ntarget_kind' + ), + ) + with pytest.raises(SupplementReadError) as captured: + load_supplements(tmp_path, sources) + assert captured.value.field_path == "version" + assert "secret" not in str(captured.value) + write_rule(tmp_path, MODEL_RULE.replace('ref = "system_only_v1"', 'ref = "secret"')) + with pytest.raises(SupplementReadError) as captured: + load_supplements(tmp_path, sources) + assert captured.value.field_path == "instruction_mapping" + assert "secret" not in str(captured.value) + + +@pytest.mark.parametrize("value", ["nan", "inf", "-inf"]) +def test_capability_override_values_must_be_finite_json(tmp_path, sources, value): + override = f"""\n[[capability_overrides]] +path = "/context/combined_window" +value = {value} +reason = "test" +evidence = "test" +""" + write_rule(tmp_path, MODEL_RULE + override) + with pytest.raises( + SupplementReadError, match="SUPPLEMENT_RULE_INVALID" + ) as captured: + load_supplements(tmp_path, sources) + assert captured.value.field_path == "/capability_overrides/0/value" + + +JSON_PARAMETER = """ +[[parameters]] +key = "reasoning_effort" +visible = true +editable = true +required = false +allowed_scopes = ["outlet"] +ui_group = "reasoning" +owner = "model" +[parameters.source] +rule_id = "test" +[json_fields] +"/parameters/0/schema" = '{"enum": [null, "low"]}' +"/parameters/0/default_value" = 'null' +""" + + +def test_explicit_json_fields_preserve_null_without_string_aliases(tmp_path, sources): + write_rule(tmp_path, MODEL_RULE + JSON_PARAMETER) + parameters = ( + load_supplements(tmp_path, sources).entries["test_model_v1"].rule["parameters"] + ) + assert parameters[0]["default_value"] is None + assert parameters[0]["schema"]["enum"] == (None, "low") + content = JSON_PARAMETER.replace( + '\'{"enum": [null, "low"]}\'', '\'{"type": "string"}\'' + ) + content = content.replace("= 'null'", "= '\"null\"'") + write_rule(tmp_path, MODEL_RULE + content) + assert ( + load_supplements(tmp_path, sources) + .entries["test_model_v1"] + .rule["parameters"][0]["default_value"] + == "null" + ) + + +@pytest.mark.parametrize( + "encoded", ["NaN", "Infinity", "1e9999", '{"key": 0, "key": 1}', "invalid"] +) +def test_embedded_json_is_strict_and_sanitized(tmp_path, sources, encoded): + write_rule( + tmp_path, MODEL_RULE + JSON_PARAMETER.replace("= 'null'", f"= '{encoded}'") + ) + with pytest.raises(SupplementReadError, match="SUPPLEMENT_JSON_FIELD_INVALID"): + load_supplements(tmp_path, sources) + + +@pytest.mark.parametrize( + "pointer", + [ + "/parameters/00/default_value", + "/parameters/1/default_value", + "/parameters/0/key", + "/capability_overrides/0/schema", + ], +) +def test_json_fields_cannot_patch_arbitrary_paths(tmp_path, sources, pointer): + write_rule( + tmp_path, + MODEL_RULE + JSON_PARAMETER.replace("/parameters/0/default_value", pointer), + ) + with pytest.raises(SupplementReadError, match="SUPPLEMENT_JSON_FIELD_INVALID"): + load_supplements(tmp_path, sources) + + +def test_json_and_toml_cannot_both_define_the_same_field(tmp_path, sources): + content = JSON_PARAMETER.replace( + 'key = "reasoning_effort"', 'key = "reasoning_effort"\ndefault_value = "low"' + ) + write_rule(tmp_path, MODEL_RULE + content) + with pytest.raises(SupplementReadError, match="SUPPLEMENT_JSON_FIELD_CONFLICT"): + load_supplements(tmp_path, sources)