diff --git a/nemoguardrails/server/experimental/_json_payload.py b/nemoguardrails/server/experimental/_json_payload.py new file mode 100644 index 0000000000..c562e10a3a --- /dev/null +++ b/nemoguardrails/server/experimental/_json_payload.py @@ -0,0 +1,93 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import json +import math +from collections.abc import Mapping, Sequence +from typing import Any + +from nemoguardrails.server.experimental.provider.types import JsonObject + + +class InvalidJson(ValueError): + """Report a payload that is not standards-compliant JSON.""" + + +class UnsupportedJsonShape(ValueError): + """Report valid JSON that cannot be handled without ambiguity.""" + + +def _unique_object(pairs: Sequence[tuple[str, Any]]) -> JsonObject: + """Build an object while rejecting duplicate member names. + + Names that differ only by case are left to the projection models, which + close every reviewed object. Opaque provider data may legitimately use + such names. + """ + result: JsonObject = {} + for key, value in pairs: + if key in result: + raise UnsupportedJsonShape(f"Duplicate JSON member {key!r} is not supported.") + result[key] = value + return result + + +def _reject_nonstandard_number(value: str): + """Reject JSON constants that are not part of the standard grammar.""" + raise InvalidJson(f"Non-standard JSON number {value!r} is not supported.") + + +def _parse_finite_float(value: str) -> float: + """Parse a JSON float only when it has a finite value.""" + parsed = float(value) + if not math.isfinite(parsed): + raise InvalidJson(f"Non-finite JSON number {value!r} is not supported.") + return parsed + + +def _parse_integer(value: str) -> int: + """Parse an integer while normalizing interpreter size-limit failures.""" + try: + return int(value) + except ValueError as error: + raise InvalidJson("The JSON integer exceeds the supported size.") from error + + +def parse_json_object(body: bytes) -> JsonObject: + """Parse a JSON object without accepting ambiguous or nonstandard input.""" + + if not isinstance(body, bytes): + raise TypeError("A provider JSON body must be bytes.") + try: + payload = json.loads( + body, + object_pairs_hook=_unique_object, + parse_constant=_reject_nonstandard_number, + parse_float=_parse_finite_float, + parse_int=_parse_integer, + ) + except UnsupportedJsonShape: + raise + except (UnicodeDecodeError, json.JSONDecodeError, RecursionError, InvalidJson) as error: + raise InvalidJson("The payload must be valid JSON.") from error + if not isinstance(payload, dict): + raise UnsupportedJsonShape("The JSON payload must be an object.") + return payload + + +def encode_json_object(payload: Mapping[str, Any]) -> bytes: + """Serialize a modified provider object as compact UTF-8 JSON.""" + + return json.dumps(payload, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode() diff --git a/nemoguardrails/server/experimental/contracts/README.md b/nemoguardrails/server/experimental/contracts/README.md index f8ac3798c2..a25203021c 100644 --- a/nemoguardrails/server/experimental/contracts/README.md +++ b/nemoguardrails/server/experimental/contracts/README.md @@ -42,11 +42,13 @@ allowed, but every other key starting with `x-nemo` is reserved and rejected. | `classification: constrained` | Restricts a value or shape without making it a rail subject. | | `gate: disabled` | Marks a constrained, null-only field for an unsupported feature. | | `classification: opaque` | Leaves the value under provider authority, without guardrail inspection. | -| `opaque_fields` | Lists reviewed opaque names on this object without declaring each property. | +| `opaque_fields` | Legacy object-level inventory; canonical exports use opaque-classified properties. | | `reason` | Explains a restriction: capability, provider integrity, or projection policy. | Opaque does not mean safe or trusted. Its nested content is not independently -reviewed. Opaque names should not overlap declared properties or use wildcards. +reviewed. Canonical exports declare each reviewed opaque name in `properties`. +A legacy `opaque_fields` inventory must not overlap declared properties or use +wildcards. `extension: true` identifies a local compatibility field outside the pinned provider schema, not an exemption from review. @@ -65,8 +67,18 @@ enables a detector or a runtime capability. `model` identifies the Python model; `source` identifies a provider schema component. `unknown_fields: configurable` marks an object whose unknown-field -handling depends on runtime validation context. Generic JSON Schema validators -do not enforce this annotation. `additionalProperties` remains the ordinary +handling depends on runtime validation context. Validation rejects unknown +members by default; only trusted configuration may allow them on such an object. +Objects without the marker always reject them. Generic JSON Schema validators +do not enforce this annotation. Reviewed opaque names appear as properties +classified `opaque`, so `additionalProperties` describes only unreviewed members. + +A member whose name matches a listed property only after Unicode case folding +(`casefold()`), such as `Tools` beside `tools`, is always rejected, even where +unknown members are allowed: some providers match names case-insensitively. +Whitespace remains part of the name. JSON Schema could state this with +`propertyNames`, but the contract vocabulary does not include it, so runtimes +and generators must implement the rule themselves. `additionalProperties` remains the ordinary schema-level object closure rule. Provider-required opaque fields may be left to provider validation; a projection is not a full provider request validator. @@ -105,3 +117,7 @@ choose deployment rails or enable runtime profile selection. This is a NeMo Guardrails document using JSON Schema vocabulary, not an OpenAPI document or an OpenAPI Overlay. Generic schema tools do not execute its guardrail annotations. + +## Provider boundaries + +- [OpenAI Chat Completions](openai/README.md) diff --git a/nemoguardrails/server/experimental/contracts/openai/README.md b/nemoguardrails/server/experimental/contracts/openai/README.md new file mode 100644 index 0000000000..615a11b20b --- /dev/null +++ b/nemoguardrails/server/experimental/contracts/openai/README.md @@ -0,0 +1,54 @@ +# OpenAI Chat Completions contract + +Operation `createChatCompletion` uses the `single_text.v1` capability profile +for its guarded text boundary. This page summarizes that boundary and its +provider source. Neither the summary nor the source pin enables runtime +behavior. See [Guard contracts](../README.md) for the document vocabulary. + +## Guarded boundary + +| Area | Single-text boundary | +| --- | --- | +| Request | One user message with non-empty string content at `messages[0].content`. | +| Buffered response | One assistant choice with string content at `choices[0].message.content`. Empty or null content means the response has nothing to inspect, because every other reviewed field is content-free. The projection reports this through `has_text`; relaying such a response without output checks is left to the integration. | +| Constrained values | User/assistant roles, single-item arrays, and request `n` constrained to one. | +| Unsupported content | Tool, audio, multimodal, refusal, participant name, and separate reasoning content where explicitly disabled. | +| Opaque data | Reviewed provider-owned metadata and controls, not additional guarded subjects. | +| Closed objects | The request, its user message, the buffered response, and its choice and assistant message reject members outside OpenAI's fields. The user message, choice, and assistant message are configurable: only trusted configuration may allow unknown members there. Fields specific to compatible servers belong to their own reviewed extensions. | + +This request boundary does not accept conversation histories, system/developer +messages alongside the user message, or multimodal content blocks. Disabled +fields are null-only: a non-null value does not become acceptable merely because +it requests plain text or no tools. A few unsupported features are constrained +instead, because OpenAI's schema also allows a value that carries no content: +response `tool_calls` and `annotations` accept null or an empty list, and +request `logprobs` accepts null or `false`. + +Log probabilities are unsupported: they carry token text that rails do not +inspect. Request `logprobs` accepts only `false` or null, request `top_logprobs` +accepts only null, and buffered response `choices[0].logprobs` accepts only null. +The request check applies even when output inspection is off. + +Text replacement eligibility is separate from endpoint support for replacement +outcomes. Response annotations must be null or empty, because citation text is +not inspected. The policy still declares that non-empty annotations would block +text replacement. Unrelated provider data must remain intact. + +The Python projections, bindings, and endpoint determine exact acceptance and +runtime behavior. Their machine-readable contract belongs with the integration, +not in a separately maintained handwritten policy here. Recognizing the request +`stream` flag does not itself provide a streaming endpoint. Streaming needs its +own event classification, lifecycle handling, and implementation tests; the +buffered boundary is not a claim about accepted stream events. + +## Provider provenance + +[source.yaml](source.yaml) records the OpenAPI document's immutable revision, +document version, download location, and expected SHA-256 digest. The pin is a +review baseline, not a claim to cover every current OpenAI field or compatible +provider. + +The offline metadata tests check pin structure and URL/revision consistency. +They do not fetch the document or verify its bytes against the digest. Reviewing +a provider update requires checking the source artifact and changes to guarded, +constrained, disabled, and opaque fields separately from format validation. diff --git a/nemoguardrails/server/experimental/contracts/openai/source.yaml b/nemoguardrails/server/experimental/contracts/openai/source.yaml new file mode 100644 index 0000000000..c128f5633b --- /dev/null +++ b/nemoguardrails/server/experimental/contracts/openai/source.yaml @@ -0,0 +1,5 @@ +document_url: https://github.com/openai/openai-openapi/blob/df63773f69f542ef875b9f00c3837c25ba5f4f2a/openapi.yaml +download_url: https://raw.githubusercontent.com/openai/openai-openapi/df63773f69f542ef875b9f00c3837c25ba5f4f2a/openapi.yaml +revision: df63773f69f542ef875b9f00c3837c25ba5f4f2a +document_version: 2.3.0 +sha256: f2dae1a9aced09b91310db89edda51bf1e36ecbfb05230c3a50b239c07708469 diff --git a/nemoguardrails/server/experimental/provider/payload.py b/nemoguardrails/server/experimental/provider/payload.py new file mode 100644 index 0000000000..9f5198658b --- /dev/null +++ b/nemoguardrails/server/experimental/provider/payload.py @@ -0,0 +1,394 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Payload projection models and guarded text locations.""" + +from collections.abc import Sequence +from dataclasses import dataclass +from enum import Enum +from typing import Annotated, ClassVar, Literal, cast + +from pydantic import BaseModel, BeforeValidator, ConfigDict, ValidationError, ValidationInfo, model_validator +from typing_extensions import Self + +from nemoguardrails.server.experimental.provider.types import GuardedMessage, JsonObject, UnknownContentFieldPolicy + +_UNKNOWN_CONTENT_FIELDS_CONTEXT = "unknown_content_fields" + + +def _require_boolean(value: object) -> object: + """Reject integer values where a strict boolean is required.""" + if not isinstance(value, bool): + raise ValueError("value must be a boolean") + return value + + +StrictFalse = Annotated[Literal[False], BeforeValidator(_require_boolean)] + + +class PayloadCapabilityProfile(str, Enum): + """Identify a closed payload capability understood by the framework.""" + + SINGLE_TEXT_V1 = "single_text.v1" + + +@dataclass(frozen=True, slots=True) +class ProjectionFieldCoverage: + """Classify every reviewed field of one provider object.""" + + guarded_fields: frozenset[str] = frozenset() + constrained_fields: frozenset[str] = frozenset() + opaque_fields: frozenset[str] = frozenset() + local_extension_fields: frozenset[str] = frozenset() + + def __post_init__(self) -> None: + """Require each reviewed field to have exactly one classification.""" + classifications = ( + self.guarded_fields, + self.constrained_fields, + self.opaque_fields, + self.local_extension_fields, + ) + repeated = set() + for index, fields in enumerate(classifications): + for other_fields in classifications[index + 1 :]: + repeated.update(fields & other_fields) + if repeated: + raise ValueError(f"Projection fields have multiple classifications: {sorted(repeated)}") + + @property + def reviewed_fields(self) -> frozenset[str]: + """Return every field accounted for by this coverage declaration.""" + return self.guarded_fields | self.constrained_fields | self.opaque_fields | self.local_extension_fields + + +@dataclass(frozen=True, slots=True) +class ProjectionModelContract: + """Bind field coverage to one nested projection model.""" + + model: type[BaseModel] + coverage: ProjectionFieldCoverage + source_schema: str | None = None + + def __post_init__(self) -> None: + """Validate an explicitly declared provider source schema.""" + if self.source_schema is not None and not self.source_schema.strip(): + raise ValueError("A projection source schema must be non-empty when declared.") + + +@dataclass(frozen=True, slots=True) +class PayloadProjectionContract: + """Describe the runtime semantics of one guarded payload projection.""" + + projection_id: str + direction: Literal["request", "response"] + profile: PayloadCapabilityProfile + root: ProjectionFieldCoverage + content_models: tuple[ProjectionModelContract, ...] = () + + def __post_init__(self) -> None: + """Validate the projection identity and nested model declarations.""" + if not self.projection_id.strip(): + raise ValueError("A payload projection contract must have an identifier.") + models = [entry.model for entry in self.content_models] + if len(models) != len(set(models)): + raise ValueError("A payload projection contract must describe each content model once.") + + +def validate_payload_projection_contract( + model: type[BaseModel], + expected_direction: Literal["request", "response"], +) -> PayloadProjectionContract: + """Validate and return the semantic contract attached to a payload model.""" + + contract = getattr(model, "projection_contract", None) + if not isinstance(contract, PayloadProjectionContract): + raise ValueError(f"Payload model {model.__name__!r} must declare a projection contract.") + if contract.direction != expected_direction: + raise ValueError( + f"Payload model {model.__name__!r} declares {contract.direction!r}, expected {expected_direction!r}." + ) + + def field_names(projection_model: type[BaseModel]) -> frozenset[str]: + """Return serialized field names for one projection model.""" + return frozenset( + field.alias if field.alias is not None else name for name, field in projection_model.model_fields.items() + ) + + missing_root_fields = field_names(model) - contract.root.reviewed_fields + if missing_root_fields: + raise ValueError( + f"Payload contract {contract.projection_id!r} omits root fields: {sorted(missing_root_fields)}" + ) + for entry in contract.content_models: + missing_fields = field_names(entry.model) - entry.coverage.reviewed_fields + if missing_fields: + raise ValueError( + f"Payload contract {contract.projection_id!r} omits {entry.model.__name__} fields: " + f"{sorted(missing_fields)}" + ) + return contract + + +def guarded_schema_error(error: ValidationError, subject: str) -> str: + """Describe one projection failure without echoing provider content.""" + + issue = error.errors(include_url=False, include_context=False, include_input=False)[0] + location = ".".join(str(part) for part in issue["loc"]) + location_text = f" at {location}" if location else "" + return f"The {subject} does not match the guarded content schema{location_text}: {issue['msg']}." + + +class GuardedProjectionModel(BaseModel): + """Provide common validation behavior for guarded payload projections.""" + + model_config = ConfigDict(extra="allow", frozen=True) + projection_contract: ClassVar[PayloadProjectionContract | None] = None + + @classmethod + def validate_payload( + cls, + document: object, + *, + unknown_content_fields: UnknownContentFieldPolicy = UnknownContentFieldPolicy.FORBID, + ) -> Self: + """Validate a provider payload with the requested unknown-field policy. + + FORBID is the default. ALLOW opens only objects whose policy declares + configurable unknown fields, and is reserved for trusted configuration. + """ + return cls.model_validate(document, context={_UNKNOWN_CONTENT_FIELDS_CONTEXT: unknown_content_fields}) + + +def unknown_content_field_policy(info: ValidationInfo) -> UnknownContentFieldPolicy: + """Return the unknown-field policy requested for a validation, closed by default.""" + policy = (info.context or {}).get(_UNKNOWN_CONTENT_FIELDS_CONTEXT, UnknownContentFieldPolicy.FORBID) + return UnknownContentFieldPolicy(policy) + + +class GuardedContentModel(GuardedProjectionModel): + """Mark a content-bearing provider object. + + Its object policy decides whether unreviewed members are accepted. + """ + + +@dataclass(slots=True) +class GuardedMessageTarget: + """Identify the exact provider message field guarded by the pipeline.""" + + role: Literal["user", "assistant"] + _object: JsonObject + _member: str + allows_replacement: bool + + @property + def has_text(self) -> bool: + """Return whether the target holds text for content checks. + + A validated projection reaches an empty or null target only when its + location allows empty text. The projection has then shown that no + other reviewed field carries content, so there is nothing to inspect. + """ + content = self._object.get(self._member) + return isinstance(content, str) and content != "" + + @property + def message(self) -> GuardedMessage: + """Return the message currently stored at the target.""" + content = self._object[self._member] + if not isinstance(content, str): + raise TypeError("Guarded message content must be a string.") + return GuardedMessage(role=self.role, content=content) + + def replace_content(self, content: str) -> None: + """Replace guarded content when the contract permits modification.""" + if not self.allows_replacement: + raise ValueError("The guarded provider content does not support replacement.") + self._object[self._member] = content + + +@dataclass(frozen=True, slots=True) +class GuardedArraySelection: + """Select one array member by provider field values.""" + + match: tuple[tuple[str, str | int | bool], ...] + cardinality: Literal["exactly_one"] + + def select(self, items: Sequence[object]) -> object: + """Return the single array item matching the declared discriminator.""" + missing = object() + matches = [] + for item in items: + values = { + field: item.get(field, missing) if isinstance(item, dict) else getattr(item, field, missing) + for field, _ in self.match + } + if all( + type(values[field]) is type(expected) and values[field] == expected for field, expected in self.match + ): + matches.append(item) + if len(matches) != 1: + raise ValueError("Guarded provider content requires exactly one matching array member.") + return matches[0] + + +@dataclass(frozen=True, slots=True) +class GuardedTextLocation: + """Locate one guarded text member in a validated provider payload.""" + + role: Literal["user", "assistant"] + object_path: tuple[str | int | GuardedArraySelection, ...] + member: str + allows_replacement: bool + replacement_blocked_by: str | None = None + allows_empty: bool = False + + def locate(self, payload: JsonObject) -> GuardedMessageTarget: + """Locate guarded text in the original provider payload.""" + value: object = payload + for segment in self.object_path: + if isinstance(segment, str) and isinstance(value, dict): + value = cast(JsonObject, value)[segment] + elif isinstance(segment, int) and isinstance(value, list): + value = cast(list[object], value)[segment] + elif isinstance(segment, GuardedArraySelection) and isinstance(value, list): + value = segment.select(cast(list[object], value)) + else: + raise TypeError("Guarded text location does not match the provider payload.") + if not isinstance(value, dict): + raise TypeError("Guarded text location must resolve to a provider object.") + allows_replacement = self.allows_replacement and not ( + self.replacement_blocked_by is not None and bool(value.get(self.replacement_blocked_by)) + ) + return GuardedMessageTarget( + self.role, + cast(JsonObject, value), + self.member, + allows_replacement=allows_replacement, + ) + + def validate_projection(self, projection: BaseModel) -> None: + """Require the projected location to resolve to text.""" + value: object = projection + for segment in self.object_path: + if isinstance(segment, str) and isinstance(value, BaseModel): + value = getattr(value, segment) + elif isinstance(segment, int) and isinstance(value, list): + value = cast(list[object], value)[segment] + elif isinstance(segment, GuardedArraySelection) and isinstance(value, list): + value = segment.select(cast(list[object], value)) + else: + raise ValueError("Guarded text location does not match the provider projection.") + if not isinstance(value, BaseModel): + raise ValueError("Guarded text location does not resolve to projected text.") + text = getattr(value, self.member, None) + if not isinstance(text, str) and not (self.allows_empty and text is None): + raise ValueError("Guarded text location does not resolve to projected text.") + + def matches_projection(self, projection: BaseModel) -> bool: + """Return whether this location resolves to text in the projection.""" + + def unwrap_root(value: object) -> object: + """Unwrap nested Pydantic root models.""" + while isinstance(value, BaseModel) and type(value).__pydantic_root_model__: + value = getattr(value, "root") + return value + + try: + value: object = projection + for segment in self.object_path: + value = unwrap_root(value) + if isinstance(segment, str) and isinstance(value, BaseModel): + value = getattr(value, segment) + elif isinstance(segment, int) and isinstance(value, list): + value = cast(list[object], value)[segment] + elif isinstance(segment, GuardedArraySelection) and isinstance(value, list): + value = segment.select(cast(list[object], value)) + else: + return False + value = unwrap_root(value) + return isinstance(value, BaseModel) and isinstance(unwrap_root(getattr(value, self.member)), str) + except (AttributeError, IndexError, ValueError): + return False + + +@dataclass(frozen=True, slots=True) +class GuardedTextAlternatives: + """Select one of several provider encodings for the same guarded text.""" + + locations: tuple[GuardedTextLocation, ...] + + def _projection_location(self, projection: BaseModel) -> GuardedTextLocation: + """Return the single location matching the projected representation.""" + matches = [location for location in self.locations if location.matches_projection(projection)] + if len(matches) != 1: + raise ValueError("Guarded text alternatives require exactly one matching representation.") + return matches[0] + + def locate(self, projection: BaseModel, payload: JsonObject) -> GuardedMessageTarget: + """Locate guarded text using the projected provider representation.""" + return self._projection_location(projection).locate(payload) + + def validate_projection(self, projection: BaseModel) -> None: + """Require exactly one alternative to match the projection.""" + self._projection_location(projection) + + +class GuardedPayloadModel(GuardedProjectionModel): + """Validate and locate guarded text in a provider payload projection.""" + + guarded_text_location: ClassVar[GuardedTextLocation | GuardedTextAlternatives | None] = None + + @classmethod + def has_guarded_text_binding(cls) -> bool: + """Return whether the model declares how to locate guarded text.""" + return ( + cls.guarded_text_location is not None + or cls.locate_guarded_message is not GuardedPayloadModel.locate_guarded_message + ) + + @model_validator(mode="after") + def validate_guarded_text_target(self) -> Self: + """Validate the declared guarded text location after model parsing.""" + if self.guarded_text_location is not None: + self.guarded_text_location.validate_projection(self) + return self + + def locate_guarded_message(self, payload: JsonObject) -> GuardedMessageTarget: + """Locate the guarded message in the original provider payload.""" + if self.guarded_text_location is None: + raise TypeError("The guarded payload model does not declare a guarded text location.") + if isinstance(self.guarded_text_location, GuardedTextAlternatives): + return self.guarded_text_location.locate(self, payload) + return self.guarded_text_location.locate(payload) + + +class GuardedRequestModel(GuardedPayloadModel): + """Validate a provider request and determine its response mode.""" + + stream_selector_field: ClassVar[str | None] = None + + @classmethod + def has_response_mode_binding(cls) -> bool: + """Return whether the request declares how to detect streaming.""" + return cls.stream_selector_field is not None or cls.streams_response is not GuardedRequestModel.streams_response + + @property + def streams_response(self) -> bool: + """Return the provider response mode selected by this request.""" + if self.stream_selector_field is None: + raise TypeError("The guarded request model does not declare a stream selector field.") + return bool(getattr(self, self.stream_selector_field)) diff --git a/nemoguardrails/server/experimental/provider/projection_policy.py b/nemoguardrails/server/experimental/provider/projection_policy.py new file mode 100644 index 0000000000..bade5406e4 --- /dev/null +++ b/nemoguardrails/server/experimental/provider/projection_policy.py @@ -0,0 +1,462 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Field and object policies for provider payload projections. + +Model declarations provide field coverage, guarded text locations, and +exported contract schemas. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any, ClassVar, Literal, cast, get_args + +from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, model_validator +from pydantic.fields import FieldInfo +from typing_extensions import Self + +from nemoguardrails.server.experimental.provider.payload import ( + GuardedTextLocation, + PayloadCapabilityProfile, + PayloadProjectionContract, + ProjectionFieldCoverage, + ProjectionModelContract, + unknown_content_field_policy, +) +from nemoguardrails.server.experimental.provider.types import UnknownContentFieldPolicy + +EXTENSION = "x-nemo-guardrails" +CONTRACT_VERSION = "1.0.0-alpha.1" + +# Formats from guard-contract.schema.json, checked when policy is declared. +_REASON = re.compile(r"(core_capability|provider_integrity|projection_policy)\.[a-z][a-z0-9_]*") +_SOURCE_COMPONENT = re.compile(r"[A-Za-z_][A-Za-z0-9._-]*") + + +def _check_reason(reason: str) -> None: + """Require a structured reason such as core_capability.tool_content.""" + if not _REASON.fullmatch(reason): + raise ValueError(f"Reason {reason!r} must be a structured reason such as 'core_capability.tool_content'") + + +@dataclass(frozen=True) +class ObjectPolicy: + """Describe the reviewed boundary of one JSON object. + + Attributes: + source: Provider component schema name, without a JSON Pointer prefix. + This is provenance metadata, not a schema lookup or validation step. + opaque: Reviewed passthrough fields not declared as Pydantic fields. + Names must be unique and must not overlap declared fields. + unknown_fields: How unreviewed members are handled. "forbid" always + rejects them. "configurable" rejects them unless validation + explicitly allows unknown content fields, which is reserved for + trusted configuration. Members listed in opaque are reviewed. + + This metadata does not select guardrails, a deployment configuration, or a + capability profile. Field-specific policy belongs on the field annotations. + """ + + source: str | None = None + opaque: tuple[str, ...] = () + unknown_fields: Literal["forbid", "configurable"] = "forbid" + + def __post_init__(self) -> None: + """Reject values the exported contract format cannot represent.""" + if self.source is not None and not _SOURCE_COMPONENT.fullmatch(self.source): + raise ValueError(f"Source {self.source!r} must be a bare component schema name") + if any(not name or name == "*" for name in self.opaque): + raise ValueError("Opaque fields must be named; wildcards are not supported") + + +class PolicyModel(BaseModel): + """Require declared fields to carry guard policy metadata. + + Combine this base with the appropriate runtime projection base. At subclass + creation it checks field classifications, opaque inventory overlap, disabled + field defaults, and unsupported aliases. At validation it enforces the + object's unknown-field policy. It does not prove upstream field coverage or + serialize arbitrary custom validators into the exported schema. + """ + + # Unknown members are kept so the object policy, not Pydantic, decides them. + model_config = ConfigDict(extra="allow") + policy: ClassVar[ObjectPolicy] = ObjectPolicy() + + @classmethod + def __pydantic_init_subclass__(cls, **kwargs: Any) -> None: + """Check local policy consistency after Pydantic has assembled fields.""" + super().__pydantic_init_subclass__(**kwargs) + if len(set(cls.policy.opaque)) != len(cls.policy.opaque): + raise ValueError(f"{cls.__name__}: duplicate opaque fields") + overlap = set(cls.policy.opaque) & cls.model_fields.keys() + if overlap: + raise ValueError(f"{cls.__name__}: opaque overlaps declared fields: {sorted(overlap)}") + for name, field in cls.model_fields.items(): + metadata = field_policy(field) + if metadata.get("classification") not in {"guarded", "constrained", "opaque"}: + raise ValueError(f"{cls.__name__}.{name}: missing field policy") + if field.alias or field.validation_alias or field.serialization_alias: + raise ValueError(f"{cls.__name__}.{name}: aliases are not supported") + if metadata.get("gate") == "disabled" and ( + field.annotation is not type(None) or field.is_required() or field.default is not None + ): + raise ValueError(f"{cls.__name__}.{name}: disabled fields must be optional and null-only") + + @model_validator(mode="after") + def reject_unreviewed_fields(self, info: ValidationInfo) -> Self: + """Reject members outside the reviewed fields unless policy allows them. + + A case variant of a reviewed name, such as ``Tools``, is always + rejected: some providers match member names case-insensitively, so it + could reach a field this policy disables or inspects. + """ + reviewed = {name.casefold(): name for name in (*type(self).model_fields, *self.policy.opaque)} + unreviewed = sorted(set(self.model_extra or {}) - set(self.policy.opaque)) + for name in unreviewed: + reviewed_name = reviewed.get(name.casefold()) + if reviewed_name is not None: + raise ValueError(f"field {name!r} differs from reviewed field {reviewed_name!r} only by case") + allowed = ( + self.policy.unknown_fields == "configurable" + and unknown_content_field_policy(info) == UnknownContentFieldPolicy.ALLOW + ) + if unreviewed and not allowed: + raise ValueError(f"unreviewed fields are forbidden: {', '.join(unreviewed)}") + return self + + +def field_policy(field: FieldInfo) -> dict[str, Any]: + """Read a field's guard metadata, returning an empty mapping if absent. + + The returned mapping may belong to the field; callers must not mutate it. + """ + extra = field.json_schema_extra + if not isinstance(extra, dict): + return {} + metadata = extra.get(EXTENSION) + return cast(dict[str, Any], metadata) if isinstance(metadata, dict) else {} + + +# Pydantic constraints whose exported keywords (minLength/minItems, +# maxLength/maxItems, pattern) exist in the guard contract field vocabulary. +_EXPORTABLE_CONSTRAINTS = frozenset({"min_length", "max_length", "pattern"}) + + +def _field(metadata: dict[str, Any], constraints: dict[str, Any]) -> Any: + """Build Annotated field metadata while keeping defaults in assignments.""" + if {"default", "default_factory", "alias"} & constraints.keys(): + raise ValueError("Declare defaults explicitly on the field; aliases are not supported") + unsupported = constraints.keys() - _EXPORTABLE_CONSTRAINTS + if unsupported: + raise ValueError(f"Constraints not expressible in the guard contract: {sorted(unsupported)}") + return Field(json_schema_extra={EXTENSION: metadata}, **constraints) + + +def guarded( + role: Literal["user", "assistant"] | None = None, + *, + replaceable: bool | None = None, + blocked_by: str | None = None, + replacement_reason: str | None = None, + **constraints: Any, +) -> Any: + """Mark guarded content or a container leading to guarded content. + + Args: + role: Text subject role. Omit for structural container fields. + replaceable: Whether the subject permits replacement. The buffered + binding defaults to read-only when this is not declared. + blocked_by: Sibling field whose non-empty value prevents replacement. + replacement_reason: Structured reason for the replacement restriction. + **constraints: min_length, max_length, or pattern; other Pydantic + constraints have no guard contract keyword and are rejected. + + Use inside Annotated and put defaults on the field assignment. Declaring + replacement eligibility does not enable runtime replacement support. + """ + metadata: dict[str, Any] = {"classification": "guarded"} + if role is None and any(value is not None for value in (replaceable, blocked_by, replacement_reason)): + raise ValueError("Text replacement policy requires a subject role") + if role is not None: + subject: dict[str, Any] = {"kind": "text", "role": role} + if replaceable is not None: + subject["replaceable"] = replaceable + if blocked_by is not None: + subject["replacement_blocked_by"] = blocked_by + if replacement_reason is not None: + _check_reason(replacement_reason) + subject["replacement_reason"] = replacement_reason + metadata["subject"] = subject + return _field(metadata, constraints) + + +def constrained(*, reason: str | None = None, **constraints: Any) -> Any: + """Mark a field restricted by its type and optional Pydantic constraints. + + The optional reason explains the restriction in the exported contract. + This helper adds no restriction by itself; the annotation and constraints + must express it. Defaults belong on the field assignment. + """ + metadata: dict[str, Any] = {"classification": "constrained"} + if reason is not None: + _check_reason(reason) + metadata["reason"] = reason + return _field(metadata, constraints) + + +def disabled(reason: str, *, extension: bool = False) -> Any: + """Declare an unsupported feature as an omittable, null-only field. + + Use field: Annotated[None, disabled(reason)] = None. Non-null values are + rejected by the field type, and PolicyModel checks this declaration. + A nonblank structured reason explains the restriction. Set extension for a + locally recognized field outside the pinned provider schema. + """ + _check_reason(reason) + metadata: dict[str, Any] = {"classification": "constrained", "gate": "disabled", "reason": reason} + if extension: + metadata["extension"] = True + return _field(metadata, {}) + + +def opaque() -> Any: + """Mark a declared field as provider-owned data outside guarded content. + + Use field: Annotated[Any, opaque()] = None for an unconstrained optional + value. ObjectPolicy.opaque is the alternative for reviewed fields that do + not need a model attribute. Do not declare the same field in both places. + """ + return _field({"classification": "opaque"}, {}) + + +def model_graph(root: type[PolicyModel]) -> dict[str, type[PolicyModel]]: + """Collect the root and nested policy models in traversal order. + + Models are keyed by class name. Reject nested models without policy and + distinct models sharing a name. Repeated references are visited once; + recursive schema export is rejected separately. + """ + models: dict[str, type[PolicyModel]] = {} + + def visit(annotation: Any) -> None: + """Follow model fields and container or union type arguments once.""" + if isinstance(annotation, type) and issubclass(annotation, BaseModel): + if not issubclass(annotation, PolicyModel): + raise ValueError("Nested models must declare field policies") + name = annotation.__name__ + if name in models: + if models[name] is not annotation: + raise ValueError("Model names must be unique within a payload") + return + models[name] = annotation + for field in annotation.model_fields.values(): + visit(field.annotation) + for argument in get_args(annotation): + visit(argument) + + visit(root) + return models + + +def field_coverage(model: type[PolicyModel]) -> ProjectionFieldCoverage: + """Derive disjoint runtime field inventories from one object's declarations. + + Local extensions occupy a separate inventory even when their annotation + is constrained. The opaque inventory includes declared opaque attributes + and ObjectPolicy.opaque names. This does not check an upstream schema. + """ + groups: dict[str, set[str]] = { + "guarded": set(), + "constrained": set(), + "opaque": set(model.policy.opaque), + "extension": set(), + } + for name, field in model.model_fields.items(): + metadata = field_policy(field) + classification = "extension" if metadata.get("extension") else metadata["classification"] + groups[classification].add(name) + return ProjectionFieldCoverage( + guarded_fields=frozenset(groups["guarded"]), + constrained_fields=frozenset(groups["constrained"]), + opaque_fields=frozenset(groups["opaque"]), + local_extension_fields=frozenset(groups["extension"]), + ) + + +def payload_contract( + model: type[PolicyModel], + *, + projection_id: str, + direction: Literal["request", "response"], + profile: PayloadCapabilityProfile = PayloadCapabilityProfile.SINGLE_TEXT_V1, +) -> PayloadProjectionContract: + """Build runtime coverage metadata for a request or response projection. + + Args: + model: Policy-annotated root, with policy-annotated nested models. + projection_id: Stable identifier for this payload projection. + direction: Whether the projection describes a request or response. + profile: Framework-defined capability implemented by the runtime. + + Returns: + Root and nested-model coverage, ready to attach in a binding module. + Text extraction is derived separately by text_location. + """ + return PayloadProjectionContract( + projection_id=projection_id, + direction=direction, + profile=profile, + root=field_coverage(model), + content_models=tuple( + ProjectionModelContract(model=nested, source_schema=nested.policy.source, coverage=field_coverage(nested)) + for nested in model_graph(model).values() + if nested is not model + ), + ) + + +def export_payload_schema(model: type[PolicyModel], *, projection_id: str) -> dict[str, Any]: + """Export a fresh, inline schema with object and field guard annotations. + + Args: + model: Trusted policy-annotated model to describe. + projection_id: Exported root title; the class name remains model metadata. + + Returns: + A payload schema, not a complete operation contract. + Defaults and nullable types are retained. Disjoint nullable anyOf + branches are rendered as oneOf for the current contract vocabulary. + Reviewed opaque names are exported as opaque-classified properties, + and additionalProperties follows the object's unknown-field policy. + + Raises: + ValueError: Nested models are not policy-annotated, names collide, + references recurse, or an anyOf union cannot be safely converted. + + This function does not validate the complete contract format, inspect an + upstream provider schema, or capture arbitrary Python validator behavior. + The operation exporter is responsible for format validation. + """ + document = model.model_json_schema() + definitions = document.pop("$defs", {}) + models = model_graph(model) + + def expand(node: dict[str, Any], active: tuple[str, ...] = ()) -> dict[str, Any]: + """Inline references and attach object policy without mutating the input.""" + if "$ref" in node: + name = node["$ref"].removeprefix("#/$defs/") + if name in active: + raise ValueError("Recursive models are not supported by the contract export") + return expand({**definitions[name], **{k: v for k, v in node.items() if k != "$ref"}}, (*active, name)) + result = dict(node) + metadata = dict(result.get(EXTENSION, {})) + if "properties" in result: + policy = models[result["title"]].policy + if policy.source: + metadata["source"] = "#/components/schemas/" + policy.source + result["additionalProperties"] = policy.unknown_fields == "configurable" + if policy.unknown_fields == "configurable": + metadata["unknown_fields"] = policy.unknown_fields + # Opaque values stay runtime extras, but the export lists them as + # properties so additionalProperties describes only unreviewed members. + properties = { + **result["properties"], + **{name: {EXTENSION: {"classification": "opaque"}} for name in policy.opaque}, + } + result["properties"] = {name: expand(child, active) for name, child in sorted(properties.items())} + else: + result.pop("title", None) + if "items" in result: + result["items"] = expand(result["items"], active) + if "anyOf" in result: + alternatives = result.pop("anyOf") + non_null = [branch for branch in alternatives if branch != {"type": "null"}] + if ( + len(alternatives) != 2 + or len(non_null) != 1 + or non_null[0].get("type") not in {"array", "object", "string", "boolean", "integer", "number"} + ): + raise ValueError("Contract export supports only disjoint nullable anyOf unions") + result["oneOf"] = alternatives + for keyword in ("anyOf", "oneOf", "allOf"): + if keyword in result: + result[keyword] = [expand(child, active) for child in result[keyword]] + if metadata: + result[EXTENSION] = metadata + return result + + exported = expand(document) + exported["title"] = projection_id + exported.setdefault(EXTENSION, {})["model"] = model.__name__ + return exported + + +def text_location(model: type[PolicyModel]) -> GuardedTextLocation: + """Derive one buffered text target by following guarded object fields. + + Traversed arrays must constrain their size to exactly one item. The subject + must be a named string field, optionally nullable; its role and replacement + restrictions become runtime location metadata. A nullable subject, or one + without a minimum length, may hold no text. This helper does not provide union selectors or + streaming classification and does not itself attach the result to a model. + + Raises: + ValueError: No unique text subject can be derived, a traversed array is + not constrained to one item, a replacement blocker is not a field of + the subject's object, or the schema cannot be exported. + """ + locations: list[GuardedTextLocation] = [] + + def visit(node: dict[str, Any], path: tuple[str | int, ...]) -> None: + """Accumulate subject locations along guarded object and array paths.""" + subject = node.get(EXTENSION, {}).get("subject") + if subject: + branches = node.get("oneOf", [node]) + types = [branch.get("type") for branch in branches] + if types not in (["string"], ["string", "null"], ["null", "string"]): + raise ValueError("A guarded subject must be a named string field") + if not path or not isinstance(path[-1], str): + raise ValueError("A guarded subject must be a named string field") + text = branches[types.index("string")] + locations.append( + GuardedTextLocation( + role=subject["role"], + object_path=path[:-1], + member=path[-1], + allows_replacement=subject.get("replaceable", False), + replacement_blocked_by=subject.get("replacement_blocked_by"), + # A nullable subject, or one without a minimum length, may hold no text. + allows_empty="null" in types or text.get("minLength", 0) < 1, + ) + ) + if node.get("type") == "array": + if node.get("minItems") != 1 or node.get("maxItems") != 1: + raise ValueError("Buffered text extraction requires exactly one array item") + visit(node["items"], (*path, 0)) + for name, child in node.get("properties", {}).items(): + blocker = child.get(EXTENSION, {}).get("subject", {}).get("replacement_blocked_by") + # A misspelled blocker would never match and silently allow replacement. + if blocker is not None and blocker not in node["properties"]: + raise ValueError(f"Replacement blocker {blocker!r} is not a field of the subject's object") + if child.get(EXTENSION, {}).get("classification") == "guarded": + visit(child, (*path, name)) + + visit(export_payload_schema(model, projection_id=model.__name__), ()) + if len(locations) != 1: + raise ValueError("Buffered text extraction requires exactly one guarded subject") + return locations[0] diff --git a/nemoguardrails/server/experimental/provider/types.py b/nemoguardrails/server/experimental/provider/types.py index b454144dfa..e0d52faf93 100644 --- a/nemoguardrails/server/experimental/provider/types.py +++ b/nemoguardrails/server/experimental/provider/types.py @@ -16,7 +16,17 @@ """Define provider-neutral values shared by the guarded proxy pipeline.""" from dataclasses import dataclass -from typing import Literal +from enum import Enum +from typing import Any, Literal + +JsonObject = dict[str, Any] + + +class UnknownContentFieldPolicy(str, Enum): + """Control whether unreviewed fields may accompany guarded content.""" + + ALLOW = "allow" + FORBID = "forbid" @dataclass(frozen=True, slots=True) diff --git a/nemoguardrails/server/experimental/providers/__init__.py b/nemoguardrails/server/experimental/providers/__init__.py new file mode 100644 index 0000000000..e8c0ab466e --- /dev/null +++ b/nemoguardrails/server/experimental/providers/__init__.py @@ -0,0 +1,16 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Provider payload implementations for transparent proxy.""" diff --git a/nemoguardrails/server/experimental/providers/openai/__init__.py b/nemoguardrails/server/experimental/providers/openai/__init__.py new file mode 100644 index 0000000000..5b8359bd38 --- /dev/null +++ b/nemoguardrails/server/experimental/providers/openai/__init__.py @@ -0,0 +1,16 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""OpenAI payload policies and source metadata.""" diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md b/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md new file mode 100644 index 0000000000..bd569b5635 --- /dev/null +++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md @@ -0,0 +1,41 @@ +# Chat Completions buffered policy + +The handwritten request and response projection models are the source of truth +for the buffered field policy. Types express accepted values, annotations express +guarded/constrained/disabled/opaque policy, and assignments express defaults: + +```python +n: Annotated[Literal[1], BeforeValidator(_require_int), constrained(reason="core_capability.single_text_target")] = 1 +stream: Annotated[StrictBool, constrained()] = False +audio: Annotated[None, disabled("core_capability.audio_content")] = None +``` + +The shared [projection policy module](../../../provider/projection_policy.py) +provides the declaration helpers, runtime metadata derivation, and schema export. +`ObjectPolicy` records object-level provider schema names, reviewed opaque fields, and +unknown-field policy. It is not a deployment profile selector. The capability +profile remains `single_text.v1`; the exported document format remains experimental +`1.0.0-alpha.1`. + +The request/response binding modules retain their existing runtime classes. +The provider revision pin lives once in +[`providers/openai/source.py`](../source.py), mirroring `source.yaml`. Their coverage inventories and exact text locations +are derived from the model annotations. Extraction supports one required item +at each traversed array boundary; unsupported or ambiguous boundaries fail. +No YAML is loaded to construct these bindings. + +`export_payload_schema` exports the declared schema and policy for readers. +It does not prove upstream provider compatibility or serialize arbitrary Python +validators. +Streaming classification, stateful hooks, and endpoint construction are outside +this buffered projection layer. + +Existing validation behavior is preserved: omitted disabled fields default to +null, explicit non-null disabled values fail, `stream` is a strict boolean, +and response annotations accept only null or an empty list. `n` accepts only the +integer `1`; booleans, strings, and floats are rejected. Generic JSON Schema +and Python validation are not claimed to be interchangeable. + +The [contract guide](../../../contracts/README.md) defines the document format; +the [OpenAI boundary summary](../../../contracts/openai/README.md) records its +scope and provider provenance. Neither is loaded by the models or bindings. diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/__init__.py b/nemoguardrails/server/experimental/providers/openai/chat_completions/__init__.py new file mode 100644 index 0000000000..f652b03f6b --- /dev/null +++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/__init__.py @@ -0,0 +1,16 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Buffered OpenAI Chat Completions projections and bindings.""" diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/request_binding.py b/nemoguardrails/server/experimental/providers/openai/chat_completions/request_binding.py new file mode 100644 index 0000000000..a9dbd0f38b --- /dev/null +++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/request_binding.py @@ -0,0 +1,47 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Guarded text bindings for OpenAI Chat Completions requests.""" + +from nemoguardrails.server.experimental.provider.projection_policy import payload_contract, text_location +from nemoguardrails.server.experimental.providers.openai.chat_completions.request_projection import ( + ChatCompletionsGuardedRequestProjection, +) + +CAPABILITY_PROFILE = "single_text.v1" +REQUEST_SOURCE_SCHEMA = "CreateChatCompletionRequest" +PAYLOAD_CONTRACT = payload_contract( + ChatCompletionsGuardedRequestProjection, + projection_id="openai.chat_completions.request.text.v1", + direction="request", +) +REQUEST_GUARDED_FIELDS = PAYLOAD_CONTRACT.root.guarded_fields +REQUEST_CONSTRAINED_FIELDS = PAYLOAD_CONTRACT.root.constrained_fields +REQUEST_OPAQUE_FIELDS = PAYLOAD_CONTRACT.root.opaque_fields +REQUEST_CONTENT_SCHEMAS = tuple( + (entry.source_schema, entry.coverage.reviewed_fields - entry.coverage.local_extension_fields) + for entry in PAYLOAD_CONTRACT.content_models + if entry.source_schema is not None +) +GUARDED_TEXT_LOCATION = text_location(ChatCompletionsGuardedRequestProjection) +STREAM_SELECTOR_FIELD = "stream" + + +class ChatCompletionsGuardedRequest(ChatCompletionsGuardedRequestProjection): + """Bind the handwritten projection to its guarded runtime semantics.""" + + guarded_text_location = GUARDED_TEXT_LOCATION + projection_contract = PAYLOAD_CONTRACT + stream_selector_field = STREAM_SELECTOR_FIELD diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/request_projection.py b/nemoguardrails/server/experimental/providers/openai/chat_completions/request_projection.py new file mode 100644 index 0000000000..032300d8a5 --- /dev/null +++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/request_projection.py @@ -0,0 +1,109 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Field policies for guarded OpenAI Chat Completions requests.""" + +from typing import Annotated, ClassVar, Literal + +from pydantic import BeforeValidator, StrictBool + +from nemoguardrails.server.experimental.provider.payload import GuardedContentModel, GuardedRequestModel, StrictFalse +from nemoguardrails.server.experimental.provider.projection_policy import ( + ObjectPolicy, + PolicyModel, + constrained, + disabled, + guarded, +) + + +def _require_int(value: object) -> object: + """Reject booleans, strings, and floats before Literal[1] can coerce them.""" + if type(value) is not int: + raise ValueError("n must be the integer 1") + return value + + +class ChatCompletionsUserMessageProjection(PolicyModel, GuardedContentModel): + """Accept one user text message with reviewed optional provider metadata.""" + + policy: ClassVar[ObjectPolicy] = ObjectPolicy( + source="ChatCompletionRequestUserMessage", unknown_fields="configurable" + ) + content: Annotated[str, guarded("user", replaceable=True, min_length=1)] + # The participant name is shown to the model but is not inspected text. + name: Annotated[None, disabled("core_capability.participant_name")] = None + role: Annotated[Literal["user"], constrained()] + + +class ChatCompletionsGuardedRequestProjection(PolicyModel, GuardedRequestModel): + """Declare the single-message request shape and explicit feature restrictions. + + The request binding attaches extraction, coverage, and response-mode metadata. + Recognizing the stream flag does not imply that an endpoint supports streaming. + The request is closed to OpenAI's fields: compatible servers render some + other fields, such as chat_template_kwargs or documents, into the prompt. + """ + + policy: ClassVar[ObjectPolicy] = ObjectPolicy( + opaque=( + "safety_identifier", + "logit_bias", + "presence_penalty", + "store", + "stop", + "top_p", + "verbosity", + "seed", + "stream_options", + "moderation", + "service_tier", + "user", + "prompt_cache_options", + "temperature", + "max_tokens", + "max_completion_tokens", + "frequency_penalty", + "prompt_cache_key", + "model", + "metadata", + "prompt_cache_retention", + ) + ) + messages: Annotated[list[ChatCompletionsUserMessageProjection], guarded(min_length=1, max_length=1)] + n: Annotated[ + Literal[1], BeforeValidator(_require_int), constrained(reason="core_capability.single_text_target") + ] = 1 + stream: Annotated[StrictBool, constrained()] = False + audio: Annotated[None, disabled("core_capability.audio_content")] = None + function_call: Annotated[None, disabled("core_capability.tool_content")] = None + functions: Annotated[None, disabled("core_capability.tool_content")] = None + modalities: Annotated[None, disabled("core_capability.multimodal_content")] = None + parallel_tool_calls: Annotated[None, disabled("core_capability.tool_content")] = None + prediction: Annotated[None, disabled("core_capability.predicted_content")] = None + response_format: Annotated[None, disabled("projection_policy.plain_text_output")] = None + tool_choice: Annotated[None, disabled("core_capability.tool_content")] = None + tools: Annotated[None, disabled("core_capability.tool_content")] = None + web_search_options: Annotated[None, disabled("core_capability.tool_content")] = None + # Response logprobs carry token text that output rails do not inspect, and + # the response projection accepts only null logprobs. Reject requests for + # them before dispatch instead of failing the provider response afterwards. + logprobs: Annotated[StrictFalse | None, constrained(reason="provider_integrity.token_logprobs")] = None + top_logprobs: Annotated[None, disabled("provider_integrity.token_logprobs")] = None + # Some compatible servers pass reasoning_effort into the chat template, so + # only OpenAI's enumerated values are accepted. + reasoning_effort: Annotated[ + Literal["none", "minimal", "low", "medium", "high", "xhigh", "max"] | None, constrained() + ] = None diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/response_binding.py b/nemoguardrails/server/experimental/providers/openai/chat_completions/response_binding.py new file mode 100644 index 0000000000..0ff2b69551 --- /dev/null +++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/response_binding.py @@ -0,0 +1,45 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Guarded text bindings for buffered OpenAI Chat Completions responses.""" + +from nemoguardrails.server.experimental.provider.projection_policy import payload_contract, text_location +from nemoguardrails.server.experimental.providers.openai.chat_completions.response_projection import ( + ChatCompletionsGuardedResponseProjection, +) + +CAPABILITY_PROFILE = "single_text.v1" +RESPONSE_SOURCE_SCHEMA = "CreateChatCompletionResponse" +PAYLOAD_CONTRACT = payload_contract( + ChatCompletionsGuardedResponseProjection, + projection_id="openai.chat_completions.response.text.v1", + direction="response", +) +RESPONSE_GUARDED_FIELDS = PAYLOAD_CONTRACT.root.guarded_fields +RESPONSE_CONSTRAINED_FIELDS = PAYLOAD_CONTRACT.root.constrained_fields +RESPONSE_OPAQUE_FIELDS = PAYLOAD_CONTRACT.root.opaque_fields +RESPONSE_CONTENT_SCHEMAS = tuple( + (entry.source_schema, entry.coverage.reviewed_fields - entry.coverage.local_extension_fields) + for entry in PAYLOAD_CONTRACT.content_models + if entry.source_schema is not None +) +GUARDED_TEXT_LOCATION = text_location(ChatCompletionsGuardedResponseProjection) + + +class ChatCompletionsGuardedResponse(ChatCompletionsGuardedResponseProjection): + """Bind the handwritten projection to its guarded runtime semantics.""" + + guarded_text_location = GUARDED_TEXT_LOCATION + projection_contract = PAYLOAD_CONTRACT diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/response_projection.py b/nemoguardrails/server/experimental/providers/openai/chat_completions/response_projection.py new file mode 100644 index 0000000000..d0cbf4a8c7 --- /dev/null +++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/response_projection.py @@ -0,0 +1,92 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Field policies for guarded buffered OpenAI Chat Completions responses.""" + +from typing import Annotated, Any, ClassVar, Literal + +from nemoguardrails.server.experimental.provider.payload import GuardedContentModel, GuardedPayloadModel +from nemoguardrails.server.experimental.provider.projection_policy import ( + ObjectPolicy, + PolicyModel, + constrained, + disabled, + guarded, + opaque, +) + + +class ChatCompletionsAssistantMessageProjection(PolicyModel, GuardedContentModel): + """Accept assistant text while preventing replacement of annotated content.""" + + policy: ClassVar[ObjectPolicy] = ObjectPolicy(source="ChatCompletionResponseMessage", unknown_fields="configurable") + # OpenAI allows empty or null content, for example when generation stops + # at a length limit or a content filter. Every other reviewed field is + # content-free here, so such a response has nothing to inspect. + content: Annotated[ + str | None, + guarded( + "assistant", + replaceable=True, + blocked_by="annotations", + replacement_reason="provider_integrity.annotated_text", + ), + ] + role: Annotated[Literal["assistant"], constrained()] + # Citation annotations carry provider text, such as titles, that output + # rails do not inspect. Web search is disabled on the request, so an + # OpenAI response without citations has null or empty annotations. + annotations: Annotated[list[Any] | None, constrained(reason="core_capability.citation_content", max_length=0)] = ( + None + ) + audio: Annotated[None, disabled("core_capability.audio_content")] = None + function_call: Annotated[None, disabled("core_capability.tool_content")] = None + reasoning_content: Annotated[None, disabled("core_capability.reasoning_content", extension=True)] = None + refusal: Annotated[None, disabled("core_capability.refusal_content")] = None + # OpenAI's schema allows an empty list here, and it carries no tool content. + tool_calls: Annotated[list[Any] | None, constrained(reason="core_capability.tool_content", max_length=0)] = None + + +class ChatCompletionsChoiceProjection(PolicyModel, GuardedContentModel): + """Expose the guarded message while preserving opaque choice metadata.""" + + policy: ClassVar[ObjectPolicy] = ObjectPolicy(unknown_fields="configurable") + message: Annotated[ChatCompletionsAssistantMessageProjection, guarded()] + finish_reason: Annotated[Any, opaque()] = None + index: Annotated[Any, opaque()] = None + logprobs: Annotated[None, disabled("provider_integrity.token_logprobs")] = None + + +class ChatCompletionsGuardedResponseProjection(PolicyModel, GuardedPayloadModel): + """Require one guarded choice and retain reviewed provider-owned response fields. + + The response is closed to OpenAI's fields: a member outside them could + carry generated text that output rails never inspect. + """ + + policy: ClassVar[ObjectPolicy] = ObjectPolicy( + opaque=( + "service_tier", + "created", + "object", + "system_fingerprint", + "usage", + "id", + "model", + "metadata", + "moderation", + ) + ) + choices: Annotated[list[ChatCompletionsChoiceProjection], guarded(min_length=1, max_length=1)] diff --git a/nemoguardrails/server/experimental/providers/openai/source.py b/nemoguardrails/server/experimental/providers/openai/source.py new file mode 100644 index 0000000000..e94dd51fbc --- /dev/null +++ b/nemoguardrails/server/experimental/providers/openai/source.py @@ -0,0 +1,30 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Pinned OpenAI OpenAPI source metadata. + +These constants mirror contracts/openai/source.yaml; a test checks that +the values match. +""" + +PROVIDER_DOCUMENT_URL = ( + "https://github.com/openai/openai-openapi/blob/df63773f69f542ef875b9f00c3837c25ba5f4f2a/openapi.yaml" +) +PROVIDER_DOWNLOAD_URL = ( + "https://raw.githubusercontent.com/openai/openai-openapi/df63773f69f542ef875b9f00c3837c25ba5f4f2a/openapi.yaml" +) +PROVIDER_REVISION = "df63773f69f542ef875b9f00c3837c25ba5f4f2a" +PROVIDER_DOCUMENT_VERSION = "2.3.0" +PROVIDER_DOCUMENT_SHA256 = "f2dae1a9aced09b91310db89edda51bf1e36ecbfb05230c3a50b239c07708469" diff --git a/tests/server/experimental/conftest.py b/tests/server/experimental/conftest.py new file mode 100644 index 0000000000..4d01399112 --- /dev/null +++ b/tests/server/experimental/conftest.py @@ -0,0 +1,63 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import pytest +from jsonschema import Draft202012Validator, validators +from jsonschema.exceptions import ValidationError as SchemaValidationError + +from nemoguardrails.server.experimental.provider.projection_policy import EXTENSION +from nemoguardrails.server.experimental.provider.types import UnknownContentFieldPolicy + + +def _reject_case_aliases(validator, reviewed, instance, schema): + """Check exact Unicode case folding against names derived from the export.""" + if not isinstance(instance, dict): + return + names = set(reviewed) + folded_names = {name.casefold() for name in names} + for name in instance: + if name not in names and name.casefold() in folded_names: + yield SchemaValidationError(f"Member {name!r} differs from a reviewed name only by case") + + +_ExportPolicyValidator = validators.extend(Draft202012Validator, {"x-test-reviewed-properties": _reject_case_aliases}) + + +def _lower_export(node: object, policy: UnknownContentFieldPolicy) -> object: + """Derive a JSON Schema validator input from an exported payload schema alone. + + Configurable objects accept unknown members only under trusted ALLOW. Any + member that matches a listed property only case-insensitively is rejected, + which the contract vocabulary does not express but its semantics require. + """ + if isinstance(node, list): + return [_lower_export(child, policy) for child in node] + if not isinstance(node, dict): + return node + lowered = {key: _lower_export(value, policy) for key, value in node.items() if key != EXTENSION} + if "properties" in node: + if node.get(EXTENSION, {}).get("unknown_fields") == "configurable": + lowered["additionalProperties"] = policy == UnknownContentFieldPolicy.ALLOW + lowered["x-test-reviewed-properties"] = list(node["properties"]) + lowered["properties"] = {name: _lower_export(child, policy) for name, child in node["properties"].items()} + return lowered + + +@pytest.fixture +def export_policy_validator(): + def build(exported, policy): + return _ExportPolicyValidator(_lower_export(exported, policy)) + + return build diff --git a/tests/server/experimental/test_openai_chat_contract.py b/tests/server/experimental/test_openai_chat_contract.py new file mode 100644 index 0000000000..c392bbf1a6 --- /dev/null +++ b/tests/server/experimental/test_openai_chat_contract.py @@ -0,0 +1,140 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for Chat contract exports and OpenAI source metadata.""" + +import json +import re +from pathlib import Path +from typing import Annotated +from urllib.parse import urlparse + +import pytest +import yaml +from jsonschema import Draft202012Validator +from pydantic import Field + +from nemoguardrails.server.experimental.provider.projection_policy import ( + CONTRACT_VERSION, + PolicyModel, + constrained, + export_payload_schema, + guarded, +) +from nemoguardrails.server.experimental.providers.openai import source as openai_pin +from nemoguardrails.server.experimental.providers.openai.chat_completions.request_binding import ( + PAYLOAD_CONTRACT as REQUEST_CONTRACT, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.request_projection import ( + ChatCompletionsGuardedRequestProjection, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.response_binding import ( + PAYLOAD_CONTRACT as RESPONSE_CONTRACT, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.response_projection import ( + ChatCompletionsGuardedResponseProjection, +) + +REPOSITORY_ROOT = Path(__file__).parents[3] +SOURCE_PATH = REPOSITORY_ROOT / "nemoguardrails/server/experimental/contracts/openai/source.yaml" + + +@pytest.fixture(scope="module") +def openai_source() -> dict[str, str]: + """Load source provenance, not a separately authored operation policy.""" + source = yaml.safe_load(SOURCE_PATH.read_text(encoding="utf-8")) + assert isinstance(source, dict) + return source + + +def test_openai_source_declares_revision_and_digest(openai_source: dict[str, str]) -> None: + """Pin metadata has the expected shape; this does not verify upstream bytes.""" + assert set(openai_source) == { + "document_url", + "download_url", + "revision", + "document_version", + "sha256", + } + assert re.fullmatch(r"[0-9a-f]{40}", openai_source["revision"]) + assert re.fullmatch(r"[0-9a-f]{64}", openai_source["sha256"]) + assert openai_source["document_version"].strip() + + +@pytest.mark.parametrize( + ("key", "host", "prefix"), + [ + ("document_url", "github.com", "/openai/openai-openapi/blob"), + ("download_url", "raw.githubusercontent.com", "/openai/openai-openapi"), + ], +) +def test_openai_source_urls_use_the_declared_revision( + openai_source: dict[str, str], key: str, host: str, prefix: str +) -> None: + """Both locations identify the same pinned file rather than a moving branch.""" + url = urlparse(openai_source[key]) + + assert url.scheme == "https" + assert url.netloc == host + assert url.path == f"{prefix}/{openai_source['revision']}/openapi.yaml" + assert not url.query + assert not url.fragment + + +def test_python_pin_matches_source_metadata(openai_source: dict[str, str]) -> None: + """The importable pin repeats source.yaml exactly, so the two cannot drift.""" + assert { + "document_url": openai_pin.PROVIDER_DOCUMENT_URL, + "download_url": openai_pin.PROVIDER_DOWNLOAD_URL, + "revision": openai_pin.PROVIDER_REVISION, + "document_version": openai_pin.PROVIDER_DOCUMENT_VERSION, + "sha256": openai_pin.PROVIDER_DOCUMENT_SHA256, + } == openai_source + + +GUARD_CONTRACT_SCHEMA = ( + Path(__file__).parents[3] / "nemoguardrails/server/experimental/contracts/guard-contract.schema.json" +) + + +def _contract_errors(request: type[PolicyModel], response: type[PolicyModel]) -> list[str]: + """Wrap two payload exports in a minimal operation contract and validate it.""" + contract = { + "version": CONTRACT_VERSION, + "operationId": "createChatCompletion", + "profile": "single_text.v1", + "request": export_payload_schema(request, projection_id=REQUEST_CONTRACT.projection_id), + "response": export_payload_schema(response, projection_id=RESPONSE_CONTRACT.projection_id), + "integration": { + "endpoint": { + "unsupported_request_code": "unsupported_request", + "unsupported_response_code": "unsupported_response", + } + }, + } + validator = Draft202012Validator(json.loads(GUARD_CONTRACT_SCHEMA.read_text(encoding="utf-8"))) + return [error.message for error in validator.iter_errors(contract)] + + +def test_chat_exports_conform_to_the_guard_contract_schema(): + assert _contract_errors(ChatCompletionsGuardedRequestProjection, ChatCompletionsGuardedResponseProjection) == [] + + +def test_contract_schema_check_catches_unexportable_constraints(): + class Response(PolicyModel): + text: Annotated[str, guarded("assistant")] + count: Annotated[int, Field(ge=0), constrained()] + + assert _contract_errors(ChatCompletionsGuardedRequestProjection, Response) diff --git a/tests/server/experimental/test_openai_chat_projections.py b/tests/server/experimental/test_openai_chat_projections.py new file mode 100644 index 0000000000..1b0788ebf4 --- /dev/null +++ b/tests/server/experimental/test_openai_chat_projections.py @@ -0,0 +1,664 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import json +import subprocess +import sys +from typing import Annotated, Literal + +import pytest +from jsonschema import Draft202012Validator +from pydantic import ValidationError + +import nemoguardrails.server.experimental._json_payload as json_payload +from nemoguardrails.server.experimental._json_payload import ( + InvalidJson, + UnsupportedJsonShape, + encode_json_object, + parse_json_object, +) +from nemoguardrails.server.experimental.provider.payload import ( + GuardedMessageTarget, + validate_payload_projection_contract, +) +from nemoguardrails.server.experimental.provider.projection_policy import ( + EXTENSION, + export_payload_schema, + field_coverage, + text_location, +) +from nemoguardrails.server.experimental.provider.types import GuardedMessage, UnknownContentFieldPolicy +from nemoguardrails.server.experimental.providers.openai.chat_completions.request_binding import ( + CAPABILITY_PROFILE as REQUEST_PROFILE, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.request_binding import ( + PAYLOAD_CONTRACT as REQUEST_CONTRACT, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.request_binding import ( + REQUEST_SOURCE_SCHEMA, + ChatCompletionsGuardedRequest, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.request_projection import ( + ChatCompletionsGuardedRequestProjection, + ChatCompletionsUserMessageProjection, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.response_binding import ( + CAPABILITY_PROFILE as RESPONSE_PROFILE, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.response_binding import ( + PAYLOAD_CONTRACT as RESPONSE_CONTRACT, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.response_binding import ( + RESPONSE_SOURCE_SCHEMA, + ChatCompletionsGuardedResponse, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.response_projection import ( + ChatCompletionsAssistantMessageProjection, + ChatCompletionsGuardedResponseProjection, +) + + +def _json_bytes(payload): + """Encode a JSON-compatible payload without insignificant whitespace.""" + return json.dumps(payload, separators=(",", ":")).encode() + + +def _request(**updates): + """Build a representative OpenAI Chat request with optional changes.""" + payload = { + "messages": [{"role": "user", "content": "question", "name": None}], + "model": "gpt-example", + "temperature": 0.2, + } + payload.update(updates) + return payload + + +def _response(**updates): + """Build a representative OpenAI Chat response with optional changes.""" + payload = { + "id": "chatcmpl-example", + "choices": [ + { + "index": 0, + "finish_reason": "stop", + "logprobs": None, + "message": { + "role": "assistant", + "content": "answer", + "annotations": [], + }, + } + ], + "usage": {"total_tokens": 2}, + } + payload.update(updates) + return payload + + +def _guarded_request(body: bytes): + """Parse, project, and locate guarded text in a Chat request.""" + payload = parse_json_object(body) + projection = ChatCompletionsGuardedRequest.validate_payload(payload) + return payload, projection, projection.locate_guarded_message(payload) + + +def _guarded_response(body: bytes): + """Parse, project, and locate guarded text in a Chat response.""" + payload = parse_json_object(body) + projection = ChatCompletionsGuardedResponse.validate_payload(payload) + return payload, projection, projection.locate_guarded_message(payload) + + +def test_request_binding_targets_original_provider_object_without_rewriting_bytes(): + """The request binding targets the decoded provider object without rewriting bytes.""" + body = b'{ "messages" : [ { "role" : "user", "content" : "question" } ], "model" : "gpt-example", "temperature" : 0.2 }' + original = bytes(body) + + payload, projection, target = _guarded_request(body) + + assert isinstance(target, GuardedMessageTarget) + assert target.message == GuardedMessage("user", "question") + assert target._object is payload["messages"][0] + assert target.allows_replacement is True + assert projection.streams_response is False + assert body == original + + +@pytest.mark.parametrize( + "payload", + [ + _request(messages=[]), + _request(messages=[{"role": "user", "content": "one"}, {"role": "user", "content": "two"}]), + _request(messages=[{"role": "assistant", "content": "question"}]), + _request(messages=[{"role": "user", "content": ""}]), + _request(stream=0), + _request(n=2), + _request(tools=[{"type": "function"}]), + _request(response_format={"type": "json_object"}), + _request(logprobs=True), + _request(logprobs=0), + _request(top_logprobs=3), + _request(logprobs=False, top_logprobs=0), + _request(reasoning_effort="high. uninspected instructions"), + _request(reasoning_effort="HIGH"), + ], +) +def test_request_projection_rejects_shapes_outside_buffered_text_profile(payload): + """The request projection rejects shapes outside its supported text profile.""" + with pytest.raises(ValidationError): + _guarded_request(_json_bytes(payload)) + + +@pytest.mark.parametrize("logprobs", [None, False]) +def test_request_projection_accepts_requests_without_logprobs(logprobs): + """Clients may state that they do not want log probabilities.""" + _, projection, _ = _guarded_request(_json_bytes(_request(logprobs=logprobs, top_logprobs=None))) + + assert projection.logprobs is logprobs + + +@pytest.mark.parametrize("name", ["caller", "uninspected instructions", {"value": "uninspected"}]) +def test_request_projection_rejects_participant_names(name): + """The participant name reaches the model, but input rails do not inspect it.""" + with pytest.raises(ValidationError): + _guarded_request(_json_bytes(_request(messages=[{"role": "user", "content": "question", "name": name}]))) + + +@pytest.mark.parametrize("effort", [None, "none", "minimal", "low", "medium", "high", "xhigh", "max"]) +def test_request_projection_accepts_openai_reasoning_efforts(effort): + """Every reasoning effort in OpenAI's schema remains accepted.""" + _, projection, _ = _guarded_request(_json_bytes(_request(reasoning_effort=effort))) + + assert projection.reasoning_effort == effort + + +def test_request_projection_reports_streaming_response_mode(): + """The request projection preserves the operation's boolean stream selector.""" + _, projection, _ = _guarded_request(_json_bytes(_request(stream=True))) + + assert projection.streams_response is True + + +def test_response_binding_targets_original_provider_object_without_rewriting_bytes(): + """The response binding targets the decoded provider object without rewriting bytes.""" + body = _json_bytes(_response()) + original = bytes(body) + + payload, _, target = _guarded_response(body) + + assert isinstance(target, GuardedMessageTarget) + assert target.message == GuardedMessage("assistant", "answer") + assert target._object is payload["choices"][0]["message"] + assert target.allows_replacement is True + assert body == original + + +def test_declared_annotation_blocking_still_prevents_replacement(): + """Non-empty annotations are rejected, but the declared blocker still applies.""" + payload = _response() + payload["choices"][0]["message"]["annotations"] = [{"type": "url_citation"}] + + target = ChatCompletionsGuardedResponse.guarded_text_location.locate(payload) + + assert target.allows_replacement is False + + +@pytest.mark.parametrize("tool_calls", [None, []]) +def test_response_projection_accepts_absent_tool_calls(tool_calls): + """Null and empty tool calls both mean that no tool was called.""" + response = _response() + response["choices"][0]["message"]["tool_calls"] = tool_calls + + _, _, target = _guarded_response(_json_bytes(response)) + + assert target.message == GuardedMessage("assistant", "answer") + + +@pytest.mark.parametrize( + ("content", "finish_reason"), + [("", "length"), (None, "content_filter"), (None, "stop")], +) +def test_response_without_text_has_nothing_to_inspect(content, finish_reason): + """Empty or null content in an otherwise content-free response yields no checked text.""" + response = _response() + response["choices"][0]["message"]["content"] = content + response["choices"][0]["finish_reason"] = finish_reason + + _, _, target = _guarded_response(_json_bytes(response)) + + assert target.has_text is False + + +@pytest.mark.parametrize( + "message", + [ + {"role": "assistant", "content": None, "refusal": "untrusted model text"}, + {"role": "assistant", "content": None, "tool_calls": [{"id": "call", "type": "function"}]}, + {"role": "assistant", "content": None, "annotations": [{"type": "url_citation"}]}, + {"role": "assistant", "content": None, "reasoning_content": "untrusted model text"}, + {"role": "assistant"}, + ], +) +def test_response_without_text_is_rejected_when_other_fields_carry_content(message): + """Null content does not excuse uninspected content elsewhere in the message.""" + with pytest.raises(ValidationError): + _guarded_response(_json_bytes(_response(choices=[{"index": 0, "message": message}]))) + + +def test_response_with_text_has_text_to_inspect(): + _, _, target = _guarded_response(_json_bytes(_response())) + + assert target.has_text is True + + +def test_response_binding_allows_unannotated_text_replacement(): + """Unannotated assistant text remains eligible for replacement.""" + response = _response() + del response["choices"][0]["message"]["annotations"] + + _, _, target = _guarded_response(_json_bytes(response)) + + assert target.allows_replacement is True + + +@pytest.mark.parametrize( + "payload", + [ + _response(choices=[]), + _response(choices=[_response()["choices"][0], _response()["choices"][0]]), + _response(choices=[{"message": {"role": "user", "content": "answer"}}]), + _response( + choices=[ + { + "message": { + "role": "assistant", + "content": "answer", + "tool_calls": [{"id": "call", "type": "function"}], + } + } + ] + ), + _response(choices=[{"message": {"role": "assistant", "content": "answer", "reasoning_content": "hidden"}}]), + _response(choices=[{"logprobs": {"content": []}, "message": {"role": "assistant", "content": "answer"}}]), + _response(choices=[{"message": {"role": "assistant", "content": "answer", "annotations": "invalid"}}]), + _response(choices=[{"message": {"role": "assistant", "content": "answer", "annotations": ["uninspected"]}}]), + _response( + choices=[ + { + "message": { + "role": "assistant", + "content": "answer", + "annotations": [ + { + "type": "url_citation", + "url_citation": { + "url": "https://x", + "title": "uninspected", + "start_index": 0, + "end_index": 1, + }, + } + ], + } + } + ] + ), + _response(choices=[{"message": {"role": "assistant", "content": "answer", "future": True}}]), + _response(choices=[{"message": {"role": "assistant", "content": "answer", "reasoning": "hidden"}}]), + _response( + choices=[ + { + "message": { + "role": "assistant", + "content": "answer", + "provider_specific_fields": {"reasoning": "hidden"}, + } + } + ] + ), + _response(choices=[{"token_ids": [1, 2], "message": {"role": "assistant", "content": "answer"}}]), + _response(future={"provider": "opaque"}), + _response(output_text="uninspected"), + _response(prompt_text="uninspected"), + _response(__verbose={"content": "uninspected"}), + _response(Choices=[]), + ], +) +def test_response_projection_rejects_shapes_outside_buffered_text_profile(payload): + """The response projection rejects shapes outside its supported text profile.""" + with pytest.raises(ValidationError): + _guarded_response(_json_bytes(payload)) + + +@pytest.mark.parametrize( + "payload", + [ + _request(future=[1, 2, 3]), + _request(chat_template_kwargs={"messages": [{"role": "user", "content": "uninspected"}]}), + _request(kv_transfer_params={"prompt_token_ids": [1, 2, 3]}), + _request(documents=[{"title": "t", "text": "uninspected"}]), + _request(Tools=[{"type": "function"}]), + _request(STREAM=True), + _request(messages=[{"role": "user", "content": "question", "Content": "attack"}]), + _request(tools=None, TOOLS=[{"type": "function"}]), + _request(**{"stream": False, "\u017ftream": True}), + _request(messages=[{"role": "user", "content": "question", "future": {"value": 1}}]), + _request(messages=[{"role": "user", "content": "question", "task": "uninspected"}]), + _request(messages=[{"role": "user", "content": "question", "Role": "system"}]), + ], +) +def test_request_projection_rejects_members_outside_openai_fields(payload): + """Compatible servers can render extra request members into the prompt unseen.""" + with pytest.raises(ValidationError, match="unreviewed fields are forbidden|only by case"): + ChatCompletionsGuardedRequest.validate_payload(payload) + + +def test_trusted_unknown_field_policy_opens_only_configurable_objects(): + """Allowing unknown fields opens configurable content objects, never the closed root.""" + request = _request(messages=[{"role": "user", "content": "question", "future": 1}]) + with pytest.raises(ValidationError, match="unreviewed fields are forbidden: future"): + ChatCompletionsGuardedRequest.validate_payload(request) + ChatCompletionsGuardedRequest.validate_payload(request, unknown_content_fields=UnknownContentFieldPolicy.ALLOW) + + with pytest.raises(ValidationError, match="unreviewed fields are forbidden: future"): + ChatCompletionsGuardedRequest.validate_payload( + _request(future=1), unknown_content_fields=UnknownContentFieldPolicy.ALLOW + ) + + +@pytest.mark.parametrize( + ("body", "error"), + [ + (b"not json", InvalidJson), + (b"[]", UnsupportedJsonShape), + (b'{"messages":[],"messages":[]}', UnsupportedJsonShape), + (b'{"value":NaN}', InvalidJson), + (b"\xff", InvalidJson), + ], +) +def test_strict_json_parser_rejects_ambiguous_or_non_object_payloads(body, error): + """Strict JSON parsing rejects ambiguous, invalid, and non-object bodies.""" + with pytest.raises(error): + parse_json_object(body) + + +def test_strict_json_parser_rejects_non_finite_numbers_and_non_bytes(): + """Overflowing floats are invalid, and only raw bytes are parsed.""" + with pytest.raises(InvalidJson): + parse_json_object(b'{"value":1e400}') + with pytest.raises(TypeError, match="must be bytes"): + parse_json_object('{"value":1}') + + +def test_modified_objects_encode_as_compact_utf8_json(): + """A modified provider object is serialized without escapes or extra whitespace.""" + assert encode_json_object({"text": "caf\u00e9", "n": [1, 2]}) == '{"text":"caf\u00e9","n":[1,2]}'.encode() + + +def test_strict_json_parser_keeps_case_variant_names_in_opaque_data(): + """Only exact duplicates are ambiguous to every parser; closed models handle case variants.""" + assert parse_json_object(b'{"metadata":{"Env":"a","env":"b"}}') == {"metadata": {"Env": "a", "env": "b"}} + + +def test_strict_json_parser_normalizes_integer_conversion_failures(monkeypatch): + """Integer conversion failures produce the parser's stable invalid-JSON outcome.""" + + def reject_integer(_value): + """Simulate an interpreter integer-size rejection.""" + raise ValueError("integer exceeds the configured digit limit") + + monkeypatch.setattr(json_payload, "int", reject_integer, raising=False) + + with pytest.raises(InvalidJson): + parse_json_object(b'{"value":1}') + + +def test_bindings_match_contract_identity_and_replacement_policy(): + """Staged bindings preserve the generated contract identity and replacement policy.""" + assert REQUEST_PROFILE == RESPONSE_PROFILE == "single_text.v1" + assert REQUEST_SOURCE_SCHEMA == "CreateChatCompletionRequest" + assert RESPONSE_SOURCE_SCHEMA == "CreateChatCompletionResponse" + assert REQUEST_CONTRACT.direction == "request" + assert RESPONSE_CONTRACT.direction == "response" + assert validate_payload_projection_contract(ChatCompletionsGuardedRequest, "request") is REQUEST_CONTRACT + assert validate_payload_projection_contract(ChatCompletionsGuardedResponse, "response") is RESPONSE_CONTRACT + assert ChatCompletionsGuardedRequest.guarded_text_location.allows_replacement is True + assert ChatCompletionsGuardedResponse.guarded_text_location.allows_replacement is True + + +@pytest.mark.parametrize( + "module", + [ + "nemoguardrails.server.experimental.provider.payload", + "nemoguardrails.server.experimental.providers.openai.chat_completions.request_binding", + "nemoguardrails.server.experimental.providers.openai.chat_completions.response_binding", + ], +) +def test_staged_projection_modules_import_in_fresh_interpreter(module): + """Each staged projection module imports in a fresh interpreter.""" + result = subprocess.run( + [sys.executable, "-c", f"import {module}"], + check=False, + capture_output=True, + text=True, + ) + + assert result.returncode == 0, result.stderr + + +def test_explicit_defaults_are_visible_to_python_and_pydantic(): + message = ChatCompletionsUserMessageProjection(content="question", role="user") + request = ChatCompletionsGuardedRequestProjection(messages=[message]) + assert request.n == 1 + assert request.stream is False + assert request.audio is None + for name, expected in (("n", 1), ("stream", False), ("audio", None)): + field = ChatCompletionsGuardedRequestProjection.model_fields[name] + assert not field.is_required() + assert field.default is expected + + +@pytest.mark.parametrize("value", [None, []]) +def test_annotations_accept_only_null_or_empty(value): + message = ChatCompletionsAssistantMessageProjection(role="assistant", content="answer", annotations=value) + assert message.annotations == value + with pytest.raises(ValidationError): + ChatCompletionsAssistantMessageProjection(role="assistant", content="answer", annotations=[{"type": "x"}]) + + +def test_n_accepts_the_integer_one(): + request = ChatCompletionsGuardedRequest.model_validate({"messages": [{"role": "user", "content": "q"}], "n": 1}) + assert request.n == 1 + + +@pytest.mark.parametrize("value", [True, False, "1", 1.0, 0, 2, None]) +def test_n_rejects_coercible_and_other_values(value): + with pytest.raises(ValidationError): + ChatCompletionsGuardedRequest.model_validate({"messages": [{"role": "user", "content": "q"}], "n": value}) + + +@pytest.mark.parametrize("value", [0, 1, "true", None]) +def test_stream_stays_strict(value): + with pytest.raises(ValidationError): + ChatCompletionsGuardedRequest.model_validate({"messages": [{"role": "user", "content": "q"}], "stream": value}) + + +@pytest.mark.parametrize("value", [[], {}, False, "audio"]) +def test_disabled_fields_reject_every_non_null_value(value): + with pytest.raises(ValidationError): + ChatCompletionsGuardedRequest.model_validate({"messages": [{"role": "user", "content": "q"}], "audio": value}) + + +def test_bindings_derive_coverage_and_targets_from_the_typed_models(): + assert REQUEST_CONTRACT.root == field_coverage(ChatCompletionsGuardedRequestProjection) + assert RESPONSE_CONTRACT.root == field_coverage(ChatCompletionsGuardedResponseProjection) + assert ChatCompletionsGuardedRequest.guarded_text_location == text_location(ChatCompletionsGuardedRequestProjection) + assert ChatCompletionsGuardedResponse.guarded_text_location == text_location( + ChatCompletionsGuardedResponseProjection + ) + response_content = RESPONSE_CONTRACT.content_models[-1] + assert response_content.model is ChatCompletionsAssistantMessageProjection + assert response_content.coverage.local_extension_fields == frozenset({"reasoning_content"}) + assert ChatCompletionsGuardedResponse.guarded_text_location.allows_empty is True + assert ChatCompletionsGuardedRequest.guarded_text_location.allows_empty is False + + +def test_export_preserves_nullable_annotation_schema(): + exported = export_payload_schema( + ChatCompletionsGuardedResponseProjection, projection_id=RESPONSE_CONTRACT.projection_id + ) + annotations = exported["properties"]["choices"]["items"]["properties"]["message"]["properties"]["annotations"] + assert annotations["default"] is None + assert {"type": "null"} in annotations["oneOf"] + assert {"type": "array", "items": {}, "maxItems": 0} in annotations["oneOf"] + + +def test_export_lists_reviewed_opaque_names_as_properties(): + request = export_payload_schema(ChatCompletionsGuardedRequestProjection, projection_id="test.request") + assert request["properties"]["model"] == {EXTENSION: {"classification": "opaque"}} + assert "opaque_fields" not in request[EXTENSION] + validator = Draft202012Validator(request) + message = {"role": "user", "content": "q"} + assert validator.is_valid({"messages": [message], "model": "m", "temperature": 0.2}) + assert not validator.is_valid({"messages": [message], "future": 1}) + + +def test_export_follows_each_object_unknown_field_policy(): + request = export_payload_schema(ChatCompletionsGuardedRequestProjection, projection_id="test.request") + response = export_payload_schema(ChatCompletionsGuardedResponseProjection, projection_id="test.response") + choice = response["properties"]["choices"]["items"] + configurable = (request["properties"]["messages"]["items"], choice, choice["properties"]["message"]) + for content in configurable: + assert content["additionalProperties"] is True + assert content[EXTENSION]["unknown_fields"] == "configurable" + for root in (request, response): + assert root["additionalProperties"] is False + assert "unknown_fields" not in root[EXTENSION] + + +def test_field_validation_matches_original_unannotated_declarations(): + from itertools import product + from typing import Any + + from pydantic import BaseModel, Field, StrictBool + + class OriginalRequestFields(BaseModel): + n: Literal[1] = 1 + stream: StrictBool = False + audio: None = None + + class OriginalResponseFields(BaseModel): + annotations: Annotated[list[Any] | None, Field(max_length=0)] = None + content: str | None + logprobs: None = None + + def validated(model, payload): + try: + return model.model_validate(payload).model_dump() + except ValidationError: + return "rejected" + + values = [None, True, False, 0, 1, 1.0, 2, "1", "true", "", [], {}, ["x"]] + # n is intentionally stricter than Literal[1]; test_n_rejects_coercible_and_other_values covers it. + for stream, audio in product(values, repeat=2): + fields = {"n": 1, "stream": stream, "audio": audio} + old = validated(OriginalRequestFields, fields) + new = validated( + ChatCompletionsGuardedRequestProjection, + { + "messages": [{"role": "user", "content": "q"}], + **fields, + }, + ) + if isinstance(new, dict): + new = {name: new[name] for name in fields} + assert old == new, fields + + for annotations, content, logprobs in product(values + ["answer"], repeat=3): + fields = {"annotations": annotations, "content": content, "logprobs": logprobs} + old = validated(OriginalResponseFields, fields) + new = validated( + ChatCompletionsGuardedResponseProjection, + { + "choices": [ + { + "message": {"role": "assistant", "content": content, "annotations": annotations}, + "logprobs": logprobs, + } + ], + }, + ) + if isinstance(new, dict): + choice = new["choices"][0] + new = {name: choice["message"][name] for name in ("annotations", "content")} + new["logprobs"] = choice["logprobs"] + assert old == new, fields + + +_MESSAGE = {"role": "user", "content": "q"} +_REQUEST = {"messages": [_MESSAGE], "model": "m"} +_CHOICE = {"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": "a"}} +_RESPONSE = {"id": "r", "choices": [_CHOICE], "usage": {"total_tokens": 1}} + + +@pytest.mark.parametrize("policy", list(UnknownContentFieldPolicy)) +@pytest.mark.parametrize( + ("model", "document"), + [ + (ChatCompletionsGuardedRequest, _REQUEST), + (ChatCompletionsGuardedRequest, {**_REQUEST, "temperature": 0.2, "metadata": {"Env": "a", "env": "b"}}), + (ChatCompletionsGuardedRequest, {**_REQUEST, "future": 1}), + (ChatCompletionsGuardedRequest, {**_REQUEST, "Tools": []}), + (ChatCompletionsGuardedRequest, {**_REQUEST, "MODEL": "other"}), + (ChatCompletionsGuardedRequest, {**_REQUEST, "messages": [{**_MESSAGE, "future": 1}]}), + (ChatCompletionsGuardedRequest, {**_REQUEST, "messages": [{**_MESSAGE, "Content": "x"}]}), + (ChatCompletionsGuardedRequest, {**_REQUEST, "messages": [{**_MESSAGE, "namK": "x"}]}), + (ChatCompletionsGuardedResponse, _RESPONSE), + (ChatCompletionsGuardedResponse, {**_RESPONSE, "future": 1}), + (ChatCompletionsGuardedResponse, {**_RESPONSE, "Usage": {}}), + (ChatCompletionsGuardedResponse, {**_RESPONSE, "choices": [{**_CHOICE, "future": 1}]}), + (ChatCompletionsGuardedResponse, {**_RESPONSE, "choices": [{**_CHOICE, "INDEX": 1}]}), + (ChatCompletionsGuardedResponse, {**_RESPONSE, "choices": [{**_CHOICE, "meßage": {"content": "x"}}]}), + (ChatCompletionsGuardedResponse, {**_RESPONSE, "choices": [{**_CHOICE, "Message\n": {"content": "x"}}]}), + ( + ChatCompletionsGuardedResponse, + {**_RESPONSE, "choices": [{**_CHOICE, "message": {**_CHOICE["message"], "future": 1}}]}, + ), + ( + ChatCompletionsGuardedResponse, + {**_RESPONSE, "choices": [{**_CHOICE, "message": {**_CHOICE["message"], "Refusal": "x"}}]}, + ), + ( + ChatCompletionsGuardedResponse, + {**_RESPONSE, "choices": [{**_CHOICE, "message": {**_CHOICE["message"], "refuſal": "x"}}]}, + ), + ], +) +def test_export_derived_acceptance_matches_handwritten_runtime(model, document, policy, export_policy_validator): + """Acceptance derived only from the export agrees with the handwritten runtime. + + This shows that the export carries the member policy the runtime enforces. + It is not compiler equivalence, which needs tests against generated models. + """ + exported = export_payload_schema(model, projection_id=model.projection_contract.projection_id) + derived = export_policy_validator(exported, policy).is_valid(document) + try: + model.validate_payload(document, unknown_content_fields=policy) + handwritten = True + except ValidationError: + handwritten = False + + assert derived is handwritten diff --git a/tests/server/experimental/test_projection_policy.py b/tests/server/experimental/test_projection_policy.py new file mode 100644 index 0000000000..9e8078a38c --- /dev/null +++ b/tests/server/experimental/test_projection_policy.py @@ -0,0 +1,328 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from pathlib import Path +from typing import Annotated, ClassVar, Literal + +import pytest +from pydantic import Field, ValidationError + +from nemoguardrails.server.experimental.provider.payload import GuardedContentModel +from nemoguardrails.server.experimental.provider.projection_policy import ( + EXTENSION, + ObjectPolicy, + PolicyModel, + constrained, + disabled, + export_payload_schema, + guarded, + payload_contract, + text_location, +) +from nemoguardrails.server.experimental.provider.types import UnknownContentFieldPolicy + + +def test_changed_python_policy_changes_coverage_and_export(): + class Message(PolicyModel): + policy: ClassVar[ObjectPolicy] = ObjectPolicy(source="Message", opaque=("provider_id",)) + text: Annotated[str, guarded("user", replaceable=False, min_length=2)] + role: Annotated[Literal["user"], constrained()] + tools: Annotated[None, disabled("core_capability.tool_content")] = None + + class Request(PolicyModel): + messages: Annotated[list[Message], guarded(min_length=1, max_length=1)] + + contract = payload_contract(Request, projection_id="test.request", direction="request") + assert contract.content_models[0].coverage.opaque_fields == frozenset({"provider_id"}) + assert text_location(Request).allows_replacement is False + exported = export_payload_schema(Request, projection_id="test.request") + message = exported["properties"]["messages"]["items"] + assert message["properties"]["text"]["minLength"] == 2 + assert message[EXTENSION]["source"] == "#/components/schemas/Message" + assert message["properties"]["tools"]["default"] is None + + +def test_missing_policy_fails_at_class_definition(): + with pytest.raises(ValueError, match="missing field policy"): + + class Invalid(PolicyModel): + text: str + + +def test_export_rejects_overlapping_unions_instead_of_changing_their_meaning(): + class Invalid(PolicyModel): + text: Annotated[str | Literal["special"], constrained()] + + with pytest.raises(ValueError, match="disjoint nullable"): + export_payload_schema(Invalid, projection_id="test.request") + + +def test_missing_disabled_default_fails_at_class_definition(): + with pytest.raises(ValueError, match="optional and null-only"): + + class Invalid(PolicyModel): + tools: Annotated[None, disabled("core_capability.tool_content")] + + +def test_opaque_overlap_fails_at_class_definition(): + with pytest.raises(ValueError, match="overlaps declared"): + + class Invalid(PolicyModel): + policy: ClassVar[ObjectPolicy] = ObjectPolicy(opaque=("text",)) + text: Annotated[str, guarded("user")] + + +def test_duplicate_opaque_inventory_fails_at_class_definition(): + with pytest.raises(ValueError, match="duplicate opaque"): + + class Invalid(PolicyModel): + policy: ClassVar[ObjectPolicy] = ObjectPolicy(opaque=("id", "id")) + + +@pytest.mark.parametrize("helper", [guarded, constrained]) +def test_helpers_do_not_hide_defaults(helper): + with pytest.raises(ValueError, match="defaults explicitly"): + helper(default=False) + + +@pytest.mark.parametrize("helper", [guarded, constrained]) +@pytest.mark.parametrize("constraint", ["ge", "le", "gt", "lt", "multiple_of", "strict"]) +def test_helpers_reject_constraints_the_contract_cannot_express(helper, constraint): + with pytest.raises(ValueError, match="not expressible"): + helper(**{constraint: 1}) + + +def test_policy_models_are_closed_unless_configurable_fields_are_allowed(): + class Message(PolicyModel, GuardedContentModel): + policy: ClassVar[ObjectPolicy] = ObjectPolicy(opaque=("model",), unknown_fields="configurable") + content: Annotated[str, guarded("user")] + + class Root(PolicyModel, GuardedContentModel): + content: Annotated[str, guarded("user")] + + allow = UnknownContentFieldPolicy.ALLOW + Message.model_validate({"content": "q", "model": "m"}) + for model in (Message, Root): + with pytest.raises(ValidationError, match="unreviewed fields are forbidden: unreviewed"): + model.model_validate({"content": "q", "unreviewed": "x"}) + Message.validate_payload({"content": "q", "unreviewed": "x"}, unknown_content_fields=allow) + with pytest.raises(ValidationError, match="unreviewed fields are forbidden"): + Root.validate_payload({"content": "q", "unreviewed": "x"}, unknown_content_fields=allow) + for variant in ({"Content": "x"}, {"MODEL": "x"}, {"cOnTeNt": "x"}): + with pytest.raises(ValidationError, match="only by case"): + Message.validate_payload({"content": "q", **variant}, unknown_content_fields=allow) + + +@pytest.mark.parametrize( + "declare", + [ + lambda: disabled("tools are off"), + lambda: disabled(" "), + lambda: constrained(reason="Core_capability.tools"), + lambda: guarded("assistant", replacement_reason="annotated text"), + ], +) +def test_helpers_reject_unstructured_reasons(declare): + with pytest.raises(ValueError, match="structured reason"): + declare() + + +@pytest.mark.parametrize( + ("policy", "message"), + [ + (lambda: ObjectPolicy(source="#/components/schemas/Message"), "bare component"), + (lambda: ObjectPolicy(source="Message Schema"), "bare component"), + (lambda: ObjectPolicy(opaque=("*",)), "wildcards"), + (lambda: ObjectPolicy(opaque=("",)), "wildcards"), + ], +) +def test_object_policy_rejects_values_the_contract_cannot_represent(policy, message): + with pytest.raises(ValueError, match=message): + policy() + + +def test_replacement_policy_requires_subject(): + with pytest.raises(ValueError, match="requires a subject"): + guarded(replaceable=True) + + +def test_extraction_rejects_non_singleton_arrays(): + class Message(PolicyModel): + text: Annotated[str, guarded("user")] + + class Request(PolicyModel): + messages: Annotated[list[Message], guarded()] + + with pytest.raises(ValueError, match="exactly one array item"): + text_location(Request) + + +def test_extraction_rejects_unknown_replacement_blocker(): + class Message(PolicyModel): + text: Annotated[str, guarded("assistant", replaceable=True, blocked_by="annotatons")] + annotations: Annotated[list[str] | None, constrained()] = None + + with pytest.raises(ValueError, match="'annotatons' is not a field"): + text_location(Message) + + +def test_extraction_accepts_opaque_replacement_blocker(): + class Message(PolicyModel): + policy: ClassVar[ObjectPolicy] = ObjectPolicy(opaque=("citations",)) + text: Annotated[str, guarded("assistant", replaceable=True, blocked_by="citations")] + + assert text_location(Message).replacement_blocked_by == "citations" + + +def test_extraction_allows_empty_text_only_where_the_subject_type_does(): + class RequiredText(PolicyModel): + text: Annotated[str, guarded("user", min_length=1)] + + class NullableText(PolicyModel): + text: Annotated[str | None, guarded("assistant")] + + assert text_location(RequiredText).allows_empty is False + assert text_location(NullableText).allows_empty is True + + class Message(PolicyModel): + text: Annotated[str, guarded("user")] + + assert text_location(Message).allows_empty is True + + +def test_extraction_rejects_multiple_subjects(): + class Request(PolicyModel): + first: Annotated[str, guarded("user")] + second: Annotated[str, guarded("user")] + + with pytest.raises(ValueError, match="exactly one guarded subject"): + text_location(Request) + + +def test_export_does_not_require_contract_yaml(monkeypatch): + class Request(PolicyModel): + text: Annotated[str, guarded("user")] + + def unexpected_read(*args, **kwargs): + raise AssertionError("Contract export must not read YAML policy") + + monkeypatch.setattr(Path, "read_text", unexpected_read) + assert export_payload_schema(Request, projection_id="test.request")["type"] == "object" + + +@pytest.mark.parametrize("policy", list(UnknownContentFieldPolicy)) +@pytest.mark.parametrize( + ("member", "forbid", "allow"), + [ + ("key", True, True), + ("Key", False, False), + ("KEY", False, False), + ("meßage", False, False), + ("meſsage", False, False), + ("meſſage", False, False), + ("Message\n", False, True), + ("key\n", False, True), + ("namK", False, True), + ], +) +def test_export_reference_matches_unicode_casefold_and_exact_names( + member, forbid, allow, policy, export_policy_validator +): + class Content(PolicyModel, GuardedContentModel): + policy: ClassVar[ObjectPolicy] = ObjectPolicy(opaque=("key",), unknown_fields="configurable") + message: Annotated[str, guarded("user")] + + document = {"message": "q", member: "x"} + exported = export_payload_schema(Content, projection_id="test.casefold") + derived = export_policy_validator(exported, policy).is_valid(document) + expected = allow if policy == UnknownContentFieldPolicy.ALLOW else forbid + assert derived is expected + if expected: + Content.validate_payload(document, unknown_content_fields=policy) + else: + with pytest.raises(ValidationError): + Content.validate_payload(document, unknown_content_fields=policy) + + +def test_policy_models_reject_aliases_and_unpolicied_nested_models(): + from pydantic import BaseModel + + with pytest.raises(ValueError, match="aliases are not supported"): + + class Aliased(PolicyModel): + text: Annotated[str, guarded("user"), Field(alias="Text")] + + class Plain(BaseModel): + text: str + + class Root(PolicyModel): + items: Annotated[list[Plain], guarded(min_length=1, max_length=1)] + + with pytest.raises(ValueError, match="Nested models must declare field policies"): + payload_contract(Root, projection_id="test", direction="request") + + +def test_model_graph_visits_shared_models_once_and_rejects_name_collisions(): + from nemoguardrails.server.experimental.provider.projection_policy import model_graph + + class Part(PolicyModel): + text: Annotated[str, guarded("user")] + + class Root(PolicyModel): + first: Annotated[Part, guarded()] + second: Annotated[Part | None, constrained()] = None + + assert list(model_graph(Root)) == ["Root", "Part"] + + def other_part() -> type[PolicyModel]: + class Part(PolicyModel): + value: Annotated[str, constrained()] + + return Part + + class Colliding(PolicyModel): + first: Annotated[Part, guarded()] + second: Annotated[other_part() | None, constrained()] = None + + with pytest.raises(ValueError, match="Model names must be unique"): + model_graph(Colliding) + + +def test_export_rejects_recursive_models(): + class Node(PolicyModel): + text: Annotated[str, guarded("user")] + children: Annotated["list[Node]", constrained()] = [] + + Node.model_rebuild() + with pytest.raises(ValueError, match="Recursive models"): + export_payload_schema(Node, projection_id="test") + + +@pytest.mark.parametrize( + "annotation", + [ + Annotated[int, guarded("user")], + Annotated[list[Annotated[str, guarded("user")]], guarded(min_length=1, max_length=1)], + ], +) +def test_extraction_requires_a_named_string_subject(annotation): + root = type( + "Root", + (PolicyModel,), + {"__annotations__": {"value": annotation}, "__module__": __name__}, + ) + + with pytest.raises(ValueError, match="named string field"): + text_location(root) diff --git a/tests/server/experimental/test_provider_payload.py b/tests/server/experimental/test_provider_payload.py new file mode 100644 index 0000000000..06f9fc27e1 --- /dev/null +++ b/tests/server/experimental/test_provider_payload.py @@ -0,0 +1,307 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from typing import Literal + +import pytest +from pydantic import BaseModel, RootModel, ValidationError + +from nemoguardrails.server.experimental.provider.payload import ( + GuardedArraySelection, + GuardedMessageTarget, + GuardedRequestModel, + GuardedTextAlternatives, + GuardedTextLocation, + PayloadCapabilityProfile, + PayloadProjectionContract, + ProjectionFieldCoverage, + ProjectionModelContract, + guarded_schema_error, + validate_payload_projection_contract, +) + + +class TextContent(RootModel[str]): + """Represent one scalar text encoding used by the test projection.""" + + pass + + +class TextBlock(BaseModel): + """Represent one discriminated text block used by the test projection.""" + + type: Literal["text"] + text: str + + +class BlockContent(RootModel[list[TextBlock]]): + """Represent an array-based text encoding used by the test projection.""" + + pass + + +class Message(BaseModel): + """Represent a message with either supported test content encoding.""" + + content: TextContent | BlockContent + + +class RequestProjection(BaseModel): + """Represent the test request containing one guarded message.""" + + messages: list[Message] + + +def request_locations(): + """Declare guarded text locations for the two test representations.""" + return GuardedTextAlternatives( + locations=( + GuardedTextLocation( + role="user", + object_path=("messages", 0), + member="content", + allows_replacement=False, + ), + GuardedTextLocation( + role="user", + object_path=("messages", 0, "content", 0), + member="text", + allows_replacement=False, + ), + ) + ) + + +@pytest.mark.parametrize( + ("projection", "payload"), + [ + ( + RequestProjection(messages=[Message(content=TextContent("hello"))]), + {"messages": [{"content": "hello"}]}, + ), + ( + RequestProjection(messages=[Message(content=BlockContent([TextBlock(type="text", text="hello")]))]), + {"messages": [{"content": [{"type": "text", "text": "hello"}]}]}, + ), + ], +) +def test_guarded_text_alternatives_select_one_root_model_representation(projection, payload): + """Alternative locations select the one representation accepted by the projection.""" + target = request_locations().locate(projection, payload) + + assert target.message.content == "hello" + + +def test_guarded_array_selection_requires_exactly_one_discriminator_match(): + """Array selection fails unless exactly one item matches its discriminator.""" + selection = GuardedArraySelection(match=(("type", "text"),), cardinality="exactly_one") + + assert selection.select([{"type": "metadata"}, {"type": "text", "text": "answer"}]) == { + "type": "text", + "text": "answer", + } + with pytest.raises(ValueError, match="exactly one"): + selection.select([{"type": "metadata"}]) + with pytest.raises(ValueError, match="exactly one"): + selection.select([{"type": "text"}, {"type": "text"}]) + + +@pytest.mark.parametrize( + ("expected", "received"), + [ + (True, 1), + (1, True), + (False, 0), + (0, False), + ], +) +def test_guarded_array_selection_distinguishes_boolean_and_integer_discriminators(expected, received): + """Boolean and integer discriminators do not match across JSON scalar types.""" + selection = GuardedArraySelection(match=(("value", expected),), cardinality="exactly_one") + + with pytest.raises(ValueError, match="exactly one"): + selection.select([{"value": received}]) + + +def _coverage(**fields): + return ProjectionFieldCoverage(**{name: frozenset(value) for name, value in fields.items()}) + + +class _Content(BaseModel): + text: str + + +class _Root(BaseModel): + items: list[_Content] + + +def test_projection_contract_metadata_rejects_inconsistent_declarations(): + """Coverage and contract metadata reject ambiguous or empty declarations.""" + with pytest.raises(ValueError, match="multiple classifications"): + _coverage(guarded_fields={"text"}, opaque_fields={"text"}) + with pytest.raises(ValueError, match="non-empty"): + ProjectionModelContract(model=_Content, coverage=_coverage(), source_schema=" ") + with pytest.raises(ValueError, match="identifier"): + PayloadProjectionContract(" ", "request", PayloadCapabilityProfile.SINGLE_TEXT_V1, _coverage()) + entry = ProjectionModelContract(model=_Content, coverage=_coverage(guarded_fields={"text"})) + with pytest.raises(ValueError, match="each content model once"): + PayloadProjectionContract( + "test", "request", PayloadCapabilityProfile.SINGLE_TEXT_V1, _coverage(), content_models=(entry, entry) + ) + + +def test_projection_contract_validation_reports_each_mismatch(): + """A payload model must carry a contract for its direction that covers every field.""" + + def bound(contract): + model = type("Bound", (_Root,), {}) + model.projection_contract = contract + return model + + def contract(root, content): + return PayloadProjectionContract( + "test", + "request", + PayloadCapabilityProfile.SINGLE_TEXT_V1, + root, + content_models=(ProjectionModelContract(model=_Content, coverage=content),), + ) + + covered = contract(_coverage(guarded_fields={"items"}), _coverage(guarded_fields={"text"})) + assert validate_payload_projection_contract(bound(covered), "request") is covered + with pytest.raises(ValueError, match="must declare a projection contract"): + validate_payload_projection_contract(_Root, "request") + with pytest.raises(ValueError, match="expected 'response'"): + validate_payload_projection_contract(bound(covered), "response") + with pytest.raises(ValueError, match="omits root fields"): + validate_payload_projection_contract( + bound(contract(_coverage(), _coverage(guarded_fields={"text"}))), "request" + ) + with pytest.raises(ValueError, match="omits _Content fields"): + validate_payload_projection_contract( + bound(contract(_coverage(guarded_fields={"items"}), _coverage())), "request" + ) + + +def test_guarded_schema_error_describes_location_without_echoing_input(): + """Projection errors name the failing location but never the provider value.""" + with pytest.raises(ValidationError) as nested: + _Root.model_validate({"items": [{"text": 1}]}) + with pytest.raises(ValidationError) as root: + _Content.model_validate("secret") + + assert guarded_schema_error(nested.value, "request") == ( + "The request does not match the guarded content schema at items.0.text: Input should be a valid string." + ) + message = guarded_schema_error(root.value, "request") + assert message.startswith("The request does not match the guarded content schema: ") + assert "secret" not in message + + +def test_message_target_reads_and_replaces_only_permitted_text(): + """Targets expose string text and replace it only when the contract allows.""" + payload = {"content": "hello"} + target = GuardedMessageTarget("user", payload, "content", allows_replacement=True) + target.replace_content("redacted") + assert payload == {"content": "redacted"} + with pytest.raises(ValueError, match="does not support replacement"): + GuardedMessageTarget("user", payload, "content", allows_replacement=False).replace_content("x") + with pytest.raises(TypeError, match="must be a string"): + GuardedMessageTarget("user", {"content": None}, "content", allows_replacement=False).message + + +class _Block(BaseModel): + type: str + text: str | None = None + + +class _Blocks(BaseModel): + blocks: list[_Block] + + +_SELECT_TEXT = GuardedArraySelection(match=(("type", "text"),), cardinality="exactly_one") + + +def test_text_location_follows_array_selections_in_payloads_and_projections(): + """Discriminated array segments resolve in both the raw payload and the projection.""" + location = GuardedTextLocation("user", ("blocks", _SELECT_TEXT), "text", allows_replacement=False) + payload = {"blocks": [{"type": "image"}, {"type": "text", "text": "hello"}]} + projection = _Blocks.model_validate(payload) + + assert location.locate(payload).message.content == "hello" + location.validate_projection(projection) + assert location.matches_projection(projection) is True + assert location.matches_projection(_Blocks(blocks=[_Block(type="image")])) is False + + +def test_text_location_rejects_paths_that_do_not_resolve_to_text(): + """Location mismatches fail instead of selecting a different provider member.""" + location = GuardedTextLocation("user", ("blocks", 0), "text", allows_replacement=False) + with pytest.raises(TypeError, match="does not match the provider payload"): + location.locate({"blocks": {"text": "hello"}}) + with pytest.raises(TypeError, match="must resolve to a provider object"): + location.locate({"blocks": ["hello"]}) + with pytest.raises(ValueError, match="does not match the provider projection"): + GuardedTextLocation("user", ("items", "text"), "text", allows_replacement=False).validate_projection( + _Root(items=[]) + ) + with pytest.raises(ValueError, match="does not resolve to projected text"): + GuardedTextLocation("user", ("items",), "text", allows_replacement=False).validate_projection(_Root(items=[])) + with pytest.raises(ValueError, match="does not resolve to projected text"): + location.validate_projection(_Blocks(blocks=[_Block(type="text")])) + GuardedTextLocation("user", ("blocks", 0), "text", allows_replacement=False, allows_empty=True).validate_projection( + _Blocks(blocks=[_Block(type="text")]) + ) + assert ( + GuardedTextLocation("user", (1,), "text", allows_replacement=False).matches_projection(_Root(items=[])) is False + ) + + +def test_text_alternatives_require_exactly_one_matching_representation(): + """Alternatives validate only when one representation matches the projection.""" + alternatives = request_locations() + alternatives.validate_projection(RequestProjection(messages=[Message(content=TextContent("hello"))])) + with pytest.raises(ValueError, match="exactly one matching representation"): + alternatives.validate_projection(RequestProjection(messages=[])) + + +class _AlternativeRequest(GuardedRequestModel): + messages: list[Message] + stream: bool = False + guarded_text_location = request_locations() + stream_selector_field = "stream" + + +class _UnboundRequest(GuardedRequestModel): + messages: list[Message] + + +def test_payload_models_report_and_use_their_bindings(): + """Payload models expose whether they declare text and response-mode bindings.""" + payload = {"messages": [{"content": "hello"}], "stream": True} + request = _AlternativeRequest.model_validate(payload) + + assert _AlternativeRequest.has_guarded_text_binding() is True + assert _AlternativeRequest.has_response_mode_binding() is True + assert request.locate_guarded_message(payload).message.content == "hello" + assert request.streams_response is True + + unbound = _UnboundRequest.model_validate(payload) + assert _UnboundRequest.has_guarded_text_binding() is False + assert _UnboundRequest.has_response_mode_binding() is False + with pytest.raises(TypeError, match="does not declare a guarded text location"): + unbound.locate_guarded_message(payload) + with pytest.raises(TypeError, match="does not declare a stream selector field"): + unbound.streams_response