diff --git a/nemoguardrails/server/experimental/_application.py b/nemoguardrails/server/experimental/_application.py index ed7f8596b8..b1c3af7898 100644 --- a/nemoguardrails/server/experimental/_application.py +++ b/nemoguardrails/server/experimental/_application.py @@ -17,10 +17,12 @@ from collections.abc import Callable, Sequence from contextlib import asynccontextmanager +from typing import overload import httpx from fastapi import APIRouter, FastAPI +from nemoguardrails.server.experimental._configuration import ProxyConfig, resolve_config_selection, resolve_provider from nemoguardrails.server.experimental._http_kernel import ( DEFAULT_MAX_REQUEST_BODY_BYTES, DEFAULT_MAX_RESPONSE_BODY_BYTES, @@ -38,6 +40,7 @@ from nemoguardrails.server.experimental.rails.config import validate_config_entries HttpTransportFactory = Callable[[], httpx.AsyncBaseTransport] +_UNSET_CONFIG_RESOLUTION_TIMEOUT = object() @asynccontextmanager @@ -80,6 +83,17 @@ async def _lifespan(application: FastAPI): del application.state.proxy_runtime +@overload +def create_proxy_application( + config: ProxyConfig, + *, + config_resolver: ConfigResolver | None = None, + local_routers: Sequence[APIRouter] = (), + http_transport_factory: HttpTransportFactory | None = None, +) -> FastAPI: ... + + +@overload def create_proxy_application( *, providers: Sequence[ProviderIntegration], @@ -92,8 +106,77 @@ def create_proxy_application( max_request_body_bytes: int = DEFAULT_MAX_REQUEST_BODY_BYTES, max_response_body_bytes: int = DEFAULT_MAX_RESPONSE_BODY_BYTES, timeout: float = 30.0, +) -> FastAPI: ... + + +def create_proxy_application( + config: ProxyConfig | None = None, + *, + providers: Sequence[ProviderIntegration] | None = None, + rails_configs: Sequence[ConfigEntry] | None = None, + config_resolver: ConfigResolver | None = None, + config_resolution_timeout: object = _UNSET_CONFIG_RESOLUTION_TIMEOUT, + local_routers: Sequence[APIRouter] = (), + path_prefix: str | None = None, + http_transport_factory: HttpTransportFactory | None = None, + max_request_body_bytes: int | None = None, + max_response_body_bytes: int | None = None, + timeout: float | None = None, ) -> FastAPI: - """Create one private proxy application with lifespan-owned HTTP dispatch.""" + """Create one proxy from a declaration or explicit Python integrations. + + A ProxyConfig cannot be combined with direct assembly settings; callbacks + and local routers remain separate Python arguments. Declarative fixed Rails + IDs bypass the user resolver, while dynamic providers require it. Direct + calls retain their existing single-entry inference and explicit-resolver + behavior. + + Configuration and route validation run before lifespan-owned HTTP and Rails + resources are created. No upstream calls are made during construction. + """ + + if config is not None: + if not isinstance(config, ProxyConfig): + raise TypeError("config must be a ProxyConfig.") + if config_resolution_timeout is not _UNSET_CONFIG_RESOLUTION_TIMEOUT: + raise TypeError("ProxyConfig cannot be combined with direct assembly settings.") + if any( + value is not None + for value in ( + providers, + rails_configs, + path_prefix, + max_request_body_bytes, + max_response_body_bytes, + timeout, + ) + ): + raise TypeError("ProxyConfig cannot be combined with direct assembly settings.") + config = ProxyConfig.model_validate(config) + config_resolver = resolve_config_selection(config, config_resolver) + providers = tuple(resolve_provider(provider) for provider in config.providers) + rails_configs = config.rails_configs + config_resolution_timeout = config.config_resolution_timeout + path_prefix = config.path_prefix + max_request_body_bytes = config.max_request_body_bytes + max_response_body_bytes = config.max_response_body_bytes + timeout = config.timeout + if providers is None: + raise TypeError("Supply a ProxyConfig or explicit providers.") + rails_configs = () if rails_configs is None else rails_configs + config_resolution_timeout = ( + DEFAULT_CONFIG_RESOLUTION_TIMEOUT + if config_resolution_timeout is _UNSET_CONFIG_RESOLUTION_TIMEOUT + else config_resolution_timeout + ) + path_prefix = "" if path_prefix is None else path_prefix + max_request_body_bytes = ( + DEFAULT_MAX_REQUEST_BODY_BYTES if max_request_body_bytes is None else max_request_body_bytes + ) + max_response_body_bytes = ( + DEFAULT_MAX_RESPONSE_BODY_BYTES if max_response_body_bytes is None else max_response_body_bytes + ) + timeout = 30.0 if timeout is None else timeout if http_transport_factory is not None and not callable(http_transport_factory): raise TypeError("The HTTP transport factory must be callable.") diff --git a/nemoguardrails/server/experimental/_configuration.py b/nemoguardrails/server/experimental/_configuration.py new file mode 100644 index 0000000000..fb7825b367 --- /dev/null +++ b/nemoguardrails/server/experimental/_configuration.py @@ -0,0 +1,179 @@ +# 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. + +"""Validate deployment declarations and resolve them to existing proxy bindings. + +These models select reviewed Python policy; they neither load Rails directories +nor construct runtime owners. Provider URLs never select a dialect implicitly. +""" + +from collections.abc import Awaitable +from dataclasses import replace +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, StrictInt, StrictStr, field_validator, model_validator +from typing_extensions import Self + +from nemoguardrails.server.experimental._http_kernel import ( + DEFAULT_MAX_REQUEST_BODY_BYTES, + DEFAULT_MAX_RESPONSE_BODY_BYTES, +) +from nemoguardrails.server.experimental._runtime import DEFAULT_CONFIG_RESOLUTION_TIMEOUT +from nemoguardrails.server.experimental.provider.endpoint import GuardedJsonEndpoint +from nemoguardrails.server.experimental.provider.integration import ProviderIntegration +from nemoguardrails.server.experimental.rails import ConfigContext, ConfigEntry, ConfigResolver + + +class ProviderConfig(BaseModel): + """Select one built-in provider instance and its guarded operations. + + An omitted dialect selects the provider's reviewed native bindings. + Operations install guard bindings, not an HTTP route allowlist. Other + routes retain passthrough behavior. A fixed Rails ID bypasses the user + resolver; otherwise the application must supply a trusted resolver. + """ + + model_config = ConfigDict(extra="forbid", frozen=True, hide_input_in_errors=True, revalidate_instances="always") + + provider: Literal["openai"] + instance_id: StrictStr + base_url: StrictStr = Field(repr=False) + operations: tuple[Literal["chat_completions", "responses"], ...] = Field(min_length=1) + mount_path: StrictStr = "" + dialect: Literal["litellm"] | None = None + rails_config_id: StrictStr | None = None + + @field_validator("operations", mode="before") + @classmethod + def ordered_operations(cls, value: object) -> object: + """Require an ordered concrete selection and reject coercible containers.""" + if not isinstance(value, (list, tuple)): + raise ValueError("Operations must be a list or tuple of names.") + return value + + @field_validator("operations") + @classmethod + def unique_operations(cls, value: tuple[str, ...]) -> tuple[str, ...]: + """Reject duplicates instead of silently collapsing the declared policy.""" + if len(value) != len(set(value)): + raise ValueError("Operations must not contain duplicate names.") + return value + + @field_validator("rails_config_id") + @classmethod + def fixed_config_id(cls, value: str | None) -> str | None: + """Accept an exact catalogue ID or an explicit dynamic selection.""" + if value is not None and (not value or value != value.strip()): + raise ValueError("The fixed Rails config ID must be non-empty and trimmed.") + return value + + +class ProxyConfig(BaseModel): + """Declare built-in instances and one shared Rails configuration catalogue. + + Construction validates data without reading configuration files. Paths are + resolved by application assembly using the existing Python path semantics. + Callbacks are supplied to the application factory, not stored in this model. + """ + + model_config = ConfigDict(extra="forbid", frozen=True, hide_input_in_errors=True, revalidate_instances="always") + + providers: tuple[ProviderConfig, ...] = Field(min_length=1) + rails_configs: tuple[ConfigEntry, ...] = Field(min_length=1) + path_prefix: StrictStr = "" + max_request_body_bytes: StrictInt = Field(default=DEFAULT_MAX_REQUEST_BODY_BYTES, gt=0) + max_response_body_bytes: StrictInt = Field(default=DEFAULT_MAX_RESPONSE_BODY_BYTES, gt=0) + timeout: float = Field(default=30.0, strict=True, gt=0, allow_inf_nan=False) + config_resolution_timeout: float = Field( + default=DEFAULT_CONFIG_RESOLUTION_TIMEOUT, strict=True, gt=0, allow_inf_nan=False + ) + + @field_validator("providers", "rails_configs", mode="before") + @classmethod + def ordered_entries(cls, value: object) -> object: + """Snapshot lists or tuples rather than consuming arbitrary iterables.""" + if not isinstance(value, (list, tuple)): + raise ValueError("Configuration entries must be supplied as a list or tuple.") + return value + + @model_validator(mode="after") + def config_references(self) -> Self: + """Check catalogue identity and fixed references without filesystem I/O.""" + ids = [entry.id for entry in self.rails_configs] + if any(not value or value != value.strip() for value in ids): + raise ValueError("Rails config IDs must be non-empty and trimmed.") + if len(ids) != len(set(ids)): + raise ValueError("Rails config IDs must be unique.") + for provider in self.providers: + if provider.rails_config_id is not None and provider.rails_config_id not in ids: + raise ValueError("A provider references an unknown Rails config ID.") + return self + + +def resolve_provider(config: ProviderConfig) -> ProviderIntegration: + """Resolve the entire selection to reviewed bindings, without fallback.""" + from nemoguardrails.server.experimental.providers.openai.chat_completions.endpoint import CHAT_COMPLETIONS_ENDPOINT + from nemoguardrails.server.experimental.providers.openai.dialects.litellm.chat_completions import ( + LITELLM_CHAT_COMPLETIONS_ENDPOINT, + ) + from nemoguardrails.server.experimental.providers.openai.dialects.litellm.responses import ( + LITELLM_RESPONSES_ENDPOINT, + ) + from nemoguardrails.server.experimental.providers.openai.integration import create_openai_integration + from nemoguardrails.server.experimental.providers.openai.responses.endpoint import RESPONSES_ENDPOINT + + bindings: dict[tuple[str, str | None], dict[str, GuardedJsonEndpoint]] = { + ("openai", None): { + "chat_completions": CHAT_COMPLETIONS_ENDPOINT, + "responses": RESPONSES_ENDPOINT, + }, + ("openai", "litellm"): { + "chat_completions": LITELLM_CHAT_COMPLETIONS_ENDPOINT, + "responses": LITELLM_RESPONSES_ENDPOINT, + }, + } + try: + selected = bindings[(config.provider, config.dialect)] + endpoints = tuple(selected[name] for name in config.operations) + except KeyError: + raise ValueError("Unsupported provider, operation, and dialect combination.") from None + integration = create_openai_integration( + instance_id=config.instance_id, base_url=config.base_url, mount_path=config.mount_path + ) + return replace(integration, endpoints=endpoints) + + +def resolve_config_selection(config: ProxyConfig, resolver: ConfigResolver | None) -> ConfigResolver: + """Adapt explicit fixed/dynamic selection to the existing runtime resolver. + + Fixed providers never call user code. Dynamic providers receive the original + context and may select any registered ID; authorization belongs to that + trusted callback. The existing runtime validates its result before dispatch. + """ + if resolver is not None and not callable(resolver): + raise TypeError("The config resolver must be callable.") + fixed = {provider.instance_id: provider.rails_config_id for provider in config.providers} + if resolver is None and any(value is None for value in fixed.values()): + raise ValueError("Providers without a fixed Rails config ID require a resolver.") + + def select(context: ConfigContext) -> str | Awaitable[str]: + config_id = fixed[context.provider_instance_id] + if config_id is not None: + return config_id + if resolver is None: + raise ValueError("No resolver is configured for this provider.") + return resolver(context) + + return select diff --git a/nemoguardrails/server/experimental/_runtime.py b/nemoguardrails/server/experimental/_runtime.py index 85d4ef284e..4d34e95aa7 100644 --- a/nemoguardrails/server/experimental/_runtime.py +++ b/nemoguardrails/server/experimental/_runtime.py @@ -36,7 +36,7 @@ _HTTP_SHUTDOWN_TIMEOUT_SECONDS = 5.0 -def _validate_config_resolution_timeout(timeout: float) -> float: +def _validate_config_resolution_timeout(timeout: object) -> float: if isinstance(timeout, bool) or not isinstance(timeout, (int, float)) or not math.isfinite(timeout) or timeout <= 0: raise ValueError("The config resolution timeout must be a finite positive number of seconds.") return float(timeout) diff --git a/nemoguardrails/server/experimental/application.md b/nemoguardrails/server/experimental/application.md new file mode 100644 index 0000000000..15045ef3a0 --- /dev/null +++ b/nemoguardrails/server/experimental/application.md @@ -0,0 +1,196 @@ +# Python proxy application + +Declare provider instances and select their reviewed operation bindings through +one application factory. Typed Python projections, classifiers, and handwritten +hooks remain the source of policy. Application configuration selects those +bindings; it does not compile or load guard contracts. + +This Python interface requires the `server` extra. It does +not add YAML loading, a CLI launcher, or a stable configuration-file format. +The application owns outbound HTTP and preloads its Rails catalogue during +lifespan startup. Configuration directories are trusted: their actions may +execute Python. Application construction validates paths but does not start +engines or contact providers. + +## Declare a built-in provider + +```python +from pathlib import Path + +from nemoguardrails.server.experimental.application import ( + ConfigEntry, + ProviderConfig, + ProxyConfig, + create_proxy_application, +) + +config = ProxyConfig( + rails_configs=( + ConfigEntry(id="standard", config_path=Path("./rails/standard")), + ), + providers=( + ProviderConfig( + provider="openai", + instance_id="gateway", + base_url="https://gateway.example/v1", + mount_path="/gateway", + operations=("chat_completions", "responses"), + rails_config_id="standard", + ), + ), +) + +app = create_proxy_application(config) +``` + +The Rails directory must exist and contain a supported configuration. Relative +paths resolve against the working directory at application construction; use +absolute paths when embedding in another application. + +This forwards provider generation transparently through HTTPX. It does not use +the Guardrails LLM framework or LangChain to generate the provider response. +Rails configuration separately determines any models, optional integrations, +credentials, or services required by its checks. Upstream authorization is +forwarded from the caller under the existing transport rules; the declaration +does not store credentials or configure inbound authentication. Establish that +authentication separately when deploying beyond a trusted local environment. + +### Operations and dialects + +| Selection | Configuration | +| --- | --- | +| Native Chat only | `operations=("chat_completions",)`; omit `dialect`. | +| Native Chat and Responses | `operations=("chat_completions", "responses")`; omit `dialect`. | +| Reviewed LiteLLM Chat and Responses | Same operations, with `dialect="litellm"`. | + +Responses-only selection is also supported. Both operations have buffered and +streaming bindings. Order is preserved; empty, duplicate, or unknown operations +are rejected. There is no default that automatically adds future operations. + +Omitted dialect means the chosen provider's native reviewed bindings, not a +global OpenAI default. Initially the built-in implementation is `openai`, with +native and explicit LiteLLM bindings. Provider, instance, upstream, and dialect +are distinct: changing the URL does not select a dialect. Unsupported names +fail rather than falling back. The LiteLLM selection describes reviewed wire +extensions, not compatibility with every LiteLLM backend. + +Native bindings still enforce the supported single-text policy; they do not +accept every feature in the upstream API. See the +[LiteLLM Chat boundary](providers/openai/dialects/litellm/README.md) and +[LiteLLM Responses boundary](providers/openai/dialects/litellm/responses.md). + +**Operation selection is not a route allowlist.** Unselected routes retain +passthrough behavior. Selected bindings also do not activate Rails checks +disabled in the chosen Rails configuration. Inspect input/output configuration +when deciding which traffic is actually checked. + +## Fixed and dynamic Rails configuration + +Define the catalogue once, even when it contains dozens of entries. +`rails_config_id` selects one fixed entry per provider instance. No repeated +lists of allowed IDs are required: + +```python +config = ProxyConfig( + rails_configs=( + ConfigEntry(id="standard", config_path=Path("./rails/standard")), + ConfigEntry(id="strict", config_path=Path("./rails/strict")), + ), + providers=( + ProviderConfig( + provider="openai", + instance_id="public", + mount_path="/public", + base_url="https://gateway.example/v1", + operations=("chat_completions", "responses"), + rails_config_id="strict", + ), + ProviderConfig( + provider="openai", + instance_id="customers", + mount_path="/customers", + base_url="https://customers.example/v1", + operations=("chat_completions",), + ), + ), +) + +def resolve_config(context): + tenant = context.request.state.authenticated_tenant + return tenant.guardrails_config_id + +app = create_proxy_application(config, config_resolver=resolve_config) +``` + +The example requires trusted middleware to establish `authenticated_tenant`. +It must not be copied directly from an untrusted request header. + +- `public` always uses `strict` and never invokes the user callback. +- `customers` has no fixed ID, so the callback selects a registered entry. +- The callback receives the original request, provider type, provider instance + ID, and canonical operation name, such as `chat_completions.create`. +- Synchronous and asynchronous callbacks are supported. A dynamic instance + requires a callback even with a one-entry catalogue. +- All-fixed applications need no callback, even with many catalogue entries. +- Unknown fixed IDs fail construction. Unknown/invalid callback results and + callback failures fail closed before upstream dispatch, without fallback. +- Selection chooses one configuration; it does not run multiple configurations + in sequence. Dynamic callbacks own authorization; registration in the shared + catalogue alone does not authorize a caller to select that configuration. + +There are no per-provider allowlist fields or declarative per-operation +overrides in this interface. A trusted dynamic resolver can use operation +context as needed. All catalogue entries are still preloaded under the existing +repository behavior, including entries not selected by a particular request. + +## Custom Python composition + +The same factory still accepts explicit integrations, ordinary resolver +callbacks, and an application-owned HTTP transport factory: + +```python +from nemoguardrails.server.experimental.providers.openai.integration import ( + create_openai_chat_integration, +) + +app = create_proxy_application( + providers=( + create_openai_chat_integration( + instance_id="gateway", + base_url="https://gateway.example/v1", + ), + ), + rails_configs=(ConfigEntry(id="standard", config_path=Path("./rails/standard")),), +) +``` + +Direct `ProviderIntegration` objects allow custom endpoints, mixed dialects, +or custom implementations without registration. The existing direct mode still +infers static selection with one catalogue entry and no resolver, requires a +resolver with multiple entries, and invokes an explicit resolver even with one +entry. Those semantics are unchanged. + +Do not mix `ProxyConfig` with direct providers, Rails entries, or limit/prefix +overrides. Put data settings in `ProxyConfig`; `config_resolver`, +`http_transport_factory`, and `local_routers` remain separate Python arguments in +either mode. Local routers register before provider catch-alls; choose local +paths outside guarded operation paths. YAML does not load Python routers. +See [HTTP transport](http-transport.md) for local routes and transport ownership. + +## Validation, defaults, and limitations + +Unknown fields are rejected, including inside catalogue entries. Configuration +models snapshot input collections and are frozen. Provider IDs and mounts must +be unique, and existing route validation rejects conflicting bindings. The +upstream URL must be HTTP(S), without userinfo, query, or fragment. Declarative +upstreams are explicit; provider-specific environment defaults are not used. + +The default application prefix is empty, the request limit is 1 MiB, the +response limit is 10 MiB, and the outbound timeout is 30 seconds. The limits +retain existing guarded-versus-passthrough behavior; they are not a newly added +global response-buffering rule. + +The new selection path reuses existing lifecycle owners. It does not fix or +claim complete safety for interrupted startup/shutdown or failed HTTP cleanup. +Those owners require their own real-resource validation. Successful in-process +buffered/SSE tests are not proof of socket-disconnect or shutdown safety. diff --git a/nemoguardrails/server/experimental/application.py b/nemoguardrails/server/experimental/application.py new file mode 100644 index 0000000000..1223b14909 --- /dev/null +++ b/nemoguardrails/server/experimental/application.py @@ -0,0 +1,28 @@ +# 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. + +"""Python configuration and assembly for the transparent proxy. + +Use ProxyConfig for reviewed built-ins, or pass explicit ProviderIntegration +objects to the same factory for custom Python composition. Both use the same +application lifespan, HTTP owner, routing, and Rails repository. This surface +does not load YAML, export contracts, or construct policy from generated files. +""" + +from nemoguardrails.server.experimental._application import create_proxy_application +from nemoguardrails.server.experimental._configuration import ProviderConfig, ProxyConfig +from nemoguardrails.server.experimental.rails import ConfigEntry + +__all__ = ["ConfigEntry", "ProviderConfig", "ProxyConfig", "create_proxy_application"] diff --git a/nemoguardrails/server/experimental/configuration-resolution.md b/nemoguardrails/server/experimental/configuration-resolution.md index eac8a409eb..b92ea8fa3b 100644 --- a/nemoguardrails/server/experimental/configuration-resolution.md +++ b/nemoguardrails/server/experimental/configuration-resolution.md @@ -141,6 +141,10 @@ fails closed through the provider-native `config_resolution_failed` response, without forwarding the request or selecting a fallback policy. Request cancellation propagates instead of becoming a configuration error. +For declarative Python assembly, set `ProxyConfig(config_resolution_timeout=..., ...)`. +As with other application settings, this value belongs in `ProxyConfig` rather +than a separate factory argument when a declaration is supplied. + The deadline cannot preempt blocking synchronous work or force a resolver that suppresses cancellation to stop. Preloading removes request-time engine initialization; it does not make external policy selection free of latency. diff --git a/nemoguardrails/server/experimental/rails/config.py b/nemoguardrails/server/experimental/rails/config.py index 0f79d0aedd..4666c5bf0a 100644 --- a/nemoguardrails/server/experimental/rails/config.py +++ b/nemoguardrails/server/experimental/rails/config.py @@ -18,15 +18,18 @@ from collections.abc import Awaitable, Callable, Sequence from dataclasses import dataclass from pathlib import Path -from typing import TypeAlias +from typing import ClassVar, TypeAlias from fastapi import Request +from pydantic import ConfigDict @dataclass(frozen=True, slots=True) class ConfigEntry: """Associate a deployment-owned identifier with a Rails config directory.""" + __pydantic_config__: ClassVar[ConfigDict] = ConfigDict(extra="forbid", revalidate_instances="always") + id: str config_path: Path diff --git a/tests/server/experimental/test_application_configuration.py b/tests/server/experimental/test_application_configuration.py new file mode 100644 index 0000000000..f5f0207ce8 --- /dev/null +++ b/tests/server/experimental/test_application_configuration.py @@ -0,0 +1,619 @@ +# 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 declarative selection through the existing proxy assembly.""" + +import asyncio +import json +from dataclasses import replace +from pathlib import Path + +import httpx +import pytest +from fastapi import APIRouter, Request +from fastapi.testclient import TestClient +from pydantic import ValidationError + +from nemoguardrails.server.experimental import _application, _configuration, application +from nemoguardrails.server.experimental.application import ( + ConfigEntry, + ProviderConfig, + ProxyConfig, + create_proxy_application, +) +from nemoguardrails.server.experimental.providers.openai.chat_completions.endpoint import CHAT_COMPLETIONS_ENDPOINT +from nemoguardrails.server.experimental.providers.openai.dialects.litellm.chat_completions import ( + LITELLM_CHAT_COMPLETIONS_ENDPOINT, +) +from nemoguardrails.server.experimental.providers.openai.dialects.litellm.responses import LITELLM_RESPONSES_ENDPOINT +from nemoguardrails.server.experimental.providers.openai.integration import create_openai_integration +from nemoguardrails.server.experimental.providers.openai.responses.endpoint import RESPONSES_ENDPOINT +from nemoguardrails.server.experimental.rails import ConfigContext +from nemoguardrails.server.experimental.rails._repository import GuardrailsRepository +from tests.server.experimental.test_openai_responses_integration import ResponseBody +from tests.server.experimental.test_openai_responses_projections import response_payload +from tests.server.experimental.test_openai_responses_streaming_projections import event_bytes + + +def provider(**overrides): + return ProviderConfig( + **{ + "provider": "openai", + "instance_id": "gateway", + "base_url": "https://provider.test/v1", + "mount_path": "/gateway", + "operations": ("chat_completions", "responses"), + "rails_config_id": "standard", + **overrides, + } + ) + + +@pytest.fixture +def entries(tmp_path): + result = [] + for name, pattern in (("standard", "BLOCK_INPUT"), ("strict", "BLOCK_INPUT|STRICT_ONLY")): + directory = tmp_path / name + directory.mkdir() + (directory / "config.yml").write_text( + f"""models: [] +rails: + config: + regex_detection: + input: + patterns: ["{pattern}"] + output: + patterns: [BLOCK_OUTPUT] + input: + flows: [regex check input] + output: + flows: [regex check output] + streaming: + enabled: true + chunk_size: 8 + context_size: 0 + stream_first: false +""", + encoding="utf-8", + ) + result.append(ConfigEntry(name, directory)) + return tuple(result) + + +def config(entries, *providers, **overrides): + return ProxyConfig(providers=providers or (provider(),), rails_configs=entries, **overrides) + + +def request_payload(operation, text="question", streaming=False): + subject = {"messages": [{"role": "user", "content": text}]} if operation == "chat_completions" else {"input": text} + return {"model": "test-model", "stream": streaming, **subject} + + +def route(operation, mount="/gateway"): + return mount + "/v1/" + ("chat/completions" if operation == "chat_completions" else "responses") + + +class Upstream: + """Provide synthetic protocol bytes through a real HTTPX client owner.""" + + def __init__(self, text="answer"): + self.text = text + self.requests = [] + self.bodies = [] + self.factory_calls = 0 + + def factory(self): + self.factory_calls += 1 + return httpx.MockTransport(self.handle) + + async def handle(self, request): + self.requests.append(request) + payload = json.loads(await request.aread()) + if request.url.path.endswith("/chat/completions"): + if payload.get("stream"): + wire = ( + b"data: " + + json.dumps( + { + "object": "chat.completion.chunk", + "choices": [{"index": 0, "delta": {"content": self.text}}], + } + ).encode() + + b"\n\ndata: [DONE]\n\n" + ) + else: + wire = json.dumps({"choices": [{"message": {"role": "assistant", "content": self.text}}]}).encode() + elif payload.get("stream"): + wire = b"".join( + ( + event_bytes("response.created", response={"output": []}, sequence_number=0), + event_bytes( + "response.output_text.delta", + delta=self.text, + item_id="msg_test", + output_index=1, + content_index=0, + sequence_number=1, + ), + event_bytes("response.completed", response=response_payload(self.text), sequence_number=2), + ) + ) + else: + wire = json.dumps(response_payload(self.text)).encode() + body = ResponseBody(wire) + self.bodies.append(body) + return httpx.Response( + 200, + headers={"content-type": "text/event-stream" if payload.get("stream") else "application/json"}, + stream=body, + ) + + +@pytest.mark.parametrize("dialect", [None, "litellm"]) +@pytest.mark.parametrize( + "operations", + [("chat_completions",), ("responses",), ("chat_completions", "responses"), ("responses", "chat_completions")], +) +def test_exact_binding_selection_and_direct_assembly_equivalence(entries, dialect, operations): + declaration = provider(dialect=dialect, operations=operations) + actual = _configuration.resolve_provider(declaration) + native = {"chat_completions": CHAT_COMPLETIONS_ENDPOINT, "responses": RESPONSES_ENDPOINT} + alternative = {"chat_completions": LITELLM_CHAT_COMPLETIONS_ENDPOINT, "responses": LITELLM_RESPONSES_ENDPOINT} + expected_endpoints = tuple((native if dialect is None else alternative)[operation] for operation in operations) + expected = replace( + create_openai_integration(instance_id="gateway", base_url="https://provider.test/v1", mount_path="/gateway"), + endpoints=expected_endpoints, + ) + assert actual == expected + assert all(left is right for left, right in zip(actual.endpoints, expected_endpoints)) + declared_app = create_proxy_application(config(entries, declaration)) + direct_app = create_proxy_application(providers=(expected,), rails_configs=(entries[0],)) + assert declared_app.openapi() == direct_app.openapi() + + +@pytest.mark.parametrize( + "field,value", + [ + ("provider", "anthropic"), + ("provider", "unknown"), + ("dialect", "openai"), + ("dialect", "unknown"), + ("operations", ()), + ("operations", ("responses", "responses")), + ("operations", ("unknown",)), + ("operations", "responses"), + ("operations", {"responses": True}), + ("operations", {"responses"}), + ("operations", iter(["responses"])), + ("rails_config_id", ""), + ("rails_config_id", " standard "), + ("rails_config_ids", ["standard"]), + ("extra", True), + ], +) +def test_invalid_provider_declarations_are_rejected(field, value): + with pytest.raises(ValidationError): + provider(**{field: value}) + + +def test_native_binding_is_explicitly_provider_relative(): + declaration = provider().model_copy(update={"provider": "anthropic"}) + with pytest.raises(ValueError, match="Unsupported"): + _configuration.resolve_provider(declaration) + + +@pytest.mark.parametrize( + "override", + [ + {"providers": ()}, + {"rails_configs": ()}, + {"providers": "openai"}, + {"rails_configs": {}}, + {"max_request_body_bytes": True}, + {"max_request_body_bytes": 0}, + {"max_response_body_bytes": "1024"}, + {"timeout": float("inf")}, + {"timeout": True}, + {"timeout": 0}, + *({"config_resolution_timeout": value} for value in (0, -1, float("inf"), float("nan"), True, None, "5")), + {"extra": "unknown"}, + ], +) +def test_invalid_application_data_is_rejected(entries, override): + with pytest.raises(ValidationError): + ProxyConfig(**{"providers": (provider(),), "rails_configs": entries, **override}) + + +def test_catalogue_references_and_nested_unknown_keys_are_validated(entries): + with pytest.raises(ValidationError, match="unknown Rails"): + config(entries, provider(rails_config_id="missing")) + with pytest.raises(ValidationError, match="unique"): + config((entries[0], entries[0])) + data = config(entries).model_dump(mode="json") + data["rails_configs"][0]["unexpected"] = True + with pytest.raises(ValidationError): + ProxyConfig.model_validate(data) + + +def test_schema_roundtrip_and_immutable_selection_without_file_loading(monkeypatch): + def forbidden(*args, **kwargs): + pytest.fail("Data validation must not load Rails or create runtime resources") + + monkeypatch.setattr(_application.ServerRuntime, "create", forbidden) + monkeypatch.setattr(GuardrailsRepository, "startup", forbidden) + operations = ["responses"] + declarations = [provider(operations=operations)] + catalogue = [ConfigEntry("standard", Path("/nonexistent/rails"))] + value = ProxyConfig(providers=declarations, rails_configs=catalogue) + operations.append("chat_completions") + declarations.clear() + catalogue.clear() + assert value.providers[0].operations == ("responses",) + assert len(value.rails_configs) == 1 + with pytest.raises(ValidationError): + value.providers[0].rails_config_id = "other" + assert ProxyConfig.model_validate_json(value.model_dump_json()) == value + assert ProxyConfig.model_json_schema()["$defs"]["ConfigEntry"]["additionalProperties"] is False + assert application.__all__ == ["ConfigEntry", "ProviderConfig", "ProxyConfig", "create_proxy_application"] + assert application.create_proxy_application is _application.create_proxy_application + + +@pytest.mark.parametrize( + "setting,value", + [ + ("providers", ()), + ("rails_configs", ()), + ("path_prefix", ""), + ("max_request_body_bytes", 1024), + ("max_response_body_bytes", 1024), + ("timeout", 30.0), + ("config_resolution_timeout", 5.0), + ], +) +def test_config_rejects_direct_settings_even_when_they_equal_defaults(entries, setting, value): + with pytest.raises(TypeError, match="direct assembly"): + create_proxy_application(config(entries), **{setting: value}) + + +@pytest.mark.parametrize( + "changes", + [ + {"base_url": "relative"}, + {"base_url": "https://user:secret@provider.test"}, + {"base_url": "https://provider.test?secret=value"}, + {"mount_path": "/trailing/"}, + {"instance_id": "invalid id"}, + ], +) +def test_invalid_routes_and_upstreams_fail_before_resource_creation(entries, monkeypatch, changes): + def forbidden(*args, **kwargs): + pytest.fail("Invalid declarations must fail before runtime ownership begins") + + monkeypatch.setattr(_application.ServerRuntime, "create", forbidden) + monkeypatch.setattr(GuardrailsRepository, "startup", forbidden) + with pytest.raises(ValueError): + create_proxy_application(config(entries, provider(**changes)), http_transport_factory=forbidden) + + +@pytest.mark.parametrize("duplicate", ["instance", "mount"]) +def test_provider_collisions_reuse_assembly_validation(entries, duplicate): + other = provider( + instance_id="gateway" if duplicate == "instance" else "other", + mount_path="/gateway" if duplicate == "mount" else "/other", + ) + with pytest.raises(ValueError, match="unique"): + create_proxy_application(config(entries, provider(), other)) + + +def test_dynamic_selection_requires_a_callable_even_with_one_entry(entries): + declaration = config((entries[0],), provider(rails_config_id=None)) + with pytest.raises(ValueError, match="require a resolver"): + create_proxy_application(declaration) + with pytest.raises(TypeError, match="callable"): + create_proxy_application(declaration, config_resolver="not-a-callback") + + +def test_factory_revalidates_copied_models_and_requires_typed_configuration(entries): + original = config(entries) + with pytest.raises(TypeError, match="ProxyConfig"): + create_proxy_application(original.model_dump()) + with pytest.raises(TypeError, match="Supply"): + create_proxy_application() + changed = original.model_copy(update={"providers": (provider().model_copy(update={"dialect": "unknown"}),)}) + with pytest.raises(ValidationError): + create_proxy_application(changed) + + +@pytest.mark.parametrize("declarative", [False, True]) +def test_local_routers_work_with_both_application_configuration_modes(entries, declarative): + local = APIRouter() + + @local.get("/health") + async def health(): + return {"status": "ok"} + + upstream = Upstream() + declaration = config(entries, provider(mount_path="")) + options = {"local_routers": (local,), "http_transport_factory": upstream.factory} + app = ( + create_proxy_application(declaration, **options) + if declarative + else create_proxy_application( + providers=tuple(_configuration.resolve_provider(item) for item in declaration.providers), + rails_configs=declaration.rails_configs, + config_resolver=lambda context: "standard", + **options, + ) + ) + with TestClient(app) as client: + assert client.get("/health").json() == {"status": "ok"} + assert client.post("/health").status_code == 405 + assert not upstream.requests + assert ( + client.post(route("chat_completions", mount=""), json=request_payload("chat_completions")).status_code + == 200 + ) + assert len(upstream.requests) == 1 + + +def test_application_settings_and_explicit_upstream_are_preserved(entries, monkeypatch): + monkeypatch.setenv("NEMO_GUARDRAILS_OPENAI_PROXY_UPSTREAM_BASE_URL", "https://unselected.test") + original = _application.include_provider_routes + observed = {} + + def capture(*args, **kwargs): + observed.update(kwargs) + return original(*args, **kwargs) + + monkeypatch.setattr(_application, "include_provider_routes", capture) + upstream = Upstream() + app = create_proxy_application( + config( + entries, + path_prefix="/proxy", + max_request_body_bytes=512, + max_response_body_bytes=2048, + timeout=7.0, + config_resolution_timeout=2.0, + ), + http_transport_factory=upstream.factory, + ) + with TestClient(app) as client: + assert _application.get_runtime(app).config_resolution_timeout == 2.0 + assert client.post("/proxy" + route("responses"), json=request_payload("responses")).status_code == 200 + assert ( + client.post("/proxy" + route("responses"), json=request_payload("responses", "a" * 600)).status_code == 413 + ) + assert observed["max_request_body_bytes"] == 512 + assert observed["max_response_body_bytes"] == 2048 + assert observed["timeout"] == 7.0 + assert observed["path_prefix"] == "/proxy" + assert len(upstream.requests) == 1 + assert str(upstream.requests[0].url) == "https://provider.test/v1/responses" + + +def test_large_catalogue_needs_no_provider_level_enumeration(entries): + catalogue = tuple(ConfigEntry(f"tenant-{index}", entries[0].config_path) for index in range(48)) + fixed = provider(instance_id="fixed", mount_path="/fixed", rails_config_id="tenant-47") + dynamic = provider(instance_id="dynamic", mount_path="/dynamic", rails_config_id=None) + app = create_proxy_application(config(catalogue, fixed, dynamic), config_resolver=lambda context: "tenant-46") + choose = app.state.proxy_config_resolver + request = Request({"type": "http"}) + assert choose(ConfigContext(request, "openai", "fixed", "responses.create")) == "tenant-47" + assert choose(ConfigContext(request, "openai", "dynamic", "responses.create")) == "tenant-46" + assert len(app.state.proxy_rails_configs) == 48 + + +@pytest.mark.parametrize("dialect", [None, "litellm"]) +@pytest.mark.parametrize("operation", ["chat_completions", "responses"]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("decision", ["allow", "input", "output"]) +def test_declarations_use_real_rails_for_buffered_and_streaming_checks( + entries, dialect, operation, streaming, decision +): + upstream = Upstream("BLOCK_OUTPUT" if decision == "output" else "answer") + declaration = config(entries, provider(dialect=dialect, operations=(operation,))) + app = create_proxy_application(declaration, http_transport_factory=upstream.factory) + assert upstream.factory_calls == 0 + with TestClient(app) as client: + response = client.post( + route(operation), + json=request_payload(operation, "BLOCK_INPUT" if decision == "input" else "question", streaming), + ) + assert upstream.factory_calls == 1 + assert all(body.closed for body in upstream.bodies) + if decision == "allow": + assert response.status_code == 200 + assert response.content == upstream.bodies[0].body + elif decision == "input": + assert response.status_code == 400 + assert not upstream.requests + assert response.json()["error"]["code"] == "content_blocked" + else: + assert response.status_code == (200 if streaming else 400) + assert b"content_blocked" in response.content + assert b"BLOCK_OUTPUT" not in response.content + + +@pytest.mark.parametrize("async_resolver", [False, True]) +def test_mixed_fixed_dynamic_instances_select_from_shared_catalogue(entries, async_resolver): + contexts = [] + + def choose(context): + contexts.append(context) + assert context.request.state.tenant == "trusted" + return "standard" + + async def choose_async(context): + return choose(context) + + upstream = Upstream() + declaration = config( + entries, + provider(instance_id="public", mount_path="/public", rails_config_id="strict"), + provider( + instance_id="internal", mount_path="/internal", base_url="https://internal.test/v1", rails_config_id=None + ), + ) + app = create_proxy_application( + declaration, config_resolver=choose_async if async_resolver else choose, http_transport_factory=upstream.factory + ) + + @app.middleware("http") + async def authenticate(request, call_next): + request.state.tenant = "trusted" + return await call_next(request) + + with TestClient(app) as client: + for operation in ("chat_completions", "responses"): + assert ( + client.post(route(operation, "/public"), json=request_payload(operation, "STRICT_ONLY")).status_code + == 400 + ) + assert ( + client.post(route(operation, "/internal"), json=request_payload(operation, "STRICT_ONLY")).status_code + == 200 + ) + assert [item.provider_instance_id for item in contexts] == ["internal", "internal"] + assert [item.operation_name for item in contexts] == ["chat_completions.create", "responses.create"] + assert all(item.provider_type == "openai" for item in contexts) + assert all(request.url.host == "internal.test" for request in upstream.requests) + assert upstream.factory_calls == 1 + + +def test_all_fixed_providers_need_no_resolver_for_multiple_entries(entries): + upstream = Upstream() + app = create_proxy_application( + config( + entries, + provider(instance_id="public", mount_path="/public", rails_config_id="strict"), + provider(instance_id="internal", mount_path="/internal", rails_config_id="standard"), + ), + http_transport_factory=upstream.factory, + ) + with TestClient(app) as client: + assert ( + client.post(route("responses", "/public"), json=request_payload("responses", "STRICT_ONLY")).status_code + == 400 + ) + assert ( + client.post(route("responses", "/internal"), json=request_payload("responses", "STRICT_ONLY")).status_code + == 200 + ) + + +def test_fixed_provider_never_invokes_explicit_user_resolver(entries): + def forbidden(context): + pytest.fail("A fixed provider must bypass the user resolver") + + upstream = Upstream() + app = create_proxy_application(config(entries), config_resolver=forbidden, http_transport_factory=upstream.factory) + with TestClient(app) as client: + assert client.post(route("responses"), json=request_payload("responses")).status_code == 200 + + +@pytest.mark.parametrize("result", ["missing", "", None, 1, RuntimeError("private resolver failure")]) +@pytest.mark.parametrize("async_resolver", [False, True]) +def test_dynamic_selection_failure_is_closed_without_fallback(entries, result, async_resolver): + def choose(context): + if isinstance(result, Exception): + raise result + return result + + async def choose_async(context): + return choose(context) + + upstream = Upstream() + app = create_proxy_application( + config(entries, provider(rails_config_id=None)), + config_resolver=choose_async if async_resolver else choose, + http_transport_factory=upstream.factory, + ) + with TestClient(app) as client: + response = client.post(route("responses"), json=request_payload("responses")) + assert response.status_code == 502 + assert response.json()["error"]["code"] == "config_resolution_failed" + assert "private resolver failure" not in response.text + assert not upstream.requests + + +@pytest.mark.parametrize("selected,unselected", [("responses", "chat_completions"), ("chat_completions", "responses")]) +@pytest.mark.parametrize("streaming", [False, True]) +def test_unselected_routes_remain_passthrough_without_rails_resolution(entries, selected, unselected, streaming): + def forbidden(context): + pytest.fail("Passthrough must not resolve Rails") + + upstream = Upstream("BLOCK_OUTPUT") + app = create_proxy_application( + config(entries, provider(operations=(selected,), rails_config_id=None)), + config_resolver=forbidden, + http_transport_factory=upstream.factory, + ) + with TestClient(app) as client: + response = client.post(route(unselected), json=request_payload(unselected, "BLOCK_INPUT", streaming)) + assert response.status_code == 200 + assert response.content == upstream.bodies[0].body + assert upstream.bodies[0].closed + + +@pytest.mark.parametrize("operation", ["chat_completions", "responses"]) +@pytest.mark.parametrize("streaming", [False, True]) +def test_declarative_resolver_deadline_stops_before_dispatch(entries, operation, streaming): + cancelled = [] + + async def resolve(_context): + try: + await asyncio.Event().wait() + finally: + cancelled.append(True) + + upstream = Upstream() + app = create_proxy_application( + config(entries, provider(rails_config_id=None), config_resolution_timeout=0.1), + config_resolver=resolve, + http_transport_factory=upstream.factory, + ) + with TestClient(app) as client: + response = client.post(route(operation), json=request_payload(operation, streaming=streaming)) + assert response.status_code == 502 + assert response.json()["error"]["code"] == "config_resolution_failed" + assert cancelled == [True] + assert not upstream.requests + + +def test_direct_mode_retains_explicit_single_entry_resolver_and_custom_binding(entries): + upstream = Upstream() + contexts = [] + custom = replace( + create_openai_integration(base_url="https://provider.test/v1"), + endpoints=(LITELLM_RESPONSES_ENDPOINT,), + ) + + def choose(context): + contexts.append(context) + return entries[0].id + + with pytest.raises(ValueError, match="Multiple Rails"): + create_proxy_application(providers=(custom,), rails_configs=entries) + app = create_proxy_application( + providers=(custom,), + rails_configs=(entries[0],), + config_resolver=choose, + http_transport_factory=upstream.factory, + ) + with TestClient(app) as client: + assert client.post("/v1/responses", json=request_payload("responses")).status_code == 200 + assert len(contexts) == 1 + assert isinstance(contexts[0], ConfigContext)