feat(guardrails): add logging_only_scope to observe input, output, or both (#43695)

* feat(guardrails): add logging_only_scope to observe one direction

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* chore(guardrails): remove unrelated test churn

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): allow native lifecycle logging-only scope

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* chore(ui): regenerate api types for logging_only_scope

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): remove callbacks when scope validation fails

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): keep output-only scans when request copy fails

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* feat(ui): configure logging_only_scope on guardrails

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): keep guardrails enforcing when logging_only_scope is invalid at load

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* chore(ui): avoid inline guardrail test fixtures

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): support directional scope in PATCH and provider UI

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(ui): format guardrail files for frontend lint

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(ui): reset unsupported directional scope selections

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): normalize logging-only scope and sanitize warning logs

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(ui): normalize scope in shared guardrail field

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(api): sync guardrail schema artifacts

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): tolerate invalid stored logging-only scopes

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(guardrails): add logging_only_scope integration audit cells

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(guardrails): fix scope test lint issues

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(guardrails): extend K4 chaos test timeout

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): restore stored guardrail row verbatim on rejected patch

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): satisfy collection lint budgets

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): validate masked params through TypeAdapter

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): tolerate invalid stored params on reads

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): keep constructor coercion in tolerant params parser

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* feat(ui): edit logging_only_scope in the custom code guardrail modal

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor(ui): extract custom code logging-only scope control

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): keep no-op PUT and logging_only scan-failure semantics stable

Three regressions from the logging_only_scope feature, fixed while keeping
the input/output/both selection working:

1. PUT with byte-identical litellm_params no longer forces a teardown +
   re-init of the live callback. reject_invalid_logging_only_scope now
   only validates; re-init still happens exactly when params/name change.
   Previously a description-only PUT re-appended the callback at the END
   of litellm.callbacks, reordering guardrails: with a BLOCK guardrail
   created before a MASK one, the mask started winning and blocked
   requests started succeeding (400 -> PUT 200 -> 200). Invalid unchanged
   scopes are still rejected with 422 without touching the live instance.

2. CustomGuardrail._scan_logged_call no longer swallows input-scan
   exceptions per branch: a raising or BLOCKED input scan aborts the
   logging_only hook again (one policy call, one verdict) for guardrails
   that never selected a logging_only_scope. Explicit input/output/both
   selection keeps selecting which scans run.

3. A failed PATCH now rolls the DB row back to exactly the stored raw
   litellm_params (a legacy 4-key row stays 4 keys) instead of expanding
   it to a full LitellmParams dump; pinned with a test. This matches the
   PUT rollback shape and is the faithful rollback.

* refactor(guardrails): type directional-scope provider list as a tuple

* fix(lint): stay within the basedpyright budget

- drop a Final annotation assigned inside the validation loop
  (reportGeneralTypeIssues over ceiling by one)
- rename the PR-introduced _configured_event_hooks to public
  configured_event_hooks; its cross-module import added the two
  reportPrivateUsage errors that pushed the rule over its ceiling

* fix(guardrails): keep the response verdict for explicit logging_only_scope=both on input-scan failure

Greptile P1: an explicitly configured both-direction observer asked for a
verdict on each direction, so a failed request scan must not silently drop
the response verdict. The implicit default (logging_only_scope None) keeps
the abort semantics of a logging_only hook whose scan raised, which is the
base behavior the earlier fix restored.

Also drops two test comments that restated their assertions (Greptile P2).

* test(guardrails): align logging_only_scope integration rows with abort-on-input-failure and encrypted params

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(guardrails): keep the response scan when the request copy fails for explicit both scope

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: yucheng <yucheng@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-07 14:03:33 -07:00 • committed by GitHub
parent eb9355d11a
commit 127278f951
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
26 changed files with 7183 additions and 78 deletions

View file

@ -24,6 +24,7 @@ from litellm.types.guardrails import (
DynamicGuardrailParams,
GuardrailEventHooks,
LitellmParams,
LoggingOnlyScope,
Mode,
)
from litellm.types.llms.openai import AllMessageValues
@ -180,6 +181,7 @@ class CustomGuardrail(CustomLogger):
use_native_lifecycle_hooks: ClassVar[bool] = False
records_own_guardrail_information: ClassVar[bool] = False
logging_only_scope: LoggingOnlyScope | None
timeout: float | httpx.Timeout | None = None
@ -256,6 +258,7 @@ class CustomGuardrail(CustomLogger):
self.run_in_parallel: bool = run_in_parallel
self.scan_raw_request: bool = scan_raw_request
self.only_scan_new_messages: bool = only_scan_new_messages
self.logging_only_scope = None
if timeout is not None:
self.timeout = timeout
@ -817,6 +820,13 @@ class CustomGuardrail(CustomLogger):
def uses_apply_guardrail_interface(self) -> bool:
return type(self).apply_guardrail is not CustomGuardrail.apply_guardrail
@classmethod
def supports_logging_only_scope(cls) -> bool:
return (
cls.apply_guardrail is not CustomGuardrail.apply_guardrail
and cls.async_logging_hook is CustomGuardrail.async_logging_hook
)
def _deployment_hook_target(self) -> "CustomLogger":
if not self.uses_apply_guardrail_interface() or self.use_native_lifecycle_hooks:
return self
@ -948,7 +958,7 @@ class CustomGuardrail(CustomLogger):
result: object,
call_type: str,
) -> tuple[dict, object]: # mutable-ok: CustomLogger.async_logging_hook contract
"""logging_only: run apply_guardrail on copies of the logged request/response and record the verdict."""
"""logging_only: scan copies of the logged request and/or response according to logging_only_scope."""
from litellm.llms import get_guardrail_translation_mapping
if not self.uses_apply_guardrail_interface():
@ -995,6 +1005,28 @@ class CustomGuardrail(CustomLogger):
"standard_logging_object": {**standard_logging_object, "guardrail_information": [*existing, *entries]},
}, result
def _copy_scratch_request_fields(
self,
kwargs: Mapping[str, object],
) -> tuple[object, object] | None:
optional_params: Final = kwargs.get("optional_params")
try:
return (
copy.deepcopy(kwargs.get("messages") or kwargs.get("input")),
copy.deepcopy(optional_params.get("tools") if isinstance(optional_params, Mapping) else None),
)
except Exception as e:
if self.logging_only_scope == "output":
return None
if self.logging_only_scope == "both":
verbose_logger.warning(
"Guardrail %s: logging_only request copy failed, skipping request scan: %s",
self.guardrail_name,
e,
)
return None
raise
async def _scan_logged_call(
self,
kwargs: dict, # mutable-ok: CustomLogger.async_logging_hook contract
@ -1003,18 +1035,25 @@ class CustomGuardrail(CustomLogger):
output_translation: "BaseTranslation",
scratch_metadata: dict, # mutable-ok: apply_guardrail records its verdict into request metadata
) -> None:
optional_params: Final = kwargs.get("optional_params") or {}
scratch_input: Final = copy.deepcopy(kwargs.get("messages") or kwargs.get("input"))
scratch_fields: Final = self._copy_scratch_request_fields(kwargs)
scratch_input, scratch_tools = scratch_fields or (None, None)
scratch_request: Final = {
"model": kwargs.get("model"),
"messages": scratch_input,
"input": scratch_input,
"tools": copy.deepcopy(optional_params.get("tools")),
"tools": scratch_tools,
"litellm_call_id": kwargs.get("litellm_call_id"),
"metadata": scratch_metadata,
}
await translation.process_input_messages(data=scratch_request, guardrail_to_apply=self)
if response is None:
if self.logging_only_scope != "output" and scratch_fields is not None:
if self.logging_only_scope == "both":
try:
await translation.process_input_messages(data=scratch_request, guardrail_to_apply=self)
except Exception as e: # noqa: BLE001 # one direction's scan failure must not drop the other direction's verdict
verbose_logger.warning("Guardrail %s: logging_only scan raised: %s", self.guardrail_name, e)
else:
await translation.process_input_messages(data=scratch_request, guardrail_to_apply=self)
if response is None or self.logging_only_scope == "input":
return
await output_translation.process_output_response(
response=copy.deepcopy(response), guardrail_to_apply=self, request_data=scratch_request

View file

@ -11572,6 +11572,23 @@
"description": "Google Cloud location/region (e.g., us-central1)",
"title": "Location"
},
"logging_only_scope": {
"anyOf": [
{
"enum": [
"input",
"output",
"both"
],
"type": "string"
},
{
"type": "null"
}
],
"description": "which direction a logging_only scan observes: 'input' (request), 'output' (response), or 'both' (default). Only applies to mode logging_only; pre_call/post_call on the same guardrail keep blocking.",
"title": "Logging Only Scope"
},
"mask_request_content": {
"anyOf": [
{
@ -12727,6 +12744,77 @@
"title": "GuardrailSubmissionSummary",
"type": "object"
},
"GuardrailUIAddGuardrailSettings": {
"properties": {
"content_filter_settings": {
"anyOf": [
{
"additionalProperties": true,
"type": "object"
},
{
"type": "null"
}
],
"title": "Content Filter Settings"
},
"pii_entity_categories": {
"items": {
"$ref": "#/components/schemas/PiiEntityCategoryMap"
},
"title": "Pii Entity Categories",
"type": "array"
},
"providers_without_directional_logging_only_scope": {
"items": {
"type": "string"
},
"title": "Providers Without Directional Logging Only Scope",
"type": "array"
},
"supported_actions": {
"items": {
"type": "string"
},
"title": "Supported Actions",
"type": "array"
},
"supported_entities": {
"items": {
"type": "string"
},
"title": "Supported Entities",
"type": "array"
},
"supported_modes": {
"items": {
"type": "string"
},
"title": "Supported Modes",
"type": "array"
},
"supported_modes_by_provider": {
"additionalProperties": {
"items": {
"type": "string"
},
"type": "array"
},
"title": "Supported Modes By Provider",
"type": "object"
}
},
"required": [
"supported_entities",
"supported_actions",
"supported_modes",
"supported_modes_by_provider",
"providers_without_directional_logging_only_scope",
"pii_entity_categories"
],
"title": "GuardrailUIAddGuardrailSettings",
"type": "object"
},
"HTTPValidationError": {
"properties": {
"detail": {
@ -13822,6 +13910,23 @@
"description": "Google Cloud location/region (e.g., us-central1)",
"title": "Location"
},
"logging_only_scope": {
"anyOf": [
{
"enum": [
"input",
"output",
"both"
],
"type": "string"
},
{
"type": "null"
}
],
"description": "which direction a logging_only scan observes: 'input' (request), 'output' (response), or 'both' (default). Only applies to mode logging_only; pre_call/post_call on the same guardrail keep blocking.",
"title": "Logging Only Scope"
},
"mask": {
"anyOf": [
{
@ -14882,6 +14987,27 @@
"title": "PiiAction",
"type": "string"
},
"PiiEntityCategoryMap": {
"properties": {
"category": {
"title": "Category",
"type": "string"
},
"entities": {
"items": {
"type": "string"
},
"title": "Entities",
"type": "array"
}
},
"required": [
"category",
"entities"
],
"title": "PiiEntityCategoryMap",
"type": "object"
},
"PiiEntityType": {
"enum": [
"CREDIT_CARD",
@ -16323,7 +16449,9 @@
"200": {
"content": {
"application/json": {
"schema": {}
"schema": {
"$ref": "#/components/schemas/GuardrailUIAddGuardrailSettings"
}
}
},
"description": "Successful Response"

View file

@ -36,9 +36,11 @@ from litellm.proxy.guardrails.guardrail_hooks.custom_code.sandbox import (
)
from litellm.proxy.guardrails.guardrail_registry import (
GuardrailRegistry,
configured_event_hooks,
contains_encrypted_marker,
decrypt_guardrail_litellm_params,
encrypt_guardrail_litellm_params,
parse_tolerant_litellm_params,
)
from litellm.proxy.guardrails.usage_endpoints import router as guardrails_usage_router
from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view
@ -283,7 +285,13 @@ async def list_guardrails_v2(
number_of_asterisks=4,
)
masked_litellm_params = (
BaseLitellmParams.model_validate(masked_litellm_params_dict) if masked_litellm_params_dict else None
parse_tolerant_litellm_params(
masked_litellm_params_dict,
guardrail.get("guardrail_name") or "Unknown",
params_model=BaseLitellmParams,
)
if masked_litellm_params_dict
else None
)
guardrail_configs.append(
GuardrailInfoResponse(
@ -324,7 +332,11 @@ async def list_guardrails_v2(
number_of_asterisks=4,
)
masked_in_memory_litellm_params_typed = (
BaseLitellmParams.model_validate(masked_in_memory_litellm_params)
parse_tolerant_litellm_params(
masked_in_memory_litellm_params,
guardrail.get("guardrail_name") or "Unknown",
params_model=BaseLitellmParams,
)
if masked_in_memory_litellm_params
else None
)
@ -425,7 +437,11 @@ async def create_guardrail(
guardrail_id: Final = result.get("guardrail_id", "Unknown")
try:
IN_MEMORY_GUARDRAIL_HANDLER.initialize_guardrail(guardrail=cast(Guardrail, result), source="db")
IN_MEMORY_GUARDRAIL_HANDLER.initialize_guardrail(
guardrail=cast(Guardrail, result),
source="db",
reject_invalid_logging_only_scope=True,
)
verbose_proxy_logger.info(
"Immediate sync: Successfully initialized guardrail '%s' (ID: %s)", guardrail_name, guardrail_id
)
@ -550,7 +566,10 @@ async def update_guardrail(
guardrail_name: Final = result.get("guardrail_name", "Unknown")
try:
IN_MEMORY_GUARDRAIL_HANDLER.sync_guardrail_from_db(guardrail=cast(Guardrail, result))
IN_MEMORY_GUARDRAIL_HANDLER.sync_guardrail_from_db(
guardrail=cast(Guardrail, result),
reject_invalid_logging_only_scope=True,
)
verbose_proxy_logger.info(
"Immediate sync: Successfully updated guardrail '%s' (ID: %s)", guardrail_name, guardrail_id
)
@ -1240,19 +1259,35 @@ async def patch_guardrail(
# Update litellm_params if default_on is provided or pii_entities_config is provided
existing_litellm_params: Final = _as_str_object_mapping(dict(existing_guardrail.get("litellm_params", {})))
litellm_params = LitellmParams(**existing_litellm_params)
if request.litellm_params is not None:
requested_litellm_params: Final = request.litellm_params.model_dump(exclude_unset=True)
litellm_params_dict: Final = litellm_params.model_dump(exclude_unset=True)
litellm_params_dict.update(requested_litellm_params)
merged_litellm_params: Final = _as_str_object_mapping(litellm_params_dict)
try:
litellm_params = LitellmParams(**merged_litellm_params)
except ValidationError as validation_error:
raise HTTPException(
status_code=422,
detail=f"Invalid guardrail configuration, update rejected: {validation_error}",
) from validation_error
current_litellm_params: Final = parse_tolerant_litellm_params(
existing_litellm_params,
existing_guardrail.get("guardrail_name") or "Unknown",
)
requested_litellm_params: Final[Mapping[str, object]] = (
MappingProxyType(request.litellm_params.model_dump(exclude_unset=True))
if request.litellm_params is not None
else MappingProxyType({})
)
merged_litellm_params: Final = _as_str_object_mapping(
MappingProxyType({**current_litellm_params.model_dump(exclude_unset=True), **requested_litellm_params})
)
try:
parsed_litellm_params: Final = LitellmParams(**merged_litellm_params)
except ValidationError as validation_error:
raise HTTPException(
status_code=422,
detail=f"Invalid guardrail configuration, update rejected: {validation_error}",
) from validation_error
clear_stored_scope: Final = (
"logging_only_scope" not in requested_litellm_params
and parsed_litellm_params.logging_only_scope is not None
and GuardrailEventHooks.logging_only.value not in configured_event_hooks(parsed_litellm_params.mode)
)
litellm_params: Final = (
LitellmParams(**MappingProxyType({**merged_litellm_params, "logging_only_scope": None}))
if clear_stored_scope
else parsed_litellm_params
)
# Update guardrail_info if provided
guardrail_info: Final = (
@ -1281,6 +1316,7 @@ async def patch_guardrail(
try:
IN_MEMORY_GUARDRAIL_HANDLER.sync_guardrail_from_db(
guardrail=guardrail,
reject_invalid_logging_only_scope="logging_only_scope" in requested_litellm_params,
)
verbose_proxy_logger.info(
"Immediate sync: Successfully updated guardrail '%s' (ID: %s)", guardrail_name, guardrail_id
@ -1294,15 +1330,7 @@ async def patch_guardrail(
# the caller instead of a misleading 200.
await GUARDRAIL_REGISTRY.update_guardrail_in_db(
guardrail_id=guardrail_id,
guardrail=Guardrail(
guardrail_id=guardrail_id,
guardrail_name=existing_guardrail.get("guardrail_name") or "",
litellm_params=LitellmParams(**existing_litellm_params),
guardrail_info=existing_guardrail.get(
"guardrail_info",
{},
),
),
guardrail=existing_guardrail,
prisma_client=prisma_client,
)
raise HTTPException(
@ -1404,7 +1432,13 @@ async def get_guardrail_info(guardrail_id: str):
number_of_asterisks=4,
)
masked_litellm_params = (
BaseLitellmParams.model_validate(masked_litellm_params_dict) if masked_litellm_params_dict else None
parse_tolerant_litellm_params(
masked_litellm_params_dict,
result.get("guardrail_name") or "Unknown",
params_model=BaseLitellmParams,
)
if masked_litellm_params_dict
else None
)
return GuardrailInfoResponse(
@ -1427,7 +1461,7 @@ async def get_guardrail_info(guardrail_id: str):
tags=["Guardrails"],
dependencies=[Depends(user_api_key_auth)],
)
async def get_guardrail_ui_settings():
async def get_guardrail_ui_settings() -> GuardrailUIAddGuardrailSettings:
"""
Get the UI settings for the guardrails
@ -1461,12 +1495,18 @@ async def get_guardrail_ui_settings():
# above; it only runs on pre_call.
{SupportedGuardrailIntegrations.HIDE_SECRETS.value: [GuardrailEventHooks.pre_call.value]}
)
providers_without_directional_logging_only_scope: Final = tuple(
provider
for provider, guardrail_class in guardrail_class_registry.items()
if not guardrail_class.supports_logging_only_scope()
)
return GuardrailUIAddGuardrailSettings(
supported_entities=[entity.value for entity in PiiEntityType],
supported_actions=[action.value for action in PiiAction],
supported_modes=[mode.value for mode in GuardrailEventHooks],
supported_modes_by_provider=supported_modes_by_provider,
providers_without_directional_logging_only_scope=providers_without_directional_logging_only_scope,
pii_entity_categories=category_maps,
content_filter_settings={
"prebuilt_patterns": get_pattern_metadata(),

View file

@ -100,7 +100,7 @@ _MCP_EVENT_HOOKS: Final = frozenset(
)
def _configured_event_hooks(mode: str | list[str] | Mode) -> tuple[str, ...]:
def configured_event_hooks(mode: str | list[str] | Mode) -> tuple[str, ...]:
if isinstance(mode, str):
return (mode,)
if isinstance(mode, list):
@ -114,7 +114,7 @@ def _configured_event_hooks(mode: str | list[str] | Mode) -> tuple[str, ...]:
def _is_mcp_only_mode(mode: str | list[str] | Mode) -> bool:
hooks: Final = _configured_event_hooks(mode)
hooks: Final = configured_event_hooks(mode)
return bool(hooks) and all(hook in _MCP_EVENT_HOOKS for hook in hooks)

View file

@ -6,7 +6,8 @@ import os
from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
from datetime import datetime, timezone
from itertools import chain, count
from typing import TYPE_CHECKING, Final, Literal, Optional, Protocol, TypeAlias, cast
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, Literal, Optional, Protocol, TypeAlias, TypeVar, cast
from pydantic import BaseModel, TypeAdapter, ValidationError
@ -58,6 +59,7 @@ from .guardrail_hooks.llm_as_a_judge import (
initialize_guardrail as initialize_llm_as_a_judge,
)
from .guardrail_initializers import (
configured_event_hooks,
initialize_bedrock,
initialize_hide_secrets,
initialize_lakera,
@ -572,9 +574,46 @@ def _as_callback_tuple(
return (initialized,)
def _configure_callback_scoping(
def _logging_only_scope_error(
custom_guardrail_callback: CustomGuardrail, guardrail_name: str, litellm_params: LitellmParams
) -> str | None:
logging_only_scope: Final = litellm_params.logging_only_scope
if logging_only_scope is not None and GuardrailEventHooks.logging_only.value not in configured_event_hooks(
litellm_params.mode
):
return (
f"Guardrail {guardrail_name}: logging_only_scope is set, but mode does not include logging_only, "
"so it would never apply. Add logging_only to mode or remove logging_only_scope."
)
if logging_only_scope in ("input", "output") and not custom_guardrail_callback.supports_logging_only_scope():
return (
f"Guardrail {guardrail_name}: logging_only_scope={logging_only_scope!r} is not supported by this "
"guardrail, whose logging_only hook scans on its own. Remove logging_only_scope."
)
return None
def _configure_callback_scoping(
custom_guardrail_callback: CustomGuardrail,
guardrail_name: str,
litellm_params: LitellmParams,
*,
reject_invalid_logging_only_scope: bool = False,
) -> None:
logging_only_scope: Final = litellm_params.logging_only_scope
logging_only_scope_error: Final = _logging_only_scope_error(
custom_guardrail_callback, guardrail_name, litellm_params
)
if logging_only_scope_error is not None:
if reject_invalid_logging_only_scope:
raise ValueError(logging_only_scope_error)
verbose_proxy_logger.error(
"%s Ignoring logging_only_scope; the guardrail keeps its configured mode.",
logging_only_scope_error.replace("\r", "").replace("\n", ""),
)
custom_guardrail_callback.logging_only_scope = None
else:
custom_guardrail_callback.logging_only_scope = logging_only_scope
for scoping_param in (
"skip_system_message_in_guardrail",
"skip_tool_message_in_guardrail",
@ -597,6 +636,28 @@ def _configure_callback_scoping(
_apply_configured_bool_overrides(custom_guardrail_callback, litellm_params)
_ParamsT = TypeVar("_ParamsT", bound=BaseModel)
def parse_tolerant_litellm_params(
litellm_params_data: Mapping[str, object],
guardrail_name: str,
params_model: type[_ParamsT] = LitellmParams,
) -> _ParamsT:
try:
return params_model(**litellm_params_data)
except ValidationError as validation_error:
if any(tuple(error["loc"]) != ("logging_only_scope",) for error in validation_error.errors()):
raise
verbose_proxy_logger.error(
"Guardrail %s: logging_only_scope=%r is not one of 'input', 'output' or 'both'. "
"Ignoring logging_only_scope; the guardrail keeps its configured mode.",
guardrail_name.replace("\r", "").replace("\n", ""),
str(litellm_params_data.get("logging_only_scope")).replace("\r", "").replace("\n", "")[:100],
)
return params_model(**MappingProxyType({**litellm_params_data, "logging_only_scope": None}))
class InMemoryGuardrailHandler:
"""
Class that handles initializing guardrails and adding them to the CallbackManager
@ -633,6 +694,8 @@ class InMemoryGuardrailHandler:
config_file_path: str | None = None,
llm_router: Optional["Router"] = None,
source: Literal["db", "config"] = "config",
*,
reject_invalid_logging_only_scope: bool = False,
) -> Guardrail | None:
"""
Initialize a guardrail from a dictionary and add it to the litellm callback manager
@ -653,7 +716,10 @@ class InMemoryGuardrailHandler:
verbose_proxy_logger.debug("litellm_params= %s", litellm_params_data)
if isinstance(litellm_params_data, dict):
litellm_params = LitellmParams(**litellm_params_data)
if reject_invalid_logging_only_scope:
litellm_params = LitellmParams(**litellm_params_data)
else:
litellm_params = parse_tolerant_litellm_params(litellm_params_data, guardrail["guardrail_name"])
else:
litellm_params = litellm_params_data
@ -679,8 +745,18 @@ class InMemoryGuardrailHandler:
config_file_path=config_file_path,
llm_router=llm_router,
)
for custom_guardrail_callback in created_callbacks:
_configure_callback_scoping(custom_guardrail_callback, guardrail["guardrail_name"], litellm_params)
try:
for custom_guardrail_callback in created_callbacks:
_configure_callback_scoping(
custom_guardrail_callback,
guardrail["guardrail_name"],
litellm_params,
reject_invalid_logging_only_scope=reject_invalid_logging_only_scope,
)
except Exception:
for custom_guardrail_callback in created_callbacks:
litellm.logging_callback_manager.remove_callback_from_all_lists(custom_guardrail_callback)
raise
parsed_guardrail: Final = Guardrail(
guardrail_id=guardrail.get("guardrail_id"),
@ -729,6 +805,28 @@ class InMemoryGuardrailHandler:
siblings: Final = self.guardrail_id_to_sibling_callbacks.get(guardrail_id, ())
return (() if primary is None else (primary,)) + siblings
def _reject_invalid_logging_only_scope(self, guardrail_id: str, guardrail: Guardrail) -> None:
"""
Strictly validate logging_only_scope on a row whose params are otherwise
unchanged, without rebuilding the live callback.
API write paths send the whole object, so an invalid scope must still be
rejected even when the write changed nothing else. But an unchanged row
must not force a teardown + re-append: initialize_guardrail appends the
rebuilt callback at the END of litellm.callbacks, so a no-op PUT would
reorder guardrails and change which one wins between a BLOCK and a MASK
guardrail over the same content.
"""
params: Final = guardrail.get("litellm_params")
if not isinstance(params, (dict, LitellmParams)):
return
litellm_params: Final = LitellmParams(**params) if isinstance(params, dict) else params
guardrail_name: Final = guardrail.get("guardrail_name", "Unknown")
for custom_guardrail_callback in self._tracked_callbacks(guardrail_id):
scope_error = _logging_only_scope_error(custom_guardrail_callback, guardrail_name, litellm_params)
if scope_error is not None:
raise ValueError(scope_error)
def initialize_custom_guardrail(
self,
guardrail: Guardrail,
@ -788,6 +886,8 @@ class InMemoryGuardrailHandler:
guardrail_id: str,
guardrail: Guardrail,
source: Literal["db", "config"] = "db",
*,
reject_invalid_logging_only_scope: bool = False,
) -> None:
"""
Update a guardrail in memory: a changed name or litellm_params rebuilds the
@ -796,8 +896,14 @@ class InMemoryGuardrailHandler:
"""
updated_guardrail: Final = cast(Guardrail, {**guardrail, "guardrail_id": guardrail_id})
if self._has_guardrail_params_changed(guardrail_id, updated_guardrail):
self.reinitialize_guardrail(guardrail=updated_guardrail, source=source)
self.reinitialize_guardrail(
guardrail=updated_guardrail,
source=source,
reject_invalid_logging_only_scope=reject_invalid_logging_only_scope,
)
return
if reject_invalid_logging_only_scope:
self._reject_invalid_logging_only_scope(guardrail_id, updated_guardrail)
self.IN_MEMORY_GUARDRAILS[guardrail_id] = updated_guardrail
self._sources[guardrail_id] = source
@ -883,6 +989,7 @@ class InMemoryGuardrailHandler:
@staticmethod
def _normalize_litellm_params_for_comparison(
params: LitellmParams | Mapping[str, object] | None,
guardrail_name: str,
) -> Mapping[str, object] | None:
"""
Render litellm_params to a canonical dict so an in-memory LitellmParams and
@ -899,7 +1006,7 @@ class InMemoryGuardrailHandler:
return params.model_dump()
if isinstance(params, dict):
try:
return LitellmParams(**params).model_dump()
return parse_tolerant_litellm_params(params, guardrail_name).model_dump()
except ValidationError as e:
verbose_proxy_logger.warning(
"Could not normalize guardrail litellm_params for comparison; treating the guardrail as changed. Error: %s",
@ -922,8 +1029,12 @@ class InMemoryGuardrailHandler:
return True
# Compare litellm_params
existing_dict: Final = self._normalize_litellm_params_for_comparison(existing.get("litellm_params"))
new_dict: Final = self._normalize_litellm_params_for_comparison(new_guardrail.get("litellm_params"))
existing_dict: Final = self._normalize_litellm_params_for_comparison(
existing.get("litellm_params"), existing.get("guardrail_name", "Unknown")
)
new_dict: Final = self._normalize_litellm_params_for_comparison(
new_guardrail.get("litellm_params"), new_guardrail.get("guardrail_name", "Unknown")
)
# Compare and identify specific differences
changed_fields = {}
@ -949,6 +1060,8 @@ class InMemoryGuardrailHandler:
guardrail: Guardrail,
config_file_path: str | None = None,
source: Literal["db", "config"] = "config",
*,
reject_invalid_logging_only_scope: bool = False,
) -> Guardrail | None:
"""
Force re-initialization of a guardrail even if it exists in memory.
@ -978,7 +1091,12 @@ class InMemoryGuardrailHandler:
# instance instead of leaving the guardrail silently removed: a guardrail
# that was enforcing must never fail open because an update was bad.
try:
return self.initialize_guardrail(guardrail=guardrail, config_file_path=config_file_path, source=source)
return self.initialize_guardrail(
guardrail=guardrail,
config_file_path=config_file_path,
source=source,
reject_invalid_logging_only_scope=reject_invalid_logging_only_scope,
)
except Exception as init_error:
if previous_guardrail is not None:
verbose_proxy_logger.exception(
@ -1003,7 +1121,9 @@ class InMemoryGuardrailHandler:
)
if existing is None or db_params is None or not contains_encrypted_marker(db_params):
return guardrail
loaded_params: Final = self._normalize_litellm_params_for_comparison(existing.get("litellm_params"))
loaded_params: Final = self._normalize_litellm_params_for_comparison(
existing.get("litellm_params"), guardrail.get("guardrail_name", "Unknown")
)
verbose_proxy_logger.warning(
"Guardrail %s has litellm_params that do not decrypt with the current key; keeping the loaded values for "
"them. Restart the proxy if the master key was rotated.",
@ -1023,7 +1143,13 @@ class InMemoryGuardrailHandler:
}
)
def sync_guardrail_from_db(self, guardrail: Guardrail, config_file_path: str | None = None) -> Guardrail | None:
def sync_guardrail_from_db(
self,
guardrail: Guardrail,
config_file_path: str | None = None,
*,
reject_invalid_logging_only_scope: bool = False,
) -> Guardrail | None:
"""
Sync a guardrail from DB - initializes if new, re-initializes if changed.
DB values that do not decrypt with the current key keep the loaded guardrail's values.
@ -1044,8 +1170,12 @@ class InMemoryGuardrailHandler:
guardrail=synced,
config_file_path=config_file_path,
source="db",
reject_invalid_logging_only_scope=reject_invalid_logging_only_scope,
)
if reject_invalid_logging_only_scope:
self._reject_invalid_logging_only_scope(guardrail_id, synced)
# Params unchanged but the entry is still DB-backed; make sure the
# source marker reflects that even if it was previously set differently
# (e.g. a config entry whose UUID later collided with a DB row).

View file

@ -901,6 +901,8 @@ class ContentFilterConfigModel(LiteLLMBaseModel):
MCP_SECURITY_ON_VIOLATION: Final = frozenset({"block", "alert"})
LoggingOnlyScope = Literal["input", "output", "both"]
class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch update guardrails
api_key: str | None = Field(default=None, description="API key for the guardrail service")
@ -1142,6 +1144,14 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up
),
)
logging_only_scope: LoggingOnlyScope | None = Field(
default=None,
description=(
"which direction a logging_only scan observes: 'input' (request), 'output' (response), or 'both' "
"(default). Only applies to mode logging_only; pre_call/post_call on the same guardrail keep blocking."
),
)
@field_validator(
"mode",
"default_action",
@ -1310,6 +1320,7 @@ class GuardrailUIAddGuardrailSettings(LiteLLMBaseModel):
supported_actions: list[str]
supported_modes: list[str]
supported_modes_by_provider: dict[str, list[str]]
providers_without_directional_logging_only_scope: tuple[str, ...]
pii_entity_categories: list[PiiEntityCategoryMap]
content_filter_settings: dict[str, object] | None = None

File diff suppressed because it is too large Load diff

View file

@ -8,6 +8,7 @@ import threading
import uuid
from collections.abc import Callable, Iterator, Mapping
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
@ -1805,8 +1806,161 @@ def test_responses_pre_call_denial_stream_survives_worker_kill(gateway: Gateway,
for response in responses:
assert response.status_code == 200, response.text
assert response.headers["content-type"].startswith("text/event-stream"), response.text
@pytest.mark.parametrize(
("logging_only_scope", "scanned_directions"),
(("input", ("request",)), ("output", ("response",)), ("both", ("request", "response"))),
)
def test_logging_only_scope_observes_only_the_configured_direction_without_blocking(
gateway: Gateway, tmp_path: Path, logging_only_scope: str, scanned_directions: tuple[str, ...]
) -> None:
identity: Final = "guardrail" + uuid.uuid4().hex
prompt: Final = "synthetic observed prompt " + identity
reply: Final = "synthetic observed reply " + identity
texts_by_direction: Final = {"request": [prompt], "response": [reply]}
def guardrail(request: Request) -> Reply:
assert request.target == "/beta/litellm_basic_guardrail_api"
return Reply(body=json.dumps({"action": "BLOCKED", "blocked_reason": "synthetic observed denial"}).encode())
def provider(request: Request) -> Reply:
assert request.target == "/v1/chat/completions"
assert json.loads(request.body)["messages"] == [{"role": "user", "content": prompt}]
return Reply(
body=json.dumps(
{
"id": identity,
"object": "chat.completion",
"created": 1,
"model": "gpt-4o-mini",
"choices": [
{"index": 0, "message": {"role": "assistant", "content": reply}, "finish_reason": "stop"}
],
"usage": {"prompt_tokens": 9, "completion_tokens": 5, "total_tokens": 14},
}
).encode()
)
with wire_server(guardrail) as policy, wire_server(provider) as upstream:
config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
config["guardrails"] = [
{
"guardrail_name": identity,
"litellm_params": {
"guardrail": "generic_guardrail_api",
"mode": "logging_only",
"logging_only_scope": logging_only_scope,
"default_on": True,
"api_base": policy.url,
"api_key": "synthetic-guardrail-key",
},
}
]
path: Final = tmp_path / "logging-only-scope.yaml"
path.write_text(yaml.safe_dump(config))
with owned_proxy(gateway, tmp_path, {}, config=path) as candidate, candidate.scenario() as scenario:
model: Final = scenario.model(api_base=upstream.url + "/v1")
response: Final = candidate.request(
"POST", "/v1/chat/completions", {"model": model, "messages": [{"role": "user", "content": prompt}]}
)
assert response.status_code == 200, response.text
assert response.json()["choices"][0]["message"]["content"] == reply, response.text
assert len(upstream.drain()) == 1
rows: Final = eventually(
lambda: read_rows('SELECT metadata FROM "LiteLLM_SpendLogs" WHERE model_group=%s', (model,)),
lambda values: len(values) == 1,
seconds=70,
)
scans: Final = tuple(json.loads(scan.body) for scan in policy.drain())
assert [(scan["input_type"], scan["texts"]) for scan in scans] == [
(direction, texts_by_direction[direction]) for direction in scanned_directions
], scans
entries: Final = object_value(rows[0]["metadata"])["guardrail_information"]
assert isinstance(entries, list), rows[0]
assert [
(entry["guardrail_name"], entry["guardrail_mode"], entry["guardrail_status"])
for entry in map(object_value, entries)
] == [(identity, "logging_only", "guardrail_intervened")] * len(scanned_directions), entries
today: Final = datetime.now(timezone.utc).date().isoformat()
guardrail_id: Final = next(
object_value(row)["guardrail_id"]
for row in candidate.get("/v2/guardrails/list")["guardrails"]
if object_value(row)["guardrail_name"] == identity
)
detail: Final = eventually(
lambda: candidate.request(
"GET",
f"/guardrails/usage/detail/{guardrail_id}",
params={"start_date": today, "end_date": today},
).json(),
lambda body: body["requestsEvaluated"] >= len(scanned_directions),
seconds=30,
return_last_on_timeout=True,
)
assert detail["requestsEvaluated"] == len(scanned_directions), detail
@pytest.mark.parametrize("logging_only_scope", ("input", "Input"))
def test_logging_only_scope_literal_or_mode_mismatch_is_ignored_at_load_and_keeps_blocking(
gateway: Gateway, tmp_path: Path, logging_only_scope: str
) -> None:
identity: Final = "guardrail" + uuid.uuid4().hex
prompt: Final = "synthetic invalid-scope prompt pineapple " + identity
def guardrail(request: Request) -> Reply:
assert request.target == "/beta/litellm_basic_guardrail_api"
return Reply(body=json.dumps({"action": "BLOCKED", "blocked_reason": "synthetic policy denial"}).encode())
def provider(request: Request) -> Reply:
assert request.target == "/v1/chat/completions"
assert json.loads(request.body)["messages"] == [{"role": "user", "content": prompt}]
return Reply(
body=json.dumps(
{
"id": identity,
"object": "chat.completion",
"created": 1,
"model": "gpt-4o-mini",
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "unchanged provider reply"},
"finish_reason": "stop",
}
],
"usage": {"prompt_tokens": 9, "completion_tokens": 5, "total_tokens": 14},
}
).encode()
)
with wire_server(guardrail) as policy, wire_server(provider) as upstream:
config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
config["guardrails"] = [
{
"guardrail_name": identity,
"litellm_params": {
"guardrail": "generic_guardrail_api",
"mode": "pre_call",
"logging_only_scope": logging_only_scope,
"default_on": True,
"blocked_words": [{"keyword": "pineapple", "action": "BLOCK"}],
"api_base": policy.url,
"api_key": "synthetic-guardrail-key",
},
}
]
path: Final = tmp_path / "invalid-scope-pre-call.yaml"
path.write_text(yaml.safe_dump(config))
with owned_proxy(gateway, tmp_path, {}, config=path) as candidate, candidate.scenario() as scenario:
model: Final = scenario.model(api_base=upstream.url + "/v1")
response: Final = candidate.request(
"POST", "/v1/chat/completions", {"model": model, "messages": [{"role": "user", "content": prompt}]}
)
assert response.status_code == 400, response.text
assert "synthetic policy denial" in response.text, response.text
assert len(policy.drain()) == 1
assert len(upstream.drain()) == 0
guardrails: Final = candidate.get("/v2/guardrails/list")["guardrails"]
assert any(object_value(row)["guardrail_name"] == identity for row in guardrails), guardrails
_TOKEN: Final = re.compile(rb"token-[0-9a-f]{32}-\d+")

View file

@ -0,0 +1,762 @@
from __future__ import annotations
import signal
import socket
import threading
import time
import uuid
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from itertools import accumulate, repeat
from pathlib import Path
from queue import SimpleQueue
from typing import Final
import httpx
import psutil
import pytest
from _logging_only_scope_support import (
JSON_OBJECT,
CallerResult,
ChaosCall,
_assert_response_id,
_call_client,
_chaos_calls,
_chaos_control_configuration,
_chaos_model_list,
_chaos_models,
_chaos_spend_minimums,
_configuration,
_direction,
_directions_for_audit_leg,
_drain_upstream,
_guardrail_entries,
_is_base_audit_leg,
_json_contains_exact_string,
_policy_call_id_matches,
_spend_rows,
_spend_rows_for_calls,
_spend_rows_matching_call,
wire_server,
)
from _logging_only_scope_support import (
_record_audit_properties as _record_audit_properties,
)
from anthropic import APIConnectionError as AnthropicAPIConnectionError
from integration._support.client import Gateway, eventually
from integration._support.process import owned_proxy, owned_proxy_process
from integration._support.wire import Reply, Request
from openai import APIConnectionError as OpenAIAPIConnectionError
from pydantic import JsonValue
def test_K1_policy_edge_restart_mid_burst_keeps_output_observation_fail_open(gateway: Gateway, tmp_path: Path) -> None:
identity: Final = f"logging-scope-k1-{uuid.uuid4().hex}"
marker: Final = uuid.uuid4().hex
def policy(_request: Request) -> Reply:
return Reply(body=b'{"action":"NONE"}')
with socket.socket() as reservation:
reservation.bind(("127.0.0.1", 0))
policy_port: Final = reservation.getsockname()[1]
with gateway.scenario() as scenario:
deployments: Final = _chaos_models(scenario, marker)
models: Final = tuple(deployment.model_name for deployment in deployments)
model_list: Final = _chaos_model_list(deployments)
chaos_calls: Final = _chaos_calls(deployments, marker)
expected_directions: Final = _directions_for_audit_leg(("request", "response"), "output")
control_config: Final = _chaos_control_configuration(tmp_path, identity, model_list)
with owned_proxy(gateway, tmp_path, {}, config=control_config, workers=2) as control_proxy:
controls: Final = tuple(
_call_client(
call.client_kind,
call.endpoint,
control_proxy,
call.model,
call.prompt,
call.stream,
f"{call.call_id}-control",
)
for call in chaos_calls
)
control_upstream: Final = _drain_upstream(gateway.upstream_url)
assert len(control_upstream) == 30, control_upstream
assert (
tuple(
sum(_json_contains_exact_string(observation["body"], call.prompt) for observation in control_upstream)
for call in chaos_calls
)
== (1,) * 30
), control_upstream
config: Final = _configuration(
tmp_path,
identity,
f"http://127.0.0.1:{policy_port}",
"output",
model_list=model_list,
)
with owned_proxy(gateway, tmp_path, {}, config=config, workers=2) as candidate:
starts: Final = tuple(threading.Event() for _ in range(3))
def run(index: int) -> tuple[int, CallerResult]:
call: Final = chaos_calls[index]
phase: Final = index // 10
assert starts[phase].wait(timeout=90), (index, phase)
return index, _call_client(
call.client_kind,
call.endpoint,
candidate,
call.model,
call.prompt,
call.stream,
call.call_id,
)
with ThreadPoolExecutor(max_workers=30) as pool:
futures: Final = tuple(pool.submit(run, index) for index in range(30))
try:
with wire_server(policy, port=policy_port) as initial_edge:
starts[0].set()
first: Final = tuple(futures[index].result(timeout=90) for index in range(10))
eventually(
lambda: initial_edge.received.qsize(),
lambda count: count == 10 * len(expected_directions),
seconds=30,
)
tuple(
_spend_rows(model, minimum)
for model, minimum in zip(models, _chaos_spend_minimums(models, chaos_calls[:10], 5))
)
starts[1].set()
middle: Final = tuple(futures[index].result(timeout=90) for index in range(10, 20))
tuple(
_spend_rows(model, minimum)
for model, minimum in zip(models, _chaos_spend_minimums(models, chaos_calls[:20], 5))
)
with wire_server(policy, port=policy_port) as recovered_edge:
starts[2].set()
recovered: Final = tuple(futures[index].result(timeout=90) for index in range(20, 30))
eventually(
lambda: recovered_edge.received.qsize(),
lambda count: count == 10 * len(expected_directions),
seconds=30,
)
tuple(_spend_rows(model, 10) for model in models)
finally:
for start in starts:
start.set()
results: Final = first + middle + recovered
assert tuple(index for index, _ in results) == tuple(range(30)), results
assert all(result.status == controls[index].status for index, result in results), results
assert all(result.text == controls[index].text for index, result in results), results
candidate_ids: Final = tuple(result.response_id for _, result in results)
assert len(set(candidate_ids)) == 30, candidate_ids
observed_upstream: Final = _drain_upstream(gateway.upstream_url)
assert len(observed_upstream) == 30, observed_upstream
assert (
tuple(
sum(
_json_contains_exact_string(observation["body"], call.prompt)
for observation in observed_upstream
)
for call in chaos_calls
)
== (1,) * 30
), observed_upstream
expected_success_ids: Final = frozenset(
call.call_id for call in chaos_calls if call.index < 10 or call.index >= 20
)
edge_payloads: Final = tuple(
JSON_OBJECT.validate_json(call.body) for call in initial_edge.drain() + recovered_edge.drain()
)
assert len(edge_payloads) == 20 * len(expected_directions), edge_payloads
successful_calls: Final = tuple(call for call in chaos_calls if call.call_id in expected_success_ids)
for call in successful_calls:
payloads_for_call: Final = tuple(
payload for payload in edge_payloads if _policy_call_id_matches(payload, call.call_id)
)
assert tuple(sorted(_direction(payload) for payload in payloads_for_call)) == tuple(
sorted(expected_directions)
), (
call.call_id,
payloads_for_call,
)
assert all(
payload["texts"]
== ([call.prompt] if _direction(payload) == "request" else [results[call.index][1].text])
for payload in payloads_for_call
), payloads_for_call
rows: Final = _spend_rows_for_calls(
models,
tuple(
(
chaos_calls[index].model,
response_id,
chaos_calls[index].call_id,
)
for index, response_id in enumerate(candidate_ids)
),
)
for index, call in enumerate(chaos_calls):
matching_rows: Final = _spend_rows_matching_call(rows, call.model, call.call_id)
assert len(matching_rows) == 1, (call.call_id, matching_rows)
row: Final = matching_rows[0]
expected_status: Final = "guardrail_failed_to_respond" if 10 <= index < 20 else "success"
entries: Final = _guardrail_entries(row)
assert tuple(
(entry["guardrail_name"], entry["guardrail_mode"], entry["guardrail_status"]) for entry in entries
) == tuple((identity, "logging_only", expected_status) for _ in expected_directions), (index, entries)
def test_K2_policy_edge_delay_does_not_delay_concurrent_callers(gateway: Gateway, tmp_path: Path) -> None:
identity: Final = f"logging-scope-k2-{uuid.uuid4().hex}"
marker: Final = uuid.uuid4().hex
def policy(_request: Request) -> Reply:
time.sleep(2)
return Reply(body=b'{"action":"NONE"}')
with gateway.scenario() as scenario:
deployments: Final = _chaos_models(scenario, marker)
models: Final = tuple(deployment.model_name for deployment in deployments)
model_list: Final = _chaos_model_list(deployments)
chaos_calls: Final = _chaos_calls(deployments, marker)
expected_directions: Final = _directions_for_audit_leg(("request", "response"), "output")
control_config: Final = _chaos_control_configuration(tmp_path, identity, model_list)
with owned_proxy(gateway, tmp_path, {}, config=control_config, workers=2) as control_proxy:
controls: Final = tuple(
_call_client(
call.client_kind,
call.endpoint,
control_proxy,
call.model,
call.prompt,
call.stream,
f"{call.call_id}-control",
)
for call in chaos_calls
)
control_upstream: Final = _drain_upstream(gateway.upstream_url)
assert len(control_upstream) == 30, control_upstream
assert (
tuple(
sum(_json_contains_exact_string(observation["body"], call.prompt) for observation in control_upstream)
for call in chaos_calls
)
== (1,) * 30
), control_upstream
with wire_server(policy) as edge:
config: Final = _configuration(tmp_path, identity, edge.url, "output", model_list=model_list)
with owned_proxy(gateway, tmp_path, {}, config=config, workers=2) as candidate:
def run(call: ChaosCall) -> tuple[int, CallerResult, float]:
started: Final = time.monotonic()
result: Final = _call_client(
call.client_kind,
call.endpoint,
candidate,
call.model,
call.prompt,
call.stream,
call.call_id,
)
return call.index, result, time.monotonic() - started
with ThreadPoolExecutor(max_workers=30) as pool:
results: Final = tuple(pool.map(run, chaos_calls))
assert tuple(index for index, _, _ in results) == tuple(range(30)), results
assert all(result.status == controls[index].status for index, result, _ in results), results
assert all(result.text == controls[index].text for index, result, _ in results), results
assert all(duration < 2 for _, _, duration in results), results
eventually(
lambda: edge.received.qsize(),
lambda count: count == 30 * len(expected_directions),
seconds=30,
)
edge_calls: Final = edge.drain()
payloads: Final = tuple(JSON_OBJECT.validate_json(call.body) for call in edge_calls)
for call in chaos_calls:
payloads_for_call: Final = tuple(
payload for payload in payloads if _policy_call_id_matches(payload, call.call_id)
)
assert tuple(sorted(_direction(payload) for payload in payloads_for_call)) == tuple(
sorted(expected_directions)
), (
call,
payloads_for_call,
)
assert all(
payload["texts"]
== ([call.prompt] if _direction(payload) == "request" else [controls[call.index].text])
for payload in payloads_for_call
), payloads_for_call
response_ids: Final = tuple(result.response_id for _, result, _ in results)
assert len(set(response_ids)) == 30, response_ids
upstream: Final = _drain_upstream(gateway.upstream_url)
assert len(upstream) == 30, upstream
assert (
tuple(
sum(_json_contains_exact_string(observation["body"], call.prompt) for observation in upstream)
for call in chaos_calls
)
== (1,) * 30
), upstream
rows: Final = _spend_rows_for_calls(
models,
tuple(
(call.model, result.response_id, call.call_id)
for call, (_, result, _) in zip(chaos_calls, results)
),
)
for call, (_, result, _) in zip(chaos_calls, results):
matching_rows: Final = _spend_rows_matching_call(rows, call.model, call.call_id)
assert len(matching_rows) == 1, (call, result.response_id, matching_rows)
row: Final = matching_rows[0]
entries: Final = _guardrail_entries(row)
assert tuple(
(entry["guardrail_name"], entry["guardrail_mode"], entry["guardrail_status"])
for entry in entries
) == tuple((identity, "logging_only", "success") for _ in expected_directions), entries
@pytest.mark.timeout(180)
def test_K3_two_worker_sigkill_checks_post_kill_spend_rows(
gateway: Gateway,
tmp_path: Path,
record_property: Callable[[str, object], None],
) -> None:
identity: Final = f"logging-scope-k3-{uuid.uuid4().hex}"
marker: Final = uuid.uuid4().hex
scan_started: Final = threading.Event()
release_scans: Final = threading.Event()
def policy(_request: Request) -> Reply:
scan_started.set()
assert release_scans.wait(timeout=60), identity
return Reply(body=b'{"action":"NONE"}')
with gateway.scenario() as scenario:
deployments: Final = _chaos_models(scenario, marker)
models: Final = tuple(deployment.model_name for deployment in deployments)
model_list: Final = _chaos_model_list(deployments)
calls: Final = _chaos_calls(deployments, marker)
expected_directions: Final = _directions_for_audit_leg(("request", "response"), "output")
expected_entries: Final = tuple((identity, "logging_only", "success") for _ in expected_directions)
control_config: Final = _chaos_control_configuration(tmp_path, identity, model_list)
with owned_proxy(gateway, tmp_path, {}, config=control_config, workers=2) as control_proxy:
controls: Final = tuple(
_call_client(
call.client_kind,
call.endpoint,
control_proxy,
call.model,
call.prompt,
call.stream,
f"{call.call_id}-control",
)
for call in calls
)
assert len(_drain_upstream(gateway.upstream_url)) == 30
with wire_server(policy) as edge:
config: Final = _configuration(tmp_path, identity, edge.url, "output", model_list=model_list)
with owned_proxy_process(gateway, tmp_path, {}, config=config, workers=2) as owned:
candidate: Final = owned.gateway
root: Final = psutil.Process(owned.process.pid)
workers: Final = eventually(
lambda: tuple(
child
for child in root.children(recursive=True)
if any("spawn_main" in part for part in child.cmdline())
),
lambda children: len(children) == 2,
seconds=30,
)
def run(call: ChaosCall) -> tuple[int, CallerResult | None, str | None]:
try:
result: Final = _call_client(
call.client_kind,
call.endpoint,
candidate,
call.model,
call.prompt,
call.stream,
call.call_id,
)
return call.index, result, None
except (OpenAIAPIConnectionError, AnthropicAPIConnectionError, httpx.RemoteProtocolError) as error:
return call.index, None, str(error)
with ThreadPoolExecutor(max_workers=30) as pool:
futures: Final = tuple(pool.submit(run, call) for call in calls)
try:
assert eventually(lambda: scan_started.is_set(), bool, seconds=30)
eventually(lambda: edge.received.qsize(), lambda count: count >= 5, seconds=30)
workers[0].send_signal(signal.SIGKILL)
killed_workers, surviving_workers = psutil.wait_procs((workers[0],), timeout=10)
assert len(killed_workers) == 1 and not surviving_workers, (
killed_workers,
surviving_workers,
)
finally:
release_scans.set()
outcomes: Final = tuple(future.result(timeout=90) for future in futures)
assert owned.process.poll() is None, "Proxy supervisor exited after a worker was killed"
successful: Final = tuple(
(calls[index], result) for index, result, error in outcomes if result is not None and error is None
)
assert successful, outcomes
assert all(
result.status == controls[call.index].status and result.text == controls[call.index].text
for call, result in successful
), successful
pre_kill_expected: Final = tuple(
(call.model, result.response_id, call.call_id) for call, result in successful
)
pre_kill_rows: Final = _spend_rows_for_calls(
models,
pre_kill_expected,
tolerate_missing=True,
)
pre_kill_rows_by_call: Final = tuple(
(call, _spend_rows_matching_call(pre_kill_rows, call.model, call.call_id)) for call in calls
)
assert all(len(rows) <= 1 for _, rows in pre_kill_rows_by_call), pre_kill_rows_by_call
pre_kill_missing_rows: Final = sum(not rows for _, rows in pre_kill_rows_by_call)
record_property("k3_pre_kill_missing_spend_rows", pre_kill_missing_rows)
for call, matching_rows in pre_kill_rows_by_call:
if not matching_rows:
continue
entries: Final = _guardrail_entries(matching_rows[0])
assert tuple(
sorted(
(
entry["guardrail_name"],
entry["guardrail_mode"],
entry["guardrail_status"],
)
for entry in entries
)
) == tuple(sorted(expected_entries)), (call, entries)
post_kill_templates: Final = calls[:6]
post_kill_calls: Final = tuple(
ChaosCall(
index=call.index,
endpoint=call.endpoint,
client_kind=call.client_kind,
model=call.model,
stream=call.stream,
prompt=f"synthetic K post-kill burst {marker}-{call.index}",
call_id=f"{marker}-k-post-kill-{call.index}",
)
for call in post_kill_templates
)
def run_post_kill(call: ChaosCall) -> CallerResult:
return _call_client(
call.client_kind,
call.endpoint,
candidate,
call.model,
call.prompt,
call.stream,
call.call_id,
)
with ThreadPoolExecutor(max_workers=len(post_kill_calls)) as pool:
post_kill_futures: Final = tuple(pool.submit(run_post_kill, call) for call in post_kill_calls)
post_kill_results: Final = tuple(future.result(timeout=90) for future in post_kill_futures)
assert all(
result.status == controls[call.index].status and result.text == controls[call.index].text
for call, result in zip(post_kill_calls, post_kill_results)
), post_kill_results
served: Final = successful + tuple(zip(post_kill_calls, post_kill_results))
response_ids: Final = tuple(result.response_id for _, result in served)
assert len(set(response_ids)) == len(response_ids), response_ids
served_calls: Final = tuple(call for call, _ in served)
requested_call_ids: Final = frozenset(call.call_id for call in calls + post_kill_calls)
upstream: Final = _drain_upstream(gateway.upstream_url)
assert all(
sum(_json_contains_exact_string(observation["body"], call.prompt) for observation in upstream) == 1
for call in served_calls
), upstream
def accumulate_policy_payloads(
collected: tuple[dict[str, JsonValue], ...], _: None
) -> tuple[dict[str, JsonValue], ...]:
return collected + tuple(JSON_OBJECT.validate_json(call.body) for call in edge.drain())
payload_batches: Final = accumulate(repeat(None), accumulate_policy_payloads, initial=())
def has_expected_post_kill_scans(collected: tuple[dict[str, JsonValue], ...], call: ChaosCall) -> bool:
payloads_for_call: Final = tuple(
payload for payload in collected if _policy_call_id_matches(payload, call.call_id)
)
return all(
sum(_direction(payload) == direction for payload in payloads_for_call)
>= expected_directions.count(direction)
for direction in expected_directions
)
payloads: Final = eventually(
lambda: next(payload_batches),
lambda collected: all(has_expected_post_kill_scans(collected, call) for call in post_kill_calls),
seconds=30,
)
assert all(
any(_policy_call_id_matches(payload, call_id) for call_id in requested_call_ids)
and _direction(payload) in expected_directions
for payload in payloads
), payloads
for call, result in zip(post_kill_calls, post_kill_results):
payloads_for_call: Final = tuple(
payload for payload in payloads if _policy_call_id_matches(payload, call.call_id)
)
assert tuple(sorted(_direction(payload) for payload in payloads_for_call)) == tuple(
sorted(expected_directions)
), (call, payloads_for_call)
assert all(
payload["texts"]
== ([call.prompt] if _direction(payload) == "request" else [controls[call.index].text])
for payload in payloads_for_call
), payloads_for_call
post_kill_expected: Final = tuple(
(call.model, result.response_id, call.call_id)
for call, result in zip(post_kill_calls, post_kill_results)
)
rows: Final = _spend_rows_for_calls(
models,
post_kill_expected,
tolerate_missing=True,
)
all_candidate_calls: Final = calls + post_kill_calls
rows_by_call: Final = tuple(
(call, _spend_rows_matching_call(rows, call.model, call.call_id)) for call in all_candidate_calls
)
assert all(len(matching_rows) <= 1 for _, matching_rows in rows_by_call), rows_by_call
for call, matching_rows in rows_by_call:
if not matching_rows:
continue
entries: Final = _guardrail_entries(matching_rows[0])
assert tuple(
sorted(
(
entry["guardrail_name"],
entry["guardrail_mode"],
entry["guardrail_status"],
)
for entry in entries
)
) == tuple(sorted(expected_entries)), (call, entries)
for call, result in successful:
matching_rows: Final = _spend_rows_matching_call(rows, call.model, call.call_id)
if matching_rows:
_assert_response_id(
call.endpoint,
str(matching_rows[0]["request_id"]),
result.response_id,
marker if call.endpoint == "responses" else call.call_id,
)
for call, result in zip(post_kill_calls, post_kill_results):
matching_rows: Final = _spend_rows_matching_call(rows, call.model, call.call_id)
assert len(matching_rows) == 1, (result.response_id, matching_rows)
_assert_response_id(
call.endpoint,
str(matching_rows[0]["request_id"]),
result.response_id,
marker if call.endpoint == "responses" else call.call_id,
)
@pytest.mark.timeout(180)
def test_K4_proxy_restart_after_fifteen_responses_records_lost_ids(
gateway: Gateway,
tmp_path: Path,
record_property: Callable[[str, object], None],
) -> None:
identity: Final = f"logging-scope-k4-{uuid.uuid4().hex}"
marker: Final = uuid.uuid4().hex
def policy(_request: Request) -> Reply:
return Reply(body=b'{"action":"NONE"}')
with gateway.scenario() as scenario:
deployments: Final = _chaos_models(scenario, marker)
models: Final = tuple(deployment.model_name for deployment in deployments)
model_list: Final = _chaos_model_list(deployments)
calls: Final = _chaos_calls(deployments, marker)
expected_directions: Final = _directions_for_audit_leg(("request", "response"), "output")
control_config: Final = _chaos_control_configuration(tmp_path, identity, model_list)
with owned_proxy(gateway, tmp_path, {}, config=control_config, workers=2) as control_proxy:
controls: Final = tuple(
_call_client(
call.client_kind,
call.endpoint,
control_proxy,
call.model,
call.prompt,
call.stream,
f"{call.call_id}-control",
)
for call in calls
)
control_upstream: Final = _drain_upstream(gateway.upstream_url)
assert len(control_upstream) == 30, control_upstream
assert (
tuple(
sum(_json_contains_exact_string(observation["body"], call.prompt) for observation in control_upstream)
for call in calls
)
== (1,) * 30
), control_upstream
with wire_server(policy) as edge:
config: Final = _configuration(tmp_path, identity, edge.url, "output", model_list=model_list)
restart_gate: Final = threading.Event()
second_wave_ready: Final = threading.Event()
second_wave_barrier: Final = threading.Barrier(15, action=second_wave_ready.set)
restarted_gateways: Final[SimpleQueue[Gateway]] = SimpleQueue()
def gateway_for_call(call: ChaosCall, first_gateway: Gateway) -> Gateway:
if call.index < 15:
return first_gateway
second_wave_barrier.wait(timeout=90)
assert restart_gate.wait(timeout=90)
return restarted_gateways.get()
def run(call: ChaosCall, first_gateway: Gateway) -> tuple[int, CallerResult]:
candidate: Final = gateway_for_call(call, first_gateway)
result: Final = _call_client(
call.client_kind,
call.endpoint,
candidate,
call.model,
call.prompt,
call.stream,
call.call_id,
)
return call.index, result
with ThreadPoolExecutor(max_workers=30) as pool:
with owned_proxy_process(gateway, tmp_path, {}, config=config, workers=2) as first_proxy:
first_proxy_port: Final = first_proxy.gateway.client.base_url.port
assert first_proxy_port is not None
futures: Final = tuple(pool.submit(run, call, first_proxy.gateway) for call in calls)
first_results: Final = tuple(futures[index].result(timeout=90)[1] for index in range(15))
assert all(
result.status == controls[call.index].status and result.text == controls[call.index].text
for call, result in zip(calls[:15], first_results)
), first_results
assert eventually(lambda: second_wave_ready.is_set(), bool, seconds=30)
eventually(
lambda: edge.received.qsize(),
lambda count: count == 15 * len(expected_directions),
seconds=30,
)
first_wave_expected: Final = tuple(
(call.model, result.response_id, call.call_id)
for call, result in zip(calls[:15], first_results)
)
first_wave_rows: Final = _spend_rows_for_calls(
models,
first_wave_expected,
tolerate_missing=True,
)
assert all(
len(_spend_rows_matching_call(first_wave_rows, model, call_id)) <= 1
for model, _, call_id in first_wave_expected
), first_wave_rows
first_wave_present_response_ids: Final = frozenset(
response_id
for model, response_id, call_id in first_wave_expected
if len(_spend_rows_matching_call(first_wave_rows, model, call_id)) == 1
)
first_wave_lost_response_ids: Final = (
frozenset(response_id for _, response_id, _ in first_wave_expected)
- first_wave_present_response_ids
)
record_property(
"K4_PRE_RESTART_LOST_RESPONSE_IDS",
tuple(sorted(first_wave_lost_response_ids)),
)
record_property(
f"K4_PRE_RESTART_LOST_ROW_COUNT_{'base' if _is_base_audit_leg() else 'head'}",
len(first_wave_lost_response_ids),
)
assert first_proxy.process.poll() is not None, first_proxy.process.pid
eventually(
lambda: tuple(
connection
for connection in psutil.net_connections(kind="tcp")
if connection.status == psutil.CONN_LISTEN and connection.laddr.port == first_proxy_port
),
lambda listeners: not listeners,
seconds=30,
)
with owned_proxy_process(gateway, tmp_path, {}, config=config, workers=2) as restarted_proxy:
for _ in range(15):
restarted_gateways.put(restarted_proxy.gateway)
restart_gate.set()
second_results: Final = tuple(futures[index].result(timeout=90)[1] for index in range(15, 30))
assert all(
result.status == controls[call.index].status and result.text == controls[call.index].text
for call, result in zip(calls[15:], second_results)
), second_results
eventually(
lambda: edge.received.qsize(),
lambda count: count == 30 * len(expected_directions),
seconds=30,
)
results: Final = first_results + second_results
expected_response_ids: Final = frozenset(result.response_id for result in results)
assert len(expected_response_ids) == 30, results
upstream: Final = _drain_upstream(gateway.upstream_url)
assert len(upstream) == 30, upstream
assert (
tuple(
sum(_json_contains_exact_string(observation["body"], call.prompt) for observation in upstream)
for call in calls
)
== (1,) * 30
), upstream
edge_payloads: Final = tuple(JSON_OBJECT.validate_json(call.body) for call in edge.drain())
assert len(edge_payloads) == 30 * len(expected_directions), edge_payloads
for call, result in zip(calls, results):
payloads_for_call: Final = tuple(
payload for payload in edge_payloads if _policy_call_id_matches(payload, call.call_id)
)
assert tuple(sorted(_direction(payload) for payload in payloads_for_call)) == tuple(
sorted(expected_directions)
), (
call,
payloads_for_call,
)
assert all(
payload["texts"] == ([call.prompt] if _direction(payload) == "request" else [result.text])
for payload in payloads_for_call
), payloads_for_call
post_restart_expected: Final = tuple(
(call.model, result.response_id, call.call_id) for call, result in zip(calls[15:], second_results)
)
rows: Final = _spend_rows_for_calls(
models,
post_restart_expected,
tolerate_missing=True,
)
record_property("K4_RESPONSE_IDS", tuple(sorted(expected_response_ids)))
for call, result in zip(calls, results):
matching_rows: Final = _spend_rows_matching_call(rows, call.model, call.call_id)
assert len(matching_rows) <= 1, (result.response_id, matching_rows)
if call.index >= 15:
assert len(matching_rows) == 1, (call.call_id, result.response_id, matching_rows)
if matching_rows:
entries: Final = _guardrail_entries(matching_rows[0])
assert tuple(
(entry["guardrail_name"], entry["guardrail_mode"], entry["guardrail_status"])
for entry in entries
) == tuple((identity, "logging_only", "success") for _ in expected_directions), (
call.call_id,
entries,
)

File diff suppressed because it is too large Load diff

File diff suppressed because it is too large Load diff

View file

@ -12,7 +12,7 @@ from litellm.integrations.custom_guardrail import (
)
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.proxy._types import CallTypes, UserAPIKeyAuth
from litellm.types.guardrails import GuardrailEventHooks, Mode
from litellm.types.guardrails import GuardrailEventHooks, LoggingOnlyScope, Mode
from litellm.types.utils import (
Choices,
GenericGuardrailAPIInputs,
@ -2577,6 +2577,32 @@ class TestLoggingOnlyApplyGuardrail:
assert "standard_logging_guardrail_information" not in kwargs["litellm_params"]["metadata"]
assert kwargs["standard_logging_object"] == {"guardrail_information": None}
@pytest.mark.parametrize(
"scope,expected_calls",
(
(None, [("request", ["hello there"]), ("response", ["general kenobi"])]),
("both", [("request", ["hello there"]), ("response", ["general kenobi"])]),
("input", [("request", ["hello there"])]),
("output", [("response", ["general kenobi"])]),
),
)
@pytest.mark.asyncio
async def test_logging_only_scope_scans_configured_directions(
self,
scope: LoggingOnlyScope | None,
expected_calls: list[tuple[str, list[str]]],
) -> None:
guardrail: Final = _ApplyOnlyObserver()
guardrail.logging_only_scope = scope
kwargs, response = _logged_call([{"role": "user", "content": "hello there"}])
out_kwargs, _ = await guardrail.async_logging_hook(kwargs, response, CallTypes.acompletion.value)
assert guardrail.calls == expected_calls
entries: Final = out_kwargs["standard_logging_object"]["guardrail_information"]
assert len(entries) == len(expected_calls)
assert [entry["guardrail_mode"] for entry in entries] == ["logging_only"] * len(expected_calls)
@pytest.mark.asyncio
async def test_appends_to_pre_call_verdicts_without_duplicating_them(self):
guardrail = _ApplyOnlyObserver()
@ -2604,6 +2630,39 @@ class TestLoggingOnlyApplyGuardrail:
assert out_kwargs is kwargs
assert out_response is response
@pytest.mark.asyncio
async def test_output_scope_scans_response_when_request_copy_fails(self):
import threading
guardrail: Final = _ApplyOnlyObserver()
guardrail.logging_only_scope = "output"
call: Final = _logged_call([{"role": "user", "content": "hello there", "lock": threading.Lock()}])
kwargs: Final = call[0]
response: Final = call[1]
out_kwargs, _ = await guardrail.async_logging_hook(kwargs, response, CallTypes.acompletion.value)
assert guardrail.calls == [("response", ["general kenobi"])]
entries: Final = out_kwargs["standard_logging_object"]["guardrail_information"]
assert [entry["guardrail_name"] for entry in entries] == ["apply-only-observer"]
assert [entry["guardrail_status"] for entry in entries] == ["success"]
@pytest.mark.asyncio
async def test_request_copy_failure_does_not_drop_the_response_scan_for_explicit_both_scope(self):
import threading
guardrail: Final = _ApplyOnlyObserver()
guardrail.logging_only_scope = "both"
call: Final = _logged_call([{"role": "user", "content": "hello there", "lock": threading.Lock()}])
kwargs: Final = call[0]
response: Final = call[1]
out_kwargs, _ = await guardrail.async_logging_hook(kwargs, response, CallTypes.acompletion.value)
assert guardrail.calls == [("response", ["general kenobi"])]
entries: Final = out_kwargs["standard_logging_object"]["guardrail_information"]
assert [entry["guardrail_status"] for entry in entries] == ["success"]
@pytest.mark.asyncio
async def test_block_verdict_is_recorded_without_raising(self):
guardrail = _ApplyOnlyObserver(block=True)
@ -2615,6 +2674,45 @@ class TestLoggingOnlyApplyGuardrail:
entries = out_kwargs["standard_logging_object"]["guardrail_information"]
assert [e["guardrail_status"] for e in entries] == ["guardrail_intervened"]
@pytest.mark.asyncio
async def test_input_scan_error_aborts_the_response_scan(self):
class _FailingObserver(_ApplyOnlyObserver):
@log_guardrail_information
async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None):
self.calls.append((input_type, list(inputs.get("texts") or [])))
if input_type == "request":
raise RuntimeError("guardrail service unavailable")
return GenericGuardrailAPIInputs(texts=[])
guardrail = _FailingObserver()
kwargs, response = _logged_call([{"role": "user", "content": "hello there"}])
out_kwargs, _ = await guardrail.async_logging_hook(kwargs, response, CallTypes.acompletion.value)
assert guardrail.calls == [("request", ["hello there"])]
entries = out_kwargs["standard_logging_object"]["guardrail_information"]
assert [e["guardrail_status"] for e in entries] == ["guardrail_failed_to_respond"]
@pytest.mark.asyncio
async def test_input_scan_error_does_not_drop_the_response_scan_for_explicit_both_scope(self):
class _FailingBothObserver(_ApplyOnlyObserver):
@log_guardrail_information
async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None):
self.calls.append((input_type, list(inputs.get("texts") or [])))
if input_type == "request":
raise RuntimeError("guardrail service unavailable")
return GenericGuardrailAPIInputs(texts=[])
guardrail = _FailingBothObserver()
guardrail.logging_only_scope = "both"
kwargs, response = _logged_call([{"role": "user", "content": "hello there"}])
out_kwargs, _ = await guardrail.async_logging_hook(kwargs, response, CallTypes.acompletion.value)
assert guardrail.calls == [("request", ["hello there"]), ("response", ["general kenobi"])]
entries = out_kwargs["standard_logging_object"]["guardrail_information"]
assert [e["guardrail_status"] for e in entries] == ["guardrail_failed_to_respond", "success"]
@pytest.mark.asyncio
async def test_call_type_without_translation_is_skipped(self):
guardrail = _ApplyOnlyObserver()
@ -2932,6 +3030,17 @@ class _NativeLifecycleLoggingGuardrail(CustomGuardrail):
return inputs
@pytest.mark.asyncio
async def test_native_lifecycle_guardrail_logging_only_scope_scans_only_input():
guardrail: Final = _NativeLifecycleLoggingGuardrail()
guardrail.logging_only_scope = "input"
kwargs, response = _logged_call([{"role": "user", "content": "native lifecycle input"}])
await guardrail.async_logging_hook(kwargs, response, CallTypes.acompletion.value)
assert guardrail.calls == [("request", ["native lifecycle input"])]
@pytest.mark.asyncio
async def test_native_lifecycle_guardrail_logging_only_scans_assembled_response():
"""A use_native_lifecycle_hooks guardrail accepts mode logging_only and its

View file

@ -1,7 +1,9 @@
import json
import time
from collections.abc import Callable
from datetime import datetime
from typing import Dict, List, Optional
from types import MappingProxyType
from typing import Dict, Final, List, Optional
from unittest.mock import AsyncMock
import pytest
@ -10,6 +12,7 @@ import yaml
from fastapi import HTTPException
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.guardrails.guardrail_endpoints import (
CreateGuardrailRequest,
@ -46,10 +49,12 @@ from litellm.proxy.guardrails.guardrail_registry import (
from litellm.types.guardrails import (
ApplyGuardrailRequest,
BaseLitellmParams,
GuardrailEventHooks,
Guardrail,
GuardrailInfoResponse,
LitellmParams,
)
from litellm.types.utils import GenericGuardrailAPIInputs
# Mock data for testing
MOCK_DB_GUARDRAIL = {
@ -64,6 +69,46 @@ MOCK_DB_GUARDRAIL = {
"updated_at": datetime.now(),
}
_INVALID_SCOPE_LITELLM_PARAMS: Final = MappingProxyType(
{
"guardrail": "litellm_content_filter",
"mode": "pre_call",
"logging_only_scope": "Input",
"blocked_words": [{"keyword": "synthetic blocked phrase", "action": "BLOCK"}],
}
)
_INVALID_SCOPE_DB_GUARDRAIL: Final = MappingProxyType(
{
"guardrail_id": "invalid-scope-db-guardrail",
"guardrail_name": "Invalid scope DB guardrail",
"litellm_params": _INVALID_SCOPE_LITELLM_PARAMS,
"guardrail_info": MappingProxyType({}),
}
)
_INVALID_SCOPE_IN_MEMORY_GUARDRAIL: Final = MappingProxyType(
{
"guardrail_id": "invalid-scope-in-memory-guardrail",
"guardrail_name": "Invalid scope in-memory guardrail",
"litellm_params": _INVALID_SCOPE_LITELLM_PARAMS,
"guardrail_info": MappingProxyType({}),
}
)
_VALID_SCOPE_DB_GUARDRAIL: Final = MappingProxyType(
{
"guardrail_id": "valid-scope-db-guardrail",
"guardrail_name": "Valid scope DB guardrail",
"litellm_params": MappingProxyType(
{
"guardrail": "litellm_content_filter",
"mode": "pre_call",
"logging_only_scope": "output",
"blocked_words": [{"keyword": "synthetic blocked phrase", "action": "BLOCK"}],
}
),
"guardrail_info": MappingProxyType({}),
}
)
MOCK_CONFIG_GUARDRAIL = {
"guardrail_id": "test-config-guardrail",
"guardrail_name": "Test Config Guardrail",
@ -89,6 +134,44 @@ MOCK_PATCH_REQUEST = PatchGuardrailRequest(
)
class _PatchScopeSupportedGuardrail(CustomGuardrail):
async def apply_guardrail(
self,
inputs: GenericGuardrailAPIInputs,
request_data: dict[str, object],
input_type: str,
logging_obj: object | None = None,
) -> GenericGuardrailAPIInputs:
return inputs
class _PatchScopeUnsupportedGuardrail(_PatchScopeSupportedGuardrail):
async def async_logging_hook(
self,
kwargs: dict[str, object],
result: object,
call_type: str,
) -> tuple[dict[str, object], object]:
return kwargs, result
def _patch_scope_initializer(
callback_type: type[CustomGuardrail],
) -> Callable[[LitellmParams, Guardrail], CustomGuardrail]:
def _initializer(litellm_params: LitellmParams, guardrail: Guardrail) -> CustomGuardrail:
import litellm
callback = callback_type(
guardrail_name=guardrail["guardrail_name"],
event_hook=litellm_params.mode,
default_on=litellm_params.default_on,
)
litellm.logging_callback_manager.add_litellm_callback(callback)
return callback
return _initializer
@pytest.fixture
def mock_prisma_client(mocker):
"""Mock Prisma client for testing"""
@ -128,6 +211,37 @@ def mock_guardrail_registry(mocker):
return mock_registry
def _setup_patch_scope_guardrail(
mocker,
monkeypatch,
mock_guardrail_registry,
callback_type: type[CustomGuardrail],
guardrail_type: str,
litellm_params: dict[str, object],
) -> tuple[InMemoryGuardrailHandler, Guardrail]:
from litellm.proxy.guardrails import guardrail_registry as registry_module
guardrail: Guardrail = {
"guardrail_id": "patch-scope-test",
"guardrail_name": "Patch scope test",
"litellm_params": {"guardrail": guardrail_type, **litellm_params},
"guardrail_info": {},
}
mock_guardrail_registry.get_guardrail_by_id_from_db.return_value = guardrail
mock_guardrail_registry.update_guardrail_in_db.return_value = guardrail
monkeypatch.setitem(
registry_module.guardrail_initializer_registry,
guardrail_type,
_patch_scope_initializer(callback_type),
)
handler = InMemoryGuardrailHandler()
handler.initialize_guardrail(guardrail=guardrail, source="db")
mocker.patch("litellm.proxy.proxy_server.prisma_client", mocker.Mock())
mocker.patch("litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY", mock_guardrail_registry)
mocker.patch("litellm.proxy.guardrails.guardrail_registry.IN_MEMORY_GUARDRAIL_HANDLER", handler)
return handler, guardrail
@pytest.mark.asyncio
async def test_list_guardrails_v2_with_db_and_config(mocker, mock_prisma_client, mock_in_memory_handler):
"""Test listing guardrails from both DB and config"""
@ -157,6 +271,54 @@ async def test_list_guardrails_v2_with_db_and_config(mocker, mock_prisma_client,
assert isinstance(config_guardrail.litellm_params, BaseLitellmParams)
@pytest.mark.asyncio
async def test_list_guardrails_v2_normalizes_invalid_scope_and_keeps_other_db_rows(
mocker, mock_prisma_client, mock_in_memory_handler
):
mock_prisma_client.db.litellm_guardrailstable.find_many.return_value = (
_INVALID_SCOPE_DB_GUARDRAIL,
_VALID_SCOPE_DB_GUARDRAIL,
)
mock_in_memory_handler.list_in_memory_guardrails.return_value = ()
mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client)
mocker.patch(
"litellm.proxy.guardrails.guardrail_registry.IN_MEMORY_GUARDRAIL_HANDLER",
mock_in_memory_handler,
)
response: Final = await list_guardrails_v2(user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN))
invalid_scope_row: Final = next(
guardrail for guardrail in response.guardrails if guardrail.guardrail_id == "invalid-scope-db-guardrail"
)
assert invalid_scope_row.litellm_params is not None
assert invalid_scope_row.litellm_params.logging_only_scope is None
assert invalid_scope_row.litellm_params.mode == "pre_call"
assert invalid_scope_row.litellm_params.guardrail == "litellm_content_filter"
assert any(guardrail.guardrail_id == "valid-scope-db-guardrail" for guardrail in response.guardrails)
@pytest.mark.asyncio
async def test_list_guardrails_v2_normalizes_invalid_scope_in_memory(
mocker, mock_prisma_client, mock_in_memory_handler
):
mock_prisma_client.db.litellm_guardrailstable.find_many.return_value = ()
mock_in_memory_handler.list_in_memory_guardrails.return_value = (_INVALID_SCOPE_IN_MEMORY_GUARDRAIL,)
mocker.patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client)
mocker.patch(
"litellm.proxy.guardrails.guardrail_registry.IN_MEMORY_GUARDRAIL_HANDLER",
mock_in_memory_handler,
)
response: Final = await list_guardrails_v2(user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN))
assert len(response.guardrails) == 1
assert response.guardrails[0].litellm_params is not None
assert response.guardrails[0].litellm_params.logging_only_scope is None
assert response.guardrails[0].litellm_params.mode == "pre_call"
assert response.guardrails[0].litellm_params.guardrail == "litellm_content_filter"
@pytest.mark.asyncio
async def test_list_guardrails_v2_skips_stale_db_backed_in_memory_entries(mocker):
"""
@ -421,6 +583,28 @@ async def test_get_guardrail_info_from_db(mocker, mock_prisma_client):
assert response.guardrail_info == {"description": "Test guardrail from DB"}
@pytest.mark.asyncio
async def test_get_guardrail_info_normalizes_invalid_scope_from_db(
mocker, mock_guardrail_registry, mock_in_memory_handler
):
mock_guardrail_registry.get_guardrail_by_id_from_db.return_value = _INVALID_SCOPE_DB_GUARDRAIL
mock_in_memory_handler.get_guardrail_by_id.return_value = None
mocker.patch("litellm.proxy.proxy_server.prisma_client", mocker.Mock())
mocker.patch("litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY", mock_guardrail_registry)
mocker.patch(
"litellm.proxy.guardrails.guardrail_registry.IN_MEMORY_GUARDRAIL_HANDLER",
mock_in_memory_handler,
)
response: Final = await get_guardrail_info("invalid-scope-db-guardrail")
assert response.guardrail_id == "invalid-scope-db-guardrail"
assert response.litellm_params is not None
assert response.litellm_params.logging_only_scope is None
assert response.litellm_params.mode == "pre_call"
assert response.litellm_params.guardrail == "litellm_content_filter"
@pytest.mark.asyncio
async def test_get_guardrail_info_from_config(mocker, mock_prisma_client, mock_in_memory_handler):
"""Test getting guardrail info from config when not found in DB"""
@ -1127,7 +1311,10 @@ async def test_update_guardrail_endpoint(
prisma_client=mocker.ANY,
)
mock_in_memory_handler.sync_guardrail_from_db.assert_called_once_with(guardrail=mocker.ANY)
mock_in_memory_handler.sync_guardrail_from_db.assert_called_once_with(
guardrail=mocker.ANY,
reject_invalid_logging_only_scope=True,
)
if scenario == "success_sync_fails_unexpected_error":
assert mock_logger is not None
@ -1256,7 +1443,10 @@ async def test_patch_guardrail_endpoint(
mock_guardrail_registry.update_guardrail_in_db.assert_called_once()
mock_in_memory_handler.sync_guardrail_from_db.assert_called_once_with(guardrail=mocker.ANY)
mock_in_memory_handler.sync_guardrail_from_db.assert_called_once_with(
guardrail=mocker.ANY,
reject_invalid_logging_only_scope=False,
)
if scenario == "success_sync_fails_unexpected_error":
assert mock_logger is not None
@ -1280,6 +1470,192 @@ async def test_patch_guardrail_rejects_mcp_only_on_violation_with_422(mocker, mo
mock_guardrail_registry.update_guardrail_in_db.assert_not_called()
@pytest.mark.asyncio
async def test_patch_guardrail_rejects_invalid_logging_only_scope_with_422(mocker, mock_guardrail_registry):
mocker.patch("litellm.proxy.proxy_server.prisma_client", mocker.Mock())
mocker.patch(
"litellm.proxy.guardrails.guardrail_endpoints.GUARDRAIL_REGISTRY",
mock_guardrail_registry,
)
mock_in_memory_handler = mocker.Mock(spec=InMemoryGuardrailHandler)
mock_in_memory_handler.sync_guardrail_from_db.side_effect = ValueError(
"Guardrail test-db-guardrail: logging_only_scope is set, but mode does not include logging_only"
)
mocker.patch(
"litellm.proxy.guardrails.guardrail_registry.IN_MEMORY_GUARDRAIL_HANDLER",
mock_in_memory_handler,
)
request = PatchGuardrailRequest(litellm_params=BaseLitellmParams(mode="pre_call", logging_only_scope="input"))
with pytest.raises(HTTPException) as exc_info:
await patch_guardrail("test-guardrail-id", request, user_api_key_dict=MOCK_ADMIN_USER)
assert exc_info.value.status_code == 422
assert "update rejected" in str(exc_info.value.detail)
mock_in_memory_handler.sync_guardrail_from_db.assert_called_once_with(
guardrail=mocker.ANY,
reject_invalid_logging_only_scope=True,
)
@pytest.mark.asyncio
async def test_patch_guardrail_clears_scope_when_logging_only_mode_is_removed(
mocker, monkeypatch, mock_guardrail_registry
):
handler, stored_guardrail = _setup_patch_scope_guardrail(
mocker,
monkeypatch,
mock_guardrail_registry,
_PatchScopeSupportedGuardrail,
"patch_scope_supported_test",
{
"mode": ["pre_call", "logging_only"],
"logging_only_scope": "output",
"default_on": True,
},
)
request = PatchGuardrailRequest(litellm_params=BaseLitellmParams(mode=["pre_call"]))
try:
result = await patch_guardrail(
stored_guardrail["guardrail_id"],
request,
user_api_key_dict=MOCK_ADMIN_USER,
)
assert result["guardrail_id"] == stored_guardrail["guardrail_id"]
persisted_guardrail = mock_guardrail_registry.update_guardrail_in_db.call_args.kwargs["guardrail"]
assert persisted_guardrail["litellm_params"].logging_only_scope is None
callback = handler.guardrail_id_to_custom_guardrail[stored_guardrail["guardrail_id"]]
assert callback.logging_only_scope is None
assert callback.should_run_guardrail(data={}, event_type=GuardrailEventHooks.pre_call) is True
finally:
handler.delete_in_memory_guardrail(stored_guardrail["guardrail_id"])
@pytest.mark.asyncio
async def test_patch_guardrail_tolerates_stored_unsupported_scope_on_unrelated_update(
mocker, monkeypatch, mock_guardrail_registry, caplog
):
handler, stored_guardrail = _setup_patch_scope_guardrail(
mocker,
monkeypatch,
mock_guardrail_registry,
_PatchScopeUnsupportedGuardrail,
"patch_scope_unsupported_test",
{"mode": "logging_only", "logging_only_scope": "output", "default_on": True},
)
caplog.clear()
request = PatchGuardrailRequest(litellm_params=BaseLitellmParams(default_on=False))
try:
result = await patch_guardrail(
stored_guardrail["guardrail_id"],
request,
user_api_key_dict=MOCK_ADMIN_USER,
)
assert result["guardrail_id"] == stored_guardrail["guardrail_id"]
callback = handler.guardrail_id_to_custom_guardrail[stored_guardrail["guardrail_id"]]
assert callback.logging_only_scope is None
assert any("Ignoring logging_only_scope" in record.getMessage() for record in caplog.records)
finally:
handler.delete_in_memory_guardrail(stored_guardrail["guardrail_id"])
@pytest.mark.asyncio
async def test_patch_guardrail_tolerates_invalid_stored_scope_on_unrelated_update(
mocker, monkeypatch, mock_guardrail_registry
):
handler, stored_guardrail = _setup_patch_scope_guardrail(
mocker,
monkeypatch,
mock_guardrail_registry,
_PatchScopeUnsupportedGuardrail,
"patch_scope_invalid_literal_test",
{"mode": "logging_only", "logging_only_scope": "sideways", "default_on": True},
)
request = PatchGuardrailRequest(litellm_params=BaseLitellmParams(default_on=False))
try:
result = await patch_guardrail(
stored_guardrail["guardrail_id"],
request,
user_api_key_dict=MOCK_ADMIN_USER,
)
assert result["guardrail_id"] == stored_guardrail["guardrail_id"]
persisted_guardrail = mock_guardrail_registry.update_guardrail_in_db.call_args.kwargs["guardrail"]
assert persisted_guardrail["litellm_params"].logging_only_scope is None
assert persisted_guardrail["litellm_params"].default_on is False
finally:
handler.delete_in_memory_guardrail(stored_guardrail["guardrail_id"])
@pytest.mark.asyncio
async def test_patch_guardrail_rejects_explicit_unsupported_scope_and_rolls_back(
mocker, monkeypatch, mock_guardrail_registry
):
handler, stored_guardrail = _setup_patch_scope_guardrail(
mocker,
monkeypatch,
mock_guardrail_registry,
_PatchScopeUnsupportedGuardrail,
"patch_scope_unsupported_test",
{"mode": "logging_only", "logging_only_scope": "output", "default_on": True},
)
request = PatchGuardrailRequest(litellm_params=BaseLitellmParams(logging_only_scope="output"))
try:
with pytest.raises(HTTPException) as exc_info:
await patch_guardrail(
stored_guardrail["guardrail_id"],
request,
user_api_key_dict=MOCK_ADMIN_USER,
)
assert exc_info.value.status_code == 422
assert mock_guardrail_registry.update_guardrail_in_db.call_count == 2
restored_guardrail = mock_guardrail_registry.update_guardrail_in_db.call_args_list[-1].kwargs["guardrail"]
assert restored_guardrail["litellm_params"] == stored_guardrail["litellm_params"]
callback = handler.guardrail_id_to_custom_guardrail[stored_guardrail["guardrail_id"]]
assert callback.logging_only_scope is None
finally:
handler.delete_in_memory_guardrail(stored_guardrail["guardrail_id"])
@pytest.mark.asyncio
async def test_patch_guardrail_rejected_update_restores_invalid_stored_scope_verbatim(
mocker, monkeypatch, mock_guardrail_registry
):
handler, stored_guardrail = _setup_patch_scope_guardrail(
mocker,
monkeypatch,
mock_guardrail_registry,
_PatchScopeUnsupportedGuardrail,
"patch_scope_invalid_rollback_test",
{"mode": "logging_only", "logging_only_scope": "sideways", "default_on": True},
)
request = PatchGuardrailRequest(litellm_params=BaseLitellmParams(logging_only_scope="output"))
try:
with pytest.raises(HTTPException) as exc_info:
await patch_guardrail(
stored_guardrail["guardrail_id"],
request,
user_api_key_dict=MOCK_ADMIN_USER,
)
assert exc_info.value.status_code == 422
assert mock_guardrail_registry.update_guardrail_in_db.call_count == 2
restored_guardrail = mock_guardrail_registry.update_guardrail_in_db.call_args_list[-1].kwargs["guardrail"]
assert restored_guardrail["litellm_params"] == stored_guardrail["litellm_params"]
callback = handler.guardrail_id_to_custom_guardrail[stored_guardrail["guardrail_id"]]
assert callback.logging_only_scope is None
finally:
handler.delete_in_memory_guardrail(stored_guardrail["guardrail_id"])
@pytest.mark.parametrize(
"scenario,expected_result,expected_exception",
[
@ -2465,6 +2841,13 @@ async def test_ui_settings_map_matches_runtime_supported_event_hooks():
from litellm.proxy.guardrails.guardrail_registry import guardrail_class_registry
result = await get_guardrail_ui_settings()
expected_without_directional_scope = {
provider
for provider, guardrail_class in guardrail_class_registry.items()
if not guardrail_class.supports_logging_only_scope()
}
assert set(result.providers_without_directional_logging_only_scope) == expected_without_directional_scope
assert "xecguard" in result.providers_without_directional_logging_only_scope
for provider, guardrail_class in guardrail_class_registry.items():
declared = guardrail_class.get_supported_event_hooks()

View file

@ -1,15 +1,20 @@
import json
from collections.abc import Iterable, Iterator
from unittest.mock import AsyncMock, MagicMock, patch
from typing import ClassVar, Final
from unittest.mock import AsyncMock, MagicMock
import pytest
from pydantic import ValidationError
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.proxy.guardrails.guardrail_registry import (
get_guardrail_initializer_from_hooks,
GuardrailRegistry,
InMemoryGuardrailHandler,
get_guardrail_initializer_from_hooks,
parse_tolerant_litellm_params,
)
from litellm.types.guardrails import GuardrailEventHooks, Guardrail, LitellmParams
from litellm.types.guardrails import Guardrail, GuardrailEventHooks, LitellmParams, LoggingOnlyScope, Mode
from litellm.types.utils import GenericGuardrailAPIInputs
def test_get_guardrail_initializer_from_hooks():
@ -400,6 +405,85 @@ def test_sync_guardrail_from_db_marks_source_db_when_unchanged():
assert handler.get_source("collide") == "db"
def test_sync_guardrail_from_db_reject_flag_keeps_callback_order_on_noop_update():
"""
The PUT endpoint syncs the whole object with reject_invalid_logging_only_scope=True.
That strictness must not force a teardown + re-append of an unchanged guardrail:
initialize_guardrail appends the rebuilt callback at the END of litellm.callbacks,
so a description-only PUT would reorder guardrails and change which one wins
between a BLOCK and a MASK guardrail over the same content.
"""
import litellm
registry_module = _register_mode_following_initializer("mode_following_test")
lists = _all_callback_lists()
snapshots = [list(cb_list) for cb_list in lists]
sentinel: Final = CustomGuardrail(
guardrail_name="order-sentinel",
supported_event_hooks=[GuardrailEventHooks.pre_call],
event_hook=GuardrailEventHooks.pre_call,
)
try:
handler = InMemoryGuardrailHandler()
handler.initialize_guardrail(guardrail=_mode_following_db_row("123", "pre_call"), source="db")
original = handler.guardrail_id_to_custom_guardrail["123"]
assert original is not None
litellm.callbacks.append(sentinel)
index_before = litellm.callbacks.index(original)
handler.sync_guardrail_from_db(
guardrail=_mode_following_db_row("123", "pre_call", "description-only edit"),
reject_invalid_logging_only_scope=True,
)
assert handler.guardrail_id_to_custom_guardrail["123"] is original
assert litellm.callbacks.index(original) == index_before
assert _live_instances_named("mode-following") == 1
finally:
registry_module.guardrail_initializer_registry.pop("mode_following_test", None)
for cb_list, snapshot in zip(lists, snapshots):
cb_list[:] = snapshot
def test_sync_guardrail_from_db_reject_flag_still_rejects_invalid_unchanged_scope():
"""
A PUT sends the whole object, so an unchanged row that already carries an
invalid logging_only_scope (tolerated at load) must still be rejected on the
strict sync path, without rebuilding the live callback.
"""
registry_module = _register_mode_following_initializer("mode_following_test")
lists = _all_callback_lists()
snapshots = [list(cb_list) for cb_list in lists]
row: Final = Guardrail(
guardrail_id="123",
guardrail_name="mode-following",
litellm_params={
"guardrail": "mode_following_test",
"mode": "pre_call",
"default_on": True,
"logging_only_scope": "input",
},
guardrail_info={},
)
try:
handler = InMemoryGuardrailHandler()
handler.initialize_guardrail(guardrail=row, source="db")
original = handler.guardrail_id_to_custom_guardrail["123"]
assert original is not None
assert original.logging_only_scope is None # tolerated at load
with pytest.raises(ValueError, match="logging_only_scope is set"):
handler.sync_guardrail_from_db(guardrail=row, reject_invalid_logging_only_scope=True)
# Rejected without touching the live instance.
assert handler.guardrail_id_to_custom_guardrail["123"] is original
assert _live_instances_named("mode-following") == 1
finally:
registry_module.guardrail_initializer_registry.pop("mode_following_test", None)
for cb_list, snapshot in zip(lists, snapshots):
cb_list[:] = snapshot
@pytest.fixture
def rotation_handler() -> Iterator[InMemoryGuardrailHandler]:
registry_module = _register_mode_following_initializer("rotation_test")
@ -528,6 +612,49 @@ def test_unchanged_db_params_do_not_register_as_changed():
assert handler._has_guardrail_params_changed(gid, new) is False
def test_db_poll_does_not_reinitialize_config_guardrail_without_default_on():
handler = InMemoryGuardrailHandler()
guardrail_id: Final = "config-default-on-guardrail"
guardrail_name: Final = "config-default-on-guardrail"
params: Final = {
"guardrail": "litellm_content_filter",
"mode": "pre_call",
"logging_only_scope": "Input",
"blocked_words": [{"keyword": "synthetic blocked phrase", "action": "BLOCK"}],
}
callback_lists: Final = _all_callback_lists()
callback_snapshots: Final = [list(callback_list) for callback_list in callback_lists]
try:
existing: Final = handler.initialize_guardrail(
guardrail=Guardrail(
guardrail_id=guardrail_id,
guardrail_name=guardrail_name,
litellm_params=params,
),
source="config",
)
assert existing is not None
assert existing["litellm_params"].default_on is False
assert existing["litellm_params"].logging_only_scope is None
synced: Final = handler.sync_guardrail_from_db(
Guardrail(
guardrail_id=guardrail_id,
guardrail_name=guardrail_name,
litellm_params=params,
)
)
assert synced is existing
assert handler.IN_MEMORY_GUARDRAILS[guardrail_id] is existing
assert handler._sources[guardrail_id] == "db"
finally:
handler.delete_in_memory_guardrail(guardrail_id)
for callback_list, snapshot in zip(callback_lists, callback_snapshots):
callback_list[:] = snapshot
def test_changed_db_params_register_as_changed():
"""Normalizing both sides must still surface a genuine config change."""
handler = InMemoryGuardrailHandler()
@ -565,6 +692,24 @@ def test_unnormalizable_db_params_register_as_changed_without_raising():
assert handler._has_guardrail_params_changed(gid, new) is True
def test_invalid_scope_literal_db_params_compare_equal_after_normalization():
handler = InMemoryGuardrailHandler()
raw = _db_litellm_params()
gid = "77777777-7777-7777-7777-777777777777"
handler.IN_MEMORY_GUARDRAILS[gid] = Guardrail(
guardrail_id=gid,
guardrail_name="cf",
litellm_params=LitellmParams(**{**raw, "logging_only_scope": None}),
)
new = Guardrail(
guardrail_id=gid,
guardrail_name="cf",
litellm_params={**raw, "logging_only_scope": "Input"},
)
assert handler._has_guardrail_params_changed(gid, new) is False
def _all_callback_lists():
import litellm
@ -1025,6 +1170,250 @@ class TestScanOnlyToolResultsInitRefusal:
)
class _LoggingOnlyScopeSupportedGuardrail(CustomGuardrail):
async def apply_guardrail(
self,
inputs: GenericGuardrailAPIInputs,
request_data: dict[str, object],
input_type: str,
logging_obj: object | None = None,
) -> GenericGuardrailAPIInputs:
return inputs
class _LoggingOnlyScopeUnsupportedGuardrail(_LoggingOnlyScopeSupportedGuardrail):
async def async_logging_hook(
self,
kwargs: dict[str, object],
result: object,
call_type: str,
) -> tuple[dict[str, object], object]:
return kwargs, result
class _LoggingOnlyScopeNativeGuardrail(_LoggingOnlyScopeSupportedGuardrail):
use_native_lifecycle_hooks: ClassVar[bool] = True
def _invalid_scope_content_filter_guardrail() -> Guardrail:
return Guardrail(
guardrail_id="invalid-scope-content-filter-test",
guardrail_name="invalid-scope-content-filter",
litellm_params={
"guardrail": "litellm_content_filter",
"mode": "pre_call",
"logging_only_scope": "Input",
"blocked_words": [{"keyword": "pineapple", "action": "BLOCK"}],
},
)
class TestLoggingOnlyScopeValidation:
@pytest.mark.parametrize(
("scope", "expected_scope"),
(("input", "input"), ("Input", None)),
)
def test_tolerant_parser_preserves_default_on_constructor_coercion(
self, scope: str, expected_scope: str | None
) -> None:
params: Final = {
"guardrail": "litellm_content_filter",
"mode": "pre_call",
"logging_only_scope": scope,
"blocked_words": [{"keyword": "synthetic blocked phrase", "action": "BLOCK"}],
}
parsed: Final = parse_tolerant_litellm_params(params, "test-content-filter")
expected: Final = LitellmParams(**{**params, "logging_only_scope": expected_scope}).model_dump()
assert parsed.default_on is False
assert parsed.model_dump() == expected
def _initialize(
self,
mode: str | list[str] | Mode,
scope: LoggingOnlyScope | None,
callback_type: type[CustomGuardrail] = _LoggingOnlyScopeSupportedGuardrail,
reject_invalid_logging_only_scope: bool = False,
assert_registered: bool = False,
) -> CustomGuardrail:
import litellm
from litellm.proxy.guardrails import guardrail_registry as registry_module
guardrail_type: Final = "logging_only_scope_test"
created_callbacks: Final[list[CustomGuardrail]] = []
def _initializer(litellm_params: LitellmParams, guardrail: Guardrail) -> CustomGuardrail:
supported_event_hooks: Final = (
[GuardrailEventHooks.logging_only] if callback_type.use_native_lifecycle_hooks else None
)
callback: Final = callback_type(
guardrail_name=guardrail["guardrail_name"],
event_hook=litellm_params.mode,
default_on=True,
supported_event_hooks=supported_event_hooks,
)
litellm.logging_callback_manager.add_litellm_callback(callback)
created_callbacks.append(callback)
return callback
registry_module.guardrail_initializer_registry[guardrail_type] = _initializer
lists: Final = _all_callback_lists()
snapshots: Final = [list(callback_list) for callback_list in lists]
try:
handler: Final = InMemoryGuardrailHandler()
result: Final = handler.initialize_guardrail(
guardrail={
"guardrail_name": "logging-only-scope-guardrail",
"litellm_params": {
"guardrail": guardrail_type,
"mode": mode,
"logging_only_scope": scope,
},
},
reject_invalid_logging_only_scope=reject_invalid_logging_only_scope,
)
assert result is not None
callback: Final = handler.guardrail_id_to_custom_guardrail[result["guardrail_id"]]
assert callback is not None
if assert_registered:
assert callback in lists[0]
return callback
except ValueError:
callback: Final = created_callbacks[0]
assert all(callback not in callback_list for callback_list in lists)
raise
finally:
for callback_list, snapshot in zip(lists, snapshots):
callback_list[:] = snapshot
registry_module.guardrail_initializer_registry.pop(guardrail_type, None)
def test_scope_without_logging_only_mode_is_ignored_at_load(self) -> None:
callback: Final = self._initialize(mode="pre_call", scope="input", assert_registered=True)
assert callback.logging_only_scope is None
assert callback.should_run_guardrail(data={}, event_type=GuardrailEventHooks.pre_call) is True
def test_scope_without_logging_only_mode_is_rejected_for_api_writes(self) -> None:
with pytest.raises(ValueError, match="logging_only_scope is set") as exc_info:
self._initialize(mode="pre_call", scope="input", reject_invalid_logging_only_scope=True)
assert str(exc_info.value) == (
"Guardrail logging-only-scope-guardrail: logging_only_scope is set, but mode does not include "
"logging_only, so it would never apply. Add logging_only to mode or remove logging_only_scope."
)
@pytest.mark.parametrize(
"mode",
(
"logging_only",
["pre_call", "logging_only"],
Mode(tags={"audit": "logging_only"}, default="pre_call"),
),
)
def test_scope_accepts_logging_only_in_supported_mode_forms(self, mode: str | list[str] | Mode) -> None:
callback: Final = self._initialize(mode=mode, scope="input")
assert callback.logging_only_scope == "input"
def test_directional_scope_is_ignored_at_load_when_guardrail_owns_logging_hook(self) -> None:
callback: Final = self._initialize(
mode="logging_only",
scope="input",
callback_type=_LoggingOnlyScopeUnsupportedGuardrail,
assert_registered=True,
)
assert callback.logging_only_scope is None
def test_directional_scope_rejected_for_api_writes_when_guardrail_owns_logging_hook(self) -> None:
with pytest.raises(ValueError, match="logging_only_scope='input' is not supported") as exc_info:
self._initialize(
mode="logging_only",
scope="input",
callback_type=_LoggingOnlyScopeUnsupportedGuardrail,
reject_invalid_logging_only_scope=True,
)
assert str(exc_info.value) == (
"Guardrail logging-only-scope-guardrail: logging_only_scope='input' is not supported by this "
"guardrail, whose logging_only hook scans on its own. Remove logging_only_scope."
)
def test_both_scope_accepted_when_guardrail_owns_logging_hook(self) -> None:
callback: Final = self._initialize(
mode="logging_only",
scope="both",
callback_type=_LoggingOnlyScopeUnsupportedGuardrail,
)
assert callback.logging_only_scope == "both"
def test_output_scope_accepted_for_native_lifecycle_guardrail(self) -> None:
callback: Final = self._initialize(
mode="logging_only",
scope="output",
callback_type=_LoggingOnlyScopeNativeGuardrail,
)
assert callback.logging_only_scope == "output"
def test_invalid_scope_fails_litellm_params_validation(self) -> None:
with pytest.raises(ValidationError):
LitellmParams(guardrail="test", mode="logging_only", logging_only_scope="request")
def test_invalid_scope_literal_keeps_content_filter_registered_and_blocking(self) -> None:
import litellm
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import (
ContentFilterGuardrail,
)
handler: Final = InMemoryGuardrailHandler()
callback_lists: Final = _all_callback_lists()
callback_snapshots: Final = [list(callback_list) for callback_list in callback_lists]
guardrail: Final = _invalid_scope_content_filter_guardrail()
try:
result: Final = handler.initialize_guardrail(guardrail=guardrail, source="config")
assert result is not None
callback: Final = handler.guardrail_id_to_custom_guardrail[result["guardrail_id"]]
assert isinstance(callback, ContentFilterGuardrail)
assert callback in litellm.callbacks
assert callback.logging_only_scope is None
assert callback.event_hook == GuardrailEventHooks.pre_call
assert callback._check_blocked_words("pineapple") is not None
finally:
handler.delete_in_memory_guardrail(guardrail["guardrail_id"])
for callback_list, snapshot in zip(callback_lists, callback_snapshots):
callback_list[:] = snapshot
def test_invalid_scope_literal_is_rejected_for_strict_initialization_without_callback_leakage(self) -> None:
handler: Final = InMemoryGuardrailHandler()
callback_lists: Final = _all_callback_lists()
callback_snapshots: Final = [list(callback_list) for callback_list in callback_lists]
with pytest.raises(ValueError, match="logging_only_scope"):
handler.initialize_guardrail(
guardrail=_invalid_scope_content_filter_guardrail(),
source="config",
reject_invalid_logging_only_scope=True,
)
assert all(callback_list == snapshot for callback_list, snapshot in zip(callback_lists, callback_snapshots))
def test_invalid_scope_literal_does_not_tolerate_other_litellm_params_errors(self) -> None:
with pytest.raises(ValidationError):
parse_tolerant_litellm_params(
{
"guardrail": "litellm_content_filter",
"mode": "pre_call",
"logging_only_scope": "Input",
"default_on": "not-a-bool",
},
"invalid-scope-content-filter",
)
@pytest.mark.asyncio
async def test_update_guardrail_in_db_raises_when_row_missing():
prisma_client = MagicMock()
@ -1044,6 +1433,41 @@ async def test_update_guardrail_in_db_raises_when_row_missing():
)
@pytest.mark.asyncio
async def test_update_guardrail_in_db_persists_raw_sparse_params_verbatim():
"""
After a rejected PATCH, the endpoint rolls back by writing the stored row's
raw litellm_params through update_guardrail_in_db. A raw dict must be
persisted exactly as stored — a legacy 4-key row stays a 4-key row — instead
of being round-tripped through LitellmParams.model_dump(), which materializes
every field default and rewrites a row the admin never wrote.
"""
prisma_client = MagicMock()
prisma_client.db.litellm_guardrailstable.update = AsyncMock(
return_value={"guardrail_id": "legacy-row", "guardrail_name": "legacy-one"}
)
legacy_params: Final = {
"guardrail": "litellm_content_filter",
"mode": "pre_call",
"guardrail_name": "legacy-one",
"blocked_words": [{"keyword": "x", "action": "BLOCK"}],
}
await GuardrailRegistry().update_guardrail_in_db(
guardrail_id="legacy-row",
guardrail=Guardrail(
guardrail_id="legacy-row",
guardrail_name="legacy-one",
litellm_params=legacy_params,
guardrail_info={},
),
prisma_client=prisma_client,
)
persisted: Final = prisma_client.db.litellm_guardrailstable.update.call_args.kwargs["data"]
assert json.loads(persisted["litellm_params"]) == legacy_params
def test_reinitialize_guardrail_restores_previous_on_failure():
"""A reinitialization whose new params make the guardrail constructor raise must
restore the previous instance instead of leaving the guardrail silently removed:
@ -1185,7 +1609,6 @@ _ENCRYPTED_PREFIX = "litellm_enc::"
class _Row(dict[str, object]):
def __getattr__(self, name: str) -> object:
return self[name]

View file

@ -1,11 +1,17 @@
"use client";
import { CircleHelp } from "lucide-react";
import React, { useId } from "react";
import React, { useEffect, useId } from "react";
import { useController, type Control, type ControllerRenderProps, type RegisterOptions } from "react-hook-form";
import { Field, FieldDescription, FieldError, FieldLabel } from "@/components/ui/field";
import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
import { Tooltip, TooltipContent, TooltipTrigger } from "@/components/ui/tooltip";
import {
getLoggingOnlyScopeOptions,
modeIncludesLoggingOnly,
normalizeLoggingOnlyScopeChoice,
type LoggingOnlyScopeChoice,
} from "./guardrail_info_helpers";
export interface GuardrailCriterion {
name: string;
@ -15,6 +21,7 @@ export interface GuardrailCriterion {
export interface GuardrailFormValues extends Record<string, unknown> {
criteria?: GuardrailCriterion[];
logging_only_scope_choice?: LoggingOnlyScopeChoice;
}
export type GuardrailFormControl = Control<GuardrailFormValues>;
export type GuardrailFieldRules = Pick<RegisterOptions<GuardrailFormValues, string>, "validate">;
@ -38,6 +45,13 @@ export const asText = (value: unknown): string => {
return "";
};
const LOGGING_ONLY_SCOPE_CHOICES: ReadonlySet<string> = new Set(
getLoggingOnlyScopeOptions(true).map(({ value }) => value),
);
const isLoggingOnlyScopeChoice = (value: unknown): value is LoggingOnlyScopeChoice =>
typeof value === "string" && LOGGING_ONLY_SCOPE_CHOICES.has(value);
export const asStringArray = (value: unknown): string[] => {
if (Array.isArray(value)) return value.filter((entry): entry is string => typeof entry === "string");
if (typeof value === "string" && value !== "") return [value];
@ -123,3 +137,55 @@ export const SkipMessageSelect: React.FC<{ control: GuardrailFieldControlProps }
</Select>
);
};
export const LoggingOnlyScopeSelect: React.FC<{
control: GuardrailFieldControlProps;
directionalScopeSupported: boolean;
}> = ({ control, directionalScopeSupported }) => {
const { id, value, onChange, "aria-invalid": ariaInvalid, "aria-describedby": ariaDescribedBy } = control;
const items = getLoggingOnlyScopeOptions(directionalScopeSupported);
useEffect(() => {
const currentChoice = isLoggingOnlyScopeChoice(value) ? value : "default";
const choice = normalizeLoggingOnlyScopeChoice(currentChoice, directionalScopeSupported);
if (choice !== value) onChange(choice);
}, [value, directionalScopeSupported, onChange]);
return (
<Select items={items} value={asText(value) || "default"} onValueChange={onChange}>
<SelectTrigger id={id} aria-invalid={ariaInvalid} aria-describedby={ariaDescribedBy} className="w-full">
<SelectValue placeholder="Select an option" />
</SelectTrigger>
<SelectContent>
{items.map((item) => (
<SelectItem key={item.value} value={item.value}>
{item.label}
</SelectItem>
))}
</SelectContent>
</Select>
);
};
export const LoggingOnlyScopeField: React.FC<{
control: GuardrailFormControl;
mode: unknown;
directionalScopeSupported: boolean;
}> = ({ control, mode, directionalScopeSupported }) => {
if (!modeIncludesLoggingOnly(mode)) return null;
return (
<GuardrailField
control={control}
name="logging_only_scope_choice"
label={labelWithHint(
"Logging only scope",
"Which direction a logging_only scan observes. Observe-only scans never block; pre_call and post_call on this guardrail still block.",
)}
>
{(fieldControl) => (
<LoggingOnlyScopeSelect control={fieldControl} directionalScopeSupported={directionalScopeSupported} />
)}
</GuardrailField>
);
};

View file

@ -0,0 +1,43 @@
import React from "react";
import { Badge } from "@/components/ui/badge";
import { Card } from "@/components/ui/card";
import { formatGuardrailMode, formatLoggingOnlyScope, modeIncludesLoggingOnly } from "./guardrail_info_helpers";
type GuardrailModeParams = {
mode?: unknown;
default_on?: boolean;
logging_only_scope?: string | null;
};
export const GuardrailModeCard: React.FC<{ litellmParams: GuardrailModeParams }> = ({ litellmParams }) => (
<Card className="block p-6">
<p>Mode</p>
<div className="mt-2">
<h3 className="text-lg font-medium">{formatGuardrailMode(litellmParams.mode) || "-"}</h3>
<Badge variant={litellmParams.default_on ? "secondary" : "outline"}>
{litellmParams.default_on ? "Default On" : "Default Off"}
</Badge>
</div>
{modeIncludesLoggingOnly(litellmParams.mode) && (
<div className="mt-4">
<p>Logging only scope</p>
<h3 className="text-lg font-medium">{formatLoggingOnlyScope(litellmParams.logging_only_scope)}</h3>
</div>
)}
</Card>
);
export const GuardrailModeRows: React.FC<{ litellmParams: GuardrailModeParams }> = ({ litellmParams }) => (
<>
<div>
<p className="font-medium">Mode</p>
<div>{formatGuardrailMode(litellmParams.mode) || "-"}</div>
</div>
{modeIncludesLoggingOnly(litellmParams.mode) && (
<div>
<p className="font-medium">Logging only scope</p>
<div>{formatLoggingOnlyScope(litellmParams.logging_only_scope)}</div>
</div>
)}
</>
);

View file

@ -37,6 +37,7 @@ const uiSettings = {
supported_entities: [],
supported_actions: [],
supported_modes: ["pre_call", "post_call"],
providers_without_directional_logging_only_scope: [],
pii_entity_categories: [],
};
@ -105,6 +106,67 @@ describe("AddGuardrailForm create payload characterization", () => {
expect(payload()).toMatchObject({ litellm_params: { mode: ["pre_call", "post_call"] } });
});
it("sends the selected output logging-only scope", async () => {
vi.mocked(networking.getGuardrailUISettings).mockResolvedValue({
...uiSettings,
supported_modes: ["pre_call", "logging_only"],
});
const user = userEvent.setup({ delay: null });
renderForm();
await user.type(await screen.findByLabelText("Guardrail Name"), "my-bedrock");
await pickProvider(user, "Bedrock Guardrail");
await user.click(screen.getByLabelText("Mode"));
await user.click((await screen.findAllByText("logging_only")).at(-1) as HTMLElement);
await chooseSelectOption(user, await screen.findByLabelText("Logging only scope"), "Output only (response)");
await user.type(await screen.findByPlaceholderText("The guardrail id on Bedrock"), "gr-123");
await user.click(screen.getByRole("button", { name: "Next" }));
await user.click(await screen.findByRole("button", { name: "Create Guardrail" }));
await waitFor(() => expect(networking.createGuardrailCall).toHaveBeenCalledTimes(1));
expect(payload()?.litellm_params.logging_only_scope).toBe("output");
});
it("hides directional scope choices for providers that do not support them", async () => {
vi.mocked(networking.getGuardrailUISettings).mockResolvedValue({
...uiSettings,
supported_modes: ["pre_call", "logging_only"],
providers_without_directional_logging_only_scope: ["xecguard"],
});
vi.mocked(networking.getGuardrailProviderSpecificParams).mockResolvedValue({
...providerParams,
xecguard: { ui_friendly_name: "XecGuard" },
});
const user = userEvent.setup({ delay: null });
renderForm();
await pickProvider(user, "XecGuard");
await user.click(screen.getByLabelText("Mode"));
await user.click((await screen.findAllByText("logging_only")).at(-1) as HTMLElement);
await user.click(await screen.findByLabelText("Logging only scope"));
expect(screen.queryByRole("option", { name: "Input only (request)" })).not.toBeInTheDocument();
expect(screen.queryByRole("option", { name: "Output only (response)" })).not.toBeInTheDocument();
expect(screen.getByRole("option", { name: "Both (request and response)" })).toBeInTheDocument();
});
it("hides logging-only scope and omits it from a pre-call payload", async () => {
const user = userEvent.setup({ delay: null });
renderForm();
await user.type(await screen.findByLabelText("Guardrail Name"), "my-bedrock");
await pickProvider(user, "Bedrock Guardrail");
expect(screen.queryByLabelText("Logging only scope")).not.toBeInTheDocument();
await user.type(await screen.findByPlaceholderText("The guardrail id on Bedrock"), "gr-123");
await user.click(screen.getByRole("button", { name: "Next" }));
await user.click(await screen.findByRole("button", { name: "Create Guardrail" }));
await waitFor(() => expect(networking.createGuardrailCall).toHaveBeenCalledTimes(1));
expect(payload()?.litellm_params).not.toHaveProperty("logging_only_scope");
});
it("blocks Next when the user deselects every mode", async () => {
const user = userEvent.setup({ delay: null });
renderForm();

View file

@ -1,5 +1,5 @@
import React, { useEffect, useMemo, useState } from "react";
import { useForm, type UseFormReturn } from "react-hook-form";
import { useForm, useWatch, type UseFormReturn } from "react-hook-form";
import { toast } from "@/lib/toast";
import {
createGuardrailCall,
@ -10,18 +10,23 @@ import {
import ContentFilterConfiguration from "./content_filter/ContentFilterConfiguration";
import { type CompetitorIntentConfig } from "./content_filter/CompetitorIntentConfiguration";
import {
choiceToLoggingOnlyScope,
choiceToSkipSystemForCreate,
choiceToSkipToolForCreate,
getGuardrailLogo,
getGuardrailProviders,
getSupportedModesForProvider,
guardrail_provider_map,
modeIncludesLoggingOnly,
populateGuardrailProviderMap,
populateGuardrailProviders,
shouldRenderContentFilterConfigSettings,
shouldRenderLLMJudgeFields,
shouldRenderPIIConfigSettings,
supportsDirectionalLoggingOnlyScope,
toModeArray,
type LoggingOnlyScope,
type LoggingOnlyScopeChoice,
} from "./guardrail_info_helpers";
import { Logo } from "@/components/molecules/logo/Logo";
import { MultiSelect } from "@/components/shared/MultiSelect";
@ -49,6 +54,7 @@ import {
requiredRule,
type GuardrailCriterion,
type GuardrailFormValues,
LoggingOnlyScopeField,
SkipMessageSelect,
} from "./GuardrailFormField";
import GuardrailOptionalParams from "./guardrail_optional_params";
@ -92,6 +98,7 @@ interface GuardrailSettings {
supported_actions: string[];
supported_modes: string[];
supported_modes_by_provider?: Record<string, string[]>;
providers_without_directional_logging_only_scope?: string[];
pii_entity_categories: Array<{
category: string;
entities: string[];
@ -162,6 +169,7 @@ type SkipMessageChoice = "inherit" | "yes" | "no";
const INITIAL_VALUES: GuardrailFormValues = {
mode: "pre_call",
default_on: false,
logging_only_scope_choice: "default",
skip_system_message_choice: "inherit",
skip_tool_message_choice: "inherit",
};
@ -201,6 +209,7 @@ interface ProviderParamsResponse {
const AddGuardrailForm: React.FC<AddGuardrailFormProps> = ({ visible, onClose, accessToken, onSuccess, preset }) => {
const form = useForm<GuardrailFormValues>({ defaultValues: INITIAL_VALUES });
const watchedMode = useWatch({ control: form.control, name: "mode" });
const [loading, setLoading] = useState(false);
const [selectedProvider, setSelectedProvider] = useState<string | null>(null);
const [guardrailSettings, setGuardrailSettings] = useState<GuardrailSettings | null>(null);
@ -236,6 +245,7 @@ const AddGuardrailForm: React.FC<AddGuardrailFormProps> = ({ visible, onClose, a
const providerValue = guardrail_provider_map[selectedProvider];
return (providerValue || "").toLowerCase() === "tool_permission";
}, [selectedProvider]);
const directionalScopeSupported = supportsDirectionalLoggingOnlyScope(guardrailSettings, selectedProvider);
// Fetch guardrail UI settings + provider params on mount / accessToken change
useEffect(() => {
@ -279,6 +289,7 @@ const AddGuardrailForm: React.FC<AddGuardrailFormProps> = ({ visible, onClose, a
guardrail_name: preset.guardrailNameSuggestion,
mode: preset.mode,
default_on: preset.defaultOn,
logging_only_scope_choice: "default",
skip_system_message_choice: "inherit",
skip_tool_message_choice: "inherit",
};
@ -441,6 +452,7 @@ const AddGuardrailForm: React.FC<AddGuardrailFormProps> = ({ visible, onClose, a
guardrail_name: string;
litellm_params: {
guardrail: string;
logging_only_scope?: LoggingOnlyScope | null;
[key: string]: unknown; // Allow dynamic properties
};
guardrail_info: Record<string, unknown>;
@ -464,6 +476,13 @@ const AddGuardrailForm: React.FC<AddGuardrailFormProps> = ({ visible, onClose, a
guardrailData.litellm_params.skip_tool_message_in_guardrail = skipToolForCreate;
}
const loggingOnlyScope = choiceToLoggingOnlyScope(
values.logging_only_scope_choice as LoggingOnlyScopeChoice | undefined,
);
if (modeIncludesLoggingOnly(values.mode) && loggingOnlyScope !== null) {
guardrailData.litellm_params.logging_only_scope = loggingOnlyScope;
}
// For Presidio PII, add the entity and action configurations
if (providerKey === "PresidioPII" && selectedEntities.length > 0) {
const piiEntitiesConfig: { [key: string]: string } = {};
@ -798,6 +817,12 @@ const AddGuardrailForm: React.FC<AddGuardrailFormProps> = ({ visible, onClose, a
{(fieldControl) => <SkipMessageSelect control={fieldControl} />}
</GuardrailField>
<LoggingOnlyScopeField
control={form.control}
mode={watchedMode}
directionalScopeSupported={directionalScopeSupported}
/>
{/* Use the GuardrailProviderFields component to render provider-specific fields */}
{showProviderFields && (
<GuardrailProviderFields

View file

@ -0,0 +1,58 @@
import React from "react";
import {
choiceToLoggingOnlyScope,
getLoggingOnlyScopeOptions,
getLoggingOnlyScopeUpdate,
type LoggingOnlyScope,
type LoggingOnlyScopeChoice,
} from "../guardrail_info_helpers";
import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
const LOGGING_ONLY_SCOPE_ITEMS = getLoggingOnlyScopeOptions(true);
const LOGGING_ONLY_SCOPE_CHOICES: ReadonlySet<string> = new Set(LOGGING_ONLY_SCOPE_ITEMS.map(({ value }) => value));
const isLoggingOnlyScopeChoice = (value: unknown): value is LoggingOnlyScopeChoice =>
typeof value === "string" && LOGGING_ONLY_SCOPE_CHOICES.has(value);
export const CustomCodeLoggingOnlyScopeSelect: React.FC<{
value: LoggingOnlyScopeChoice;
onChange: (choice: LoggingOnlyScopeChoice) => void;
}> = ({ value, onChange }) => (
<div className="w-[200px]">
<label className="mb-1 block text-xs font-medium text-muted-foreground">Logging only scope</label>
<Select
items={LOGGING_ONLY_SCOPE_ITEMS}
value={value}
onValueChange={(nextValue) => {
if (isLoggingOnlyScopeChoice(nextValue)) onChange(nextValue);
}}
>
<SelectTrigger className="w-full" aria-label="Logging only scope">
<SelectValue />
</SelectTrigger>
<SelectContent>
{LOGGING_ONLY_SCOPE_ITEMS.map((option) => (
<SelectItem key={option.value} value={option.value}>
{option.label}
</SelectItem>
))}
</SelectContent>
</Select>
</div>
);
export const getCustomCodeLoggingOnlyScopeCreate = (
mode: string[],
choice: LoggingOnlyScopeChoice,
): { logging_only_scope?: LoggingOnlyScope } => {
if (!mode.includes("logging_only")) return {};
const scope = choiceToLoggingOnlyScope(choice);
return scope === null ? {} : { logging_only_scope: scope };
};
export const getCustomCodeLoggingOnlyScopeUpdate = (
mode: string[],
litellmParams: { logging_only_scope?: string | null } | null | undefined,
choice: LoggingOnlyScopeChoice,
): { logging_only_scope?: LoggingOnlyScope | null } =>
mode.includes("logging_only") ? getLoggingOnlyScopeUpdate(litellmParams, choice) : {};

View file

@ -1,5 +1,5 @@
import { describe, it, expect, vi, beforeEach } from "vitest";
import { render, screen, waitFor } from "@testing-library/react";
import { fireEvent, render, screen, waitFor } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import CustomCodeModal from "./CustomCodeModal";
import { createGuardrailCall, updateGuardrailCall, testCustomCodeGuardrail } from "@/components/networking";
@ -130,6 +130,127 @@ describe("CustomCodeModal", () => {
expect(screen.queryByText("logging_only")).not.toBeInTheDocument();
});
it("should update the logging-only scope in edit mode", async () => {
const user = userEvent.setup();
renderModal({
editData: {
guardrail_id: "g-1",
guardrail_name: "existing-guardrail",
litellm_params: {
mode: ["logging_only"],
logging_only_scope: "input",
custom_code: "def apply_guardrail(): pass",
},
},
});
const scopeSelect = await screen.findByRole("combobox", { name: "Logging only scope" });
expect(scopeSelect).toHaveTextContent("Input only (request)");
await user.click(scopeSelect);
await user.click(await screen.findByRole("option", { name: "Output only (response)" }));
await user.click(screen.getByRole("button", { name: /update guardrail/i }));
await waitFor(() => expect(mockUpdate).toHaveBeenCalledTimes(1));
expect(mockUpdate.mock.calls[0][2]).toMatchObject({
litellm_params: { logging_only_scope: "output" },
});
});
it("should clear the logging-only scope when Default is selected in edit mode", async () => {
const user = userEvent.setup();
renderModal({
editData: {
guardrail_id: "g-1",
guardrail_name: "existing-guardrail",
litellm_params: {
mode: ["logging_only"],
logging_only_scope: "input",
custom_code: "def apply_guardrail(): pass",
},
},
});
await user.click(await screen.findByRole("combobox", { name: "Logging only scope" }));
await user.click(await screen.findByRole("option", { name: "Default (request and response)" }));
await user.click(screen.getByRole("button", { name: /update guardrail/i }));
await waitFor(() => expect(mockUpdate).toHaveBeenCalledTimes(1));
expect(mockUpdate.mock.calls[0][2]).toHaveProperty("litellm_params.logging_only_scope", null);
});
it("should omit an unchanged logging-only scope from the edit payload", async () => {
renderModal({
editData: {
guardrail_id: "g-1",
guardrail_name: "existing-guardrail",
litellm_params: {
mode: ["logging_only"],
logging_only_scope: "input",
custom_code: "def apply_guardrail(): pass",
},
},
});
fireEvent.change(await screen.findByPlaceholderText("e.g., block-pii-custom"), {
target: { value: "renamed-guardrail" },
});
await userEvent.setup().click(screen.getByRole("button", { name: /update guardrail/i }));
await waitFor(() => expect(mockUpdate).toHaveBeenCalledTimes(1));
expect(mockUpdate.mock.calls[0][2]).not.toHaveProperty("litellm_params.logging_only_scope");
});
it("should not show logging-only scope for other modes in edit mode", async () => {
renderModal({
editData: {
guardrail_id: "g-1",
guardrail_name: "existing-guardrail",
litellm_params: { mode: "pre_call", custom_code: "def apply_guardrail(): pass" },
},
});
expect(await screen.findByText("Edit Custom Guardrail")).toBeInTheDocument();
expect(screen.queryByRole("combobox", { name: "Logging only scope" })).not.toBeInTheDocument();
});
it("should create a logging-only guardrail with its selected scope", async () => {
const user = userEvent.setup();
renderModal();
await user.click(screen.getAllByRole("combobox")[0]);
await user.keyboard("logging_only");
await user.click(await screen.findByRole("option", { name: "logging_only" }));
await user.click(await screen.findByRole("combobox", { name: "Logging only scope" }));
await user.click(await screen.findByRole("option", { name: "Input only (request)" }));
fireEvent.change(screen.getByPlaceholderText("e.g., block-pii-custom"), {
target: { value: "logging-only-guardrail" },
});
await user.click(screen.getByRole("button", { name: /save guardrail/i }));
await waitFor(() => expect(mockCreate).toHaveBeenCalledTimes(1));
expect(mockCreate.mock.calls[0][1]).toMatchObject({
litellm_params: { logging_only_scope: "input" },
});
});
it("should omit the default logging-only scope when creating a guardrail", async () => {
const user = userEvent.setup();
renderModal();
await user.click(screen.getAllByRole("combobox")[0]);
await user.keyboard("logging_only");
await user.click(await screen.findByRole("option", { name: "logging_only" }));
await screen.findByRole("combobox", { name: "Logging only scope" });
fireEvent.change(screen.getByPlaceholderText("e.g., block-pii-custom"), {
target: { value: "logging-only-guardrail" },
});
await user.click(screen.getByRole("button", { name: /save guardrail/i }));
await waitFor(() => expect(mockCreate).toHaveBeenCalledTimes(1));
expect(mockCreate.mock.calls[0][1]).not.toHaveProperty("litellm_params.logging_only_scope");
});
it("should expand the test section and run a test against the backend", async () => {
const user = userEvent.setup();
mockTest.mockResolvedValue({ success: true, result: { action: "allow" } } as never);

View file

@ -2,6 +2,13 @@ import React, { useState, useRef, useEffect } from "react";
import { CheckCircle2, ChevronRight, Code, ExternalLink, PlayCircle, Save, Users, XCircle } from "lucide-react";
import { createGuardrailCall, updateGuardrailCall, testCustomCodeGuardrail } from "@/components/networking";
import { toast } from "@/lib/toast";
import { loggingOnlyScopeToChoice } from "../guardrail_info_helpers";
import type { LoggingOnlyScope, LoggingOnlyScopeChoice } from "../guardrail_info_helpers";
import {
CustomCodeLoggingOnlyScopeSelect,
getCustomCodeLoggingOnlyScopeCreate,
getCustomCodeLoggingOnlyScopeUpdate,
} from "./CustomCodeLoggingOnlyScope";
import { Button } from "@/components/ui/button";
import { Collapsible, CollapsibleContent, CollapsibleTrigger } from "@/components/ui/collapsible";
import {
@ -178,6 +185,7 @@ export interface EditGuardrailData {
mode?: string | string[];
default_on?: boolean;
custom_code?: string;
logging_only_scope?: LoggingOnlyScope | null;
[key: string]: any;
};
}
@ -196,6 +204,7 @@ const CustomCodeModal: React.FC<CustomCodeModalProps> = ({ visible, onClose, onS
const isEditMode = !!editData;
const [guardrailName, setGuardrailName] = useState("");
const [mode, setMode] = useState<string[]>(["pre_call"]);
const [loggingOnlyScopeChoice, setLoggingOnlyScopeChoice] = useState<LoggingOnlyScopeChoice>("default");
const [defaultOn, setDefaultOn] = useState(false);
const [selectedTemplate, setSelectedTemplate] = useState<string>("empty");
const [code, setCode] = useState(CODE_TEMPLATES.empty.code);
@ -320,6 +329,7 @@ const CustomCodeModal: React.FC<CustomCodeModalProps> = ({ visible, onClose, onS
// Edit mode: populate with existing data
setGuardrailName(editData.guardrail_name || "");
setMode(normalizeMode(editData.litellm_params?.mode));
setLoggingOnlyScopeChoice(loggingOnlyScopeToChoice(editData.litellm_params?.logging_only_scope));
setDefaultOn(editData.litellm_params?.default_on || false);
setCode(editData.litellm_params?.custom_code || CODE_TEMPLATES.empty.code);
setSelectedTemplate(""); // No template selected in edit mode
@ -327,6 +337,7 @@ const CustomCodeModal: React.FC<CustomCodeModalProps> = ({ visible, onClose, onS
// Create mode: reset to defaults
setGuardrailName("");
setMode(["pre_call"]);
setLoggingOnlyScopeChoice("default");
setDefaultOn(false);
setSelectedTemplate("empty");
setCode(CODE_TEMPLATES.empty.code);
@ -384,6 +395,7 @@ const CustomCodeModal: React.FC<CustomCodeModalProps> = ({ visible, onClose, onS
const updateData: any = {
litellm_params: {
custom_code: code,
...getCustomCodeLoggingOnlyScopeUpdate(mode, editData.litellm_params, loggingOnlyScopeChoice),
},
};
@ -411,6 +423,7 @@ const CustomCodeModal: React.FC<CustomCodeModalProps> = ({ visible, onClose, onS
mode: mode,
default_on: defaultOn,
custom_code: code,
...getCustomCodeLoggingOnlyScopeCreate(mode, loggingOnlyScopeChoice),
},
guardrail_info: {},
};
@ -547,6 +560,9 @@ const CustomCodeModal: React.FC<CustomCodeModalProps> = ({ visible, onClose, onS
</ComboboxContent>
</Combobox>
</div>
{mode.includes("logging_only") && (
<CustomCodeLoggingOnlyScopeSelect value={loggingOnlyScopeChoice} onChange={setLoggingOnlyScopeChoice} />
)}
<div className="w-[180px]">
<label className="mb-1 block text-xs font-medium text-muted-foreground">Template</label>
<Select

View file

@ -25,6 +25,7 @@ const uiSettings = {
supported_actions: [],
pii_entity_categories: [],
supported_modes: ["pre_call", "post_call"],
providers_without_directional_logging_only_scope: [],
};
const bedrockParams = {
@ -164,6 +165,62 @@ describe("GuardrailInfoView update payload characterization", () => {
expect(lastPayload()).toEqual({ litellm_params: { skip_system_message_in_guardrail: true } });
});
it("shows and updates the logging-only scope", async () => {
const guardrailParams = {
guardrailIdentifier: "gr-abc",
api_key: "sk-old",
mode: "logging_only",
logging_only_scope: "input",
};
vi.mocked(networking.getGuardrailInfo).mockResolvedValue(guardrail(guardrailParams));
const user = userEvent.setup({ delay: null });
renderView();
expect(await screen.findAllByText("Input only (request)")).toHaveLength(2);
await openEditor(user);
await chooseSelectOption(user, screen.getByLabelText("Logging only scope"), "Output only (response)");
await saveChanges(user);
await waitFor(() => expect(networking.updateGuardrailCall).toHaveBeenCalledTimes(1));
expect(lastPayload()).toEqual({ litellm_params: { logging_only_scope: "output" } });
});
it("clears the logging-only scope when the edit choice returns to default", async () => {
const guardrailParams = {
guardrailIdentifier: "gr-abc",
api_key: "sk-old",
mode: "logging_only",
logging_only_scope: "input",
};
vi.mocked(networking.getGuardrailInfo).mockResolvedValue(guardrail(guardrailParams));
const user = userEvent.setup({ delay: null });
renderView();
await openEditor(user);
await chooseSelectOption(user, screen.getByLabelText("Logging only scope"), "Default (request and response)");
await saveChanges(user);
await waitFor(() => expect(networking.updateGuardrailCall).toHaveBeenCalledTimes(1));
expect(lastPayload()).toEqual({ litellm_params: { logging_only_scope: null } });
});
it("clears a stored directional scope for a provider that does not support it", async () => {
vi.mocked(networking.getGuardrailUISettings).mockResolvedValue({
...uiSettings,
providers_without_directional_logging_only_scope: ["bedrock"],
});
vi.mocked(networking.getGuardrailInfo).mockResolvedValue(
guardrail({ guardrailIdentifier: "gr-abc", mode: "logging_only", logging_only_scope: "output" }),
);
const user = userEvent.setup({ delay: null });
renderView();
await openEditor(user);
await saveChanges(user);
await waitFor(() => expect(networking.updateGuardrailCall).toHaveBeenCalledTimes(1));
expect(lastPayload()).toEqual({ litellm_params: { logging_only_scope: null } });
});
it("parses the guardrail information textarea into an object", async () => {
const user = userEvent.setup({ delay: null });
renderView();

View file

@ -28,16 +28,20 @@ import {
readRecord,
requiredRule,
type GuardrailFormValues,
LoggingOnlyScopeField,
SkipMessageSelect,
} from "./GuardrailFormField";
import ContentFilterManager, { formatContentFilterDataForAPI } from "./content_filter/ContentFilterManager";
import CustomCodeModal, { EditGuardrailData } from "./custom_code/CustomCodeModal";
import { GuardrailModeCard, GuardrailModeRows } from "./GuardrailModeDisplay";
import {
formatGuardrailMode,
getLoggingOnlyScopeUpdate,
getGuardrailLogoAndName,
guardrail_provider_map,
loggingOnlyScopeToChoice,
skipSystemMessageToChoice,
skipToolMessageToChoice,
supportsDirectionalLoggingOnlyScope,
type SkipSystemMessageChoice,
type SkipToolMessageChoice,
} from "./guardrail_info_helpers";
@ -81,6 +85,7 @@ const GuardrailInfoView: React.FC<GuardrailInfoProps> = ({ guardrailId, onClose,
entities: string[];
}>;
supported_modes: string[];
providers_without_directional_logging_only_scope?: string[];
content_filter_settings?: {
prebuilt_patterns: Array<{
name: string;
@ -109,6 +114,8 @@ const GuardrailInfoView: React.FC<GuardrailInfoProps> = ({ guardrailId, onClose,
const [toolPermissionConfig, setToolPermissionConfig] = useState<ToolPermissionConfig>(emptyToolPermissionConfig);
const [toolPermissionDirty, setToolPermissionDirty] = useState(false);
const [customCodeModalVisible, setCustomCodeModalVisible] = useState(false);
const guardrailProvider = guardrailData?.litellm_params?.guardrail ?? null;
const directionalScopeSupported = supportsDirectionalLoggingOnlyScope(guardrailSettings, guardrailProvider);
// Content Filter data ref (managed by ContentFilterManager)
const contentFilterDataRef = React.useRef<{
@ -219,6 +226,8 @@ const GuardrailInfoView: React.FC<GuardrailInfoProps> = ({ guardrailId, onClose,
if (!guardrailData) return;
form.setValue("guardrail_name", guardrailData.guardrail_name);
form.setValue("default_on", guardrailData.litellm_params?.default_on);
const storedLoggingOnlyScope = guardrailData.litellm_params?.logging_only_scope;
form.setValue("logging_only_scope_choice", loggingOnlyScopeToChoice(storedLoggingOnlyScope));
form.setValue(
"skip_system_message_choice",
skipSystemMessageToChoice(guardrailData.litellm_params?.skip_system_message_in_guardrail),
@ -282,7 +291,7 @@ const GuardrailInfoView: React.FC<GuardrailInfoProps> = ({ guardrailId, onClose,
// Prepare update data object - only include changed fields
const updateData: any = {
litellm_params: {},
litellm_params: getLoggingOnlyScopeUpdate(guardrailData.litellm_params, values.logging_only_scope_choice),
};
// Only include guardrail_name if it has changed
@ -556,17 +565,7 @@ const GuardrailInfoView: React.FC<GuardrailInfoProps> = ({ guardrailId, onClose,
</div>
</Card>
<Card className="block p-6">
<p>Mode</p>
<div className="mt-2">
<h3 className="text-lg font-medium">
{formatGuardrailMode(guardrailData.litellm_params?.mode) || "-"}
</h3>
<Badge variant={guardrailData.litellm_params?.default_on ? "secondary" : "outline"}>
{guardrailData.litellm_params?.default_on ? "Default On" : "Default Off"}
</Badge>
</div>
</Card>
<GuardrailModeCard litellmParams={guardrailData.litellm_params} />
<Card className="block p-6">
<p>Created At</p>
@ -745,6 +744,11 @@ const GuardrailInfoView: React.FC<GuardrailInfoProps> = ({ guardrailId, onClose,
>
{(fieldControl) => <SkipMessageSelect control={fieldControl} />}
</GuardrailField>
<LoggingOnlyScopeField
control={form.control}
mode={guardrailData.litellm_params?.mode}
directionalScopeSupported={directionalScopeSupported}
/>
{guardrailData.litellm_params?.guardrail === "presidio" && (
<>
<SectionHeading>PII Protection</SectionHeading>
@ -856,10 +860,7 @@ const GuardrailInfoView: React.FC<GuardrailInfoProps> = ({ guardrailId, onClose,
<p className="font-medium">Provider</p>
<div>{displayName}</div>
</div>
<div>
<p className="font-medium">Mode</p>
<div>{formatGuardrailMode(guardrailData.litellm_params?.mode) || "-"}</div>
</div>
<GuardrailModeRows litellmParams={guardrailData.litellm_params} />
<div>
<p className="font-medium">Default On</p>
<Badge variant={guardrailData.litellm_params?.default_on ? "secondary" : "outline"}>

View file

@ -15,6 +15,14 @@ import {
skipToolMessageToChoice,
choiceToSkipToolForCreate,
formatGuardrailMode,
loggingOnlyScopeToChoice,
choiceToLoggingOnlyScope,
getLoggingOnlyScopeUpdate,
getLoggingOnlyScopeOptions,
formatLoggingOnlyScope,
modeIncludesLoggingOnly,
normalizeLoggingOnlyScopeChoice,
supportsDirectionalLoggingOnlyScope,
} from "./guardrail_info_helpers";
describe("guardrail_info_helpers", () => {
@ -27,6 +35,7 @@ describe("guardrail_info_helpers", () => {
"PresidioPII",
"Bedrock",
"Lakera",
"Xecguard",
"LitellmContentFilter",
"ToolPermission",
"BlockCodeExecution",
@ -239,6 +248,100 @@ describe("guardrail_info_helpers", () => {
});
});
describe("logging-only scope helpers", () => {
it("normalizes directional choices only when the provider does not support them", () => {
expect(normalizeLoggingOnlyScopeChoice("input", false)).toBe("default");
expect(normalizeLoggingOnlyScopeChoice("output", false)).toBe("default");
expect(normalizeLoggingOnlyScopeChoice("both", false)).toBe("both");
expect(normalizeLoggingOnlyScopeChoice("default", false)).toBe("default");
expect(normalizeLoggingOnlyScopeChoice("input", true)).toBe("input");
expect(normalizeLoggingOnlyScopeChoice("output", true)).toBe("output");
});
it("maps API scope values to choices and back", () => {
expect(loggingOnlyScopeToChoice("input")).toBe("input");
expect(loggingOnlyScopeToChoice("output")).toBe("output");
expect(loggingOnlyScopeToChoice("both")).toBe("both");
expect(loggingOnlyScopeToChoice(undefined)).toBe("default");
expect(loggingOnlyScopeToChoice(null)).toBe("default");
expect(loggingOnlyScopeToChoice("invalid")).toBe("default");
expect(choiceToLoggingOnlyScope("default")).toBeNull();
expect(choiceToLoggingOnlyScope(undefined)).toBeNull();
expect(choiceToLoggingOnlyScope("input")).toBe("input");
expect(choiceToLoggingOnlyScope("output")).toBe("output");
expect(choiceToLoggingOnlyScope("both")).toBe("both");
expect(getLoggingOnlyScopeUpdate({ logging_only_scope: "input" }, "input")).toEqual({});
expect(getLoggingOnlyScopeUpdate({ logging_only_scope: "input" }, "output")).toEqual({
logging_only_scope: "output",
});
expect(getLoggingOnlyScopeUpdate({ logging_only_scope: "input" }, "default")).toEqual({
logging_only_scope: null,
});
});
it("formats every scope and falls back to default for missing or unknown values", () => {
expect(formatLoggingOnlyScope("input")).toBe("Input only (request)");
expect(formatLoggingOnlyScope("output")).toBe("Output only (response)");
expect(formatLoggingOnlyScope("both")).toBe("Both (request and response)");
expect(formatLoggingOnlyScope(undefined)).toBe("Default (request and response)");
expect(formatLoggingOnlyScope(null)).toBe("Default (request and response)");
expect(formatLoggingOnlyScope("invalid")).toBe("Default (request and response)");
});
it("detects logging_only in string, array, and tagged mode values", () => {
expect(modeIncludesLoggingOnly("logging_only")).toBe(true);
expect(modeIncludesLoggingOnly(["pre_call", "logging_only"])).toBe(true);
expect(
modeIncludesLoggingOnly({
tags: { "Service-Type: internal-service": "logging_only" },
default: "pre_call",
}),
).toBe(true);
expect(modeIncludesLoggingOnly("pre_call")).toBe(false);
});
it("filters directional options for unsupported providers and keeps all options otherwise", () => {
expect(getLoggingOnlyScopeOptions(false).map((option) => option.value)).toEqual(["default", "both"]);
expect(getLoggingOnlyScopeOptions(true).map((option) => option.value)).toEqual([
"default",
"input",
"output",
"both",
]);
expect(
supportsDirectionalLoggingOnlyScope(
{ providers_without_directional_logging_only_scope: ["xecguard"] },
"Xecguard",
),
).toBe(false);
expect(
supportsDirectionalLoggingOnlyScope(
{ providers_without_directional_logging_only_scope: ["xecguard"] },
"xecguard",
),
).toBe(false);
expect(
supportsDirectionalLoggingOnlyScope(
{ providers_without_directional_logging_only_scope: ["xecguard"] },
"Bedrock",
),
).toBe(true);
expect(
supportsDirectionalLoggingOnlyScope(
{ providers_without_directional_logging_only_scope: ["xecguard"] },
"unknown-provider",
),
).toBe(true);
expect(supportsDirectionalLoggingOnlyScope(null, "Xecguard")).toBe(true);
expect(getLoggingOnlyScopeOptions(supportsDirectionalLoggingOnlyScope(null, "Xecguard"))).toEqual(
getLoggingOnlyScopeOptions(true),
);
});
});
describe("skipSystemMessageToChoice / choiceToSkipSystemForCreate", () => {
it("maps API values to form choices and back for create", () => {
expect(skipSystemMessageToChoice(undefined)).toBe("inherit");

View file

@ -116,6 +116,74 @@ export const toModeArray = (raw: unknown): string[] => {
return [];
};
export type LoggingOnlyScope = "input" | "output" | "both";
export type LoggingOnlyScopeChoice = "default" | LoggingOnlyScope;
export type LoggingOnlyScopeOption = { label: string; value: LoggingOnlyScopeChoice };
export const normalizeLoggingOnlyScopeChoice = (
choice: LoggingOnlyScopeChoice,
directionalScopeSupported: boolean,
): LoggingOnlyScopeChoice =>
directionalScopeSupported || choice === "default" || choice === "both" ? choice : "default";
const LOGGING_ONLY_SCOPE_OPTIONS: LoggingOnlyScopeOption[] = [
{ label: "Default (request and response)", value: "default" },
{ label: "Input only (request)", value: "input" },
{ label: "Output only (response)", value: "output" },
{ label: "Both (request and response)", value: "both" },
];
export const loggingOnlyScopeToChoice = (v: string | null | undefined): LoggingOnlyScopeChoice =>
v === "input" || v === "output" || v === "both" ? v : "default";
export const choiceToLoggingOnlyScope = (choice: LoggingOnlyScopeChoice | undefined): LoggingOnlyScope | null =>
choice === "input" || choice === "output" || choice === "both" ? choice : null;
export const getLoggingOnlyScopeUpdate = (
litellmParams: { logging_only_scope?: string | null } | null | undefined,
choice: LoggingOnlyScopeChoice | undefined,
): { logging_only_scope?: LoggingOnlyScope | null } => {
if (choice === undefined || choice === loggingOnlyScopeToChoice(litellmParams?.logging_only_scope)) return {};
return { logging_only_scope: choiceToLoggingOnlyScope(choice) };
};
export const formatLoggingOnlyScope = (v: string | null | undefined): string => {
if (v === "input") return "Input only (request)";
if (v === "output") return "Output only (response)";
if (v === "both") return "Both (request and response)";
return "Default (request and response)";
};
export const modeIncludesLoggingOnly = (raw: unknown): boolean => {
if (toModeArray(raw).includes("logging_only")) return true;
if (raw === null || typeof raw !== "object") return false;
const { tags, default: fallback } = raw as { tags?: Record<string, unknown>; default?: unknown };
const taggedModes =
tags && typeof tags === "object"
? Object.values(tags).some((mode) => toModeArray(mode).includes("logging_only"))
: false;
return toModeArray(fallback).includes("logging_only") || taggedModes;
};
export const getLoggingOnlyScopeOptions = (directionalScopeSupported: boolean): LoggingOnlyScopeOption[] =>
directionalScopeSupported
? LOGGING_ONLY_SCOPE_OPTIONS
: LOGGING_ONLY_SCOPE_OPTIONS.filter((option) => option.value === "default" || option.value === "both");
export const supportsDirectionalLoggingOnlyScope = (
settings: { providers_without_directional_logging_only_scope?: string[] } | null,
selectedProvider: string | null,
): boolean => {
const providerKey = selectedProvider
? (
guardrail_provider_map[selectedProvider] ??
Object.values(guardrail_provider_map).find((value) => value.toLowerCase() === selectedProvider.toLowerCase())
)?.toLowerCase()
: null;
return !providerKey || !settings?.providers_without_directional_logging_only_scope?.includes(providerKey);
};
export const formatGuardrailMode = (raw: unknown): string => {
const flat: string[] = toModeArray(raw);
if (flat.length > 0) return flat.join(", ");

View file

@ -27074,6 +27074,11 @@ export interface components {
* @description Google Cloud location/region (e.g., us-central1)
*/
location?: string | null;
/**
* Logging Only Scope
* @description which direction a logging_only scan observes: 'input' (request), 'output' (response), or 'both' (default). Only applies to mode logging_only; pre_call/post_call on the same guardrail keep blocking.
*/
logging_only_scope?: ("input" | "output" | "both") | null;
/**
* Mask Request Content
* @description Will mask request content if guardrail makes any changes
@ -33335,6 +33340,27 @@ export interface components {
/** Output Text */
output_text: string;
};
/** GuardrailUIAddGuardrailSettings */
GuardrailUIAddGuardrailSettings: {
/** Content Filter Settings */
content_filter_settings?: {
[key: string]: unknown;
} | null;
/** Pii Entity Categories */
pii_entity_categories: components["schemas"]["PiiEntityCategoryMap"][];
/** Providers Without Directional Logging Only Scope */
providers_without_directional_logging_only_scope: string[];
/** Supported Actions */
supported_actions: string[];
/** Supported Entities */
supported_entities: string[];
/** Supported Modes */
supported_modes: string[];
/** Supported Modes By Provider */
supported_modes_by_provider: {
[key: string]: string[];
};
};
/**
* HTTPAuthSecurityScheme
* @description Defines a security scheme using HTTP authentication.
@ -37044,6 +37070,11 @@ export interface components {
* @description Google Cloud location/region (e.g., us-central1)
*/
location?: string | null;
/**
* Logging Only Scope
* @description which direction a logging_only scan observes: 'input' (request), 'output' (response), or 'both' (default). Only applies to mode logging_only; pre_call/post_call on the same guardrail keep blocking.
*/
logging_only_scope?: ("input" | "output" | "both") | null;
/**
* Mask
* @description Enable content masking using Lasso classifix API
@ -40678,6 +40709,13 @@ export interface components {
* @enum {string}
*/
PiiAction: "BLOCK" | "MASK";
/** PiiEntityCategoryMap */
PiiEntityCategoryMap: {
/** Category */
category: string;
/** Entities */
entities: string[];
};
/**
* PiiEntityType
* @enum {string}
@ -60814,7 +60852,7 @@ export interface operations {
[name: string]: unknown;
};
content: {
"application/json": unknown;
"application/json": components["schemas"]["GuardrailUIAddGuardrailSettings"];
};
};
};