From 70df7d5e4637e4cccc1a3c98ef7e6a3bda7f523d Mon Sep 17 00:00:00 2001
From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
Date: Thu, 8 Oct 2026 17:12:28 +0200
Subject: [PATCH 1/7] feat(server): derive stream text paths from typed
 policies

Add provider-neutral helpers that derive stream event metadata from the
typed projection declarations instead of a second authored policy.

stream_text_path follows guarded fields directly, without schema
generation, so unrelated field types cannot affect the binding. It
follows guarded models, nullable model containers, and lists bounded to
at most one item. Unlike buffered extraction, empty arrays, null
containers, and null text are valid and become missing text for the
event rule to classify. Union containers, unbounded arrays, non-string
subjects, recursion, and ambiguous subjects are rejected.

stream_shape_coverage collects event shapes by role from the declared
rules, missing-text fallbacks, sentinels, and non-data shapes.

Signed-off-by: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
---
 .../experimental/provider/stream_policy.py    | 128 ++++++++++++
 .../server/experimental/test_stream_policy.py | 184 ++++++++++++++++++
 2 files changed, 312 insertions(+)
 create mode 100644 nemoguardrails/server/experimental/provider/stream_policy.py
 create mode 100644 tests/server/experimental/test_stream_policy.py

diff --git a/nemoguardrails/server/experimental/provider/stream_policy.py b/nemoguardrails/server/experimental/provider/stream_policy.py
new file mode 100644
index 0000000000..7aa3f24b14
--- /dev/null
+++ b/nemoguardrails/server/experimental/provider/stream_policy.py
@@ -0,0 +1,128 @@
+# 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_policy
+from nemoguardrails.server.experimental.provider.stream import StreamEventRole, StreamShapeCoverage
+from nemoguardrails.server.experimental.provider.stream_classifier import StreamEventRule, StreamTextPath
+
+
+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, or a recursive path. 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")
+            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]),
+    )
diff --git a/tests/server/experimental/test_stream_policy.py b/tests/server/experimental/test_stream_policy.py
new file mode 100644
index 0000000000..d53f66ce6f
--- /dev/null
+++ b/tests/server/experimental/test_stream_policy.py
@@ -0,0 +1,184 @@
+# 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 typing import Annotated
+
+import pytest
+from pydantic import create_model
+
+from nemoguardrails.server.experimental.provider import projection_policy
+from nemoguardrails.server.experimental.provider.projection_policy import (
+    PolicyModel,
+    constrained,
+    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 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)

From 9475a838598c096f46a43a81af5b65b6c77e3a64 Mon Sep 17 00:00:00 2001
From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
Date: Mon, 5 Oct 2026 16:17:03 +0200
Subject: [PATCH 2/7] feat(server): add OpenAI Chat stream projections and
 hooks

Declare the Chat single_text_delta.v1 stream boundary with typed
projections and ObjectPolicy. The root, choice, and delta reject
unreviewed members; reviewed root metadata passes through unchanged. A
chunk carries zero or one choice whose index is exactly the integer 0
and whose logprobs are null, as in buffered responses. Missing or null
content is metadata, empty string content is guarded text, and provider
error messages are constrained, not guarded output. The error envelope
is closed, and its error object is configurable but closed by default.

Bind the projections to explicit stream event rules for content,
provider errors, and the [DONE] sentinel, with field and shape coverage
derived from the declarations. The OpenAI pin comes from
providers/openai/source.py.

Keep lifecycle checks in handwritten hooks: require an actual [DONE]
for successful completion, distinguish it from keepalives, reject
payloads after termination or a provider error, and frame native
errors.

Tests cover classification, unchecked sibling content, completion and
termination, raw preservation, isolated hook state, rejected log
probabilities and unreviewed members, and member-policy agreement
between the export and the projections. JSON Schema accepts index 0.0
where the runtime does not, a known value-level gap.

Signed-off-by: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
---
 .../openai/chat_completions/README.md         |  41 +-
 .../chat_completions/stream_classifier.py     |  94 +++
 .../openai/chat_completions/stream_hooks.py   |  63 ++
 .../chat_completions/stream_projection.py     | 100 ++++
 .../test_openai_chat_streaming_projections.py | 543 ++++++++++++++++++
 5 files changed, 837 insertions(+), 4 deletions(-)
 create mode 100644 nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py
 create mode 100644 nemoguardrails/server/experimental/providers/openai/chat_completions/stream_hooks.py
 create mode 100644 nemoguardrails/server/experimental/providers/openai/chat_completions/stream_projection.py
 create mode 100644 tests/server/experimental/test_openai_chat_streaming_projections.py

diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md b/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md
index 40be254617..38091afec1 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,39 @@ 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, 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.
+
+[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..7be64c07aa
--- /dev/null
+++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py
@@ -0,0 +1,94 @@
+# 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,
+    build_stream_classifier,
+)
+from nemoguardrails.server.experimental.provider.stream_policy import 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_SOURCE_SCHEMA = ChatCompletionsStreamPayloadProjection.policy.source
+REJECTED_PROVIDER_SCHEMAS = ()
+STREAM_SENTINELS = ((b"[DONE]", "[DONE]"),)
+STREAM_NON_DATA_SHAPE = "[DONE]"
+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_FIELD_COVERAGE = field_coverage(ChatCompletionsStreamPayloadProjection)
+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=STREAM_FIELD_COVERAGE,
+)
+
+CAPABILITY_PROFILE = STREAM_CONTRACT.profile.value
+GUARDED_SHAPES = STREAM_CONTRACT.shapes.guarded_shapes
+SNAPSHOT_SHAPES = STREAM_CONTRACT.shapes.snapshot_shapes
+OPAQUE_SHAPES = STREAM_CONTRACT.shapes.opaque_shapes
+PROVIDER_ERROR_SHAPES = STREAM_CONTRACT.shapes.provider_error_shapes
+STREAM_FIELDS = (
+    STREAM_FIELD_COVERAGE.guarded_fields,
+    STREAM_FIELD_COVERAGE.constrained_fields,
+    STREAM_FIELD_COVERAGE.opaque_fields,
+)
+
+STREAM_CLASSIFIER = build_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..6739724503
--- /dev/null
+++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_projection.py
@@ -0,0 +1,100 @@
+# 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 pydantic import BeforeValidator
+
+from nemoguardrails.server.experimental.provider.payload import GuardedContentModel, GuardedProjectionModel
+from nemoguardrails.server.experimental.provider.projection_policy import (
+    ObjectPolicy,
+    PolicyModel,
+    constrained,
+    disabled,
+    guarded,
+    opaque,
+)
+
+
+def _require_int(value: object) -> object:
+    """Reject booleans, strings, and floats before Literal[0] can coerce them."""
+    if type(value) is not int:
+        raise ValueError("index must be the integer 0")
+    return value
+
+
+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
+    tool_calls: Annotated[None, disabled("core_capability.tool_content")] = 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[Literal[0], BeforeValidator(_require_int), constrained()]
+    # 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..7ab5da5dd2
--- /dev/null
+++ b/tests/server/experimental/test_openai_chat_streaming_projections.py
@@ -0,0 +1,543 @@
+# 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 (
+    CAPABILITY_PROFILE,
+    GUARDED_SHAPES,
+    OPAQUE_SHAPES,
+    PROVIDER_ERROR_SHAPES,
+    SNAPSHOT_SHAPES,
+    STREAM_CLASSIFIER,
+    STREAM_CONTRACT,
+    STREAM_FIELDS,
+    STREAM_SOURCE_SCHEMA,
+)
+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)
+    assert shape in GUARDED_SHAPES | OPAQUE_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 == "[DONE]"
+    assert classified.role is StreamEventRole.OPAQUE_METADATA
+
+
+@pytest.mark.parametrize(
+    "native",
+    [
+        chunk({"reasoning_content": "unguarded"}),
+        chunk({"content": "answer", "tool_calls": []}),
+        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 CAPABILITY_PROFILE == "single_text_delta.v1"
+    assert STREAM_SOURCE_SCHEMA == "CreateChatCompletionStreamResponse"
+    assert STREAM_CONTRACT.projection_id == "openai.chat_completions.stream.text.v1"
+    assert STREAM_CONTRACT.fields is not None
+    assert STREAM_FIELDS == (
+        STREAM_CONTRACT.fields.guarded_fields,
+        STREAM_CONTRACT.fields.constrained_fields,
+        STREAM_CONTRACT.fields.opaque_fields,
+    )
+
+
+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_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_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"]["const"] == 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 == "[DONE]"
+    assert STREAM_CLASSIFIER.sentinels == ((b"[DONE]", "[DONE]"),)
+    assert GUARDED_SHAPES == frozenset({"chat.completion.chunk:content"})
+    assert OPAQUE_SHAPES == frozenset({"[DONE]", "chat.completion.chunk:metadata"})
+    assert PROVIDER_ERROR_SHAPES == frozenset({"error"})
+    assert 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)

From 1cd68f3e9a488aa89d26f74bc5da956160ce9581 Mon Sep 17 00:00:00 2001
From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
Date: Fri, 9 Oct 2026 16:30:38 +0200
Subject: [PATCH 3/7] fix(server): validate typed stream bindings against field
 policy

Signed-off-by: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
---
 .../experimental/provider/stream_policy.py    | 63 +++++++++++++++++-
 .../openai/chat_completions/README.md         |  6 ++
 .../chat_completions/stream_classifier.py     |  9 ++-
 .../server/experimental/test_stream_policy.py | 65 +++++++++++++++++++
 4 files changed, 138 insertions(+), 5 deletions(-)

diff --git a/nemoguardrails/server/experimental/provider/stream_policy.py b/nemoguardrails/server/experimental/provider/stream_policy.py
index 7aa3f24b14..76a246acb9 100644
--- a/nemoguardrails/server/experimental/provider/stream_policy.py
+++ b/nemoguardrails/server/experimental/provider/stream_policy.py
@@ -24,9 +24,20 @@
 from types import UnionType
 from typing import Annotated, Any, Union, get_args, get_origin
 
-from nemoguardrails.server.experimental.provider.projection_policy import PolicyModel, field_policy
+from nemoguardrails.server.experimental.provider.projection_policy import (
+    PolicyModel,
+    field_coverage,
+    field_policy,
+    model_graph,
+)
 from nemoguardrails.server.experimental.provider.stream import StreamEventRole, StreamShapeCoverage
-from nemoguardrails.server.experimental.provider.stream_classifier import StreamEventRule, StreamTextPath
+from nemoguardrails.server.experimental.provider.stream_classifier import (
+    StreamClassifier,
+    StreamClassifierDefinition,
+    StreamEventRule,
+    StreamTextPath,
+    build_stream_classifier,
+)
 
 
 def stream_text_path(model: type[PolicyModel]) -> StreamTextPath:
@@ -126,3 +137,51 @@ def stream_shape_coverage(
         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 38091afec1..f7b34c4ba9 100644
--- a/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md
+++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md
@@ -78,6 +78,12 @@ 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 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
diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py
index 7be64c07aa..4d33119d42 100644
--- a/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py
+++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py
@@ -29,9 +29,12 @@
 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
+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,
@@ -83,7 +86,7 @@
     STREAM_FIELD_COVERAGE.opaque_fields,
 )
 
-STREAM_CLASSIFIER = build_stream_classifier(
+STREAM_CLASSIFIER = build_policy_stream_classifier(
     StreamClassifierDefinition(
         subject=PROJECTION_ID,
         contract=STREAM_CONTRACT,
diff --git a/tests/server/experimental/test_stream_policy.py b/tests/server/experimental/test_stream_policy.py
index d53f66ce6f..2c429a70b2 100644
--- a/tests/server/experimental/test_stream_policy.py
+++ b/tests/server/experimental/test_stream_policy.py
@@ -15,6 +15,7 @@
 
 """Exercise typed stream metadata derivation without provider services."""
 
+from dataclasses import replace
 from typing import Annotated
 
 import pytest
@@ -24,6 +25,7 @@
 from nemoguardrails.server.experimental.provider.projection_policy import (
     PolicyModel,
     constrained,
+    field_coverage,
     guarded,
     text_location,
 )
@@ -182,3 +184,66 @@ class Node(PolicyModel):
         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"

From 87ceca49b386eea73a71fb5c215d832df2772bfd Mon Sep 17 00:00:00 2001
From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
Date: Fri, 9 Oct 2026 18:16:43 +0200
Subject: [PATCH 4/7] fix(server): accept empty tool_calls in Chat stream
 deltas

The pinned OpenAI schema types ChatCompletionStreamResponseDelta.tool_calls
as an array with no minimum length, so an empty list is a valid value
that carries no tool content. The stream delta declared it null-only and
rejected such chunks, unlike buffered responses.

Constrain delta tool_calls to null or an empty list, matching the
buffered message. Any tool call chunk is still rejected.

Signed-off-by: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
---
 .../openai/chat_completions/stream_projection.py       |  3 ++-
 .../test_openai_chat_streaming_projections.py          | 10 +++++++++-
 2 files changed, 11 insertions(+), 2 deletions(-)

diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_projection.py b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_projection.py
index 6739724503..c5b2357aa6 100644
--- a/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_projection.py
+++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_projection.py
@@ -54,7 +54,8 @@ class ChatCompletionsStreamDeltaProjection(PolicyModel, GuardedContentModel):
     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
-    tool_calls: Annotated[None, disabled("core_capability.tool_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 ChatCompletionsStreamErrorDetailsProjection(PolicyModel, GuardedContentModel):
diff --git a/tests/server/experimental/test_openai_chat_streaming_projections.py b/tests/server/experimental/test_openai_chat_streaming_projections.py
index 7ab5da5dd2..c38a328298 100644
--- a/tests/server/experimental/test_openai_chat_streaming_projections.py
+++ b/tests/server/experimental/test_openai_chat_streaming_projections.py
@@ -121,7 +121,7 @@ def test_stream_classifier_treats_non_data_events_as_declared_opaque_shape():
     "native",
     [
         chunk({"reasoning_content": "unguarded"}),
-        chunk({"content": "answer", "tool_calls": []}),
+        chunk({"content": "answer", "tool_calls": [{"index": 0, "id": "call", "type": "function"}]}),
         chunk({"content": "answer", "future_content": "unguarded"}),
         {
             "object": "chat.completion.chunk",
@@ -541,3 +541,11 @@ def test_strict_index_is_a_known_export_value_gap():
     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"

From feafbf5fd8fbf1df15ff65b768d47ec8412fa747 Mon Sep 17 00:00:00 2001
From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
Date: Fri, 9 Oct 2026 18:20:04 +0200
Subject: [PATCH 5/7] fix(server): validate replacement blockers on stream
 subjects

Stream subjects now share the buffered check that a replacement blocker
names a declared or reviewed opaque sibling. Streaming replacement stays
unsupported, but a misspelled blocker is still a policy error.

Signed-off-by: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
---
 .../provider/projection_policy.py             | 20 +++++++++++++++----
 .../experimental/provider/stream_policy.py    |  5 ++++-
 .../server/experimental/test_stream_policy.py | 18 +++++++++++++++++
 3 files changed, 38 insertions(+), 5 deletions(-)

diff --git a/nemoguardrails/server/experimental/provider/projection_policy.py b/nemoguardrails/server/experimental/provider/projection_policy.py
index 11302db010..198f830c0c 100644
--- a/nemoguardrails/server/experimental/provider/projection_policy.py
+++ b/nemoguardrails/server/experimental/provider/projection_policy.py
@@ -462,6 +462,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 +523,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 +542,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
index 76a246acb9..5822e3f48e 100644
--- a/nemoguardrails/server/experimental/provider/stream_policy.py
+++ b/nemoguardrails/server/experimental/provider/stream_policy.py
@@ -29,6 +29,7 @@
     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 (
@@ -53,7 +54,8 @@ def stream_text_path(model: type[PolicyModel]) -> StreamTextPath:
     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, or a recursive path. Object-union selectors and
+            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] = []
@@ -102,6 +104,7 @@ def visit_model(current: type[PolicyModel], path: StreamTextPath, active: tuple[
             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, (), ())
diff --git a/tests/server/experimental/test_stream_policy.py b/tests/server/experimental/test_stream_policy.py
index 2c429a70b2..94505f784c 100644
--- a/tests/server/experimental/test_stream_policy.py
+++ b/tests/server/experimental/test_stream_policy.py
@@ -97,6 +97,24 @@ def test_stream_path_requires_one_subject(fields):
         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)
 

From 9a05c7ae745a12747ff07cb2cd07a43ac6b26222 Mon Sep 17 00:00:00 2001
From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
Date: Mon, 12 Oct 2026 13:21:55 +0200
Subject: [PATCH 6/7] refactor(server): keep stream declarations independent of
 export vocabulary

Field helpers rejected any Pydantic constraint the guard contract could not
express, which pushed runtime declarations toward custom validators: the
stream choice index used a BeforeValidator to require the integer zero.
Helpers now accept the constraints declarations use (length, pattern, ge, le
and strict), so the index is declared as a strict integer bounded to zero.
Whether the export format can carry a constraint is checked at export time.
Strictness has no JSON Schema keyword and is enforced at runtime only.

The Chat stream classifier also drops unused aliases kept for an older
compiler, derives field coverage directly, and names data-less keepalives
"keepalive" instead of reusing the "[DONE]" sentinel's name. Runtime
behavior is unchanged.

Signed-off-by: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
---
 .../provider/projection_policy.py             | 16 ++++---
 .../openai/chat_completions/README.md         |  9 ++++
 .../chat_completions/stream_classifier.py     | 18 +-------
 .../chat_completions/stream_projection.py     | 11 +----
 .../test_openai_chat_streaming_projections.py | 45 ++++++++-----------
 .../experimental/test_projection_policy.py    | 17 +++++--
 6 files changed, 53 insertions(+), 63 deletions(-)

diff --git a/nemoguardrails/server/experimental/provider/projection_policy.py b/nemoguardrails/server/experimental/provider/projection_policy.py
index 198f830c0c..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.
diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md b/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md
index f7b34c4ba9..ba1eae510e 100644
--- a/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md
+++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/README.md
@@ -70,6 +70,11 @@ 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
@@ -81,6 +86,10 @@ 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.
 
diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py
index 4d33119d42..67b44b7a25 100644
--- a/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py
+++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_classifier.py
@@ -41,10 +41,8 @@
 )
 
 PROJECTION_ID = "openai.chat_completions.stream.text.v1"
-STREAM_SOURCE_SCHEMA = ChatCompletionsStreamPayloadProjection.policy.source
-REJECTED_PROVIDER_SCHEMAS = ()
 STREAM_SENTINELS = ((b"[DONE]", "[DONE]"),)
-STREAM_NON_DATA_SHAPE = "[DONE]"
+STREAM_NON_DATA_SHAPE = "keepalive"
 STREAM_RULES = (
     StreamEventRule(
         shape="chat.completion.chunk:content",
@@ -63,7 +61,6 @@
     ),
 )
 
-STREAM_FIELD_COVERAGE = field_coverage(ChatCompletionsStreamPayloadProjection)
 STREAM_CONTRACT = StreamProjectionContract(
     projection_id=PROJECTION_ID,
     profile=StreamCapabilityProfile.SINGLE_TEXT_DELTA_V1,
@@ -72,18 +69,7 @@
         sentinels=STREAM_SENTINELS,
         non_data_shape=STREAM_NON_DATA_SHAPE,
     ),
-    fields=STREAM_FIELD_COVERAGE,
-)
-
-CAPABILITY_PROFILE = STREAM_CONTRACT.profile.value
-GUARDED_SHAPES = STREAM_CONTRACT.shapes.guarded_shapes
-SNAPSHOT_SHAPES = STREAM_CONTRACT.shapes.snapshot_shapes
-OPAQUE_SHAPES = STREAM_CONTRACT.shapes.opaque_shapes
-PROVIDER_ERROR_SHAPES = STREAM_CONTRACT.shapes.provider_error_shapes
-STREAM_FIELDS = (
-    STREAM_FIELD_COVERAGE.guarded_fields,
-    STREAM_FIELD_COVERAGE.constrained_fields,
-    STREAM_FIELD_COVERAGE.opaque_fields,
+    fields=field_coverage(ChatCompletionsStreamPayloadProjection),
 )
 
 STREAM_CLASSIFIER = build_policy_stream_classifier(
diff --git a/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_projection.py b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_projection.py
index c5b2357aa6..febb2c88b2 100644
--- a/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_projection.py
+++ b/nemoguardrails/server/experimental/providers/openai/chat_completions/stream_projection.py
@@ -23,8 +23,6 @@
 
 from typing import Annotated, Any, ClassVar, Literal
 
-from pydantic import BeforeValidator
-
 from nemoguardrails.server.experimental.provider.payload import GuardedContentModel, GuardedProjectionModel
 from nemoguardrails.server.experimental.provider.projection_policy import (
     ObjectPolicy,
@@ -36,13 +34,6 @@
 )
 
 
-def _require_int(value: object) -> object:
-    """Reject booleans, strings, and floats before Literal[0] can coerce them."""
-    if type(value) is not int:
-        raise ValueError("index must be the integer 0")
-    return value
-
-
 class ChatCompletionsStreamDeltaProjection(PolicyModel, GuardedContentModel):
     """Accept assistant text or metadata-only deltas within a closed boundary."""
 
@@ -83,7 +74,7 @@ class ChatCompletionsStreamChoiceProjection(PolicyModel, GuardedContentModel):
 
     delta: Annotated[ChatCompletionsStreamDeltaProjection, guarded()]
     finish_reason: Annotated[Any, opaque()] = None
-    index: Annotated[Literal[0], BeforeValidator(_require_int), constrained()]
+    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
diff --git a/tests/server/experimental/test_openai_chat_streaming_projections.py b/tests/server/experimental/test_openai_chat_streaming_projections.py
index c38a328298..5dec6dffec 100644
--- a/tests/server/experimental/test_openai_chat_streaming_projections.py
+++ b/tests/server/experimental/test_openai_chat_streaming_projections.py
@@ -48,15 +48,7 @@
     STREAM_SELECTOR_FIELD,
 )
 from nemoguardrails.server.experimental.providers.openai.chat_completions.stream_classifier import (
-    CAPABILITY_PROFILE,
-    GUARDED_SHAPES,
-    OPAQUE_SHAPES,
-    PROVIDER_ERROR_SHAPES,
-    SNAPSHOT_SHAPES,
     STREAM_CLASSIFIER,
-    STREAM_CONTRACT,
-    STREAM_FIELDS,
-    STREAM_SOURCE_SCHEMA,
 )
 from nemoguardrails.server.experimental.providers.openai.chat_completions.stream_hooks import (
     ChatCompletionsStreamHooks,
@@ -107,13 +99,14 @@ def test_stream_classifier_covers_the_declared_event_inventory(native, shape, ro
     classified = STREAM_CLASSIFIER.classify_event(event(native))
 
     assert (classified.shape, classified.role, classified.text) == (shape, role, text)
-    assert shape in GUARDED_SHAPES | OPAQUE_SHAPES | PROVIDER_ERROR_SHAPES
+    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 == "[DONE]"
+    assert classified.shape == "keepalive"
     assert classified.role is StreamEventRole.OPAQUE_METADATA
 
 
@@ -147,15 +140,10 @@ def test_stream_classifier_accepts_reviewed_error_extensions_without_guarding_th
 
 
 def test_stream_contract_preserves_declared_identity_and_field_inventory():
-    assert CAPABILITY_PROFILE == "single_text_delta.v1"
-    assert STREAM_SOURCE_SCHEMA == "CreateChatCompletionStreamResponse"
-    assert STREAM_CONTRACT.projection_id == "openai.chat_completions.stream.text.v1"
-    assert STREAM_CONTRACT.fields is not None
-    assert STREAM_FIELDS == (
-        STREAM_CONTRACT.fields.guarded_fields,
-        STREAM_CONTRACT.fields.constrained_fields,
-        STREAM_CONTRACT.fields.opaque_fields,
-    )
+    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():
@@ -171,7 +159,7 @@ def test_stream_models_classify_every_declared_field_and_derive_root_coverage():
             for field in model.model_fields.values():
                 assert field_policy(field)["classification"] in {"guarded", "constrained", "opaque"}
     coverage = field_coverage(ChatCompletionsStreamPayloadProjection)
-    assert STREAM_CONTRACT.fields == coverage
+    assert STREAM_CLASSIFIER.contract.fields == coverage
     assert coverage.guarded_fields == frozenset({"choices"})
     assert coverage.constrained_fields == frozenset({"object"})
     assert coverage.opaque_fields == frozenset(
@@ -183,7 +171,9 @@ def test_stream_models_classify_every_declared_field_and_derive_root_coverage():
 
 
 def test_stream_schema_preserves_closure_defaults_and_text_policy():
-    schema = export_payload_schema(ChatCompletionsStreamPayloadProjection, projection_id=STREAM_CONTRACT.projection_id)
+    schema = export_payload_schema(
+        ChatCompletionsStreamPayloadProjection, projection_id=STREAM_CLASSIFIER.contract.projection_id
+    )
     choices = schema["properties"]["choices"]
     choice = choices["items"]
     delta = choice["properties"]["delta"]
@@ -191,7 +181,7 @@ def test_stream_schema_preserves_closure_defaults_and_text_policy():
     assert "minItems" not in choices
     assert choice["additionalProperties"] is delta["additionalProperties"] is False
     assert choice["required"] == ["delta", "index"]
-    assert choice["properties"]["index"]["const"] == 0
+    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"}]
@@ -242,12 +232,13 @@ def test_stream_declarations_preserve_transport_shapes_and_provenance():
     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 == "[DONE]"
+    assert STREAM_CLASSIFIER.non_data_shape == "keepalive"
     assert STREAM_CLASSIFIER.sentinels == ((b"[DONE]", "[DONE]"),)
-    assert GUARDED_SHAPES == frozenset({"chat.completion.chunk:content"})
-    assert OPAQUE_SHAPES == frozenset({"[DONE]", "chat.completion.chunk:metadata"})
-    assert PROVIDER_ERROR_SHAPES == frozenset({"error"})
-    assert SNAPSHOT_SHAPES == frozenset()
+    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()
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})
 
 

From 21cc830a8d18c313fee9dd2545b1c6a935431381 Mon Sep 17 00:00:00 2001
From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
Date: Tue, 13 Oct 2026 09:43:57 +0200
Subject: [PATCH 7/7] test(server): cover typed stream policy checks

Adds cases for annotated array items, containers that do not lead to a
model, untyped event models, text events without one assistant subject,
and non-canonical or repeated sentinels. No runtime behavior changes.

Signed-off-by: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com>
---
 .../server/experimental/test_stream_policy.py | 63 ++++++++++++++++++-
 1 file changed, 62 insertions(+), 1 deletion(-)

diff --git a/tests/server/experimental/test_stream_policy.py b/tests/server/experimental/test_stream_policy.py
index 94505f784c..0b603b6f28 100644
--- a/tests/server/experimental/test_stream_policy.py
+++ b/tests/server/experimental/test_stream_policy.py
@@ -19,7 +19,7 @@
 from typing import Annotated
 
 import pytest
-from pydantic import create_model
+from pydantic import BaseModel, create_model
 
 from nemoguardrails.server.experimental.provider import projection_policy
 from nemoguardrails.server.experimental.provider.projection_policy import (
@@ -265,3 +265,64 @@ def test_typed_classifier_preserves_nullable_missing_text_and_checks_actual_text
     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))