diff --git a/nemoguardrails/server/experimental/provider/projection_policy.py b/nemoguardrails/server/experimental/provider/projection_policy.py index 11302db010..3860942635 100644 --- a/nemoguardrails/server/experimental/provider/projection_policy.py +++ b/nemoguardrails/server/experimental/provider/projection_policy.py @@ -182,18 +182,19 @@ def field_policy(field: FieldInfo) -> dict[str, Any]: 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"}) +# Runtime constraints that declarations may use. Export separately rejects any +# schema keyword the guard contract cannot carry. JSON Schema has no keyword for +# Python strictness, so `strict` is enforced at runtime only. +_FIELD_CONSTRAINTS = frozenset({"min_length", "max_length", "pattern", "ge", "le", "strict"}) 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 + unsupported = constraints.keys() - _FIELD_CONSTRAINTS if unsupported: - raise ValueError(f"Constraints not expressible in the guard contract: {sorted(unsupported)}") + raise ValueError(f"Unsupported Pydantic constraints: {sorted(unsupported)}") return Field(json_schema_extra={EXTENSION: metadata}, **constraints) @@ -213,8 +214,9 @@ def guarded( 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. + **constraints: min_length, max_length, pattern, ge, le, or strict. + Format support is checked separately when exporting a complete + guard contract; strictness has no JSON Schema keyword. Use inside Annotated and put defaults on the field assignment. Declaring replacement eligibility does not enable runtime replacement support. @@ -462,6 +464,21 @@ def expand(node: dict[str, Any]) -> dict[str, Any]: return exported +def replacement_blocker(model: type[PolicyModel], subject: dict[str, Any]) -> str | None: + """Return a subject's replacement blocker after checking it names a reviewed sibling. + + A misspelled blocker would never match a provider member and would silently + allow replacement, so buffered and stream subjects both validate it. + + Raises: + ValueError: The blocker is not a declared or reviewed opaque field of the model. + """ + blocker = subject.get("replacement_blocked_by") + if blocker is not None and blocker not in set(model.model_fields) | set(model.policy.opaque): + raise ValueError(f"Replacement blocker {blocker!r} is not a field of the subject's object") + return blocker + + def text_location(model: type[PolicyModel]) -> GuardedTextLocation: """Derive one buffered text target directly from typed field declarations. @@ -508,7 +525,6 @@ def visit_model( """Collect subjects from the model objects that own their policies.""" if current in active: raise ValueError("Recursive guarded text paths are not supported") - reviewed = set(current.model_fields) | set(current.policy.opaque) for name, field in current.model_fields.items(): policy = field_policy(field) if policy.get("classification") != "guarded": @@ -528,9 +544,7 @@ def visit_model( ) if annotation is not str and not literal_text: raise ValueError("A guarded subject must be a named string field") - blocker = subject.get("replacement_blocked_by") - if blocker is not None and blocker not in reviewed: - raise ValueError(f"Replacement blocker {blocker!r} is not a field of the subject's object") + blocker = replacement_blocker(current, subject) locations.append( GuardedTextLocation( role=subject["role"], diff --git a/nemoguardrails/server/experimental/provider/stream_policy.py b/nemoguardrails/server/experimental/provider/stream_policy.py new file mode 100644 index 0000000000..5822e3f48e --- /dev/null +++ b/nemoguardrails/server/experimental/provider/stream_policy.py @@ -0,0 +1,190 @@ +# 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. + +"""Derive streaming metadata from typed projection fields and event rules. + +Field policy declares the guarded text location; StreamEventRule declares event +selection, roles, and missing-text handling. These helpers remove duplicate +inventories without inferring provider lifecycle behavior or upstream coverage. +They operate on trusted framework declarations, not user-supplied schemas. +""" + +from types import UnionType +from typing import Annotated, Any, Union, get_args, get_origin + +from nemoguardrails.server.experimental.provider.projection_policy import ( + PolicyModel, + field_coverage, + field_policy, + model_graph, + replacement_blocker, +) +from nemoguardrails.server.experimental.provider.stream import StreamEventRole, StreamShapeCoverage +from nemoguardrails.server.experimental.provider.stream_classifier import ( + StreamClassifier, + StreamClassifierDefinition, + StreamEventRule, + StreamTextPath, + build_stream_classifier, +) + + +def stream_text_path(model: type[PolicyModel]) -> StreamTextPath: + """Derive one named string or nullable-string subject from typed guarded fields. + + Follow guarded models, nullable model containers, and lists that allow at + most one item, without schema generation, so unrelated field types cannot + affect the binding. Unlike buffered extraction, empty arrays, null + containers, and null text are valid: the event rule decides whether absent + text is rejected or classified as metadata. This function only derives the + path; it does not validate an event or choose its role. + + Raises: + ValueError: The declaration has no unique text subject, an array that + may hold more than one item, an unsupported subject type, a union + container, a recursive path, or a replacement blocker that is + not a reviewed sibling field. Object-union selectors and + multi-item arrays need an explicit StreamEventRule.text_path. + """ + paths: list[StreamTextPath] = [] + + def unwrap(annotation: Any, metadata: list[Any]) -> tuple[Any, list[Any]]: + """Read Annotated arguments and drop one null alternative.""" + if get_origin(annotation) is Annotated: + annotation, *inner = get_args(annotation) + metadata = [*inner, *metadata] + if get_origin(annotation) in (Union, UnionType): + alternatives = [item for item in get_args(annotation) if item is not type(None)] + if len(alternatives) != 1 or len(alternatives) == len(get_args(annotation)): + raise ValueError("Stream text extraction does not infer object-union selectors") + return unwrap(alternatives[0], metadata) + return annotation, metadata + + def max_length(metadata: list[Any]) -> int | None: + """Read the last declared maximum length, matching field precedence.""" + return next((item.max_length for item in reversed(metadata) if hasattr(item, "max_length")), None) + + def visit_container( + annotation: Any, metadata: list[Any], path: StreamTextPath, active: tuple[type[PolicyModel], ...] + ) -> None: + """Follow guarded models and lists bounded to at most one item.""" + annotation, metadata = unwrap(annotation, metadata) + if get_origin(annotation) is list: + if max_length(metadata) != 1: + raise ValueError("Stream text extraction requires at most one array item") + visit_container(get_args(annotation)[0], [], (*path, 0), active) + elif isinstance(annotation, type) and issubclass(annotation, PolicyModel): + visit_model(annotation, path, active) + else: + raise ValueError("Guarded stream containers must lead to a named string field") + + def visit_model(current: type[PolicyModel], path: StreamTextPath, active: tuple[type[PolicyModel], ...]) -> None: + """Collect subjects from the models that own their policies.""" + if current in active: + raise ValueError("Recursive guarded stream text paths are not supported") + for name, field in current.model_fields.items(): + policy = field_policy(field) + if policy.get("classification") != "guarded": + continue + if not policy.get("subject"): + visit_container(field.annotation, list(field.metadata), (*path, name), (*active, current)) + continue + annotation, _ = unwrap(field.annotation, list(field.metadata)) + if annotation is not str: + raise ValueError("A guarded stream subject must be a named string field") + replacement_blocker(current, policy["subject"]) + paths.append((*path, name)) + + visit_model(model, (), ()) + if len(paths) != 1: + raise ValueError("Stream text extraction requires exactly one guarded subject") + return paths[0] + + +def stream_shape_coverage( + rules: tuple[StreamEventRule, ...], + *, + sentinels: tuple[tuple[bytes, str], ...] = (), + non_data_shape: str | None = None, +) -> StreamShapeCoverage: + """Collect event shapes by role, including fallbacks and opaque transport. + + A shape shared by different roles is rejected by StreamShapeCoverage. + Selection and fallback consistency are checked by build_stream_classifier; + deriving coverage does not replace that validation or prove provider-wide + event coverage. + """ + groups: dict[StreamEventRole, set[str]] = {role: set() for role in StreamEventRole} + for rule in rules: + groups[rule.role].add(rule.shape) + if rule.missing_text_role is not None: + groups[rule.missing_text_role].add(rule.missing_text_shape or rule.shape) + groups[StreamEventRole.OPAQUE_METADATA].update(shape for _, shape in sentinels) + if non_data_shape is not None: + groups[StreamEventRole.OPAQUE_METADATA].add(non_data_shape) + return StreamShapeCoverage( + guarded_shapes=frozenset(groups[StreamEventRole.GUARDED_TEXT]), + snapshot_shapes=frozenset(groups[StreamEventRole.TEXT_SNAPSHOT]), + opaque_shapes=frozenset(groups[StreamEventRole.OPAQUE_METADATA]), + provider_error_shapes=frozenset(groups[StreamEventRole.PROVIDER_ERROR]), + ) + + +def validate_stream_policy(definition: StreamClassifierDefinition | StreamClassifier) -> None: + """Require a typed classifier's runtime bindings to agree with field policy. + + Text-bearing rules must follow the unique assistant subject derived from + guarded fields. Non-text rules cannot hide a declared subject. Coverage + must describe exactly the bound rules and transport shapes. This validates + the derivable typed subset; custom selectors and handwritten classifiers + can use the lower-level runtime interfaces but are not implicitly exportable. + No schemas are generated and no protocol hooks are instantiated. + """ + expected_shapes = stream_shape_coverage( + definition.rules, sentinels=definition.sentinels, non_data_shape=definition.non_data_shape + ) + if definition.contract.shapes != expected_shapes: + raise ValueError("Stream coverage differs from the bound event rules") + text_models: set[type[PolicyModel]] = set() + for rule in definition.rules: + if not issubclass(rule.model, PolicyModel): + raise TypeError("Typed stream bindings require policy-annotated event models") + subjects = [ + field_policy(field)["subject"] + for model in model_graph(rule.model).values() + for field in model.model_fields.values() + if field_policy(field).get("subject") + ] + if rule.role in {StreamEventRole.GUARDED_TEXT, StreamEventRole.TEXT_SNAPSHOT}: + if len(subjects) != 1 or subjects[0].get("role") != "assistant": + raise ValueError("A text event requires one guarded subject with the assistant role") + if rule.text_path != stream_text_path(rule.model): + raise ValueError(f"Stream event {rule.shape!r}: text binding differs from field policy") + text_models.add(rule.model) + elif subjects: + raise ValueError(f"Stream event {rule.shape!r}: a non-text role hides a guarded subject") + if definition.contract.fields is not None: + if len(text_models) != 1 or definition.contract.fields != field_coverage(next(iter(text_models))): + raise ValueError("Stream field coverage differs from field policy") + sentinels = [value for value, _ in definition.sentinels] + if any(not value or value != value.strip() for value in sentinels) or len(set(sentinels)) != len(sentinels): + raise ValueError("Stream sentinels must be unique nonempty canonical byte strings") + + +def build_policy_stream_classifier(definition: StreamClassifierDefinition) -> StreamClassifier: + """Build a classifier only after generic and typed-policy checks pass.""" + classifier = build_stream_classifier(definition) + validate_stream_policy(classifier) + return classifier diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md b/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md index 40be254617..ba1eae510e 100644 --- a/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md +++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md @@ -1,4 +1,4 @@ -# Chat Completions buffered policy +# Chat Completions projection policy The handwritten request and response projection models are the source of truth for the buffered field policy. Types express accepted values, annotations express @@ -37,9 +37,9 @@ 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 +Existing buffered 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. Canonical exports are closed by default and explicitly declare configurable unknown-field handling @@ -54,6 +54,54 @@ 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. +## Streaming declarations + +[stream_projection.py](stream_projection.py) uses the same field annotations and +`ObjectPolicy` for the experimental `single_text_delta.v1` boundary. A chunk +accepts zero or one choice with a strict integer index of zero, and `logprobs` +must be null. The root, choice, and delta reject unreviewed members; reviewed +root metadata such as `id`, `model`, and `usage` passes through unchanged. +Missing or null content is metadata, while empty string content is still guarded +text. Error messages are constrained fields, not guarded assistant output. The +error envelope rejects unreviewed members, and its error object rejects them +unless trusted configuration allows unknown fields; the stream classifier +always validates with the closed default. + +[stream_classifier.py](stream_classifier.py) owns explicit `StreamEventRule` +declarations for content and provider errors, along with the opaque `[DONE]` +sentinel. Field coverage comes from the models; shape coverage comes from event +rules. Data-less SSE events have a distinct `keepalive` shape; they are not +terminal `[DONE]` events. Numeric bounds and strictness use ordinary Pydantic +constraints, independent of the inspection document's supported vocabulary. +Unsupported document representations fail at export, not runtime construction. +Shape coverage includes +rules, missing-text fallbacks, and transport declarations. The shared +[stream policy helpers](../../../provider/stream_policy.py) derive a text path +from typed fields, without schema generation, only when there is one subject +and every traversed array has at most one item. A null container or null text +on that path is missing text. Union containers and multi-item selectors remain +explicit event-rule paths. Buffered extraction still requires exactly one item; +the streaming helper does not relax it. + +The typed classifier builder checks the text path, assistant subject, event roles, +and coverage against these declarations before runtime use. A mismatched path +cannot silently turn guarded text into a metadata-only event. The lower-level +classifier resolves text from the original JSON before validation and rejects a +projection that changes that text, including through container transformations. +Bytes are relayed unchanged only after their original guarded text is checked. +The lower-level +classifier API remains available for handwritten bindings; those do not +automatically satisfy typed-policy validation or contract export. + +[stream_hooks.py](stream_hooks.py) remains handwritten. It enforces terminal +ordering, distinguishes real `[DONE]` events from keepalives, accepts provider +errors as terminal outcomes, and frames native errors. These checks require +history and do not belong in field annotations or a stateless classifier. + +The projections and classifier can be used independently of endpoint wiring. +This stage does not bind them to the HTTP endpoint or extend the shared buffered +contract exporter with streaming declarations. + ## Buffered contract export The shared [contract exporter](../../../provider/contract_export.py) takes the diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py new file mode 100644 index 0000000000..67b44b7a25 --- /dev/null +++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py @@ -0,0 +1,83 @@ +# 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. + +"""Bind typed Chat stream projections to explicit event and transport rules. + +The declarations own event selection and missing-text behavior. Field coverage, +shape inventories, and the text path are derived; stateful lifecycle checks stay +in stream_hooks. Importing this module does not enable an endpoint. +""" + +from nemoguardrails.server.experimental.provider.projection_policy import field_coverage +from nemoguardrails.server.experimental.provider.stream import ( + StreamCapabilityProfile, + StreamEventRole, + StreamProjectionContract, +) +from nemoguardrails.server.experimental.provider.stream_classifier import ( + StreamClassifierDefinition, + StreamEventRule, +) +from nemoguardrails.server.experimental.provider.stream_policy import ( + build_policy_stream_classifier, + stream_shape_coverage, + stream_text_path, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.stream_projection import ( + ChatCompletionsStreamErrorProjection, + ChatCompletionsStreamPayloadProjection, +) + +PROJECTION_ID = "openai.chat_completions.stream.text.v1" +STREAM_SENTINELS = ((b"[DONE]", "[DONE]"),) +STREAM_NON_DATA_SHAPE = "keepalive" +STREAM_RULES = ( + StreamEventRule( + shape="chat.completion.chunk:content", + model=ChatCompletionsStreamPayloadProjection, + role=StreamEventRole.GUARDED_TEXT, + match=(("object", "chat.completion.chunk"),), + text_path=stream_text_path(ChatCompletionsStreamPayloadProjection), + missing_text_role=StreamEventRole.OPAQUE_METADATA, + missing_text_shape="chat.completion.chunk:metadata", + ), + StreamEventRule( + shape="error", + model=ChatCompletionsStreamErrorProjection, + role=StreamEventRole.PROVIDER_ERROR, + required_fields=frozenset({"error"}), + ), +) + +STREAM_CONTRACT = StreamProjectionContract( + projection_id=PROJECTION_ID, + profile=StreamCapabilityProfile.SINGLE_TEXT_DELTA_V1, + shapes=stream_shape_coverage( + STREAM_RULES, + sentinels=STREAM_SENTINELS, + non_data_shape=STREAM_NON_DATA_SHAPE, + ), + fields=field_coverage(ChatCompletionsStreamPayloadProjection), +) + +STREAM_CLASSIFIER = build_policy_stream_classifier( + StreamClassifierDefinition( + subject=PROJECTION_ID, + contract=STREAM_CONTRACT, + non_data_shape=STREAM_NON_DATA_SHAPE, + sentinels=STREAM_SENTINELS, + rules=STREAM_RULES, + ) +) diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_hooks.py b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_hooks.py new file mode 100644 index 0000000000..1242362ce2 --- /dev/null +++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_hooks.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. + +"""Bind OpenAI Chat Completions protocol hooks to the staged classifier.""" + +from nemoguardrails.server.experimental.provider.sse import ServerSentEvent +from nemoguardrails.server.experimental.provider.stream import ( + GuardedStreamEvent, + StreamEventRole, + UnsupportedProviderStream, +) + + +class ChatCompletionsStreamHooks: + """Require successful streams to end with ``[DONE]``. + + A provider error may end the stream without a sentinel. Data-less keepalives + are permitted throughout; payload events cannot follow an error or sentinel, + except for a sentinel after an error. State belongs to one upstream stream. + """ + + def __init__(self) -> None: + self._done = False + self._provider_error = False + + def observe_event(self, event: ServerSentEvent, classified: GuardedStreamEvent) -> None: + """Track the actual sentinel and reject payloads after termination.""" + + data = event.data + if data is None: + return + if self._done: + raise UnsupportedProviderStream("A Chat Completions data event follows after [DONE].") + if data.strip() == b"[DONE]": + self._done = True + return + if self._provider_error: + raise UnsupportedProviderStream("A Chat Completions payload follows after a provider error.") + if classified.role is StreamEventRole.PROVIDER_ERROR: + self._provider_error = True + + def validate_end_of_stream(self) -> None: + """Reject successful streams that end before their sentinel.""" + + if not self._done and not self._provider_error: + raise UnsupportedProviderStream("The Chat Completions stream ended before [DONE].") + + def encode_error(self, rendered_body: bytes) -> tuple[bytes, bytes]: + """Encode an OpenAI-compatible error and terminal chat event.""" + + return b"data: " + rendered_body + b"\n\n", b"data: [DONE]\n\n" diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_projection.py b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_projection.py new file mode 100644 index 0000000000..febb2c88b2 --- /dev/null +++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_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. + +"""Declare guarded text, closed content boundaries, and opaque stream metadata. + +These handwritten models own per-event field policy. The classifier binds them +to event roles; handwritten hooks enforce ordering and terminal-state behavior. +""" + +from __future__ import annotations + +from typing import Annotated, Any, ClassVar, Literal + +from nemoguardrails.server.experimental.provider.payload import GuardedContentModel, GuardedProjectionModel +from nemoguardrails.server.experimental.provider.projection_policy import ( + ObjectPolicy, + PolicyModel, + constrained, + disabled, + guarded, + opaque, +) + + +class ChatCompletionsStreamDeltaProjection(PolicyModel, GuardedContentModel): + """Accept assistant text or metadata-only deltas within a closed boundary.""" + + policy: ClassVar[ObjectPolicy] = ObjectPolicy(source="ChatCompletionStreamResponseDelta") + + audio: Annotated[None, disabled("core_capability.audio_content", extension=True)] = None + content: Annotated[str | None, guarded("assistant")] = 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 + role: Annotated[Literal["assistant"] | None, constrained()] = 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 ChatCompletionsStreamErrorDetailsProjection(PolicyModel, GuardedContentModel): + """Validate the native error message without treating it as guarded text.""" + + policy: ClassVar[ObjectPolicy] = ObjectPolicy(source="Error", unknown_fields="configurable") + + code: Annotated[Any, opaque()] = None + message: Annotated[str, constrained()] + misalignment: Annotated[Any, opaque()] = None + param: Annotated[Any, opaque()] = None + type: Annotated[Any, opaque()] = None + + +class ChatCompletionsStreamErrorProjection(PolicyModel, GuardedProjectionModel): + """Accept only an error envelope, never an error mixed with generated content.""" + + policy: ClassVar[ObjectPolicy] = ObjectPolicy(source="ErrorResponse") + + error: Annotated[ChatCompletionsStreamErrorDetailsProjection, constrained()] + + +class ChatCompletionsStreamChoiceProjection(PolicyModel, GuardedContentModel): + """Require integer choice zero and reject unreviewed choice-level content.""" + + delta: Annotated[ChatCompletionsStreamDeltaProjection, guarded()] + finish_reason: Annotated[Any, opaque()] = None + index: Annotated[int, constrained(strict=True, ge=0, le=0)] + # Log probabilities carry token text that output rails do not inspect, as in + # buffered responses; requests for them are already rejected. + logprobs: Annotated[None, disabled("provider_integrity.token_logprobs")] = None + + +class ChatCompletionsStreamPayloadProjection(PolicyModel, GuardedProjectionModel): + """Allow one text choice or an empty usage chunk, preserving root metadata.""" + + policy: ClassVar[ObjectPolicy] = ObjectPolicy( + source="CreateChatCompletionStreamResponse", + opaque=("created", "id", "model", "moderation", "obfuscation", "service_tier", "system_fingerprint", "usage"), + ) + + choices: Annotated[list[ChatCompletionsStreamChoiceProjection], guarded(max_length=1)] + object: Annotated[Literal["chat.completion.chunk"], constrained()] diff --git a/tests/server/experimental/test_openai_chat_streaming_projections.py b/tests/server/experimental/test_openai_chat_streaming_projections.py new file mode 100644 index 0000000000..5dec6dffec --- /dev/null +++ b/tests/server/experimental/test_openai_chat_streaming_projections.py @@ -0,0 +1,542 @@ +# 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 pathlib import Path + +import pytest +import yaml +from jsonschema import Draft202012Validator +from pydantic import ValidationError + +from nemoguardrails.server.experimental._content_checker import ContentAllowed, StreamBufferingPolicy +from nemoguardrails.server.experimental._guarded_stream import guard_provider_stream +from nemoguardrails.server.experimental.provider.projection_policy import ( + EXTENSION, + export_payload_schema, + field_coverage, + field_policy, + model_graph, +) +from nemoguardrails.server.experimental.provider.sse import ServerSentEvent +from nemoguardrails.server.experimental.provider.stream import ( + ClassifiedStreamAdapter, + StreamEventRole, + UnsupportedProviderStream, + create_classified_stream_adapter_factory, +) +from nemoguardrails.server.experimental.provider.types import GuardedMessage, UnknownContentFieldPolicy +from nemoguardrails.server.experimental.providers.openai.chat_completions.request_binding import ( + PAYLOAD_CONTRACT, + REQUEST_CONSTRAINED_FIELDS, + REQUEST_GUARDED_FIELDS, + REQUEST_OPAQUE_FIELDS, + STREAM_SELECTOR_FIELD, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.stream_classifier import ( + STREAM_CLASSIFIER, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.stream_hooks import ( + ChatCompletionsStreamHooks, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.stream_projection import ( + ChatCompletionsStreamDeltaProjection, + ChatCompletionsStreamErrorProjection, + ChatCompletionsStreamPayloadProjection, +) +from nemoguardrails.server.experimental.providers.openai.errors import render_openai_error + + +def event(payload): + data = payload if isinstance(payload, bytes) else json.dumps(payload, separators=(",", ":")).encode() + return ServerSentEvent.from_bytes(b"data: " + data + b"\n\n") + + +def chunk(delta, **fields): + return { + "id": "chatcmpl-stream", + "object": "chat.completion.chunk", + "choices": [{"index": 0, "delta": delta, **fields}], + "system_fingerprint": "fp-stream", + } + + +@pytest.mark.parametrize( + ("native", "shape", "role", "text"), + [ + (chunk({"role": "assistant"}), "chat.completion.chunk:metadata", StreamEventRole.OPAQUE_METADATA, None), + (chunk({"content": "answer"}), "chat.completion.chunk:content", StreamEventRole.GUARDED_TEXT, "answer"), + ( + {"object": "chat.completion.chunk", "choices": [], "usage": {"output_tokens": 1}}, + "chat.completion.chunk:metadata", + StreamEventRole.OPAQUE_METADATA, + None, + ), + ( + {"error": {"message": "upstream failed", "type": "server_error"}}, + "error", + StreamEventRole.PROVIDER_ERROR, + None, + ), + (b"[DONE]", "[DONE]", StreamEventRole.OPAQUE_METADATA, None), + ], +) +def test_stream_classifier_covers_the_declared_event_inventory(native, shape, role, text): + classified = STREAM_CLASSIFIER.classify_event(event(native)) + + assert (classified.shape, classified.role, classified.text) == (shape, role, text) + shapes = STREAM_CLASSIFIER.contract.shapes + assert shape in shapes.guarded_shapes | shapes.opaque_shapes | shapes.provider_error_shapes + + +def test_stream_classifier_treats_non_data_events_as_declared_opaque_shape(): + classified = STREAM_CLASSIFIER.classify_event(ServerSentEvent.from_bytes(b": keepalive\n\n")) + + assert classified.shape == "keepalive" + assert classified.role is StreamEventRole.OPAQUE_METADATA + + +@pytest.mark.parametrize( + "native", + [ + chunk({"reasoning_content": "unguarded"}), + chunk({"content": "answer", "tool_calls": [{"index": 0, "id": "call", "type": "function"}]}), + chunk({"content": "answer", "future_content": "unguarded"}), + { + "object": "chat.completion.chunk", + "choices": [ + {"index": 0, "delta": {"content": "first"}}, + {"index": 1, "delta": {"content": "second"}}, + ], + }, + {"error": {"message": "failed"}, "choices": [{"index": 0, "delta": {"content": "hidden"}}]}, + ], +) +def test_stream_classifier_rejects_content_outside_the_closed_profile(native): + with pytest.raises(UnsupportedProviderStream): + STREAM_CLASSIFIER.classify_event(event(native)) + + +def test_stream_classifier_accepts_reviewed_error_extensions_without_guarding_them(): + classified = STREAM_CLASSIFIER.classify_event( + event({"error": {"message": "failed", "misalignment": {"provider": "opaque"}}}) + ) + + assert classified.role is StreamEventRole.PROVIDER_ERROR + + +def test_stream_contract_preserves_declared_identity_and_field_inventory(): + assert STREAM_CLASSIFIER.contract.profile.value == "single_text_delta.v1" + assert ChatCompletionsStreamPayloadProjection.policy.source == "CreateChatCompletionStreamResponse" + assert STREAM_CLASSIFIER.contract.projection_id == "openai.chat_completions.stream.text.v1" + assert STREAM_CLASSIFIER.contract.fields == field_coverage(ChatCompletionsStreamPayloadProjection) + + +def test_shared_request_binding_owns_the_stream_selector_and_field_inventory(): + assert STREAM_SELECTOR_FIELD == "stream" + assert REQUEST_GUARDED_FIELDS == PAYLOAD_CONTRACT.root.guarded_fields + assert REQUEST_CONSTRAINED_FIELDS == PAYLOAD_CONTRACT.root.constrained_fields + assert REQUEST_OPAQUE_FIELDS == PAYLOAD_CONTRACT.root.opaque_fields + + +def test_stream_models_classify_every_declared_field_and_derive_root_coverage(): + for root in (ChatCompletionsStreamPayloadProjection, ChatCompletionsStreamErrorProjection): + for model in model_graph(root).values(): + for field in model.model_fields.values(): + assert field_policy(field)["classification"] in {"guarded", "constrained", "opaque"} + coverage = field_coverage(ChatCompletionsStreamPayloadProjection) + assert STREAM_CLASSIFIER.contract.fields == coverage + assert coverage.guarded_fields == frozenset({"choices"}) + assert coverage.constrained_fields == frozenset({"object"}) + assert coverage.opaque_fields == frozenset( + {"created", "id", "model", "moderation", "obfuscation", "service_tier", "system_fingerprint", "usage"} + ) + assert field_coverage(ChatCompletionsStreamDeltaProjection).local_extension_fields == frozenset( + {"audio", "reasoning_content"} + ) + + +def test_stream_schema_preserves_closure_defaults_and_text_policy(): + schema = export_payload_schema( + ChatCompletionsStreamPayloadProjection, projection_id=STREAM_CLASSIFIER.contract.projection_id + ) + choices = schema["properties"]["choices"] + choice = choices["items"] + delta = choice["properties"]["delta"] + assert choices["maxItems"] == 1 + assert "minItems" not in choices + assert choice["additionalProperties"] is delta["additionalProperties"] is False + assert choice["required"] == ["delta", "index"] + assert choice["properties"]["index"]["minimum"] == choice["properties"]["index"]["maximum"] == 0 + content = delta["properties"]["content"] + assert content["default"] is None + assert content["oneOf"] == [{"type": "string"}, {"type": "null"}] + assert content[EXTENSION]["subject"] == {"kind": "text", "role": "assistant"} + assert delta[EXTENSION]["unknown_fields"] == "forbid" + assert choice["properties"]["logprobs"][EXTENSION]["gate"] == "disabled" + for native in (chunk({}), chunk({"content": None}), chunk({"content": ""}), chunk({"content": "safe"})): + Draft202012Validator(schema).validate(native) + error = export_payload_schema(ChatCompletionsStreamErrorProjection, projection_id="error") + assert error["additionalProperties"] is False + assert error["properties"]["error"]["additionalProperties"] is False + assert error["properties"]["error"][EXTENSION]["unknown_fields"] == "configurable" + assert error["properties"]["error"]["properties"]["message"][EXTENSION]["classification"] == "constrained" + Draft202012Validator(error).validate({"error": {"message": "failed", "type": {"opaque": True}}}) + + +def test_stream_hooks_frame_provider_native_error_and_terminal_events(): + body = b'{"error":{"code":"failed"}}' + + assert ChatCompletionsStreamHooks().encode_error(body) == ( + b'data: {"error":{"code":"failed"}}\n\n', + b"data: [DONE]\n\n", + ) + + +@pytest.mark.parametrize( + "module", + [ + "nemoguardrails.server.experimental.providers.openai.chat_completions.request_binding", + "nemoguardrails.server.experimental.providers.openai.chat_completions.stream_projection", + "nemoguardrails.server.experimental.providers.openai.chat_completions.stream_classifier", + "nemoguardrails.server.experimental.providers.openai.chat_completions.stream_hooks", + ], +) +def test_stream_modules_import_in_a_fresh_interpreter(module): + result = subprocess.run( + [sys.executable, "-c", f"import {module}"], + check=False, + capture_output=True, + text=True, + ) + + assert result.returncode == 0, result.stderr + + +def test_stream_declarations_preserve_transport_shapes_and_provenance(): + """Check the staged stream declarations without a second authored policy.""" + contracts = Path(__file__).parents[3] / "nemoguardrails/server/experimental/contracts/openai" + source = yaml.safe_load((contracts / "source.yaml").read_text(encoding="utf-8")) + assert STREAM_CLASSIFIER.require_event_type is False + assert STREAM_CLASSIFIER.non_data_shape == "keepalive" + assert STREAM_CLASSIFIER.sentinels == ((b"[DONE]", "[DONE]"),) + shapes = STREAM_CLASSIFIER.contract.shapes + assert shapes.guarded_shapes == frozenset({"chat.completion.chunk:content"}) + assert shapes.opaque_shapes == frozenset({"[DONE]", "keepalive", "chat.completion.chunk:metadata"}) + assert shapes.provider_error_shapes == frozenset({"error"}) + assert shapes.snapshot_shapes == frozenset() + content, error = STREAM_CLASSIFIER.rules + assert content.match == (("object", "chat.completion.chunk"),) + assert content.required_fields == frozenset() + assert content.text_path == ("choices", 0, "delta", "content") + assert content.missing_text_shape == "chat.completion.chunk:metadata" + assert content.missing_text_role is StreamEventRole.OPAQUE_METADATA + assert error.match == () + assert error.required_fields == frozenset({"error"}) + assert error.role is StreamEventRole.PROVIDER_ERROR + + +@pytest.mark.parametrize( + ("delta", "text"), + [ + ({}, None), + ({"content": None}, None), + ({"role": None}, None), + ({"role": "assistant", "content": ""}, ""), + ({"content": "answer"}, "answer"), + ], +) +def test_stream_text_preserves_missing_null_and_empty_string_semantics(delta, text): + """Null or absent content is metadata; an empty string is still guarded text.""" + classified = STREAM_CLASSIFIER.classify_event(event(chunk(delta))) + assert classified.text == text + expected = StreamEventRole.OPAQUE_METADATA if text is None else StreamEventRole.GUARDED_TEXT + assert classified.role is expected + + +@pytest.mark.parametrize("index", [False, True, "0", 0.0, 1]) +def test_stream_choice_index_requires_the_declared_integer_zero(index): + native = chunk({"content": "hidden"}) + native["choices"][0]["index"] = index + + with pytest.raises(UnsupportedProviderStream): + STREAM_CLASSIFIER.classify_event(event(native)) + + +@pytest.mark.parametrize("delta", [{"content": "safe"}, {"role": "assistant"}, {}]) +@pytest.mark.parametrize( + "fields", + [ + {"message": {"role": "assistant", "content": "unchecked"}}, + {"text": "unchecked"}, + {"future_content": "unchecked"}, + ], +) +def test_stream_classifier_rejects_unreviewed_choice_fields(delta, fields): + with pytest.raises(UnsupportedProviderStream): + STREAM_CLASSIFIER.classify_event(event(chunk(delta, **fields))) + + +def test_stream_classifier_preserves_reviewed_choice_metadata(): + native = chunk({"content": "safe"}, finish_reason="stop", logprobs=None) + + classified = STREAM_CLASSIFIER.classify_event(event(native)) + + assert classified.text == "safe" + + +@pytest.mark.parametrize("logprobs", [{"content": [{"token": "uninspected"}]}, {"content": []}]) +def test_stream_classifier_rejects_log_probabilities(logprobs): + """Stream log probabilities carry token text, as in buffered responses.""" + with pytest.raises(UnsupportedProviderStream): + STREAM_CLASSIFIER.classify_event(event(chunk({"content": "safe"}, logprobs=logprobs))) + + +@pytest.mark.parametrize("member", [{"future": 1}, {"Choices": []}]) +def test_stream_classifier_rejects_unreviewed_root_members(member): + """The stream root is closed like the buffered response root.""" + with pytest.raises(UnsupportedProviderStream): + STREAM_CLASSIFIER.classify_event(event({**chunk({"content": "safe"}), **member})) + + +def observe(hooks, value): + parsed = value if isinstance(value, ServerSentEvent) else event(value) + hooks.observe_event(parsed, STREAM_CLASSIFIER.classify_event(parsed)) + + +@pytest.mark.parametrize("keepalive", [b": keepalive\n\n", b"\n", b"id: keepalive\n\n"]) +def test_stream_hooks_require_actual_done_after_keepalives(keepalive): + hooks = ChatCompletionsStreamHooks() + observe(hooks, ServerSentEvent.from_bytes(keepalive)) + + with pytest.raises(UnsupportedProviderStream, match="before.*DONE"): + hooks.validate_end_of_stream() + + observe(hooks, chunk({"content": "safe"})) + observe(hooks, b"[DONE]") + observe(hooks, ServerSentEvent.from_bytes(keepalive)) + hooks.validate_end_of_stream() + + +@pytest.mark.parametrize("tail", [b"[DONE]", chunk({"content": "later"}), chunk({}), {"error": {"message": "late"}}]) +def test_stream_hooks_reject_data_after_done(tail): + hooks = ChatCompletionsStreamHooks() + observe(hooks, b"[DONE]") + + with pytest.raises(UnsupportedProviderStream, match="after.*DONE"): + observe(hooks, tail) + + +@pytest.mark.parametrize("done", [False, True]) +def test_stream_hooks_preserve_provider_error_endings(done): + hooks = ChatCompletionsStreamHooks() + observe(hooks, {"error": {"message": "upstream failed"}}) + observe(hooks, ServerSentEvent.from_bytes(b": keepalive\n\n")) + if done: + observe(hooks, b"[DONE]") + + hooks.validate_end_of_stream() + + +@pytest.mark.parametrize("tail", [chunk({"content": "later"}), chunk({}), {"error": {"message": "again"}}]) +def test_stream_hooks_reject_payloads_after_provider_error(tail): + hooks = ChatCompletionsStreamHooks() + observe(hooks, {"error": {"message": "upstream failed"}}) + + with pytest.raises(UnsupportedProviderStream, match="after.*error"): + observe(hooks, tail) + + +def test_stream_adapter_factory_creates_independent_completion_state(): + factory = create_classified_stream_adapter_factory(STREAM_CLASSIFIER, ChatCompletionsStreamHooks) + first, second = factory(), factory() + first.classify_event(event(b"[DONE]")) + second.classify_event(event(chunk({"content": "safe"}))) + + first.validate_end_of_stream() + with pytest.raises(UnsupportedProviderStream): + second.validate_end_of_stream() + second.classify_event(event(b"[DONE]")) + second.validate_end_of_stream() + + +class StreamChecker: + def __init__(self): + self.checked = [] + + async def check_output(self, check): + self.checked.append(check.output_content) + return ContentAllowed() + + +async def guarded_events(events, chunk_size=1): + closed = [] + + async def source(): + try: + for item in events: + yield item.raw + finally: + closed.append(True) + + checker = StreamChecker() + result = b"".join( + [ + raw + async for raw in guard_provider_stream( + source(), + checker=checker, + streaming_policy=StreamBufferingPolicy(chunk_size, 0), + input_message=GuardedMessage("user", "question"), + adapter=ClassifiedStreamAdapter(STREAM_CLASSIFIER, ChatCompletionsStreamHooks()), + render_outcome=render_openai_error, + ) + ] + ) + assert closed == [True] + return result, checker.checked + + +@pytest.mark.asyncio +@pytest.mark.parametrize("delta", [{"content": "safe"}, {"role": "assistant"}]) +async def test_guarded_stream_hides_unreviewed_choice_content(delta): + invalid = event(chunk(delta, message={"content": "unchecked"})) + + result, checked = await guarded_events([invalid, event(b"[DONE]")]) + + assert invalid.raw not in result + assert b"unchecked" not in result + assert checked == [] + assert b"unsupported_chat_completions_response_shape" in result + assert result.endswith(event(b"[DONE]").raw) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("chunk_size", [1, 2]) +async def test_guarded_stream_reports_missing_done_without_releasing_pending_text(chunk_size): + text = event(chunk({"content": "safe"})) + + result, checked = await guarded_events([text], chunk_size) + + assert (text.raw in result) is (chunk_size == 1) + assert checked == (["safe"] if chunk_size == 1 else []) + assert b"unsupported_chat_completions_response_shape" in result + assert result.endswith(event(b"[DONE]").raw) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("tail", [event(b"[DONE]"), event(chunk({"content": "later"}))]) +async def test_guarded_stream_hides_invalid_events_after_done(tail): + text = event(chunk({"content": "safe"})) + + result, checked = await guarded_events([text, event(b"[DONE]"), tail]) + + assert result.startswith(text.raw) + assert checked == ["safe"] + assert b"later" not in result + assert b"unsupported_chat_completions_response_shape" in result + assert result.count(event(b"[DONE]").raw) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("done", [False, True]) +async def test_guarded_stream_preserves_provider_errors_without_synthesizing_another(done): + events = [event({"error": {"message": "upstream failed"}})] + if done: + events.append(event(b"[DONE]")) + + result, checked = await guarded_events(events) + + assert result == b"".join(item.raw for item in events) + assert checked == [] + + +@pytest.mark.asyncio +async def test_guarded_stream_preserves_valid_sequence_and_reviewed_metadata(): + events = [ + ServerSentEvent.from_bytes(b": keepalive\r\n\r\n"), + event(chunk({"role": "assistant"})), + event(chunk({"content": "safe"}, finish_reason=None, logprobs=None)), + event(chunk({}, finish_reason="stop")), + event({"object": "chat.completion.chunk", "choices": [], "usage": {"total_tokens": 1}}), + event(b"[DONE]"), + ServerSentEvent.from_bytes(b": trailing keepalive\n\n"), + ] + + result, checked = await guarded_events(events) + + assert result == b"".join(item.raw for item in events) + assert checked == ["safe"] + + +_CHUNK = chunk({"content": "safe"}, finish_reason="stop") +_ERROR = {"error": {"message": "failed", "type": "server_error"}} + + +@pytest.mark.parametrize("policy", list(UnknownContentFieldPolicy)) +@pytest.mark.parametrize( + ("model", "document"), + [ + (ChatCompletionsStreamPayloadProjection, _CHUNK), + (ChatCompletionsStreamPayloadProjection, {**_CHUNK, "usage": {"total_tokens": 1}}), + (ChatCompletionsStreamPayloadProjection, {**_CHUNK, "future": 1}), + (ChatCompletionsStreamPayloadProjection, {**_CHUNK, "Usage": {}}), + (ChatCompletionsStreamPayloadProjection, chunk({"content": "safe"}, future=1)), + (ChatCompletionsStreamPayloadProjection, chunk({"content": "safe", "future": 1})), + (ChatCompletionsStreamPayloadProjection, chunk({"content": "safe", "Content": "x"})), + (ChatCompletionsStreamErrorProjection, _ERROR), + (ChatCompletionsStreamErrorProjection, {"error": {**_ERROR["error"], "future": 1}}), + (ChatCompletionsStreamErrorProjection, {"error": {**_ERROR["error"], "Message": "x"}}), + (ChatCompletionsStreamErrorProjection, {**_ERROR, "future": 1}), + ], +) +def test_stream_export_member_policy_matches_projections(model, document, policy, export_policy_validator): + """Export and typed projections agree on which object members are accepted. + + This covers object-member policy only, not full value equivalence (see the + integer index test below). The policy is applied by direct projection + validation; the stream classifier itself always validates with the closed + default, so trusted ALLOW is not exercised through the classifier here. + """ + exported = export_payload_schema(model, projection_id=model.__name__) + derived = export_policy_validator(exported, policy).is_valid(document) + try: + model.validate_payload(document, unknown_content_fields=policy) + runtime = True + except ValidationError: + runtime = False + + assert derived is runtime + + +def test_strict_index_is_a_known_export_value_gap(): + """JSON Schema accepts 0.0 as integer 0; the runtime keeps the stricter integer check.""" + exported = export_payload_schema(ChatCompletionsStreamPayloadProjection, projection_id="stream") + native = chunk({"content": "safe"}) + native["choices"][0]["index"] = 0.0 + + assert Draft202012Validator(exported).is_valid(native) + with pytest.raises(ValidationError): + ChatCompletionsStreamPayloadProjection.validate_payload(native) + + +@pytest.mark.parametrize("tool_calls", [None, []]) +def test_stream_delta_accepts_absent_tool_calls(tool_calls): + """Null and empty tool calls carry no tool content, as in buffered responses.""" + classified = STREAM_CLASSIFIER.classify_event(event(chunk({"content": "answer", "tool_calls": tool_calls}))) + + assert classified.text == "answer" diff --git a/tests/server/experimental/test_projection_policy.py b/tests/server/experimental/test_projection_policy.py index d471ebd679..b8cfc0c050 100644 --- a/tests/server/experimental/test_projection_policy.py +++ b/tests/server/experimental/test_projection_policy.py @@ -148,9 +148,20 @@ def test_helpers_do_not_hide_defaults(helper): @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"): +def test_helpers_support_runtime_numeric_constraints(helper): + class NumericPolicy(PolicyModel): + value: Annotated[int, helper(ge=0, le=2, strict=True)] + + assert NumericPolicy.model_validate({"value": 1}).value == 1 + for invalid in (-1, 3, True, "1", 1.0): + with pytest.raises(ValueError): + NumericPolicy.model_validate({"value": invalid}) + + +@pytest.mark.parametrize("helper", [guarded, constrained]) +@pytest.mark.parametrize("constraint", ["min_lenght", "gt", "lt", "multiple_of", "allow_inf_nan", "max_digits"]) +def test_helpers_reject_unsupported_constraint_names(helper, constraint): + with pytest.raises(ValueError, match="Unsupported Pydantic constraints"): helper(**{constraint: 1}) diff --git a/tests/server/experimental/test_stream_policy.py b/tests/server/experimental/test_stream_policy.py new file mode 100644 index 0000000000..0b603b6f28 --- /dev/null +++ b/tests/server/experimental/test_stream_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. + +"""Exercise typed stream metadata derivation without provider services.""" + +from dataclasses import replace +from typing import Annotated + +import pytest +from pydantic import BaseModel, create_model + +from nemoguardrails.server.experimental.provider import projection_policy +from nemoguardrails.server.experimental.provider.projection_policy import ( + PolicyModel, + constrained, + field_coverage, + guarded, + text_location, +) +from nemoguardrails.server.experimental.provider.stream import ( + StreamCapabilityProfile, + StreamEventRole, + StreamProjectionContract, +) +from nemoguardrails.server.experimental.provider.stream_classifier import ( + StreamClassifierDefinition, + StreamEventRule, + build_stream_classifier, +) +from nemoguardrails.server.experimental.provider.stream_policy import stream_shape_coverage, stream_text_path + + +class TextDelta(PolicyModel): + text: Annotated[str | None, guarded("assistant")] = None + + +class DeltaEnvelope(PolicyModel): + deltas: Annotated[list[TextDelta], guarded(max_length=1)] + + +def test_stream_path_allows_empty_arrays_and_nullable_text_without_weakening_buffered_paths(): + assert stream_text_path(DeltaEnvelope) == ("deltas", 0, "text") + assert DeltaEnvelope(deltas=[]).deltas == [] + assert TextDelta().text is None + assert TextDelta(text="").text == "" + with pytest.raises(ValueError, match="exactly one array item"): + text_location(DeltaEnvelope) + # Buffered extraction also accepts a nullable subject, reported as allowing empty text. + assert text_location(TextDelta).allows_empty is True + + +def test_stream_path_follows_nullable_model_containers(): + """A null container yields missing text; the event rule decides how that is classified.""" + + class Envelope(PolicyModel): + delta: Annotated[TextDelta | None, guarded()] = None + + assert stream_text_path(Envelope) == ("delta", "text") + + +@pytest.mark.parametrize("maximum", [None, 0, 2]) +def test_stream_path_rejects_arrays_without_an_explicit_single_item_boundary(maximum): + class Envelope(PolicyModel): + deltas: Annotated[list[TextDelta], guarded(max_length=maximum)] + + with pytest.raises(ValueError, match="at most one array item"): + stream_text_path(Envelope) + + +@pytest.mark.parametrize("annotation", [int, list[str], str | int]) +def test_stream_path_rejects_non_string_or_ambiguous_subjects(annotation): + model = create_model( + "InvalidSubject", __base__=PolicyModel, text=(Annotated[annotation, guarded("assistant")], ...) + ) + with pytest.raises(ValueError): + stream_text_path(model) + + +@pytest.mark.parametrize("fields", [{}, {"a": TextDelta, "b": TextDelta}]) +def test_stream_path_requires_one_subject(fields): + model = create_model( + "Envelope", __base__=PolicyModel, **{name: (Annotated[child, guarded()], ...) for name, child in fields.items()} + ) + with pytest.raises(ValueError, match="exactly one guarded subject"): + stream_text_path(model) + + +def test_stream_path_validates_replacement_blocker_like_buffered_paths(): + """Streams do not replace text, but a misspelled blocker is still a policy error.""" + + class Typo(PolicyModel): + annotations: Annotated[list[str] | None, constrained()] = None + text: Annotated[str | None, guarded("assistant", replaceable=True, blocked_by="annotatons")] = None + + class Sibling(PolicyModel): + annotations: Annotated[list[str] | None, constrained()] = None + text: Annotated[str | None, guarded("assistant", replaceable=True, blocked_by="annotations")] = None + + with pytest.raises(ValueError, match="Replacement blocker 'annotatons'"): + stream_text_path(Typo) + with pytest.raises(ValueError, match="Replacement blocker 'annotatons'"): + text_location(Typo) + assert stream_text_path(Sibling) == ("text",) + + +def rule(shape, role, **kwargs): + return StreamEventRule(shape=shape, model=TextDelta, role=role, match=(("type", shape),), **kwargs) + + +def test_shape_coverage_includes_rules_snapshots_fallbacks_and_transport(): + rules = ( + rule( + "text", + StreamEventRole.GUARDED_TEXT, + text_path=("text",), + missing_text_role=StreamEventRole.OPAQUE_METADATA, + missing_text_shape="metadata", + ), + rule( + "snapshot", + StreamEventRole.TEXT_SNAPSHOT, + text_path=("text",), + missing_text_role=StreamEventRole.TEXT_SNAPSHOT, + missing_text_value="", + ), + rule("error", StreamEventRole.PROVIDER_ERROR), + rule("usage", StreamEventRole.OPAQUE_METADATA), + ) + sentinels = ((b"[DONE]", "done"),) + coverage = stream_shape_coverage(rules, sentinels=sentinels, non_data_shape="keepalive") + assert coverage.guarded_shapes == frozenset({"text"}) + assert coverage.snapshot_shapes == frozenset({"snapshot"}) + assert coverage.provider_error_shapes == frozenset({"error"}) + assert coverage.opaque_shapes == frozenset({"metadata", "usage", "done", "keepalive"}) + contract = StreamProjectionContract("example", StreamCapabilityProfile.SINGLE_TEXT_DELTA_V1, coverage) + classifier = build_stream_classifier( + StreamClassifierDefinition("example", contract, rules, sentinels=sentinels, non_data_shape="keepalive") + ) + assert classifier.contract.shapes == coverage + + +def test_shape_coverage_rejects_conflicting_roles(): + rules = (rule("text", StreamEventRole.GUARDED_TEXT, text_path=("text",)),) + with pytest.raises(ValueError, match="multiple classifications"): + stream_shape_coverage(rules, non_data_shape="text") + + +def test_derived_coverage_does_not_bypass_classifier_consistency_checks(): + rules = (rule("text", StreamEventRole.GUARDED_TEXT),) + contract = StreamProjectionContract( + "example", StreamCapabilityProfile.SINGLE_TEXT_DELTA_V1, stream_shape_coverage(rules) + ) + with pytest.raises(ValueError, match="inconsistent text role and path"): + build_stream_classifier(StreamClassifierDefinition("example", contract, rules)) + + +def test_stream_path_ignores_unrelated_field_types(): + """A constrained union elsewhere in the model cannot break the guarded text binding.""" + + class Envelope(PolicyModel): + deltas: Annotated[list[TextDelta], guarded(max_length=1)] + seed: Annotated[str | int | None, constrained()] = None + + assert stream_text_path(Envelope) == ("deltas", 0, "text") + + +def test_stream_path_does_not_generate_schemas(monkeypatch): + """Runtime bindings come from typed fields, not from contract export.""" + + def unexpected(*args, **kwargs): + raise AssertionError("stream text paths must not generate schemas") + + monkeypatch.setattr(projection_policy, "export_payload_schema", unexpected) + monkeypatch.setattr(PolicyModel, "model_json_schema", classmethod(unexpected)) + assert stream_text_path(DeltaEnvelope) == ("deltas", 0, "text") + + +def test_stream_path_rejects_union_containers_and_recursion(): + class Other(PolicyModel): + text: Annotated[str, guarded("assistant")] + + class Either(PolicyModel): + delta: Annotated[TextDelta | Other, guarded()] + + class Node(PolicyModel): + child: Annotated["Node | None", guarded()] = None + + Node.model_rebuild() + with pytest.raises(ValueError, match="object-union selectors"): + stream_text_path(Either) + with pytest.raises(ValueError, match="Recursive"): + stream_text_path(Node) + + +@pytest.mark.parametrize("mutation", ["path", "role", "coverage", "inventory"]) +def test_typed_classifier_rejects_policy_drift(mutation): + """Binding mistakes must not downgrade guarded content to metadata.""" + from nemoguardrails.server.experimental.provider import stream_policy + + event = StreamEventRule( + "text", + DeltaEnvelope, + StreamEventRole.GUARDED_TEXT, + required_fields=frozenset({"deltas"}), + text_path=("deltas", 0, "text"), + missing_text_role=StreamEventRole.OPAQUE_METADATA, + missing_text_shape="metadata", + ) + coverage = stream_shape_coverage((event,)) + contract = StreamProjectionContract( + "example", StreamCapabilityProfile.SINGLE_TEXT_DELTA_V1, coverage, field_coverage(DeltaEnvelope) + ) + if mutation == "path": + event = replace(event, text_path=("deltas", 1, "text")) + elif mutation == "role": + event = replace( + event, role=StreamEventRole.OPAQUE_METADATA, text_path=None, missing_text_role=None, missing_text_shape=None + ) + contract = replace(contract, shapes=stream_shape_coverage((event,))) + elif mutation == "coverage": + contract = replace( + contract, fields=replace(contract.fields, guarded_fields=frozenset(), opaque_fields=frozenset({"deltas"})) + ) + else: + contract = replace(contract, shapes=replace(coverage, opaque_shapes=frozenset({"metadata", "undeclared"}))) + definition = StreamClassifierDefinition("example", contract, (event,)) + with pytest.raises(ValueError, match="field policy|guarded subject|event rules"): + stream_policy.build_policy_stream_classifier(definition) + + +def test_typed_classifier_preserves_nullable_missing_text_and_checks_actual_text(): + """Validated bindings retain explicit missing-text behavior.""" + from nemoguardrails.server.experimental.provider import stream_policy + from nemoguardrails.server.experimental.provider.sse import ServerSentEvent + + event = StreamEventRule( + "text", + DeltaEnvelope, + StreamEventRole.GUARDED_TEXT, + required_fields=frozenset({"deltas"}), + text_path=stream_text_path(DeltaEnvelope), + missing_text_role=StreamEventRole.OPAQUE_METADATA, + missing_text_shape="metadata", + ) + contract = StreamProjectionContract( + "example", StreamCapabilityProfile.SINGLE_TEXT_DELTA_V1, stream_shape_coverage((event,)) + ) + classifier = stream_policy.build_policy_stream_classifier(StreamClassifierDefinition("example", contract, (event,))) + assert ( + classifier.classify_event(ServerSentEvent.from_bytes(b'data: {"deltas": []}\n\n')).role + is StreamEventRole.OPAQUE_METADATA + ) + classified = classifier.classify_event(ServerSentEvent.from_bytes(b'data: {"deltas": [{"text": "checked"}]}\n\n')) + assert classified.role is StreamEventRole.GUARDED_TEXT + assert classified.text == "checked" + + +def test_stream_path_reads_annotations_on_array_items(): + class Envelope(PolicyModel): + deltas: Annotated[list[Annotated[TextDelta, guarded()]], guarded(max_length=1)] + + assert stream_text_path(Envelope) == ("deltas", 0, "text") + + +def test_stream_path_rejects_containers_that_do_not_lead_to_a_model(): + class Envelope(PolicyModel): + deltas: Annotated[list[int], guarded(max_length=1)] + + with pytest.raises(ValueError, match="must lead to a named string field"): + stream_text_path(Envelope) + + +def policy_definition(event, **changes): + contract = StreamProjectionContract( + "example", StreamCapabilityProfile.SINGLE_TEXT_DELTA_V1, stream_shape_coverage((event,), **changes) + ) + return StreamClassifierDefinition("example", contract, (event,), **changes) + + +def test_typed_classifier_requires_policy_models(): + from nemoguardrails.server.experimental.provider import stream_policy + + class Untyped(BaseModel): + kind: str + + event = StreamEventRule("meta", Untyped, StreamEventRole.OPAQUE_METADATA, match=(("kind", "meta"),)) + with pytest.raises(TypeError, match="policy-annotated event models"): + stream_policy.build_policy_stream_classifier(policy_definition(event)) + + +def test_typed_text_events_require_one_assistant_subject(): + from nemoguardrails.server.experimental.provider import stream_policy + + class UserText(PolicyModel): + text: Annotated[str, guarded("user")] + + event = StreamEventRule( + "text", UserText, StreamEventRole.GUARDED_TEXT, required_fields=frozenset({"text"}), text_path=("text",) + ) + with pytest.raises(ValueError, match="one guarded subject with the assistant role"): + stream_policy.build_policy_stream_classifier(policy_definition(event)) + + +@pytest.mark.parametrize("sentinels", [((b" [DONE]", "done"),), ((b"[DONE]", "done"), (b"[DONE]", "done"))]) +def test_typed_classifier_requires_canonical_unique_sentinels(sentinels): + from nemoguardrails.server.experimental.provider import stream_policy + + event = StreamEventRule( + "text", + DeltaEnvelope, + StreamEventRole.GUARDED_TEXT, + required_fields=frozenset({"deltas"}), + text_path=stream_text_path(DeltaEnvelope), + ) + with pytest.raises(ValueError, match="unique nonempty canonical"): + stream_policy.build_policy_stream_classifier(policy_definition(event, sentinels=sentinels))