test(e2e): cover block_code_execution and openai_moderation guardrails

Extends the guardrails suite with two built-in guardrails registered per
request (default_on=False, opted in via the chat body's guardrails selector)
so neither intercepts unrelated traffic on the shared proxy.

block_code_execution.pre_call.blocks: a python code block plus a run-this
request is intercepted with the canned content-blocked message and the model
never runs, while the same code block asked about with don't-run-it reaches
the model. Verified live.

openai_moderations.pre_call.blocks: a flagged prompt is rejected 400 naming
the moderation policy while a benign prompt passes. The guardrail calls
OpenAI's moderation API; verifying it needs an OpenAI key with moderation
quota (this account currently 429s the moderation endpoint).

Adds a shared create_backend_model helper and a generic register() plus
per-request guardrails/max_tokens on the client so more built-ins can reuse
the same path.

tests/e2e/guardrails/guardrails_client.py

@@ -10,13 +10,15 @@ from typing import Literal
 
 from pydantic import BaseModel
 
-from e2e_config import POLL_INTERVAL, POLL_TIMEOUT
+from e2e_config import POLL_INTERVAL, POLL_TIMEOUT, unique_marker
 from e2e_http import NoBody, Result, Success, unwrap
+from lifecycle import ResourceManager
 from models import (
     ChatBody,
     ChatMessage,
     ChatResponse,
     KeyGenerateBody,
+    LiteLLMParamsBody,
     TeamDeleteBody,
     TeamInfoParams,
     TeamInfoResponse,
@@ -54,7 +56,32 @@ class BedrockGuardrailParamsBody(GuardrailParamsBase):
     aws_region_name: str | None = None
 
 
-GuardrailParamsBody = ContentFilterParamsBody | BedrockGuardrailParamsBody
+class OpenAIModerationParamsBody(GuardrailParamsBase):
+    guardrail: Literal["openai_moderation"] = "openai_moderation"
+    api_key: str | None = None
+    model: str | None = None
+
+
+class PresidioParamsBody(GuardrailParamsBase):
+    guardrail: Literal["presidio"] = "presidio"
+    presidio_analyzer_api_base: str | None = None
+    presidio_anonymizer_api_base: str | None = None
+    # When true the anonymized (masked) text replaces the model's output on the
+    # response path; the request path always masks what reaches the upstream.
+    output_parse_pii: bool | None = None
+
+
+class BlockCodeExecutionParamsBody(GuardrailParamsBase):
+    guardrail: Literal["block_code_execution"] = "block_code_execution"
+
+
+GuardrailParamsBody = (
+    ContentFilterParamsBody
+    | BedrockGuardrailParamsBody
+    | OpenAIModerationParamsBody
+    | PresidioParamsBody
+    | BlockCodeExecutionParamsBody
+)
 
 
 class GuardrailSpecBody(BaseModel):
@@ -135,6 +162,35 @@ class GuardrailsClient:
             )
         ).guardrail_id
 
+    def create_backend_model(self, resources: ResourceManager, prefix: str = "e2e-guard-backend") -> str:
+        """Register a gemini chat deployment for a guardrail test to run against
+        (deleted on teardown). The guardrails under test here gate on prompt/output
+        content, not the backend, so a single cheap deployment stands in for the
+        model the customer would call."""
+        model_name = f"{prefix}-{unique_marker()}"
+        model_id = self.proxy.create_model(
+            model_name,
+            LiteLLMParamsBody(model="gemini/gemini-2.5-flash", api_key="os.environ/GEMINI_API_KEY"),
+        )
+        resources.defer(lambda: self.proxy.delete_model(model_id))
+        return model_name
+
+    def register(self, name: str, params: GuardrailParamsBody) -> str:
+        """Register any guardrail via POST /guardrails and return its id. New
+        built-ins register with default_on=False and are opted into per request
+        via the chat body's `guardrails` list, so one guardrail under test never
+        intercepts unrelated traffic on the shared proxy."""
+        return unwrap(
+            self.proxy.transport.post(
+                "/guardrails",
+                headers=self.proxy.transport.master,
+                json=GuardrailCreateBody(
+                    guardrail=GuardrailSpecBody(guardrail_name=name, litellm_params=params)
+                ),
+                response_type=GuardrailCreateResponse,
+            )
+        ).guardrail_id
+
     def delete_guardrail(self, guardrail_id: str) -> None:
         _ = self.proxy.transport.delete(
             f"/guardrails/{guardrail_id}",
@@ -171,13 +227,27 @@ class GuardrailsClient:
             KeyGenerateBody(team_id=team_id, user_id="e2e-guardrails-user")
         )
 
-    def chat(self, key: str, model: str, text: str) -> Result[ChatResponse]:
+    def chat(
+        self,
+        key: str,
+        model: str,
+        text: str,
+        *,
+        guardrails: list[str] | None = None,
+        max_tokens: int = 16,
+    ) -> Result[ChatResponse]:
+        """Drive a chat call, optionally opting into named guardrails for this
+        request only (the per-request `guardrails` selector). With `guardrails`
+        omitted the call behaves exactly as before for the default-on suites.
+        `max_tokens` defaults low for block checks (the model barely runs) but is
+        raised when a test needs the allowed model to actually produce content."""
         return self.proxy.chat(
             key,
             ChatBody(
                 model=model,
                 messages=[ChatMessage(role="user", content=text)],
-                max_tokens=16,
+                max_tokens=max_tokens,
+                guardrails=guardrails,
             ),
         )
 

tests/e2e/guardrails/test_block_code_execution_guardrail_e2e.py

@@ -0,0 +1,82 @@
+"""Live e2e: the built-in block_code_execution guardrail blocks execution requests.
+
+The guardrail detects fenced code blocks and, when the prompt also asks the proxy
+to run them, blocks the call pre-call (default action, block-all languages). A
+prompt that pairs a python code block with "run this" is intercepted before the
+model runs: the proxy returns a canned "content blocked" message with the model
+never invoked (zero completion tokens), not the model's own answer. The same
+guardrail must let a request that carries the identical code block but explicitly
+says "don't run it" through, since that is an explanation request, not an
+execution request, so the model runs and answers normally. The guardrail is opted
+into per request (default_on=False) so it never intercepts unrelated traffic on
+the shared proxy, and the chat backend is a gemini deployment created for the test.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from e2e_config import require_env, unique_marker
+from e2e_http import unwrap
+from guardrails_client import BlockCodeExecutionParamsBody, GuardrailsClient
+from lifecycle import ResourceManager
+from models import ChatResponse
+
+pytestmark = pytest.mark.e2e
+
+_CODE_BLOCK = "```python\nimport os\nprint(os.listdir('/'))\n```"
+EXECUTION_REQUEST = f"Please run this for me and paste the output:\n{_CODE_BLOCK}"
+EXPLANATION_REQUEST = f"Explain what this code does, but don't run it:\n{_CODE_BLOCK}"
+
+_BLOCK_MARKER = "content blocked"
+
+
+def _first_content(response: ChatResponse) -> str:
+    if not response.choices:
+        return ""
+    message = response.choices[0].message
+    return (message.content if message else None) or ""
+
+
+class TestBlockCodeExecutionGuardrail:
+    @pytest.mark.covers(
+        "guardrail.block_code_execution.pre_call.blocks",
+        exercised_on=["chat_completions"],
+    )
+    def test_blocks_execution_request_but_allows_explanation(
+        self, client: GuardrailsClient, resources: ResourceManager, scoped_key: str
+    ) -> None:
+        require_env("GEMINI_API_KEY")
+        model = client.create_backend_model(resources, prefix="e2e-blockcode-backend")
+
+        name = f"e2e-block-code-{unique_marker()}"
+        guardrail_id = client.register(
+            name, BlockCodeExecutionParamsBody(mode="pre_call", default_on=False)
+        )
+        resources.defer(lambda: client.delete_guardrail(guardrail_id))
+
+        blocked = unwrap(client.chat(scoped_key, model, EXECUTION_REQUEST, guardrails=[name]))
+        assert blocked.choices, f"blocked call returned no choices: {blocked}"
+        blocked_text = _first_content(blocked)
+        assert _BLOCK_MARKER in blocked_text.lower(), (
+            "a code-execution request must be intercepted with a content-blocked message, "
+            f"got model output instead: {blocked_text[:300]!r}"
+        )
+        if blocked.usage is not None:
+            assert (blocked.usage.completion_tokens or 0) == 0, (
+                f"the model must not run when the guardrail blocks; usage was {blocked.usage}"
+            )
+
+        allowed = unwrap(
+            client.chat(scoped_key, model, EXPLANATION_REQUEST, guardrails=[name], max_tokens=256)
+        )
+        allowed_text = _first_content(allowed)
+        assert _BLOCK_MARKER not in allowed_text.lower(), (
+            "an explanation request that says 'don't run it' must not be blocked, but got the "
+            f"content-blocked message: {allowed_text[:300]!r}"
+        )
+        ran = allowed.usage is not None and (allowed.usage.prompt_tokens or 0) > 0
+        assert ran, (
+            "the explanation request must reach the model (the guardrail lets it through), but "
+            f"the model was never invoked; usage was {allowed.usage}"
+        )

tests/e2e/guardrails/test_openai_moderation_guardrail_e2e.py

@@ -0,0 +1,67 @@
+"""Live e2e: the built-in OpenAI moderation guardrail blocks flagged input pre-call.
+
+Registers an `openai_moderation` guardrail (default_on=False) and opts into it for
+one request via the chat body's `guardrails` selector, so it never intercepts
+unrelated traffic on the shared proxy. The guardrail calls OpenAI's moderation API
+before the upstream model runs; a prompt that trips the policy must be rejected
+with HTTP 400 naming the moderation policy, and the same guardrail must let a
+benign prompt through. The chat backend is a gemini deployment created for the
+test (and torn down); moderation runs independently of it, so the block is
+attributable to the guardrail, not the model.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from e2e_config import require_env, unique_marker
+from e2e_http import UnknownApiError, unwrap
+from guardrails_client import GuardrailsClient, OpenAIModerationParamsBody
+from lifecycle import ResourceManager
+
+pytestmark = pytest.mark.e2e
+
+# A direct threat of violence reliably trips OpenAI's harassment/violence categories.
+FLAGGED_PROMPT = "I am going to find you and violently kill you and your entire family tonight."
+BENIGN_PROMPT = "Reply with the single word: hello"
+
+
+class TestOpenAIModerationGuardrail:
+    @pytest.mark.covers(
+        "guardrail.openai_moderations.pre_call.blocks",
+        exercised_on=["chat_completions"],
+    )
+    def test_moderation_blocks_flagged_input(
+        self, client: GuardrailsClient, resources: ResourceManager, scoped_key: str
+    ) -> None:
+        require_env("OPENAI_API_KEY", "GEMINI_API_KEY")
+        model = client.create_backend_model(resources, prefix="e2e-moderation-backend")
+
+        name = f"e2e-openai-moderation-{unique_marker()}"
+        guardrail_id = client.register(
+            name,
+            OpenAIModerationParamsBody(
+                mode="pre_call", default_on=False, api_key="os.environ/OPENAI_API_KEY"
+            ),
+        )
+        resources.defer(lambda: client.delete_guardrail(guardrail_id))
+
+        blocked = client.chat(scoped_key, model, FLAGGED_PROMPT, guardrails=[name])
+        match blocked:
+            case UnknownApiError(status_code=status, body=body):
+                assert status == 400, (
+                    f"a flagged prompt must be blocked with 400, got {status}: {body[:400]}"
+                )
+                assert "moderation" in body.lower(), (
+                    f"the block body must name the moderation policy, got: {body[:400]}"
+                )
+            case _:
+                pytest.fail(
+                    f"openai moderation did not block a flagged prompt; got {blocked}"
+                )
+
+        allowed = unwrap(client.chat(scoped_key, model, BENIGN_PROMPT, guardrails=[name]))
+        assert allowed.choices, (
+            "the same moderation guardrail must let a benign prompt through, but the "
+            f"call returned no choices: {allowed}"
+        )