feat(server): add OpenAI Chat stream projections and hooks

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"],

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

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