diff --git a/nemoguardrails/server/experimental/__main__.py b/nemoguardrails/server/experimental/__main__.py
new file mode 100644
index 0000000000..055c9e1aa2
--- /dev/null
+++ b/nemoguardrails/server/experimental/__main__.py
@@ -0,0 +1,21 @@
+# 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.
+
+"""Launch the server-owned CLI at its temporary module location."""
+
+from nemoguardrails.server.experimental.cli import main
+
+if __name__ == "__main__":
+    main()
diff --git a/nemoguardrails/server/experimental/_application_manifest.py b/nemoguardrails/server/experimental/_application_manifest.py
new file mode 100644
index 0000000000..e33ab0950e
--- /dev/null
+++ b/nemoguardrails/server/experimental/_application_manifest.py
@@ -0,0 +1,177 @@
+# 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.
+
+"""Load bounded deployment YAML into the existing application configuration.
+
+The manifest is data, not a Python import or policy-authoring mechanism. Rails
+directories remain trusted executable configuration, loaded only by the serving
+application's lifespan. Validation errors deliberately omit input values.
+"""
+
+from pathlib import Path
+from typing import Literal
+
+import yaml
+from pydantic import Field, StrictStr, ValidationError, field_validator
+from yaml.events import AliasEvent
+from yaml.nodes import MappingNode, Node, ScalarNode, SequenceNode
+
+from nemoguardrails.server.experimental._configuration import ProviderConfig, ProxyConfig
+from nemoguardrails.server.experimental.rails import ConfigEntry
+
+MAX_MANIFEST_BYTES = 1024 * 1024
+MAX_MANIFEST_DEPTH = 32
+MAX_MANIFEST_NODES = 10_000
+
+
+class ManifestError(ValueError):
+    """Report a safe diagnostic without source excerpts or input values."""
+
+
+class ManifestProviderConfig(ProviderConfig):
+    """Restrict the Python provider declaration to a required fixed Rails ID."""
+
+    rails_config_id: StrictStr = Field(...)
+
+
+class ApplicationManifest(ProxyConfig):
+    """Add an explicit file-format revision to the Python declaration.
+
+    Revision 1 does not promise API stability. The standalone loader requires
+    fixed Rails selections because it cannot supply trusted Python callbacks.
+    """
+
+    version: Literal[1]
+    providers: tuple[ManifestProviderConfig, ...] = Field(min_length=1)
+
+    @field_validator("version", mode="before")
+    @classmethod
+    def integer_version(cls, value: object) -> object:
+        """Reject boolean and floating-point spellings of the version."""
+        if type(value) is not int:
+            raise ValueError("The manifest version must be an integer.")
+        return value
+
+
+def _location(node: Node) -> str:
+    """Return source coordinates without exposing a source excerpt."""
+    return f"line {node.start_mark.line + 1}, column {node.start_mark.column + 1}"
+
+
+class _ManifestLoader(yaml.SafeLoader):
+    """Bound the YAML tree before construction and prohibit indirection."""
+
+    def __init__(self, source: str):
+        super().__init__(source)
+        self._depth = 0
+        self._nodes = 0
+
+    def compose_node(self, parent, index):
+        """Reject aliases, explicit tags, anchors and oversized nesting."""
+        event = self.peek_event()
+        location = f"line {event.start_mark.line + 1}, column {event.start_mark.column + 1}"
+        if isinstance(event, AliasEvent) or getattr(event, "anchor", None) is not None:
+            raise ManifestError(f"{location}: YAML anchors and aliases are not supported.")
+        if getattr(event, "tag", None) is not None:
+            raise ManifestError(f"{location}: explicit YAML tags are not supported.")
+        self._nodes += 1
+        self._depth += 1
+        try:
+            if self._depth > MAX_MANIFEST_DEPTH or self._nodes > MAX_MANIFEST_NODES:
+                raise ManifestError(f"{location}: manifest structure exceeds the supported limits.")
+            return super().compose_node(parent, index)
+        finally:
+            self._depth -= 1
+
+
+def _construct(loader: _ManifestLoader, node: Node, locations: dict, path: tuple = ()) -> object:
+    """Construct only ordinary scalar/list/object data and record coordinates."""
+    locations[path] = _location(node)
+    if isinstance(node, MappingNode):
+        result = {}
+        for key, value in node.value:
+            if not isinstance(key, ScalarNode) or key.tag != "tag:yaml.org,2002:str":
+                raise ManifestError(f"{_location(key)}: mapping keys must be strings; merges are not supported.")
+            if key.value in result:
+                raise ManifestError(f"{_location(key)}: duplicate mapping key.")
+            result[key.value] = _construct(loader, value, locations, (*path, key.value))
+        return result
+    if isinstance(node, SequenceNode):
+        return [_construct(loader, value, locations, (*path, index)) for index, value in enumerate(node.value)]
+    if not isinstance(node, ScalarNode) or node.tag not in {
+        "tag:yaml.org,2002:str",
+        "tag:yaml.org,2002:int",
+        "tag:yaml.org,2002:float",
+        "tag:yaml.org,2002:bool",
+        "tag:yaml.org,2002:null",
+    }:
+        raise ManifestError(f"{_location(node)}: unsupported YAML value.")
+    value = loader.construct_object(node)
+    if isinstance(value, str) and "${" in value:
+        raise ManifestError(f"{_location(node)}: environment interpolation is not supported.")
+    return value
+
+
+def load_manifest(path: Path) -> ProxyConfig:
+    """Read one UTF-8 manifest without executing Rails or constructing owners.
+
+    Relative Rails paths are anchored to the resolved manifest's directory,
+    including when the caller supplies a symlink. Application assembly remains
+    responsible for checking directory availability, routes and upstream URLs.
+    """
+    try:
+        path = path.resolve(strict=True)
+        if not path.is_file():
+            raise ManifestError("The manifest must be a regular file.")
+        with path.open("rb") as source:
+            raw = source.read(MAX_MANIFEST_BYTES + 1)
+        if len(raw) > MAX_MANIFEST_BYTES:
+            raise ManifestError("The manifest exceeds the 1 MiB size limit.")
+        source_text = raw.decode("utf-8")
+    except (OSError, UnicodeError, RuntimeError):
+        raise ManifestError("Cannot read the manifest as a UTF-8 file.") from None
+
+    locations: dict[tuple, str] = {}
+    loader = None
+    try:
+        loader = _ManifestLoader(source_text)
+        node = loader.get_single_node()
+        if not isinstance(node, MappingNode):
+            raise ManifestError("The manifest root must be an object.")
+        data = _construct(loader, node, locations)
+    except yaml.YAMLError as error:
+        mark = getattr(error, "problem_mark", None)
+        location = f"line {mark.line + 1}, column {mark.column + 1}: " if mark is not None else ""
+        raise ManifestError(f"{location}invalid YAML.") from None
+    finally:
+        if loader is not None:
+            loader.dispose()
+    try:
+        manifest = ApplicationManifest.model_validate(data)
+    except ValidationError as error:
+        details = []
+        for failure in error.errors(include_url=False, include_context=False, include_input=False)[:8]:
+            location = tuple(failure["loc"])
+            while location not in locations and location:
+                location = location[:-1]
+            details.append(f"{locations.get(location, 'manifest')}: {failure['type']}")
+        raise ManifestError("Invalid manifest: " + "; ".join(details) + ". Check the manifest schema.") from None
+
+    entries = tuple(
+        ConfigEntry(entry.id, (path.parent / entry.config_path).resolve()) for entry in manifest.rails_configs
+    )
+    return ProxyConfig.model_validate(
+        {**manifest.model_dump(exclude={"version", "rails_configs"}), "rails_configs": entries}
+    )
diff --git a/nemoguardrails/server/experimental/_configuration_error.py b/nemoguardrails/server/experimental/_configuration_error.py
new file mode 100644
index 0000000000..e7a2e27111
--- /dev/null
+++ b/nemoguardrails/server/experimental/_configuration_error.py
@@ -0,0 +1,18 @@
+# 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.
+
+
+class DeploymentValidationError(ValueError):
+    """Invalid deployment configuration with a message containing no input values."""
diff --git a/nemoguardrails/server/experimental/_provider_routes.py b/nemoguardrails/server/experimental/_provider_routes.py
index cdd40564e1..8a0aada715 100644
--- a/nemoguardrails/server/experimental/_provider_routes.py
+++ b/nemoguardrails/server/experimental/_provider_routes.py
@@ -21,6 +21,7 @@
 
 from fastapi import APIRouter, FastAPI, routing
 
+from nemoguardrails.server.experimental._configuration_error import DeploymentValidationError
 from nemoguardrails.server.experimental._content_checker import ContentCheckerResolver
 from nemoguardrails.server.experimental._guarded_proxy import create_guarded_http_operation
 from nemoguardrails.server.experimental._http_kernel import (
@@ -66,13 +67,13 @@ def map_upstream_path(path: str, raw_path: bytes) -> str:
 
 def _validate_provider_plans(plans: Sequence[ProviderRoutePlan]) -> None:
     if not plans:
-        raise ValueError("At least one provider integration is required.")
+        raise DeploymentValidationError("At least one provider integration is required.")
     mount_paths = [plan.mount_path for plan in plans]
     if len(mount_paths) != len(set(mount_paths)):
-        raise ValueError("Provider mount paths must be unique.")
+        raise DeploymentValidationError("Provider mount paths must be unique.")
     instance_ids = [plan.provider.instance_id for plan in plans]
     if len(instance_ids) != len(set(instance_ids)):
-        raise ValueError("Provider instance ids must be unique.")
+        raise DeploymentValidationError("Provider instance ids must be unique.")
 
     operation_ids = [
         f"{plan.provider.instance_id}.{endpoint.operation_name}"
@@ -80,7 +81,7 @@ def _validate_provider_plans(plans: Sequence[ProviderRoutePlan]) -> None:
         for endpoint in plan.provider.endpoints
     ]
     if len(operation_ids) != len(set(operation_ids)):
-        raise ValueError("Guarded endpoint operation identifiers must be globally unique.")
+        raise DeploymentValidationError("Guarded endpoint operation identifiers must be globally unique.")
     routes = [
         (method, _route_shape(operation_path.route_path))
         for plan in plans
@@ -88,15 +89,13 @@ def _validate_provider_plans(plans: Sequence[ProviderRoutePlan]) -> None:
         for method in operation_path.methods | operation_path.passthrough_methods
     ]
     if len(routes) != len(set(routes)):
-        raise ValueError("Guarded endpoint paths must be globally unique after provider mounting.")
+        raise DeploymentValidationError("Guarded endpoint paths must be globally unique after provider mounting.")
     for plan in plans:
         for other in plans:
             if not other.mount_path.startswith(plan.mount_path + "/"):
                 continue
             if plan.ownership.intersects_mount(other.mount_path):
-                raise ValueError(
-                    f"Provider mount {other.mount_path!r} would shadow a guarded route of instance {plan.provider.instance_id!r}."
-                )
+                raise DeploymentValidationError("A provider mount would shadow a guarded route of another instance.")
 
 
 def _registered_http_routes(router: APIRouter | FastAPI, path_prefix: str = "") -> dict[str, set[str]]:
@@ -177,7 +176,7 @@ def include_provider_routes(
     if normalized_prefix and (
         not normalized_prefix.startswith("/") or not normalized_prefix.isascii() or "//" in normalized_prefix
     ):
-        raise ValueError("The provider path prefix must be empty or one canonical absolute ASCII path.")
+        raise DeploymentValidationError("The provider path prefix must be empty or one canonical absolute ASCII path.")
     reserved_routes = _registered_http_routes(application)
     plans = []
     for provider in providers:
diff --git a/nemoguardrails/server/experimental/_provider_transport.py b/nemoguardrails/server/experimental/_provider_transport.py
index 73d020efd0..24e75a5899 100644
--- a/nemoguardrails/server/experimental/_provider_transport.py
+++ b/nemoguardrails/server/experimental/_provider_transport.py
@@ -21,6 +21,7 @@
 
 import httpx
 
+from nemoguardrails.server.experimental._configuration_error import DeploymentValidationError
 from nemoguardrails.server.experimental._http_kernel import (
     DEFAULT_MAX_RESPONSE_BODY_BYTES,
     BufferedHttpRequest,
@@ -133,16 +134,21 @@ def __init__(
 
         if not isinstance(http_client, httpx.AsyncClient) and not callable(http_client):
             raise TypeError("The outbound HTTP client must be an httpx.AsyncClient or a callable resolver.")
-        parsed_base_url = httpx.URL(upstream.base_url)
+        try:
+            parsed_base_url = httpx.URL(upstream.base_url)
+        except httpx.InvalidURL:
+            raise DeploymentValidationError(
+                "The outbound base URL must be a valid absolute HTTP or HTTPS URL."
+            ) from None
         if not parsed_base_url.is_absolute_url or parsed_base_url.scheme not in {"http", "https"}:
-            raise ValueError("The outbound base URL must be an absolute HTTP or HTTPS URL.")
+            raise DeploymentValidationError("The outbound base URL must be an absolute HTTP or HTTPS URL.")
         if parsed_base_url.userinfo:
-            raise ValueError("The outbound base URL must not contain credentials.")
+            raise DeploymentValidationError("The outbound base URL must not contain credentials.")
         if parsed_base_url.query or parsed_base_url.fragment:
-            raise ValueError("The outbound base URL must not contain a query or fragment.")
+            raise DeploymentValidationError("The outbound base URL must not contain a query or fragment.")
         normalized_mount = upstream.mount_path.rstrip("/")
         if normalized_mount and (not normalized_mount.startswith("/") or not normalized_mount.isascii()):
-            raise ValueError("The outbound mount path must be an absolute ASCII path.")
+            raise DeploymentValidationError("The outbound mount path must be an absolute ASCII path.")
         if any(
             not isinstance(name, str) or not name or not name.isascii() or name != name.strip().lower()
             for name in control_headers
diff --git a/nemoguardrails/server/experimental/application.md b/nemoguardrails/server/experimental/application.md
index 15045ef3a0..934ed7d118 100644
--- a/nemoguardrails/server/experimental/application.md
+++ b/nemoguardrails/server/experimental/application.md
@@ -1,17 +1,179 @@
-# Python proxy application
+# 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 Python interface and server-owned YAML launcher require the `server` extra.
+The manifest's numeric revision identifies its format, not a stability promise.
 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.
 
+## Run from YAML
+
+The server owns its launcher; it is not a subcommand of the library CLI.
+Its current module location is temporary. A later independent server package
+will register its own executable against the same CLI implementation.
+
+From the repository root, install dependencies and copy the complete example
+into a deployment-owned directory:
+
+```bash
+uv sync --locked --extra server
+cp -R nemoguardrails/server/experimental/examples/basic proxy-example
+cd proxy-example
+```
+
+Keep the copied `proxy.yaml` and its sibling `rails` directory together. Edit
+your copy rather than the package's bundled example. From this directory:
+
+```bash
+uv run --locked python -m nemoguardrails.server.experimental validate proxy.yaml
+uv run --locked python -m nemoguardrails.server.experimental inspect proxy.yaml
+uv run --locked python -m nemoguardrails.server.experimental serve proxy.yaml
+```
+
+The [example manifest](examples/basic/proxy.yaml)
+selects native OpenAI Chat Completions and Responses bindings against a local
+upstream at `http://127.0.0.1:9000/v1`; it does not start an upstream. Both
+operations support the reviewed single-text buffered and streaming modes.
+Set `base_url` to your deployed provider endpoint and select `dialect: litellm`
+only if that provider needs the reviewed dialect. A URL does not imply a dialect.
+
+```yaml
+version: 1
+rails_configs:
+  - id: standard
+    config_path: ./rails/standard
+providers:
+  - provider: openai
+    instance_id: gateway
+    base_url: https://gateway.example/v1
+    operations: [chat_completions, responses]
+    rails_config_id: standard
+```
+
+Keep the proxy manifest outside the Rails directory: Rails loads YAML files
+recursively from its own directory, and the two formats must not be mixed.
+The example uses `proxy.yaml` beside `rails/config.yml`.
+
+The YAML fields are the Python configuration fields below, plus required integer
+`version: 1`. Every provider must specify a fixed `rails_config_id` in YAML;
+there is no implicit selection, callback import or dynamic resolver syntax.
+Multiple providers and dozens of shared catalogue entries do not require
+repeating the catalogue. Use Python for trusted dynamic resolution.
+
+Relative Rails paths resolve against the manifest's directory, not the shell's
+working directory. If the manifest is a symlink, its target directory is used.
+No environment or tilde expansion is performed. Unknown fields, duplicate keys,
+explicit YAML tags, anchors, aliases, merges, non-string mapping keys and multiple
+documents are rejected. Input is UTF-8 and bounded to 1 MiB, depth 32 and 10,000
+nodes (including keys). The manifest is deployment-owned; never accept uploads
+from callers as configuration.
+
+`validate` checks the declaration, directory availability, upstream
+URL rules and route composition. It does not read Rails contents, import their
+actions, start engines, contact providers or prove checker compatibility.
+`inspect` returns JSON describing actual bindings and directional profiles,
+including passthrough behavior; it does not claim that particular rails are
+enabled. `routes` lists served guarded paths; `reserved_route_patterns` lists
+additional path patterns reserved to reject unsupported aliases. Upstream
+credentials, queries and fragments are rejected, and manifest diagnostics omit
+source values/excerpts. Inspection still contains deployment metadata,
+including upstream addresses; treat it accordingly.
+
+For editor guidance, emit JSON Schema directly from the manifest model:
+
+```bash
+uv run --locked python -m nemoguardrails.server.experimental schema > proxy.schema.json
+```
+
+This is the configuration schema, not an exported guard contract. Semantic
+checks such as fixed-ID references and routing still require `validate`.
+There is no separately maintained schema artifact.
+
+`serve` defaults to `127.0.0.1:8000`; `--host` and `--port` override it.
+On shutdown, in-flight requests have 30 seconds to finish before cancellation;
+`--timeout-graceful-shutdown` sets this deadline in seconds. Application resource
+cleanup follows request draining.
+Access logging and forwarded-proxy-header trust are disabled. Provide your own
+inbound access controls before listening beyond loopback. Caller authorization
+headers still follow the existing upstream forwarding rules; this launcher
+does not store provider keys or introduce service authentication.
+
+### Requests and blocking
+
+The [example Rails configuration](examples/basic/rails/config.yml) blocks common
+ASCII email-address patterns in guarded input and output text using local regex
+checks. It needs no rail model, additional package or rail-service API key;
+generation still uses the configured upstream. This is a small, editable policy,
+not comprehensive PII detection or semantic content moderation. It can miss
+obfuscated or internationalized addresses and can block harmless example addresses.
+Only text selected by the operation binding is checked, not every field in the
+provider payload.
+
+Output streaming explicitly uses `chunk_size: 200`, `context_size: 50` and
+`stream_first: false`. The sizes match the Rails configuration defaults, but in
+this proxy they count **nonempty provider text deltas**, not tokenizer tokens or
+SSE events. Each window is checked before its new text is released; the last
+partial window is checked at completion. Short responses may therefore be held
+until completion. Smaller windows reduce buffering delay but increase check
+frequency; these example values are not a deployment-specific tuning recommendation.
+The proxy requires `stream_first: false` when output checking is enabled.
+
+Context lets a later check detect patterns spanning windows, but cannot retract
+an already-released prefix. Finite context can also miss sufficiently long or
+fragmented patterns. Do not treat streaming regex checks as a guarantee against
+partial disclosure. Use buffered responses when the entire answer must be
+checked before any of its text is returned; the regex's detection limits still apply.
+
+With a compatible upstream running, replace `YOUR_DEPLOYED_MODEL` with its model
+ID. Add an `Authorization` header from a locally managed environment variable if
+the upstream requires one; never put credentials in the manifest.
+
+```bash
+curl -sS http://127.0.0.1:8000/v1/chat/completions \
+  -H 'Content-Type: application/json' \
+  -d '{"model":"YOUR_DEPLOYED_MODEL","messages":[{"role":"user","content":"hello"}]}'
+
+curl -N http://127.0.0.1:8000/v1/chat/completions \
+  -H 'Content-Type: application/json' \
+  -d '{"model":"YOUR_DEPLOYED_MODEL","messages":[{"role":"user","content":"hello"}],"stream":true}'
+
+curl -sS http://127.0.0.1:8000/v1/responses \
+  -H 'Content-Type: application/json' \
+  -d '{"model":"YOUR_DEPLOYED_MODEL","input":"hello","store":false}'
+
+curl -sS http://127.0.0.1:8000/v1/chat/completions \
+  -H 'Content-Type: application/json' \
+  -d '{"model":"YOUR_DEPLOYED_MODEL","messages":[{"role":"user","content":"Contact person@example.com"}]}'
+```
+
+The last request returns `content_blocked` before upstream dispatch. To check
+output blocking deterministically, use a synthetic upstream that returns
+`person@example.com`: buffered output is rejected; a streamed window containing
+the address is withheld and an error event is returned after streaming headers
+have been sent. Earlier allowed windows may already have been released. An LLM
+prompt asking it to emit an address is not a deterministic regression test.
+
+Invalid guarded content does not fall back to a weaker binding. Unselected
+operations may still pass through, and a disabled input/output check does not
+become enabled merely by selecting an operation.
+
+Profile/variant selection is not exposed. Inspection reads current profiles
+from the bindings. Future variants would select shipped, tested combinations
+per operation at startup; Rails configuration alone cannot enable multimodal
+extraction or inspection.
+
+The launcher reuses the existing lifespan, including catalogue preloading.
+Known incomplete-cleanup/retry prerequisites in the core and shared HTTP owners
+remain separate. Successful launch or disconnect smoke tests do not establish
+that interrupted cleanup is complete; run those real-owner regressions before
+claiming lifecycle readiness.
+
 ## Declare a built-in provider
 
 ```python
@@ -190,6 +352,13 @@ 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.
 
+`config_resolution_timeout` defaults to five seconds and applies separately to
+trusted resolver execution. Set it in `ProxyConfig` or the manifest, or as a
+keyword in direct Python assembly. Synchronous resolvers must remain short and
+non-blocking; the cooperative deadline cannot interrupt blocking code. See
+[configuration resolution](configuration-resolution.md) for asynchronous selection
+and fail-closed timeout behavior.
+
 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
diff --git a/nemoguardrails/server/experimental/cli.py b/nemoguardrails/server/experimental/cli.py
new file mode 100644
index 0000000000..4792b616a3
--- /dev/null
+++ b/nemoguardrails/server/experimental/cli.py
@@ -0,0 +1,171 @@
+# 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.
+
+"""Server-owned manifest commands, independent of the library's CLI.
+
+Validation and inspection construct routes but never enter application lifespan.
+Serving uses the same application assembly and its existing resource ownership.
+"""
+
+import json
+from pathlib import Path
+
+import httpx
+import typer
+from fastapi import FastAPI
+
+from nemoguardrails.server.experimental._application_manifest import (
+    ApplicationManifest,
+    ManifestError,
+    load_manifest,
+)
+from nemoguardrails.server.experimental._configuration import ProxyConfig, resolve_provider
+from nemoguardrails.server.experimental._configuration_error import DeploymentValidationError
+from nemoguardrails.server.experimental.application import create_proxy_application
+from nemoguardrails.server.experimental.provider.payload import validate_payload_projection_contract
+
+app = typer.Typer(
+    no_args_is_help=True, pretty_exceptions_enable=False, help="Validate, inspect or serve a proxy manifest."
+)
+
+
+def _prepare(path: Path) -> tuple[ProxyConfig, FastAPI]:
+    """Validate through existing assembly, suppressing raw validation inputs."""
+    try:
+        config = load_manifest(path)
+        application = create_proxy_application(config)
+        return config, application
+    except ManifestError as error:
+        typer.echo(str(error), err=True)
+    except DeploymentValidationError as error:
+        typer.echo(f"Invalid application: {error}", err=True)
+    except (ValueError, TypeError, OSError, RuntimeError, httpx.InvalidURL):
+        typer.echo(
+            "Invalid application: check Rails directories, provider URLs, instance IDs and route mounts.", err=True
+        )
+    raise typer.Exit(2)
+
+
+@app.command()
+def validate(manifest: Path = typer.Argument(..., help="Deployment-owned YAML file.")) -> None:
+    """Check manifest data, directories and routing without loading Rails code."""
+    _prepare(manifest)
+    typer.echo("Valid declaration. Rails contents, checker compatibility and startup have not been validated.")
+
+
+def inspect_config(config: ProxyConfig) -> dict:
+    """Describe selected bindings, not inferred checker or provider capabilities."""
+    providers = []
+    for declaration in config.providers:
+        provider = resolve_provider(declaration)
+        prefix = config.path_prefix.rstrip("/") + provider.upstream.mount_path
+        operations = []
+        for endpoint in provider.endpoints:
+            request = validate_payload_projection_contract(endpoint.guarded_request_model, "request")
+            response = validate_payload_projection_contract(endpoint.guarded_response_model, "response")
+            operations.append(
+                {
+                    "operation": endpoint.operation_name,
+                    "routes": [
+                        {
+                            "path": prefix + route.route_path,
+                            "guarded_methods": sorted(route.methods),
+                            "passthrough_methods": sorted(route.passthrough_methods),
+                        }
+                        for route in endpoint.operation_paths
+                        if route.route_path == endpoint.route_path
+                    ],
+                    "reserved_route_patterns": [
+                        prefix + route.route_path
+                        for route in endpoint.operation_paths
+                        if route.route_path != endpoint.route_path
+                    ],
+                    "buffered": True,
+                    "streaming": endpoint.stream is not None,
+                    "profiles": {
+                        "request": request.profile.value,
+                        "response": response.profile.value,
+                        "stream": endpoint.stream.classifier.contract.profile.value if endpoint.stream else None,
+                    },
+                }
+            )
+        providers.append(
+            {
+                "provider": provider.provider_type,
+                "instance_id": provider.instance_id,
+                "mount_path": prefix,
+                "upstream": provider.upstream.base_url,
+                "dialect": declaration.dialect or "native",
+                "rails_config_id": declaration.rails_config_id,
+                "operations": operations,
+            }
+        )
+    return {
+        "providers": providers,
+        "rails_config_ids": [entry.id for entry in config.rails_configs],
+        "limits": {
+            "max_request_body_bytes": config.max_request_body_bytes,
+            "max_response_body_bytes": config.max_response_body_bytes,
+            "timeout": config.timeout,
+            "config_resolution_timeout": config.config_resolution_timeout,
+        },
+        "routing": {
+            "unselected_routes": "passthrough within provider mounts, excluding reserved application routes",
+            "unsupported_guarded_shapes": "rejected when guard checks apply; not retried as passthrough",
+            "mount_precedence": "most-specific mount first for passthrough",
+        },
+        "checks": "Not loaded. Profiles describe bound policy, not enabled Rails checks or checker compatibility.",
+    }
+
+
+@app.command(name="inspect")
+def inspect_manifest(manifest: Path = typer.Argument(..., help="Deployment-owned YAML file.")) -> None:
+    """Show effective routes, bound profiles and fixed Rails selections as JSON."""
+    config, _ = _prepare(manifest)
+    typer.echo(json.dumps(inspect_config(config), indent=2))
+
+
+@app.command()
+def schema() -> None:
+    """Emit editor JSON Schema from the manifest model, without a second schema."""
+    typer.echo(json.dumps(ApplicationManifest.model_json_schema(), indent=2))
+
+
+@app.command()
+def serve(
+    manifest: Path = typer.Argument(..., help="Deployment-owned YAML file."),
+    host: str = typer.Option("127.0.0.1", help="Listen address; non-loopback requires deployment access controls."),
+    port: int = typer.Option(8000, min=0, max=65535, help="Listen port; 0 selects a free port."),
+    timeout_graceful_shutdown: int = typer.Option(
+        30, min=1, help="Seconds to drain in-flight requests before cancellation and application cleanup."
+    ),
+) -> None:
+    """Run the existing application with fixed bindings and lifespan ownership."""
+    import uvicorn
+
+    _, application = _prepare(manifest)
+    uvicorn.run(
+        application,
+        host=host,
+        port=port,
+        access_log=False,
+        proxy_headers=False,
+        timeout_graceful_shutdown=timeout_graceful_shutdown,
+    )
+
+
+def main() -> None:
+    """Enter the server CLI from module invocation or a future console script."""
+    app()
diff --git a/nemoguardrails/server/experimental/examples/basic/proxy.yaml b/nemoguardrails/server/experimental/examples/basic/proxy.yaml
new file mode 100644
index 0000000000..d8049ca534
--- /dev/null
+++ b/nemoguardrails/server/experimental/examples/basic/proxy.yaml
@@ -0,0 +1,12 @@
+version: 1
+rails_configs:
+  - id: standard
+    config_path: ./rails
+providers:
+  - provider: openai
+    instance_id: local
+    base_url: http://127.0.0.1:9000/v1
+    operations:
+      - chat_completions
+      - responses
+    rails_config_id: standard
diff --git a/nemoguardrails/server/experimental/examples/basic/rails/config.yml b/nemoguardrails/server/experimental/examples/basic/rails/config.yml
new file mode 100644
index 0000000000..0fce8cfa4b
--- /dev/null
+++ b/nemoguardrails/server/experimental/examples/basic/rails/config.yml
@@ -0,0 +1,21 @@
+models: []
+rails:
+  config:
+    regex_detection:
+      input:
+        patterns:
+          - '\b[A-Za-z0-9._%+-]{1,64}@[A-Za-z0-9.-]{1,253}\.[A-Za-z]{2,63}\b'
+      output:
+        patterns:
+          - '\b[A-Za-z0-9._%+-]{1,64}@[A-Za-z0-9.-]{1,253}\.[A-Za-z]{2,63}\b'
+  input:
+    flows:
+      - regex check input
+  output:
+    flows:
+      - regex check output
+    streaming:
+      enabled: true
+      chunk_size: 200
+      context_size: 50
+      stream_first: false
diff --git a/nemoguardrails/server/experimental/provider/integration.py b/nemoguardrails/server/experimental/provider/integration.py
index 85b2a65d71..6b7ea03281 100644
--- a/nemoguardrails/server/experimental/provider/integration.py
+++ b/nemoguardrails/server/experimental/provider/integration.py
@@ -19,6 +19,7 @@
 from dataclasses import dataclass
 from typing import Any, TypeAlias
 
+from nemoguardrails.server.experimental._configuration_error import DeploymentValidationError
 from nemoguardrails.server.experimental._http_paths import _route_shape
 from nemoguardrails.server.experimental.provider.endpoint import GuardedJsonEndpoint
 from nemoguardrails.server.experimental.provider.errors import ProviderErrorMapping
@@ -58,9 +59,9 @@ def __post_init__(self) -> None:
         if not self.provider_type or not self.provider_type.replace("-", "_").isidentifier():
             raise ValueError("The provider type must be a non-empty identifier.")
         if not self.instance_id or not self.instance_id.replace("-", "_").isidentifier():
-            raise ValueError("The provider instance id must be a non-empty identifier.")
+            raise DeploymentValidationError("The provider instance id must be a non-empty identifier.")
         if not isinstance(self.upstream.base_url, str) or not self.upstream.base_url.strip():
-            raise ValueError("The provider upstream base URL must not be empty.")
+            raise DeploymentValidationError("The provider upstream base URL must not be empty.")
         if not self.display_name.strip():
             raise ValueError("The provider display name must not be empty.")
         if not self.endpoints:
@@ -75,9 +76,9 @@ def __post_init__(self) -> None:
         if mount_path and (
             not mount_path.startswith("/") or mount_path.endswith("/") or not mount_path.isascii() or "//" in mount_path
         ):
-            raise ValueError("A provider mount path must be empty or one canonical absolute ASCII path.")
+            raise DeploymentValidationError("A provider mount path must be empty or one canonical absolute ASCII path.")
         if "{" in mount_path or "}" in mount_path:
-            raise ValueError("A provider mount path must not contain dynamic parameters.")
+            raise DeploymentValidationError("A provider mount path must not contain dynamic parameters.")
 
         routes = [
             (method, _route_shape(operation_path.route_path))
diff --git a/nemoguardrails/server/experimental/rails/config.py b/nemoguardrails/server/experimental/rails/config.py
index 4666c5bf0a..48a678665e 100644
--- a/nemoguardrails/server/experimental/rails/config.py
+++ b/nemoguardrails/server/experimental/rails/config.py
@@ -23,6 +23,8 @@
 from fastapi import Request
 from pydantic import ConfigDict
 
+from nemoguardrails.server.experimental._configuration_error import DeploymentValidationError
+
 
 @dataclass(frozen=True, slots=True)
 class ConfigEntry:
@@ -56,12 +58,12 @@ def validate_config_entries(entries: Sequence[ConfigEntry]) -> tuple[ConfigEntry
         if not isinstance(entry, ConfigEntry):
             raise TypeError("Rails config entries must be ConfigEntry instances.")
         if not entry.id or entry.id != entry.id.strip():
-            raise ValueError("A Rails config id must be non-empty and trimmed.")
+            raise DeploymentValidationError("A Rails config id must be non-empty and trimmed.")
         if entry.id in ids:
-            raise ValueError(f"Duplicate Rails config id: {entry.id}")
+            raise DeploymentValidationError("Duplicate Rails config id.")
         config_path = entry.config_path.expanduser().resolve()
         if not config_path.is_dir():
-            raise ValueError(f"Rails config '{entry.id}' does not reference a configuration directory.")
+            raise DeploymentValidationError("A Rails config entry does not reference a configuration directory.")
         ids.add(entry.id)
         resolved.append(ConfigEntry(id=entry.id, config_path=config_path))
     return tuple(resolved)
diff --git a/tests/server/experimental/test_application_manifest.py b/tests/server/experimental/test_application_manifest.py
new file mode 100644
index 0000000000..33f2825502
--- /dev/null
+++ b/tests/server/experimental/test_application_manifest.py
@@ -0,0 +1,468 @@
+# 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.
+
+"""Test the manifest trust boundary and the server-owned command surface."""
+
+import asyncio
+import json
+
+import httpx
+import pytest
+import yaml
+from jsonschema import Draft202012Validator
+from typer.testing import CliRunner
+
+from nemoguardrails.server.experimental import _application, _application_manifest, cli
+from nemoguardrails.server.experimental._application_manifest import ApplicationManifest, ManifestError, load_manifest
+from nemoguardrails.server.experimental._configuration import resolve_provider
+from nemoguardrails.server.experimental.application import (
+    ConfigEntry,
+    ProviderConfig,
+    ProxyConfig,
+    create_proxy_application,
+)
+from nemoguardrails.server.experimental.rails import _repository
+
+runner = CliRunner()
+
+
+@pytest.fixture
+def document(tmp_path):
+    directory = tmp_path / "rails"
+    directory.mkdir()
+    (directory / "config.yml").write_text("models: []\n", encoding="utf-8")
+    return {
+        "version": 1,
+        "rails_configs": [{"id": "standard", "config_path": "rails"}],
+        "providers": [
+            {
+                "provider": "openai",
+                "instance_id": "public",
+                "base_url": "https://provider.test/v1",
+                "operations": ["chat_completions", "responses"],
+                "rails_config_id": "standard",
+            }
+        ],
+    }
+
+
+def write_manifest(tmp_path, document):
+    path = tmp_path / "proxy.yaml"
+    path.write_text(yaml.safe_dump(json.loads(json.dumps(document)), sort_keys=False), encoding="utf-8")
+    return path
+
+
+@pytest.mark.parametrize("dialect", [None, "litellm"])
+@pytest.mark.parametrize("resolution_timeout", [0.5, 5.0])
+def test_yaml_and_python_resolve_identical_bindings_from_another_directory(
+    tmp_path, document, monkeypatch, dialect, resolution_timeout
+):
+    document["providers"][0]["dialect"] = dialect
+    document["config_resolution_timeout"] = resolution_timeout
+    path = write_manifest(tmp_path, document)
+    elsewhere = tmp_path / "elsewhere"
+    elsewhere.mkdir()
+    monkeypatch.chdir(elsewhere)
+    config = load_manifest(path)
+    expected = ProxyConfig(
+        rails_configs=(ConfigEntry("standard", tmp_path / "rails"),),
+        config_resolution_timeout=resolution_timeout,
+        providers=(
+            ProviderConfig(
+                provider="openai",
+                instance_id="public",
+                base_url="https://provider.test/v1",
+                operations=("chat_completions", "responses"),
+                rails_config_id="standard",
+                dialect=dialect,
+            ),
+        ),
+    )
+    assert config == expected
+    assert cli.inspect_config(config)["limits"]["config_resolution_timeout"] == resolution_timeout
+    assert resolve_provider(config.providers[0]) == resolve_provider(expected.providers[0])
+    actual_app = create_proxy_application(config)
+    expected_app = create_proxy_application(expected)
+    assert actual_app.openapi() == expected_app.openapi()
+
+
+@pytest.mark.parametrize(
+    "source",
+    [
+        "",
+        "[]",
+        "version: 1\nversion: 1",
+        "providers: [{instance_id: a, instance_id: b}]",
+        "version: [",
+        "version: 1\n---\nversion: 1",
+        "version: !!int 1",
+        "x: !secret value",
+        "x: !!python/object/apply:os.system [false]",
+        "x: &a [1]\ny: *a",
+        "x: *undefined",
+        "<<: {version: 1}",
+        "true: value",
+        "[a, b]: value",
+        "x: 2026-10-15",
+        "x: ${PRIVATE_VALUE}",
+        "x: " + "[" * 33 + "0" + "]" * 33,
+    ],
+)
+def test_rejects_unsafe_ambiguous_or_unsupported_yaml(tmp_path, source):
+    path = tmp_path / "proxy.yaml"
+    path.write_text(source, encoding="utf-8")
+    with pytest.raises(ManifestError):
+        load_manifest(path)
+
+
+def test_byte_and_node_limits(tmp_path, monkeypatch):
+    path = tmp_path / "proxy.yaml"
+    path.write_bytes(b"x" * (_application_manifest.MAX_MANIFEST_BYTES + 1))
+    with pytest.raises(ManifestError, match="size limit"):
+        load_manifest(path)
+    monkeypatch.setattr(_application_manifest, "MAX_MANIFEST_NODES", 8)
+    path.write_text("x: [1, 2, 3, 4, 5, 6, 7, 8]", encoding="utf-8")
+    with pytest.raises(ManifestError, match="structure"):
+        load_manifest(path)
+
+
+@pytest.mark.parametrize("kind", ["missing", "directory", "utf8"])
+def test_manifest_file_errors_are_sanitized(tmp_path, kind):
+    path = tmp_path / "SECRET"
+    if kind == "directory":
+        path.mkdir()
+    elif kind == "utf8":
+        path.write_bytes(b"\xff\xfe")
+    result = runner.invoke(cli.app, ["validate", str(path)])
+    assert result.exit_code == 2
+    assert "SECRET" not in result.output
+    assert "Traceback" not in result.output
+
+
+@pytest.mark.parametrize("version", [None, True, False, 1.0, "1", 2])
+def test_version_requires_exact_supported_integer(tmp_path, document, version):
+    document["version"] = version
+    with pytest.raises(ManifestError):
+        load_manifest(write_manifest(tmp_path, document))
+
+
+@pytest.mark.parametrize(
+    "change",
+    [
+        {"profile": "single_text.v1"},
+        {"variant": "text"},
+        {"operations": {"chat_completions": {"variant": "text"}}},
+        {"operations": []},
+        {"operations": ["chat_completions", "chat_completions"]},
+        {"operations": ["unimplemented"]},
+        {"provider": "anthropic"},
+        {"dialect": "openai"},
+        {"rails_config_id": None},
+        {"rails_config_id": "unknown"},
+        {"factory": "untrusted.module:create"},
+        {"base_url": "https://user:SECRET@provider.test/v1"},
+        {"base_url": "https://provider.test/v1?api_key=SECRET"},
+        {"base_url": "ftp://provider.test"},
+        {"base_url": "https://provider.test:SECRET"},
+        {"mount_path": "invalid"},
+    ],
+)
+def test_invalid_declarations_fail_closed_without_echoing_values(tmp_path, document, change):
+    document["providers"][0].update(change)
+    result = runner.invoke(cli.app, ["inspect", str(write_manifest(tmp_path, document))])
+    assert result.exit_code == 2
+    assert "SECRET" not in result.output
+    assert "Traceback" not in result.output
+
+
+@pytest.mark.parametrize("target", ["root", "provider", "entry"])
+def test_unknown_keys_rejected_with_coordinates_not_source_values(tmp_path, document, target):
+    mapping = {"root": document, "provider": document["providers"][0], "entry": document["rails_configs"][0]}[target]
+    mapping["SECRET_KEY"] = "SECRET_VALUE"
+    result = runner.invoke(cli.app, ["validate", str(write_manifest(tmp_path, document))])
+    assert result.exit_code == 2
+    assert ("unexpected_keyword_argument" if target == "entry" else "extra_forbidden") in result.output
+    assert "line " in result.output and "column " in result.output
+    assert "SECRET" not in result.output
+
+
+@pytest.mark.parametrize(
+    ("fault", "reason"),
+    [
+        ("duplicate_mount", "Provider mount paths must be unique."),
+        ("duplicate_id", "Provider instance ids must be unique."),
+        ("missing_directory", "A Rails config entry does not reference a configuration directory."),
+        ("prefix", "The provider path prefix must be empty or one canonical absolute ASCII path."),
+    ],
+)
+def test_assembly_validation_is_shared(tmp_path, document, fault, reason):
+    if fault.startswith("duplicate"):
+        document["providers"].append(
+            {
+                **document["providers"][0],
+                "instance_id": "other" if fault == "duplicate_mount" else "public",
+                "mount_path": "" if fault == "duplicate_mount" else "/other",
+            }
+        )
+    elif fault == "missing_directory":
+        document["rails_configs"][0]["config_path"] = "SECRET_MISSING"
+    else:
+        document["path_prefix"] = "invalid"
+    result = runner.invoke(cli.app, ["validate", str(write_manifest(tmp_path, document))])
+    assert result.exit_code == 2
+    assert f"Invalid application: {reason}" in result.output
+    assert "SECRET" not in result.output
+
+
+@pytest.mark.parametrize(
+    ("change", "reason"),
+    [
+        ({"base_url": "https://user:SECRET@provider.test/v1"}, "must not contain credentials"),
+        ({"base_url": "https://provider.test/v1?api_key=SECRET"}, "must not contain a query or fragment"),
+        ({"base_url": "https://provider.test:SECRET"}, "must be a valid absolute HTTP or HTTPS URL"),
+        ({"instance_id": "SECRET invalid"}, "instance id must be a non-empty identifier"),
+        ({"mount_path": "/SECRET/{dynamic}"}, "mount path must not contain dynamic parameters"),
+    ],
+)
+def test_deployment_errors_explain_the_reason_without_input_values(tmp_path, document, change, reason):
+    document["providers"][0].update(change)
+    result = runner.invoke(cli.app, ["validate", str(write_manifest(tmp_path, document))])
+    assert result.exit_code == 2
+    assert reason in result.output
+    assert "SECRET" not in result.output
+    assert "Traceback" not in result.output
+
+
+def test_mount_shadow_diagnostic_does_not_echo_instance_or_mount(tmp_path, document):
+    document["providers"][0]["instance_id"] = "SECRET_INSTANCE"
+    document["providers"].append({**document["providers"][0], "instance_id": "other", "mount_path": "/SECRET_MOUNT"})
+    result = runner.invoke(cli.app, ["validate", str(write_manifest(tmp_path, document))])
+    assert result.exit_code == 2
+    assert "would shadow a guarded route of another instance" in result.output
+    assert "SECRET" not in result.output
+
+
+@pytest.mark.parametrize("error_type", [ValueError, TypeError, OSError, RuntimeError, httpx.InvalidURL])
+def test_unexpected_assembly_errors_remain_redacted(tmp_path, document, monkeypatch, error_type):
+    def fail(config):
+        raise error_type("SECRET_VALUE")
+
+    monkeypatch.setattr(cli, "create_proxy_application", fail)
+    result = runner.invoke(cli.app, ["validate", str(write_manifest(tmp_path, document))])
+    assert result.exit_code == 2
+    assert "Invalid application: check Rails directories" in result.output
+    assert "SECRET" not in result.output
+    assert "Traceback" not in result.output
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("dialect", [None, "litellm"])
+@pytest.mark.parametrize(("prefix", "mount"), [("", ""), ("/proxy", "/public")])
+async def test_inspection_separates_served_routes_from_rejected_aliases(tmp_path, document, dialect, prefix, mount):
+    document["path_prefix"] = prefix
+    document["providers"][0].update(dialect=dialect, mount_path=mount)
+    path = write_manifest(tmp_path, document)
+    result = runner.invoke(cli.app, ["inspect", str(path)])
+    assert result.exit_code == 0, result.output
+    operations = json.loads(result.output)["providers"][0]["operations"]
+    application = create_proxy_application(load_manifest(path))
+    registered = application.openapi()["paths"]
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=application), base_url="http://test") as client:
+        for operation in operations:
+            suffix = "/responses" if operation["operation"] == "responses.create" else "/chat/completions"
+            route = prefix + mount + "/v1" + suffix
+            assert operation["routes"] == [
+                {
+                    "path": route,
+                    "guarded_methods": ["POST"],
+                    "passthrough_methods": ["GET"] if suffix == "/chat/completions" else [],
+                }
+            ]
+            assert "post" in registered[route]
+            assert operation["reserved_route_patterns"] == [prefix + mount + "/{api_version}" + suffix]
+            rejected = await client.post(prefix + mount + "/v2" + suffix, json={})
+            assert rejected.status_code == 422
+
+
+@pytest.mark.parametrize("filename", ["config.yml", "nested/policy.yaml", None], ids=["named", "nested", "empty"])
+def test_validation_and_inspection_do_not_load_rails_or_create_owners(tmp_path, document, monkeypatch, filename):
+    def forbidden(*args, **kwargs):
+        pytest.fail("structural commands must not enter runtime construction or load Rails")
+
+    monkeypatch.setattr(_repository.RailsConfig, "from_path", forbidden)
+    monkeypatch.setattr(_repository, "Guardrails", forbidden)
+    monkeypatch.setattr(_application.ServerRuntime, "create", forbidden)
+    source = tmp_path / "rails/config.yml"
+    if filename is None:
+        source.unlink()
+    elif filename != "config.yml":
+        destination = tmp_path / "rails" / filename
+        destination.parent.mkdir(parents=True)
+        source.rename(destination)
+    (tmp_path / "rails/actions.py").write_text("raise RuntimeError('must not execute')", encoding="utf-8")
+    path = write_manifest(tmp_path, document)
+    assert runner.invoke(cli.app, ["validate", str(path)]).exit_code == 0
+    result = runner.invoke(cli.app, ["inspect", str(path)])
+    assert result.exit_code == 0
+    inspected = json.loads(result.output)
+    provider = inspected["providers"][0]
+    assert provider["dialect"] == "native"
+    assert provider["rails_config_id"] == "standard"
+    assert len(provider["operations"]) == 2
+    for operation in provider["operations"]:
+        assert operation["profiles"] == {
+            "request": "single_text.v1",
+            "response": "single_text.v1",
+            "stream": "single_text_delta.v1",
+        }
+        assert operation["buffered"] and operation["streaming"]
+    assert inspected["routing"]["unselected_routes"].startswith("passthrough")
+    assert inspected["checks"].startswith("Not loaded")
+
+
+def test_inspection_uses_actual_bound_profiles(tmp_path, document, monkeypatch):
+    from dataclasses import replace
+
+    from nemoguardrails.server.experimental.provider.payload import PayloadCapabilityProfile
+
+    actual = cli.resolve_provider
+
+    def resolve(config):
+        provider = actual(config)
+        endpoint = provider.endpoints[0]
+        contract = endpoint.guarded_request_model.projection_contract
+        assert contract.profile == PayloadCapabilityProfile.SINGLE_TEXT_V1
+        return replace(provider, endpoints=(replace(endpoint, stream=None),))
+
+    monkeypatch.setattr(cli, "resolve_provider", resolve)
+    inspected = cli.inspect_config(load_manifest(write_manifest(tmp_path, document)))
+    operation = inspected["providers"][0]["operations"][0]
+    assert operation["streaming"] is False
+    assert operation["profiles"]["stream"] is None
+
+
+def test_schema_is_derived_and_can_validate_the_example(document):
+    result = runner.invoke(cli.app, ["schema"])
+    assert result.exit_code == 0
+    schema = json.loads(result.output)
+    assert schema == ApplicationManifest.model_json_schema()
+    Draft202012Validator.check_schema(schema)
+    Draft202012Validator(schema).validate(document)
+    assert schema["additionalProperties"] is False
+    provider_schema = schema["$defs"]["ManifestProviderConfig"]
+    assert provider_schema["additionalProperties"] is False
+    assert "rails_config_id" in provider_schema["required"]
+    assert provider_schema["properties"]["rails_config_id"]["type"] == "string"
+    assert schema["$defs"]["ConfigEntry"]["additionalProperties"] is False
+
+
+@pytest.mark.parametrize("shutdown_timeout", [None, 1])
+def test_serve_uses_existing_assembly_once_and_loopback_defaults(tmp_path, document, monkeypatch, shutdown_timeout):
+    import uvicorn
+
+    calls = []
+    actual = cli.create_proxy_application
+
+    def create(config):
+        calls.append(config)
+        return actual(config)
+
+    observed = {}
+    monkeypatch.setattr(cli, "create_proxy_application", create)
+    monkeypatch.setattr(uvicorn, "run", lambda app, **kwargs: observed.update(app=app, **kwargs))
+    arguments = ["serve", str(write_manifest(tmp_path, document))]
+    if shutdown_timeout is not None:
+        arguments.extend(["--timeout-graceful-shutdown", str(shutdown_timeout)])
+    result = runner.invoke(cli.app, arguments)
+    assert result.exit_code == 0, result.output
+    assert len(calls) == 1
+    assert observed["host"] == "127.0.0.1"
+    assert observed["port"] == 8000
+    assert observed["access_log"] is False
+    assert observed["proxy_headers"] is False
+    assert observed["timeout_graceful_shutdown"] == (30 if shutdown_timeout is None else shutdown_timeout)
+    assert not hasattr(observed["app"].state, "proxy_runtime")
+
+
+@pytest.mark.parametrize("command", ["validate", "inspect", "serve", "schema"])
+def test_command_help(command):
+    assert runner.invoke(cli.app, [command, "--help"]).exit_code == 0
+
+
+@pytest.mark.parametrize(
+    "arguments",
+    [
+        [],
+        ["serve"],
+        ["serve", "unused", "--port", "-1"],
+        ["serve", "unused", "--port", "65536"],
+        ["serve", "unused", "--timeout-graceful-shutdown", "0"],
+        ["serve", "unused", "--timeout-graceful-shutdown", "-1"],
+    ],
+)
+def test_invalid_cli_usage(arguments):
+    assert runner.invoke(cli.app, arguments).exit_code == 2
+
+
+@pytest.mark.asyncio
+async def test_partial_catalog_startup_closes_real_core_workers_and_http_owners(tmp_path, document, monkeypatch):
+    """A later parse failure must roll back the already-started real core."""
+    from nemoguardrails.guardrails.iorails import IORails
+    from nemoguardrails.server.experimental._runtime import ServerRuntime
+
+    invalid = tmp_path / "invalid"
+    invalid.mkdir()
+    (invalid / "config.yml").write_text("models: [", encoding="utf-8")
+    document["rails_configs"].append({"id": "broken", "config_path": "invalid"})
+    path = write_manifest(tmp_path, document)
+    assert runner.invoke(cli.app, ["validate", str(path)]).exit_code == 0
+    application = create_proxy_application(load_manifest(path))
+    runtimes = []
+    cores = []
+    workers = []
+    actual_create = ServerRuntime.create
+    actual_start = IORails.start
+
+    def create(cls, *args, **kwargs):
+        runtime = actual_create(*args, **kwargs)
+        runtimes.append(runtime)
+        return runtime
+
+    async def start(core):
+        await actual_start(core)
+        cores.append(core)
+        workers.extend(core._generate_async_queue._workers)
+
+    monkeypatch.setattr(ServerRuntime, "create", classmethod(create))
+    monkeypatch.setattr(IORails, "start", start)
+    try:
+        with pytest.raises(yaml.YAMLError):
+            async with application.router.lifespan_context(application):
+                pytest.fail("invalid catalogue must not become ready")
+        assert len(cores) == len(runtimes) == 1
+        assert workers and all(worker.done() for worker in workers)
+        assert not cores[0]._running
+        assert cores[0].rails_manager._http_client._client.is_closed
+        assert runtimes[0].http_client.is_closed
+        assert not runtimes[0].content_checkers._checkers
+        assert not hasattr(application.state, "proxy_runtime")
+    finally:
+        for worker in workers:
+            worker.cancel()
+        await asyncio.gather(*workers, return_exceptions=True)
+        for core in cores:
+            await core.rails_manager._http_client.close()
+        for runtime in runtimes:
+            await runtime.close()
diff --git a/tests/server/experimental/test_guardrails_repository.py b/tests/server/experimental/test_guardrails_repository.py
index 9447fe10d7..47259cb877 100644
--- a/tests/server/experimental/test_guardrails_repository.py
+++ b/tests/server/experimental/test_guardrails_repository.py
@@ -329,7 +329,6 @@ async def shutdown():
 async def test_repository_retries_real_core_shutdown_after_queue_timeout(tmp_path, monkeypatch):
     """Retain the real engine until interrupted worker cleanup completes."""
 
-    monkeypatch.setattr(_repository, "_SHUTDOWN_TIMEOUT_SECONDS", 0.02)
     repository = GuardrailsRepository((ConfigEntry("policy", config_path(tmp_path)),))
     await repository.startup()
     facade = repository._checkers["policy"]._rails
@@ -352,7 +351,9 @@ async def unfinished_check():
     try:
         with anyio.fail_after(1):
             await entered.wait()
-            await repository.close()
+            with monkeypatch.context() as failing_close:
+                failing_close.setattr(_repository, "_SHUTDOWN_TIMEOUT_SECONDS", 0.02)
+                await repository.close()
         assert "policy" in repository._checkers
         assert any(not worker.done() for worker in workers)
 
diff --git a/tests/server/experimental/test_proxy_launcher.py b/tests/server/experimental/test_proxy_launcher.py
new file mode 100644
index 0000000000..5539d9e7f4
--- /dev/null
+++ b/tests/server/experimental/test_proxy_launcher.py
@@ -0,0 +1,357 @@
+# 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 the module launcher with real sockets and a synthetic upstream.
+
+Adapted from the preserved runnable-first prototype, using the current shared
+assembly and declarative manifest rather than its earlier single-provider API.
+The synthetic protocol events are test fixtures, not provider recordings.
+"""
+
+import asyncio
+import json
+import os
+import re
+import signal
+import subprocess
+import sys
+from contextlib import asynccontextmanager
+from pathlib import Path
+
+import anyio
+import httpx
+import pytest
+import yaml
+from aiohttp import web
+
+from tests.server.experimental.test_application_configuration import request_payload
+from tests.server.experimental.test_openai_responses_projections import response_payload
+from tests.server.experimental.test_openai_responses_streaming_projections import event_bytes
+
+ROOT = Path(__file__).resolve().parents[3]
+EXAMPLE = ROOT / "nemoguardrails/server/experimental/examples/basic"
+LAUNCHER_TIMEOUT_SECONDS = 30
+
+
+def chat_response(text="safe"):
+    """Build a minimal synthetic Chat response."""
+    return {"choices": [{"index": 0, "message": {"role": "assistant", "content": text}, "finish_reason": "stop"}]}
+
+
+def stream_bytes(operation, parts, *, complete=True):
+    """Encode valid text events using each operation's actual protocol."""
+    if operation == "responses":
+        events = [event_bytes("response.created", response={"output": []}, sequence_number=0)]
+        events.extend(
+            event_bytes(
+                "response.output_text.delta",
+                delta=text,
+                item_id="msg_test",
+                output_index=1,
+                content_index=0,
+                sequence_number=index,
+            )
+            for index, text in enumerate(parts, 1)
+        )
+        if complete:
+            events.append(
+                event_bytes(
+                    "response.completed", response=response_payload("".join(parts)), sequence_number=len(parts) + 1
+                )
+            )
+        return b"".join(events)
+    events = [{"choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}]}]
+    events.extend({"choices": [{"index": 0, "delta": {"content": part}, "finish_reason": None}]} for part in parts)
+    if complete:
+        events.append({"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]})
+    wire = b"".join(
+        b"data: " + json.dumps({"object": "chat.completion.chunk", **event}).encode() + b"\n\n" for event in events
+    )
+    return wire + (b"data: [DONE]\n\n" if complete else b"")
+
+
+@asynccontextmanager
+async def synthetic_upstream():
+    """Serve deterministic bytes and observe upstream socket disconnection."""
+    requests = []
+    disconnected = asyncio.Event()
+
+    async def generation(request):
+        payload = await request.json()
+        requests.append(payload)
+        operation = "responses" if request.path.endswith("/responses") else "chat_completions"
+        text = payload["input"] if operation == "responses" else payload["messages"][0]["content"]
+        answer = "person@example.com" if text == "emit-blocked-output" else "safe"
+        if not payload.get("stream"):
+            return web.json_response(response_payload(answer) if operation == "responses" else chat_response(answer))
+        response = web.StreamResponse(headers={"content-type": "text/event-stream"})
+        await response.prepare(request)
+        try:
+            if text == "disconnect":
+                await response.write(stream_bytes(operation, ["sa", "fe"] * 100, complete=False))
+                while True:
+                    await asyncio.sleep(0.01)
+                    await response.write(b": keepalive\n\n")
+            parts = ["person@", "example.com"] if answer == "person@example.com" else ["sa", "fe"]
+            if text == "emit-blocked-window":
+                parts = ["safe "] * 198 + ["person@", "example.com"]
+            complete = text != "emit-blocked-window"
+            for event in stream_bytes(operation, parts, complete=complete).split(b"\n\n"):
+                if event:
+                    await response.write(event + b"\n\n")
+                    await asyncio.sleep(0.005)
+            if not complete:
+                while True:
+                    await asyncio.sleep(0.01)
+                    await response.write(b": keepalive\n\n")
+        except (ConnectionResetError, asyncio.CancelledError):
+            disconnected.set()
+        return response
+
+    async def models(request):
+        requests.append({"path": request.path})
+        response = web.StreamResponse(headers={"content-type": "application/octet-stream"})
+        await response.prepare(request)
+        await response.write(b"safe")
+        try:
+            while True:
+                await asyncio.sleep(0.01)
+                await response.write(b"opaque")
+        except (ConnectionResetError, asyncio.CancelledError):
+            disconnected.set()
+        return response
+
+    application = web.Application()
+    application.router.add_post("/v1/chat/completions", generation)
+    application.router.add_post("/v1/responses", generation)
+    application.router.add_get("/v1/models", models)
+    runner = web.AppRunner(application, shutdown_timeout=1)
+    await runner.setup()
+    site = web.TCPSite(runner, "127.0.0.1", 0)
+    try:
+        await site.start()
+        yield f"http://127.0.0.1:{runner.addresses[0][1]}/v1", requests, disconnected
+    finally:
+        await runner.cleanup()
+
+
+@asynccontextmanager
+async def launched_proxy(tmp_path, upstream, dialect=None, *, timeout_graceful_shutdown=None):
+    """Launch the documented module, wait for readiness, and reap the child."""
+    manifest = yaml.safe_load((EXAMPLE / "proxy.yaml").read_text(encoding="utf-8"))
+    manifest["rails_configs"][0]["config_path"] = str(EXAMPLE / "rails")
+    manifest["providers"][0]["base_url"] = upstream
+    if dialect is not None:
+        manifest["providers"][0]["dialect"] = dialect
+    path = tmp_path / "proxy.yaml"
+    path.write_text(yaml.safe_dump(manifest), encoding="utf-8")
+    environment = dict(os.environ)
+    for key in (
+        "OPENAI_API_KEY",
+        "NVIDIA_API_KEY",
+        "NVIDIA_INFERENCE_KEY",
+        "LIVE_TEST",
+        "LIVE_TEST_MODE",
+        "TEST_LIVE_MODE",
+    ):
+        environment.pop(key, None)
+    environment["NEMO_GUARDRAILS_NO_USAGE_STATS"] = "1"
+    environment["PYTHONPATH"] = str(ROOT)
+    shutdown_arguments = (
+        [] if timeout_graceful_shutdown is None else ["--timeout-graceful-shutdown", str(timeout_graceful_shutdown)]
+    )
+    process = await asyncio.create_subprocess_exec(
+        sys.executable,
+        "-m",
+        "nemoguardrails.server.experimental",
+        "serve",
+        str(path),
+        "--port",
+        "0",
+        *shutdown_arguments,
+        cwd=tmp_path,
+        env=environment,
+        creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.STDOUT,
+    )
+    lines = []
+    drain = None
+    ready = False
+    try:
+        assert process.stdout is not None
+        with anyio.fail_after(LAUNCHER_TIMEOUT_SECONDS):
+            while True:
+                line = await process.stdout.readline()
+                assert line, b"".join(lines).decode()
+                lines.append(line)
+                match = re.search(rb"Uvicorn running on (http://127\.0\.0\.1:\d+)", line)
+                if match:
+                    ready = True
+                    url = match[1].decode()
+                    break
+        drain = asyncio.create_task(process.stdout.read())
+        yield url, process
+    finally:
+        shutdown_timed_out = False
+        if drain is None and process.stdout is not None:
+            drain = asyncio.create_task(process.stdout.read())
+        if process.returncode is None:
+            try:
+                if sys.platform == "win32":
+                    process.send_signal(signal.CTRL_BREAK_EVENT)
+                else:
+                    process.terminate()
+                await asyncio.wait_for(process.wait(), LAUNCHER_TIMEOUT_SECONDS)
+            except asyncio.TimeoutError:
+                shutdown_timed_out = True
+            finally:
+                if process.returncode is None:
+                    process.kill()
+                    await process.wait()
+        if drain is not None:
+            lines.append(await drain)
+        output = b"".join(lines).decode()
+        if shutdown_timed_out:
+            pytest.fail(f"launcher did not shut down within {LAUNCHER_TIMEOUT_SECONDS} seconds:\n{output}")
+        if ready:
+            if sys.platform != "win32":
+                assert process.returncode in (0, -signal.SIGTERM), output
+            assert "Application shutdown complete." in output, output
+            assert "cleanup timed out" not in output.lower(), output
+            assert "cleanup failed" not in output.lower(), output
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("dialect", [None, "litellm"])
+async def test_localhost_launcher_buffered_streaming_and_blocking(tmp_path, dialect):
+    async with synthetic_upstream() as (upstream, requests, _):
+        async with launched_proxy(tmp_path, upstream, dialect) as (url, _):
+            async with httpx.AsyncClient(base_url=url, trust_env=False, timeout=5) as client:
+                for operation in ("chat_completions", "responses"):
+                    route = "/v1/responses" if operation == "responses" else "/v1/chat/completions"
+                    for stream in (False, True):
+                        allowed = await client.post(route, json=request_payload(operation, streaming=stream))
+                        assert allowed.status_code == 200, allowed.text
+                        if stream:
+                            assert allowed.content == stream_bytes(operation, ["sa", "fe"])
+                        else:
+                            assert allowed.json() == (
+                                response_payload("safe") if operation == "responses" else chat_response()
+                            )
+                        count = len(requests)
+                        blocked_input = await client.post(
+                            route, json=request_payload(operation, "Contact person@example.com", stream)
+                        )
+                        assert blocked_input.status_code == 400
+                        assert blocked_input.json()["error"]["code"] == "content_blocked"
+                        assert len(requests) == count
+                        blocked_output = await client.post(
+                            route, json=request_payload(operation, "emit-blocked-output", stream)
+                        )
+                        assert blocked_output.status_code == (200 if stream else 400)
+                        assert "content_blocked" in blocked_output.text
+                        assert b"person@" not in blocked_output.content
+                        assert b"example.com" not in blocked_output.content
+                    blocked_window = await client.post(
+                        route, json=request_payload(operation, "emit-blocked-window", True)
+                    )
+                    assert blocked_window.status_code == 200
+                    assert "content_blocked" in blocked_window.text
+                    assert b"safe " not in blocked_window.content
+                    assert b"person@" not in blocked_window.content
+                    assert b"example.com" not in blocked_window.content
+                assert len(requests) == 10
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["chat_completions", "responses", "passthrough"])
+async def test_localhost_disconnect_closes_upstream_socket(tmp_path, operation):
+    async with synthetic_upstream() as (upstream, requests, disconnected):
+        async with launched_proxy(tmp_path, upstream) as (url, _):
+            async with httpx.AsyncClient(base_url=url, trust_env=False, timeout=5) as client:
+                path = {
+                    "chat_completions": "/v1/chat/completions",
+                    "responses": "/v1/responses",
+                    "passthrough": "/v1/models",
+                }[operation]
+                passthrough = operation == "passthrough"
+                kwargs = {} if passthrough else {"json": request_payload(operation, "disconnect", True)}
+                request = httpx.Request("GET" if passthrough else "POST", httpx.URL(url).join(path), **kwargs)
+                reader, writer = await asyncio.open_connection(request.url.host, request.url.port)
+                try:
+                    writer.write(
+                        request.method.encode()
+                        + b" "
+                        + request.url.raw_path
+                        + b" HTTP/1.1\r\n"
+                        + b"".join(name + b": " + value + b"\r\n" for name, value in request.headers.raw)
+                        + b"\r\n"
+                        + request.content
+                    )
+                    await writer.drain()
+                    received = b""
+                    with anyio.fail_after(5):
+                        headers = await reader.readuntil(b"\r\n\r\n")
+                        assert headers.split(b"\r\n", 1)[0].split()[1] == b"200", headers
+                        while True:
+                            chunk = await reader.read(4096)
+                            assert chunk, "the upstream ended before the downstream disconnected"
+                            received += chunk
+                            if (b"safe" if passthrough else b'"sa"') in received:
+                                break
+                    assert b'"error"' not in received
+                    writer.write_eof()
+                    with anyio.fail_after(5):
+                        await reader.read()
+                finally:
+                    writer.close()
+                    with anyio.fail_after(5):
+                        await writer.wait_closed()
+                with anyio.fail_after(5):
+                    await disconnected.wait()
+                assert len(requests) == 1
+                allowed = await client.post("/v1/chat/completions", json=request_payload("chat_completions"))
+                assert allowed.status_code == 200
+                assert allowed.json() == chat_response()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["chat_completions", "responses", "passthrough"])
+async def test_shutdown_cancels_active_stream_and_closes_upstream(tmp_path, operation):
+    async with synthetic_upstream() as (upstream, requests, disconnected):
+        async with launched_proxy(tmp_path, upstream, timeout_graceful_shutdown=1) as (url, process):
+            async with httpx.AsyncClient(base_url=url, trust_env=False, timeout=5) as client:
+                path = {
+                    "chat_completions": "/v1/chat/completions",
+                    "responses": "/v1/responses",
+                    "passthrough": "/v1/models",
+                }[operation]
+                passthrough = operation == "passthrough"
+                kwargs = {} if passthrough else {"json": request_payload(operation, "disconnect", True)}
+                async with client.stream("GET" if passthrough else "POST", path, **kwargs) as response:
+                    assert response.status_code == 200
+                    chunks = response.aiter_raw()
+                    assert await anext(chunks)
+                    assert not disconnected.is_set()
+                    assert process.returncode is None
+                    if sys.platform == "win32":
+                        process.send_signal(signal.CTRL_BREAK_EVENT)
+                    else:
+                        process.terminate()
+                    with anyio.fail_after(10):
+                        await process.wait()
+                        await disconnected.wait()
+                assert len(requests) == 1
diff --git a/tests/server/experimental/test_streaming_http.py b/tests/server/experimental/test_streaming_http.py
index a948d0735d..e403de16bf 100644
--- a/tests/server/experimental/test_streaming_http.py
+++ b/tests/server/experimental/test_streaming_http.py
@@ -409,7 +409,8 @@ async def send(_message):
 @pytest.mark.parametrize("close_mode", ["complete", "hang", "raise"])
 @pytest.mark.parametrize("with_callback", [False, True])
 async def test_response_finalizer_closes_each_owner_once(monkeypatch, same_source, close_mode, with_callback):
-    monkeypatch.setattr(_stream_cleanup, "_STREAM_CLOSE_TIMEOUT_SECONDS", 0.01)
+    if close_mode == "hang":
+        monkeypatch.setattr(_stream_cleanup, "_STREAM_CLOSE_TIMEOUT_SECONDS", 0.1)
     attempts = []
 
     class ClosingSource(Source):