diff --git a/backend/app/alembic/versions/083_add_llm_call_metadata.py b/backend/app/alembic/versions/083_add_llm_call_metadata.py new file mode 100644 index 000000000..53e4e9f0f --- /dev/null +++ b/backend/app/alembic/versions/083_add_llm_call_metadata.py @@ -0,0 +1,40 @@ +"""Add llm_call.metadata (generic extensibility catch-all) + +Revision ID: 083 +Revises: 082 +Create Date: 2026-09-08 00:00:00.000000 + +Re-adds a `metadata` JSONB column on `llm_call`, previously dropped by 079 +(as collateral of an unrelated feature revert, not because the column was a +bad idea). This time it's a generic catch-all, mirroring `llm_chain.metadata` +(added via `metadata_` in the model to dodge SQLAlchemy's reserved +`Base.metadata` attribute) — first use case: persisting input/output +guardrail results so /llm/call polling (GET /llm/call/{job_id}) can surface +them, matching what's already sent on the callback payload's `metadata` +field. +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision = "083" +down_revision = "082" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column( + "llm_call", + sa.Column( + "metadata", + postgresql.JSONB(astext_type=sa.Text()), + nullable=True, + comment="Future-proof extensibility catch-all (e.g. guardrail results)", + ), + ) + + +def downgrade(): + op.drop_column("llm_call", "metadata") diff --git a/backend/app/api/routes/guardrails.py b/backend/app/api/routes/guardrails.py index 3f0db3ad1..afd8ab4b3 100644 --- a/backend/app/api/routes/guardrails.py +++ b/backend/app/api/routes/guardrails.py @@ -1,7 +1,9 @@ import logging +from typing import Annotated, Any from uuid import UUID -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Query, Response +from fastapi.responses import JSONResponse from opentelemetry import trace from app.api.deps import AuthContextDep, SessionDep @@ -17,6 +19,7 @@ GuardrailsRequest, ) from app.services.guardrails.jobs import start_job +from app.services.llm.guardrails import proxy_guardrails_request from app.utils import APIResponse, load_description, validate_callback_url logger = logging.getLogger(__name__) @@ -91,6 +94,301 @@ def apply_guardrails_endpoint( ) +def _upstream_response(status_code: int, payload: Any) -> Response: + """An empty upstream body must stay empty (204s cannot carry one). + + Tenant-scoped data must never be cached by a shared/intermediary cache + (CWE-525); `no-store` is stronger than `private` for that guarantee. + """ + headers = {"Cache-Control": "no-store"} + if payload is None: + return Response(status_code=status_code, headers=headers) + return JSONResponse(status_code=status_code, content=payload, headers=headers) + + +# ROUTE ORDERING: every fixed single-segment path below collides with the +# GET /guardrails/{job_id} route declared after this section. FastAPI matches in +# declaration order and does not fall through when {job_id} fails UUID parsing, +# so these must stay above it. + + +@router.get( + "/guardrails", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def list_guardrails_validator_types(_current_user: AuthContextDep) -> Response: + """List the validator types supported upstream and their JSON schemas.""" + status_code, payload = proxy_guardrails_request( + "GET", + "/", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + ) + return _upstream_response(status_code, payload) + + +@router.post( + "/guardrails/ban_lists", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def create_guardrails_ban_list( + _current_user: AuthContextDep, body: dict[str, Any] +) -> Response: + status_code, payload = proxy_guardrails_request( + "POST", + "/ban_lists/", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + json_body=body, + ) + return _upstream_response(status_code, payload) + + +@router.get( + "/guardrails/ban_lists", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def list_guardrails_ban_lists( + _current_user: AuthContextDep, + domain: str | None = None, + offset: Annotated[int, Query(ge=0)] = 0, + limit: Annotated[int | None, Query(ge=1, le=100)] = None, +) -> Response: + status_code, payload = proxy_guardrails_request( + "GET", + "/ban_lists/", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + params={"domain": domain, "offset": offset, "limit": limit}, + ) + return _upstream_response(status_code, payload) + + +@router.post( + "/guardrails/llm_prompt_configs", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def create_guardrails_llm_prompt_config( + _current_user: AuthContextDep, body: dict[str, Any] +) -> Response: + status_code, payload = proxy_guardrails_request( + "POST", + "/llm_prompt_configs/", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + json_body=body, + ) + return _upstream_response(status_code, payload) + + +@router.get( + "/guardrails/llm_prompt_configs", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def list_guardrails_llm_prompt_configs( + _current_user: AuthContextDep, + validator_name: str | None = None, + offset: Annotated[int, Query(ge=0)] = 0, + limit: Annotated[int | None, Query(ge=1, le=100)] = None, +) -> Response: + status_code, payload = proxy_guardrails_request( + "GET", + "/llm_prompt_configs/", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + params={"validator_name": validator_name, "offset": offset, "limit": limit}, + ) + return _upstream_response(status_code, payload) + + +@router.post( + "/guardrails/validators/configs", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def create_guardrails_validator_config( + _current_user: AuthContextDep, body: dict[str, Any] +) -> Response: + status_code, payload = proxy_guardrails_request( + "POST", + "/validators/configs/", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + json_body=body, + ) + return _upstream_response(status_code, payload) + + +@router.get( + "/guardrails/validators/configs", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def list_guardrails_validator_configs( + _current_user: AuthContextDep, + ids: Annotated[list[UUID] | None, Query()] = None, + stage: str | None = None, + type: str | None = None, +) -> Response: + status_code, payload = proxy_guardrails_request( + "GET", + "/validators/configs/", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + params={ + "ids": [str(config_id) for config_id in ids] if ids else None, + "stage": stage, + "type": type, + }, + ) + return _upstream_response(status_code, payload) + + +@router.get( + "/guardrails/validators/configs/{config_id}", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def get_guardrails_validator_config( + _current_user: AuthContextDep, config_id: UUID +) -> Response: + status_code, payload = proxy_guardrails_request( + "GET", + f"/validators/configs/{config_id}", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + ) + return _upstream_response(status_code, payload) + + +@router.patch( + "/guardrails/validators/configs/{config_id}", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def update_guardrails_validator_config( + _current_user: AuthContextDep, config_id: UUID, body: dict[str, Any] +) -> Response: + status_code, payload = proxy_guardrails_request( + "PATCH", + f"/validators/configs/{config_id}", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + json_body=body, + ) + return _upstream_response(status_code, payload) + + +@router.delete( + "/guardrails/validators/configs/{config_id}", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def delete_guardrails_validator_config( + _current_user: AuthContextDep, config_id: UUID +) -> Response: + status_code, payload = proxy_guardrails_request( + "DELETE", + f"/validators/configs/{config_id}", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + ) + return _upstream_response(status_code, payload) + + +@router.get( + "/guardrails/ban_lists/{ban_list_id}", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def get_guardrails_ban_list( + _current_user: AuthContextDep, ban_list_id: UUID +) -> Response: + status_code, payload = proxy_guardrails_request( + "GET", + f"/ban_lists/{ban_list_id}", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + ) + return _upstream_response(status_code, payload) + + +@router.patch( + "/guardrails/ban_lists/{ban_list_id}", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def update_guardrails_ban_list( + _current_user: AuthContextDep, ban_list_id: UUID, body: dict[str, Any] +) -> Response: + status_code, payload = proxy_guardrails_request( + "PATCH", + f"/ban_lists/{ban_list_id}", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + json_body=body, + ) + return _upstream_response(status_code, payload) + + +@router.delete( + "/guardrails/ban_lists/{ban_list_id}", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def delete_guardrails_ban_list( + _current_user: AuthContextDep, ban_list_id: UUID +) -> Response: + status_code, payload = proxy_guardrails_request( + "DELETE", + f"/ban_lists/{ban_list_id}", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + ) + return _upstream_response(status_code, payload) + + +@router.get( + "/guardrails/llm_prompt_configs/{prompt_config_id}", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def get_guardrails_llm_prompt_config( + _current_user: AuthContextDep, prompt_config_id: UUID +) -> Response: + status_code, payload = proxy_guardrails_request( + "GET", + f"/llm_prompt_configs/{prompt_config_id}", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + ) + return _upstream_response(status_code, payload) + + +@router.patch( + "/guardrails/llm_prompt_configs/{prompt_config_id}", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def update_guardrails_llm_prompt_config( + _current_user: AuthContextDep, prompt_config_id: UUID, body: dict[str, Any] +) -> Response: + status_code, payload = proxy_guardrails_request( + "PATCH", + f"/llm_prompt_configs/{prompt_config_id}", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + json_body=body, + ) + return _upstream_response(status_code, payload) + + +@router.delete( + "/guardrails/llm_prompt_configs/{prompt_config_id}", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def delete_guardrails_llm_prompt_config( + _current_user: AuthContextDep, prompt_config_id: UUID +) -> Response: + status_code, payload = proxy_guardrails_request( + "DELETE", + f"/llm_prompt_configs/{prompt_config_id}", + organization_id=_current_user.organization_.id, + project_id=_current_user.project_.id, + ) + return _upstream_response(status_code, payload) + + @router.get( "/guardrails/{job_id}", response_model=APIResponse[GuardrailsJobPublic], @@ -112,7 +410,7 @@ def get_guardrails_job_status( tag="guardrails", system="guardrails", lifecycle="api.guardrails.status", - job_id=job_id, + job_id=str(job_id), project_id=project_id, organization_id=_current_user.organization_.id, ): diff --git a/backend/app/api/routes/llm.py b/backend/app/api/routes/llm.py index 94d3b341e..91ee0d67e 100644 --- a/backend/app/api/routes/llm.py +++ b/backend/app/api/routes/llm.py @@ -189,6 +189,7 @@ def get_llm_call_status( raise HTTPException(status_code=404, detail="Job not found") llm_call_response = None + call_metadata: dict | None = None if job.status.value == JobStatus.SUCCESS: llm_calls = get_llm_calls_by_job_id( session=session, job_id=job_id, project_id=project_id @@ -229,6 +230,7 @@ def get_llm_call_status( usage=Usage(**usage_payload), provider_raw_response=None, ) + call_metadata = llm_call.metadata_ job_response = LLMJobPublic( job_id=job.id, @@ -237,4 +239,4 @@ def get_llm_call_status( error_message=job.error_message, ) - return APIResponse.success_response(data=job_response) + return APIResponse.success_response(data=job_response, metadata=call_metadata) diff --git a/backend/app/celery/celery_app.py b/backend/app/celery/celery_app.py index e0ddf8dcd..0859f0c1c 100644 --- a/backend/app/celery/celery_app.py +++ b/backend/app/celery/celery_app.py @@ -13,7 +13,7 @@ from app.core.config import settings from app.core.logger import configure_logging -from app.core.sentry_filters import before_send_transaction_filter +from app.core.sentry_filters import before_send_filter, before_send_transaction_filter logger = logging.getLogger(__name__) _telemetry_initialized = False @@ -44,7 +44,11 @@ def _initialize_worker_observability() -> None: release=settings.API_VERSION, instrumenter="otel", traces_sample_rate=1.0, + # LLM input/output is end-user text; never attach request/response + # bodies to error events or trace transactions. + max_request_body_size="never", enable_logs=True, + before_send=before_send_filter, before_send_transaction=before_send_transaction_filter, integrations=[ LoggingIntegration( diff --git a/backend/app/celery/tasks/job_execution.py b/backend/app/celery/tasks/job_execution.py index 3ddd898f9..a408983c7 100644 --- a/backend/app/celery/tasks/job_execution.py +++ b/backend/app/celery/tasks/job_execution.py @@ -13,7 +13,8 @@ """ import logging -from typing import TYPE_CHECKING +from collections.abc import Callable +from typing import TYPE_CHECKING, TypeVar from asgi_correlation_id import correlation_id from celery import Task, current_task @@ -32,6 +33,8 @@ logger = logging.getLogger(__name__) +T = TypeVar("T") + # Sentinel correlation id used when no trace id is propagated from the # enqueueing request. Matches the codebase-wide "N/A" default (see # app/core/logger.py and app/celery/utils.py). @@ -43,7 +46,7 @@ def _set_trace(trace_id: str) -> None: logger.info(f"[_set_trace] Set correlation ID: {trace_id}") -def _extract_parent_context(task_instance) -> otel_context.Context: +def _extract_parent_context(task_instance: Task) -> otel_context.Context: """Extract OTel parent context from Celery headers if available.""" headers = getattr(task_instance.request, "headers", None) or {} carrier: dict[str, str] = {} @@ -62,20 +65,24 @@ def _extract_parent_context(task_instance) -> otel_context.Context: return extract(carrier) -def _run_with_otel_parent(task_instance, fn): - """Attach extracted parent context and execute function. - - When Celery auto-instrumentation is active, there is already a current - `run/...` span. Re-attaching extracted parent context here would make - service spans become siblings of `run/...` instead of children. +def _run_with_otel_parent( + task_instance: Task, fn: Callable[[], T] +) -> T: # noqa: UP047 (black doesn't support PEP 695 generics yet) + """Attach the extracted parent context and execute `fn` under it. - We only attach extracted context as a fallback when no active span exists. + opentelemetry-instrumentation-celery's own extraction (CeleryGetter) + reads headers via getattr(task.request, key), but propagation headers + live in task.request.headers — so it never finds them and its `run/...` + span is always an unparented root. We extract from `.headers` ourselves + (see _extract_parent_context) and attach that as current before running + the task body, so spans created inside `fn` correctly nest under the + enqueueing request's trace instead of starting a disconnected one. """ - current_ctx = trace.get_current_span().get_span_context() - if current_ctx and current_ctx.is_valid: + parent_ctx = _extract_parent_context(task_instance) + parent_span_ctx = trace.get_current_span(parent_ctx).get_span_context() + if not (parent_span_ctx and parent_span_ctx.is_valid): return fn() - parent_ctx = _extract_parent_context(task_instance) token = otel_context.attach(parent_ctx) try: return fn() diff --git a/backend/app/core/sentry_filters.py b/backend/app/core/sentry_filters.py index 6712803bd..6ac3720c7 100644 --- a/backend/app/core/sentry_filters.py +++ b/backend/app/core/sentry_filters.py @@ -2,6 +2,26 @@ from typing import Any +_REDACTED = "[REDACTED]" + +# sentry_sdk's CeleryIntegration attaches the task's raw args/kwargs to every +# event captured during that task as extra["celery-job"]. For these tasks, +# kwargs["request_data"] carries the caller's raw query text and (for chain +# jobs) prior block responses -- none of that should reach Sentry. +_LLM_JOB_TASK_NAMES = { + "app.celery.tasks.job_execution.run_llm_job", + "app.celery.tasks.job_execution.run_llm_chain_job", + "app.celery.tasks.job_execution.run_response_job", +} +_SENSITIVE_REQUEST_DATA_KEYS = ( + "query", + "request_metadata", + "response", + "output", + "callback_url", +) + + _SQL_OR_CONNECT = re.compile(r"^(select|insert|update|delete|connect)\b", re.IGNORECASE) _HTTP_SEND_RECEIVE = re.compile(r"http (send|receive)$", re.IGNORECASE) _DB_QUERY_SPAN = re.compile(r"^db\.query$", re.IGNORECASE) @@ -85,3 +105,46 @@ def before_send_transaction_filter( event["spans"] = filtered return event + + +def _redact_llm_job_kwargs(kwargs: dict[str, Any]) -> dict[str, Any]: + """Replaces end-user text fields and identifying URLs in an LLM job's + request_data with a placeholder, leaving non-sensitive fields (config, ids) + intact. + """ + request_data = kwargs.get("request_data") + if not isinstance(request_data, dict): + return kwargs + + redacted_request_data = dict(request_data) + for key in _SENSITIVE_REQUEST_DATA_KEYS: + if key in redacted_request_data: + redacted_request_data[key] = _REDACTED + + redacted_kwargs = dict(kwargs) + redacted_kwargs["request_data"] = redacted_request_data + return redacted_kwargs + + +def before_send_filter( + event: dict[str, Any], hint: dict[str, Any] +) -> dict[str, Any] | None: + """Strips end-user query/response text from LLM job celery-job context + before an event reaches Sentry. + """ + extra = event.get("extra") + if not isinstance(extra, dict): + return event + + celery_job = extra.get("celery-job") + if not isinstance(celery_job, dict): + return event + + if celery_job.get("task_name") not in _LLM_JOB_TASK_NAMES: + return event + + kwargs = celery_job.get("kwargs") + if isinstance(kwargs, dict): + celery_job["kwargs"] = _redact_llm_job_kwargs(kwargs) + + return event diff --git a/backend/app/core/telemetry.py b/backend/app/core/telemetry.py index 99d2fc959..c8effbfbc 100644 --- a/backend/app/core/telemetry.py +++ b/backend/app/core/telemetry.py @@ -14,6 +14,7 @@ from opentelemetry.instrumentation.logging import LoggingInstrumentor from opentelemetry.instrumentation.requests import RequestsInstrumentor from opentelemetry.instrumentation.utils import _SUPPRESS_HTTP_INSTRUMENTATION_KEY +from opentelemetry.propagate import set_global_textmap from opentelemetry.sdk.resources import SERVICE_NAME, Resource from opentelemetry.sdk.trace import TracerProvider @@ -192,9 +193,14 @@ def setup_telemetry(service_name: str | None = None) -> None: # Bridge OTel spans into Sentry as Sentry transactions and spans, with full attribute and error capture. if settings.SENTRY_DSN: - from sentry_sdk.integrations.opentelemetry import SentrySpanProcessor + from sentry_sdk.integrations.opentelemetry import ( + SentryPropagator, + SentrySpanProcessor, + ) tracer_provider.add_span_processor(SentrySpanProcessor()) + # Downstream services extract sentry-trace, not W3C traceparent. + set_global_textmap(SentryPropagator()) trace.set_tracer_provider(tracer_provider) diff --git a/backend/app/crud/llm.py b/backend/app/crud/llm.py index 44e605fb7..f96b598a1 100644 --- a/backend/app/crud/llm.py +++ b/backend/app/crud/llm.py @@ -65,6 +65,7 @@ def create_llm_call( organization_id: int, resolved_config: ConfigBlob, original_provider: str, + metadata: dict[str, Any] | None = None, ) -> LlmCall: """ Create a new LLM call record in the database. @@ -76,6 +77,7 @@ def create_llm_call( project_id: Project this LLM call belongs to organization_id: Organization this LLM call belongs to resolved_config: The resolved configuration blob (either from stored config or ad-hoc) + metadata: Extensibility catch-all dict (e.g. input guardrail results) Returns: LlmCall: The created LLM call record @@ -150,6 +152,7 @@ def create_llm_call( conversation_id=conversation_id, auto_create=auto_create, config=config_dict, + metadata_=metadata, ) session.add(db_llm_call) @@ -172,6 +175,7 @@ def update_llm_call_response( content: dict[str, Any] | None = None, usage: dict[str, Any] | None = None, conversation_id: str | None = None, + metadata: dict[str, Any] | None = None, ) -> LlmCall: """ Update an LLM call record with response data. @@ -183,6 +187,7 @@ def update_llm_call_response( content: Response content dict usage: Token usage dict conversation_id: Conversation ID if created/updated + metadata: Extensibility catch-all dict, merged into any existing value Returns: LlmCall: The updated LLM call record @@ -218,6 +223,10 @@ def update_llm_call_response( db_llm_call.usage = usage if conversation_id is not None: db_llm_call.conversation_id = conversation_id + if metadata is not None: + existing_metadata = dict(db_llm_call.metadata_ or {}) + existing_metadata.update(metadata) + db_llm_call.metadata_ = existing_metadata db_llm_call.updated_at = now() @@ -283,6 +292,7 @@ def save_rephrase_guardrail_call( resolved_config=config_blob, original_provider=str(config_blob.completion.provider), chain_id=chain_id, + metadata=request_metadata, ) try: update_llm_call_response( diff --git a/backend/app/main.py b/backend/app/main.py index 5d2a09cf0..fd97c03da 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -18,7 +18,7 @@ from app.core.exception_handlers import register_exception_handlers from app.core.logger import configure_logging from app.core.middleware import StripTrailingSlashMiddleware, http_request_logger -from app.core.sentry_filters import before_send_transaction_filter +from app.core.sentry_filters import before_send_filter, before_send_transaction_filter from app.core.telemetry import instrument_app, setup_telemetry from app.load_env import load_environment @@ -34,7 +34,11 @@ release=settings.API_VERSION, instrumenter="otel", traces_sample_rate=1.0, + # LLM input/output is end-user text; never attach request/response + # bodies to error events or trace transactions. + max_request_body_size="never", enable_logs=True, + before_send=before_send_filter, before_send_transaction=before_send_transaction_filter, integrations=[ LoggingIntegration( diff --git a/backend/app/models/llm/request.py b/backend/app/models/llm/request.py index c028a978a..63b2e7fbb 100644 --- a/backend/app/models/llm/request.py +++ b/backend/app/models/llm/request.py @@ -764,6 +764,16 @@ class LlmCall(SQLModel, table=True): ), ) + metadata_: dict[str, Any] | None = Field( + default=None, + sa_column=sa.Column( + "metadata", + JSONB, + nullable=True, + comment="Future-proof extensibility catch-all (e.g. guardrail results)", + ), + ) + # Timestamps inserted_at: datetime = Field( default_factory=now, diff --git a/backend/app/services/collections/providers/openai.py b/backend/app/services/collections/providers/openai.py index 770e0773f..84993268c 100644 --- a/backend/app/services/collections/providers/openai.py +++ b/backend/app/services/collections/providers/openai.py @@ -47,8 +47,14 @@ def upload_files( if doc.file_size_kb is None: doc.file_size_kb = round(tmp.tell() / 1024, 2) tmp.seek(0) + if "." in doc.fname: + filename, extension = doc.fname.rsplit(".", 1) + normalized_fname = f"{filename}.{extension.lower()}" + else: + normalized_fname = doc.fname uploaded = self.client.files.create( - file=(doc.fname, tmp), purpose="assistants" + file=(normalized_fname, tmp), + purpose="assistants", ) except Exception as err: logger.error( diff --git a/backend/app/services/llm/guardrails.py b/backend/app/services/llm/guardrails.py index 5b85e08b8..bf12512e7 100644 --- a/backend/app/services/llm/guardrails.py +++ b/backend/app/services/llm/guardrails.py @@ -1,16 +1,124 @@ import json import logging import time +from collections.abc import Sequence from dataclasses import dataclass, field -from typing import Any +from typing import Any, TypeGuard from uuid import UUID import httpx +from fastapi import HTTPException +from opentelemetry import trace from app.core.config import settings from app.models.llm.request import Validator logger = logging.getLogger(__name__) +tracer = trace.get_tracer(__name__) + + +GUARDRAILS_PROXY_TIMEOUT_SECONDS = 30.0 + + +def _guardrails_headers( + organization_id: int | None, project_id: int | None +) -> dict[str, str]: + """Guardrails is internal-only: tenant travels in headers set from the + auth context, never from caller-supplied body/query fields.""" + headers = { + "accept": "application/json", + "Authorization": f"Bearer {settings.KAAPI_GUARDRAILS_AUTH}", + "Content-Type": "application/json", + } + if organization_id is not None: + headers["X-ORGANIZATION-ID"] = str(organization_id) + if project_id is not None: + headers["X-PROJECT-ID"] = str(project_id) + return headers + + +def proxy_guardrails_request( + method: str, + path: str, + *, + organization_id: int, + project_id: int, + params: dict[str, Any] | None = None, + json_body: dict[str, Any] | None = None, +) -> tuple[int, Any]: + """Forward a management-API call to the guardrails service verbatim. + + No fail-open: these are synchronous CRUD calls, so upstream status codes and + bodies (including its 422s) are handed back to the caller unchanged. + """ + url = f"{settings.KAAPI_GUARDRAILS_URL}{path}" + headers = _guardrails_headers(organization_id, project_id) + # Unset query params must be omitted, not sent as empty values. + query = {k: v for k, v in (params or {}).items() if v is not None} + + logger.info( + f"[proxy_guardrails_request] Forwarding to guardrails | method: {method}, " + f"url: {url}, organization_id: {organization_id}, project_id: {project_id}" + ) + + try: + with ( + tracer.start_as_current_span( + f"guardrails.proxy {method} {path}", + attributes={ + "kaapi.organization_id": str(organization_id), + "kaapi.project_id": str(project_id), + }, + ), + httpx.Client(timeout=GUARDRAILS_PROXY_TIMEOUT_SECONDS) as client, + ): + response = client.request( + method, url, params=query, json=json_body, headers=headers + ) + except httpx.RequestError as e: + logger.error( + f"[proxy_guardrails_request] [KAAPI] Could not reach the guardrails service — " + f"retry shortly, and contact Kaapi if it persists (code: {type(e).__name__}) | " + f"method: {method}, url: {url}", + exc_info=True, + ) + raise HTTPException( + status_code=502, detail="Guardrails service unavailable" + ) from e + + if response.status_code >= 500: + logger.error( + f"[proxy_guardrails_request] [GUARDRAILS] Upstream error " + f"(code: {response.status_code}) | method: {method}, url: {url}" + ) + elif response.status_code >= 400: + logger.warning( + f"[proxy_guardrails_request] [GUARDRAILS] Request rejected " + f"(code: {response.status_code}) | method: {method}, url: {url}" + ) + + if not response.content: + return response.status_code, None + try: + return response.status_code, response.json() + except ValueError: + logger.error( + f"[proxy_guardrails_request] [GUARDRAILS] Non-JSON response body " + f"(code: {response.status_code}) | method: {method}, url: {url}" + ) + raise HTTPException( + status_code=502, detail="Guardrails service returned an invalid response" + ) from None + + +def _is_auth_error(e: Exception) -> TypeGuard[httpx.HTTPStatusError]: + # 422 included: a missing/invalid tenant header is a backend bug, not a + # transient outage, so it must not fall open like one. + return isinstance(e, httpx.HTTPStatusError) and e.response.status_code in ( + 401, + 403, + 422, + ) @dataclass @@ -95,7 +203,7 @@ def apply_guardrails( job_id, project_id, organization_id, - suppress_pass_logs=True, + suppress_pass_logs=False, output_text=output_text, ) @@ -132,13 +240,33 @@ def apply_guardrails( ) +def summarize_validator_results(outcome: GuardrailsOutcome) -> list[dict[str, Any]]: + """Per-validator name/outcome/text summary, extracted from the raw + guardrails service response, for use in llm_call metadata.""" + data = outcome.raw.get("data") or {} + validator_results = data.get("validator_results") or [] + + summaries = [] + for validator_result in validator_results: + summaries.append( + { + "name": validator_result.get("name"), + "outcome": validator_result.get("outcome"), + "error": validator_result.get("error"), + "input_text": validator_result.get("input_text"), + "output_text": validator_result.get("output_text"), + } + ) + return summaries + + def run_guardrails_validation( input_text: str, - guardrail_config: list[Validator | dict[str, Any]], + guardrail_config: Sequence[Validator | dict[str, Any]], job_id: UUID, project_id: int | None, organization_id: int | None, - suppress_pass_logs: bool = True, + suppress_pass_logs: bool = False, output_text: str | None = None, ) -> dict[str, Any]: """ @@ -166,8 +294,6 @@ def run_guardrails_validation( payload = { "request_id": str(job_id), - "project_id": project_id, - "organization_id": organization_id, "input": input_text, "validators": validators, } @@ -175,11 +301,7 @@ def run_guardrails_validation( if output_text is not None: payload["output"] = output_text - headers = { - "accept": "application/json", - "Authorization": f"Bearer {settings.KAAPI_GUARDRAILS_AUTH}", - "Content-Type": "application/json", - } + headers = _guardrails_headers(organization_id, project_id) url = f"{settings.KAAPI_GUARDRAILS_URL}/" payload_bytes = json.dumps(payload).encode() @@ -192,7 +314,18 @@ def run_guardrails_validation( started = time.monotonic() try: - with httpx.Client(timeout=45.0) as client: + with ( + tracer.start_as_current_span( + "guardrails.proxy.validate", + attributes={ + "kaapi.job_id": str(job_id), + "kaapi.organization_id": str(organization_id), + "kaapi.project_id": str(project_id), + "kaapi.validator_count": len(validators), + }, + ), + httpx.Client(timeout=45.0) as client, + ): response = client.post( url, json=payload, @@ -208,6 +341,21 @@ def run_guardrails_validation( return response.json() except Exception as e: elapsed_ms = int((time.monotonic() - started) * 1000) + if _is_auth_error(e): + # Auth failure means a broken deploy (token/IP mismatch), not a + # transient outage — fail the job instead of silently bypassing. + logger.error( + f"[run_guardrails_validation] Guardrails auth failed. " + f"job_id={job_id}, elapsed_ms={elapsed_ms}, error={e}" + ) + status_code = e.response.status_code + return { + "success": False, + "bypassed": False, + # Status only — str(e) embeds the internal service URL and this + # string is client-visible via job.error_message. + "error": f"Guardrails service rejected the request (HTTP {status_code})", + } logger.warning( f"[run_guardrails_validation] Service unavailable. Bypassing guardrails. " f"job_id={job_id}, elapsed_ms={elapsed_ms}, error={e}" @@ -247,24 +395,26 @@ def list_validators_config( if not input_validator_config_ids and not output_validator_config_ids: return [], [] - headers = { - "accept": "application/json", - "Authorization": f"Bearer {settings.KAAPI_GUARDRAILS_AUTH}", - "Content-Type": "application/json", - } + headers = _guardrails_headers(organization_id, project_id) endpoint = f"{settings.KAAPI_GUARDRAILS_URL}/validators/configs/" def _build_params(validator_ids: list[UUID]) -> dict[str, Any]: - params = { - "organization_id": organization_id, - "project_id": project_id, + return { "ids": [str(validator_config_id) for validator_config_id in validator_ids], } - return {key: value for key, value in params.items() if value is not None} try: - with httpx.Client(timeout=10.0) as client: + with ( + tracer.start_as_current_span( + "guardrails.proxy.list_validator_configs", + attributes={ + "kaapi.organization_id": str(organization_id), + "kaapi.project_id": str(project_id), + }, + ), + httpx.Client(timeout=10.0) as client, + ): def _fetch_by_ids(validator_ids: list[UUID]) -> list[dict[str, Any]]: if not validator_ids: @@ -300,6 +450,19 @@ def _fetch_by_ids(validator_ids: list[UUID]) -> list[dict[str, Any]]: return input_guardrails, output_guardrails except Exception as e: + if _is_auth_error(e): + # Propagate so job executors fail the job instead of running + # without guardrails on a misconfigured token/IP. Sanitized: + # str(e) embeds the internal service URL and the executors put + # this message into the client-visible job error. + logger.error( + f"[list_validators_config] Guardrails auth failed | " + f"organization_id={organization_id}, project_id={project_id}, " + f"endpoint={endpoint}, error={e}" + ) + raise ValueError( + f"Guardrails config fetch rejected (HTTP {e.response.status_code})" + ) from e logger.warning( "[list_validators_config] Guardrails service unavailable or invalid response. " "Proceeding without input/output guardrails. " diff --git a/backend/app/services/llm/jobs.py b/backend/app/services/llm/jobs.py index 1c32b0b5e..62f5baf34 100644 --- a/backend/app/services/llm/jobs.py +++ b/backend/app/services/llm/jobs.py @@ -67,7 +67,7 @@ Usage, ) from app.services.llm.chain.types import BlockResult -from app.services.llm.guardrails import apply_guardrails +from app.services.llm.guardrails import apply_guardrails, summarize_validator_results from app.services.llm.mappers import ( resolve_default_audio_provider, transform_kaapi_config_to_native, @@ -371,19 +371,13 @@ def apply_input_guardrails( job_id: UUID, project_id: int, organization_id: int, -) -> tuple[QueryParams, str | None, str | None]: +) -> tuple[QueryParams, str | None, str | None, dict[str, Any] | None]: """Apply input guardrails from a config_blob. Shared with llm-call and llm-chain. - Thin adapter over ``apply_guardrails`` that maps the outcome onto a - ``QueryParams`` payload. - - Returns (query, error, guardrail_direct_response) where: - - error is set when guardrails hard-block the request - - guardrail_direct_response is set when rephrase_needed=True and the safe_text - should be returned directly to the user without hitting the LLM + Returns (query, error, guardrail_direct_response, metadata). """ if not config_blob or not config_blob.input_guardrails: - return query, None, None + return query, None, None, None if not isinstance(query.input, TextInput): logger.info( @@ -391,29 +385,54 @@ def apply_input_guardrails( f"job_id={job_id}, " f"input_type={getattr(query.input, 'type', type(query.input).__name__)}" ) - return query, None, None + return query, None, None, None + original_input_text = query.input.content.value outcome = apply_guardrails( - text=query.input.content.value, + text=original_input_text, validators=config_blob.input_guardrails, job_id=job_id, project_id=project_id, organization_id=organization_id, ) + metadata = None + if outcome.applied: + metadata = { + "input_guardrail": { + "user_input": original_input_text, + "input_sent_to_llm": outcome.safe_text, + "validators": summarize_validator_results(outcome), + } + } if outcome.error is not None: - return query, outcome.error, None + return query, outcome.error, None, metadata if outcome.rephrase_needed: logger.info( f"[apply_input_guardrails] rephrase_needed=True, returning safe_text directly | job_id={job_id}" ) - return query, None, outcome.safe_text + return query, None, outcome.safe_text, metadata # No-op paths (no validators, bypassed) leave the query untouched. if outcome.applied and outcome.safe_text is not None: + if not outcome.safe_text.strip(): + # A fix-mode validator that supplies no fix_value (e.g. topic_relevance + # with no built-in fix) falls back to "" — forwarding that to the LLM + # provider fails with a confusing provider-side error instead of a + # clear guardrails-blocked one. + logger.warning( + f"[apply_input_guardrails] Guardrails reduced input to empty text; " + f"blocking request | job_id={job_id}" + ) + return ( + query, + "Input guardrails rejected the request and left no usable content.", + None, + metadata, + ) query.input.content.value = outcome.safe_text - return query, None, None + return query, None, None, metadata def apply_output_guardrails( @@ -427,9 +446,6 @@ def apply_output_guardrails( ) -> tuple[BlockResult, str | None]: """Apply output guardrails from a config_blob. Shared by /llm/call and /llm/chain. - Thin adapter over ``apply_guardrails`` that maps the outcome onto a - ``BlockResult``. - Returns (modified_result, None) on success, or (result, error_string) on failure. """ if not config_blob or not config_blob.output_guardrails: @@ -443,23 +459,68 @@ def apply_output_guardrails( ) return result, None + original_output_text = result.response.response.output.content.value outcome = apply_guardrails( text=input_text or "", validators=config_blob.output_guardrails, job_id=job_id, project_id=project_id, organization_id=organization_id, - output_text=result.response.response.output.content.value, + output_text=original_output_text, ) + if outcome.applied: + existing_metadata = result.metadata or {} + existing_metadata["output_guardrail"] = { + "llm_output_pre_guardrail": original_output_text, + "response_to_user": outcome.safe_text, + "validators": summarize_validator_results(outcome), + } + result.metadata = existing_metadata + if outcome.error is not None: return result, outcome.error if outcome.applied and outcome.safe_text is not None: + if not outcome.safe_text.strip(): + logger.warning( + f"[apply_output_guardrails] Guardrails reduced output to empty text; " + f"blocking response | job_id={job_id}" + ) + return ( + result, + "Output guardrails rejected the response and left no usable content.", + ) result.response.response.output.content.value = outcome.safe_text return result, None +def persist_output_guardrail_result( + *, + llm_call_id: UUID | None, + content: dict[str, Any] | None, + metadata: dict[str, Any] | None, +) -> None: + """Re-persists content/metadata onto the already-created LlmCall row + after output guardrails run. Uses its own session since the caller's + session may already be closed by this point.""" + if not llm_call_id: + return + if content is None and metadata is None: + return + try: + with Session(engine) as session: + update_llm_call_response( + session, llm_call_id=llm_call_id, content=content, metadata=metadata + ) + except Exception as e: + logger.error( + f"[persist_output_guardrail_result] Failed to persist guardrail " + f"result: {e} | llm_call_id={llm_call_id}", + exc_info=True, + ) + + DETECTED_LANGUAGE_FALLBACK = "en-IN" _TTS_LANGUAGE_KEYS = ("target_language_code", "language_code") @@ -582,13 +643,22 @@ def execute_llm_call( project_id=project_id, organization_id=organization_id, ) - query, input_error, guardrail_direct_response = apply_input_guardrails( + ( + query, + input_error, + guardrail_direct_response, + input_guardrail_metadata, + ) = apply_input_guardrails( config_blob=config_blob, query=query, job_id=job_id, project_id=project_id, organization_id=organization_id, ) + if input_guardrail_metadata: + if request_metadata is None: + request_metadata = {} + request_metadata.update(input_guardrail_metadata) if guardrail_direct_response is not None: # Runs before the Kaapi->native transform, so params may be # a typed model (Kaapi/proxy variants) rather than a dict. @@ -692,6 +762,7 @@ def execute_llm_call( resolved_config=config_blob, original_provider=Provider.PROXY.value, chain_id=chain_id, + metadata=request_metadata, ) llm_call_id = llm_call.id except Exception as e: @@ -853,6 +924,17 @@ def execute_llm_call( trace.Status(trace.StatusCode.ERROR, output_error) ) return BlockResult(error=output_error, llm_call_id=llm_call_id) + if config_blob.output_guardrails: + updated_content = None + if isinstance(result.response.response.output, TextOutput): + updated_content = ( + result.response.response.output.model_dump() + ) + persist_output_guardrail_result( + llm_call_id=llm_call_id, + content=updated_content, + metadata=result.metadata, + ) return result @@ -922,6 +1004,7 @@ def execute_llm_call( resolved_config=resolved_config_blob, original_provider=original_provider, chain_id=chain_id, + metadata=request_metadata, ) llm_call_id = llm_call.id _set_traceability_attributes(create_span, llm_call_id=llm_call_id) @@ -1234,6 +1317,15 @@ def execute_llm_call( trace.Status(trace.StatusCode.ERROR, output_error) ) return BlockResult(error=output_error, llm_call_id=llm_call_id) + if config_blob.output_guardrails: + updated_content = None + if isinstance(result.response.response.output, TextOutput): + updated_content = result.response.response.output.model_dump() + persist_output_guardrail_result( + llm_call_id=llm_call_id, + content=updated_content, + metadata=result.metadata, + ) return result diff --git a/backend/app/tests/api/routes/test_guardrails.py b/backend/app/tests/api/routes/test_guardrails.py index 8b34a67ce..7ac271b31 100644 --- a/backend/app/tests/api/routes/test_guardrails.py +++ b/backend/app/tests/api/routes/test_guardrails.py @@ -1,9 +1,14 @@ -from unittest.mock import patch +from contextlib import contextmanager +from typing import Any +from unittest.mock import MagicMock, patch from uuid import uuid4 +import httpx +import pytest from fastapi.testclient import TestClient from sqlmodel import Session +from app.core.config import settings from app.crud import JobCrud from app.models import Job, JobStatus, JobType, JobUpdate from app.tests.utils.auth import TestAuthContext @@ -11,6 +16,69 @@ VALIDATOR_ID = str(uuid4()) +# (client method, kaapi path, request body, upstream path) — "{id}" is filled +# with a freshly generated UUID by the tests that consume this table. +PROXY_ROUTES = [ + ("GET", "/guardrails", None, "/"), + ("POST", "/guardrails/ban_lists", {"name": "slurs"}, "/ban_lists/"), + ("GET", "/guardrails/ban_lists", None, "/ban_lists/"), + ("GET", "/guardrails/ban_lists/{id}", None, "/ban_lists/{id}"), + ("PATCH", "/guardrails/ban_lists/{id}", {"name": "renamed"}, "/ban_lists/{id}"), + ("DELETE", "/guardrails/ban_lists/{id}", None, "/ban_lists/{id}"), + ( + "POST", + "/guardrails/llm_prompt_configs", + {"validator_name": "toxicity", "prompt": "be nice"}, + "/llm_prompt_configs/", + ), + ("GET", "/guardrails/llm_prompt_configs", None, "/llm_prompt_configs/"), + ( + "GET", + "/guardrails/llm_prompt_configs/{id}", + None, + "/llm_prompt_configs/{id}", + ), + ( + "PATCH", + "/guardrails/llm_prompt_configs/{id}", + {"prompt": "be nicer"}, + "/llm_prompt_configs/{id}", + ), + ( + "DELETE", + "/guardrails/llm_prompt_configs/{id}", + None, + "/llm_prompt_configs/{id}", + ), + ( + "POST", + "/guardrails/validators/configs", + {"type": "pii", "stage": "input"}, + "/validators/configs/", + ), + ("GET", "/guardrails/validators/configs", None, "/validators/configs/"), + ( + "GET", + "/guardrails/validators/configs/{id}", + None, + "/validators/configs/{id}", + ), + ( + "PATCH", + "/guardrails/validators/configs/{id}", + {"stage": "output"}, + "/validators/configs/{id}", + ), + ( + "DELETE", + "/guardrails/validators/configs/{id}", + None, + "/validators/configs/{id}", + ), +] + +PROXY_ROUTE_IDS = [f"{method} {path}" for method, path, _, _ in PROXY_ROUTES] + def _payload(**overrides): body = { @@ -218,9 +286,345 @@ def test_get_guardrails_failed_returns_error_message( assert data["guardrails_response"] is None +# ---------- management-API proxy routes ---------- + + +class TestProxyPassthrough: + @pytest.mark.parametrize( + "status_code, body", + [ + (200, {"success": True, "data": [{"name": "pii"}]}), + (422, {"detail": [{"loc": ["body", "name"], "msg": "field required"}]}), + (404, {"detail": "Ban list not found"}), + ], + ) + def test_upstream_status_and_body_echoed( + self, + client: TestClient, + user_api_key_header: dict[str, str], + status_code: int, + body: dict[str, Any], + ) -> None: + with _mock_upstream(status_code=status_code, json_body=body): + resp = client.get("api/v1/guardrails", headers=user_api_key_header) + + assert resp.status_code == status_code + assert resp.json() == body + + def test_empty_upstream_body_returns_status_with_no_body( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + with _mock_upstream(status_code=204, content=b""): + resp = client.delete( + f"api/v1/guardrails/ban_lists/{uuid4()}", headers=user_api_key_header + ) + + assert resp.status_code == 204 + assert resp.content == b"" + + def test_create_echoes_upstream_status_and_body( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + created = {"id": str(uuid4()), "validator_name": "toxicity"} + # Upstream create routes return FastAPI's default 200. + with _mock_upstream(status_code=200, json_body=created) as calls: + resp = client.post( + "api/v1/guardrails/llm_prompt_configs", + json={"validator_name": "toxicity", "prompt": "be nice"}, + headers=user_api_key_header, + ) + + assert resp.status_code == 200 + assert resp.json() == created + assert calls[0]["kwargs"]["json"] == { + "validator_name": "toxicity", + "prompt": "be nice", + } + + def test_connect_error_returns_502( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + with _mock_upstream(raises=httpx.ConnectError("connection refused")): + resp = client.get("api/v1/guardrails", headers=user_api_key_header) + + assert resp.status_code == 502 + assert resp.json()["error"] == "Guardrails service unavailable" + + def test_non_json_upstream_body_returns_502( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + with _mock_upstream(status_code=200, content=b"gateway"): + resp = client.get("api/v1/guardrails", headers=user_api_key_header) + + assert resp.status_code == 502 + assert resp.json()["error"] == "Guardrails service returned an invalid response" + + +class TestProxyForwardedRequest: + @pytest.mark.parametrize( + "method, kaapi_path, body, upstream_path", PROXY_ROUTES, ids=PROXY_ROUTE_IDS + ) + def test_route_forwards_method_path_body_and_tenant( + self, + client: TestClient, + user_api_key: TestAuthContext, + user_api_key_header: dict[str, str], + method: str, + kaapi_path: str, + body: dict[str, Any] | None, + upstream_path: str, + ) -> None: + resource_id = uuid4() + upstream_body = {"ok": True} + with _mock_upstream(json_body=upstream_body) as calls: + resp = client.request( + method, + f"api/v1{kaapi_path.format(id=resource_id)}", + json=body, + headers=user_api_key_header, + ) + + assert resp.status_code == 200 + assert resp.json() == upstream_body + assert calls[0]["args"] == ( + method, + f"{settings.KAAPI_GUARDRAILS_URL}{upstream_path.format(id=resource_id)}", + ) + assert calls[0]["kwargs"]["json"] == body + headers = calls[0]["kwargs"]["headers"] + assert headers["X-ORGANIZATION-ID"] == str(user_api_key.organization_id) + assert headers["X-PROJECT-ID"] == str(user_api_key.project_id) + + def test_unset_limit_is_dropped_from_forwarded_params( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + with _mock_upstream(json_body={"data": []}) as calls: + client.get("api/v1/guardrails/ban_lists", headers=user_api_key_header) + + params = calls[0]["kwargs"]["params"] + assert params == {"offset": 0} + + def test_ids_forwarded_as_list( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + first, second = str(uuid4()), str(uuid4()) + with _mock_upstream(json_body={"data": []}) as calls: + client.get( + f"api/v1/guardrails/validators/configs?ids={first}&ids={second}" + "&stage=input", + headers=user_api_key_header, + ) + + assert calls[0]["kwargs"]["params"] == { + "ids": [first, second], + "stage": "input", + } + + def test_ban_list_filters_forwarded( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + with _mock_upstream(json_body={"data": []}) as calls: + client.get( + "api/v1/guardrails/ban_lists?domain=email&offset=5&limit=10", + headers=user_api_key_header, + ) + + assert calls[0]["kwargs"]["params"] == { + "domain": "email", + "offset": 5, + "limit": 10, + } + + def test_llm_prompt_config_filters_forwarded( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + with _mock_upstream(json_body={"data": []}) as calls: + client.get( + "api/v1/guardrails/llm_prompt_configs?validator_name=toxicity&limit=50", + headers=user_api_key_header, + ) + + assert calls[0]["kwargs"]["params"] == { + "validator_name": "toxicity", + "offset": 0, + "limit": 50, + } + + def test_validator_config_filters_forwarded_without_ids( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + with _mock_upstream(json_body={"data": []}) as calls: + client.get( + "api/v1/guardrails/validators/configs?stage=output&type=pii", + headers=user_api_key_header, + ) + + assert calls[0]["kwargs"]["params"] == {"stage": "output", "type": "pii"} + + def test_ids_are_normalised_to_canonical_uuid_strings( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + """Upstream expects strings; UUID parsing also canonicalises the casing.""" + config_id = uuid4() + with _mock_upstream(json_body={"data": []}) as calls: + client.get( + f"api/v1/guardrails/validators/configs?ids={str(config_id).upper()}", + headers=user_api_key_header, + ) + + assert calls[0]["kwargs"]["params"] == {"ids": [str(config_id)]} + + @pytest.mark.parametrize( + "kaapi_path", ["/guardrails/ban_lists", "/guardrails/llm_prompt_configs"] + ) + @pytest.mark.parametrize("query", ["limit=0", "limit=101", "offset=-1"]) + def test_out_of_range_pagination_rejected( + self, + client: TestClient, + user_api_key_header: dict[str, str], + kaapi_path: str, + query: str, + ) -> None: + resp = client.get(f"api/v1{kaapi_path}?{query}", headers=user_api_key_header) + assert resp.status_code == 422 + + def test_tenant_headers_come_from_auth_context_not_request( + self, + client: TestClient, + user_api_key: TestAuthContext, + user_api_key_header: dict[str, str], + ) -> None: + with _mock_upstream(status_code=200, json_body={"id": str(uuid4())}) as calls: + client.post( + "api/v1/guardrails/ban_lists?organization_id=999", + json={"name": "slurs", "organization_id": 999, "project_id": 888}, + headers=user_api_key_header, + ) + + headers = calls[0]["kwargs"]["headers"] + assert headers["X-ORGANIZATION-ID"] == str(user_api_key.organization_id) + assert headers["X-PROJECT-ID"] == str(user_api_key.project_id) + + def test_ban_list_detail_path_forwarded( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + ban_list_id = uuid4() + with _mock_upstream(json_body={"id": str(ban_list_id)}) as calls: + resp = client.patch( + f"api/v1/guardrails/ban_lists/{ban_list_id}", + json={"name": "renamed"}, + headers=user_api_key_header, + ) + + assert resp.status_code == 200 + method, url = calls[0]["args"] + assert method == "PATCH" + assert url.endswith(f"/ban_lists/{ban_list_id}") + + +class TestProxyRouteOrdering: + def test_ban_lists_list_route_wins_over_job_status_route( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + with _mock_upstream(json_body={"data": []}) as calls: + resp = client.get( + "api/v1/guardrails/ban_lists", headers=user_api_key_header + ) + + assert resp.status_code == 200 + assert calls[0]["args"][1].endswith("/ban_lists/") + + def test_validator_configs_list_and_detail_routes_do_not_collide( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + config_id = uuid4() + with _mock_upstream(json_body={"data": []}) as calls: + client.get( + "api/v1/guardrails/validators/configs", headers=user_api_key_header + ) + client.get( + f"api/v1/guardrails/validators/configs/{config_id}", + headers=user_api_key_header, + ) + + assert calls[0]["args"][1].endswith("/validators/configs/") + assert calls[1]["args"][1].endswith(f"/validators/configs/{config_id}") + + def test_llm_prompt_configs_list_route_wins_over_job_status_route( + self, client: TestClient, user_api_key_header: dict[str, str] + ) -> None: + with _mock_upstream(json_body={"data": []}) as calls: + resp = client.get( + "api/v1/guardrails/llm_prompt_configs", headers=user_api_key_header + ) + + assert resp.status_code == 200 + assert calls[0]["args"][1].endswith("/llm_prompt_configs/") + + +def test_list_ban_lists_requires_auth(client: TestClient) -> None: + resp = client.get("api/v1/guardrails/ban_lists") + assert resp.status_code in (401, 403) + + +@pytest.mark.parametrize( + "method, kaapi_path, body, _upstream_path", PROXY_ROUTES, ids=PROXY_ROUTE_IDS +) +def test_proxy_routes_require_auth( + client: TestClient, + method: str, + kaapi_path: str, + body: dict[str, Any] | None, + _upstream_path: str, +) -> None: + with _mock_upstream(json_body={"data": []}) as calls: + resp = client.request( + method, f"api/v1{kaapi_path.format(id=uuid4())}", json=body + ) + + assert resp.status_code in (401, 403) + assert calls == [] + + # ---------- helpers ---------- +@contextmanager +def _mock_upstream( + *, + status_code: int = 200, + json_body: Any = None, + content: bytes | None = None, + raises: Exception | None = None, +): + """Stub the guardrails HTTP boundary; yields the recorded client.request calls.""" + calls: list[dict[str, Any]] = [] + + response = MagicMock() + response.status_code = status_code + if content is None: + import json as _json + + response.content = _json.dumps(json_body).encode() + response.json.return_value = json_body + else: + response.content = content + response.json.side_effect = ValueError("not json") + + def _request(*args: Any, **kwargs: Any): + calls.append({"args": args, "kwargs": kwargs}) + if raises is not None: + raise raises + return response + + client = MagicMock() + client.request.side_effect = _request + + with patch("app.services.llm.guardrails.httpx.Client") as mock_client_cls: + mock_client_cls.return_value.__enter__.return_value = client + yield calls + + def _stub_job() -> Job: """Minimal in-memory Job for routes that only read id/status/timestamps.""" from datetime import datetime, timezone diff --git a/backend/app/tests/core/test_sentry_filters.py b/backend/app/tests/core/test_sentry_filters.py new file mode 100644 index 000000000..1873b7368 --- /dev/null +++ b/backend/app/tests/core/test_sentry_filters.py @@ -0,0 +1,80 @@ +from app.core.sentry_filters import before_send_filter + + +def _event_with_celery_job(task_name: str, kwargs: dict) -> dict: + return { + "extra": { + "celery-job": { + "task_name": task_name, + "args": [], + "kwargs": kwargs, + } + } + } + + +def test_before_send_filter_redacts_query_for_llm_job(): + event = _event_with_celery_job( + "app.celery.tasks.job_execution.run_llm_job", + { + "job_id": "b9ce6621-9b5a-45c4-969f-df7613ff7dc4", + "organization_id": 1, + "project_id": 1, + "request_data": { + "callback_url": "https://webhooksite.net/some-id", + "config": {"blob": {"completion": {"provider": "openai"}}}, + "query": { + "input": { + "type": "text", + "content": {"value": "Amit Gupta phone number is 919611188278"}, + } + }, + "request_metadata": None, + }, + }, + ) + + result = before_send_filter(event, {}) + + request_data = result["extra"]["celery-job"]["kwargs"]["request_data"] + assert request_data["query"] == "[REDACTED]" + # callback_url can carry identifying/credential data and is redacted too. + assert request_data["callback_url"] == "[REDACTED]" + # Non-sensitive fields are left untouched so the trace stays useful. + assert request_data["config"] == {"blob": {"completion": {"provider": "openai"}}} + + +def test_before_send_filter_redacts_query_for_chain_and_response_jobs(): + for task_name in ( + "app.celery.tasks.job_execution.run_llm_chain_job", + "app.celery.tasks.job_execution.run_response_job", + ): + event = _event_with_celery_job( + task_name, + {"request_data": {"query": {"input": "sensitive text"}}}, + ) + + result = before_send_filter(event, {}) + + request_data = result["extra"]["celery-job"]["kwargs"]["request_data"] + assert request_data["query"] == "[REDACTED]" + + +def test_before_send_filter_ignores_unrelated_tasks(): + event = _event_with_celery_job( + "app.celery.tasks.job_execution.run_doctransform_job", + {"request_data": {"query": {"input": "not an llm job"}}}, + ) + + result = before_send_filter(event, {}) + + request_data = result["extra"]["celery-job"]["kwargs"]["request_data"] + assert request_data["query"] == {"input": "not an llm job"} + + +def test_before_send_filter_passes_through_events_without_celery_job(): + event = {"message": "some unrelated error"} + + result = before_send_filter(event, {}) + + assert result == event diff --git a/backend/app/tests/services/collections/providers/test_openai_provider.py b/backend/app/tests/services/collections/providers/test_openai_provider.py index d31fb0ae3..8d8cf42c0 100644 --- a/backend/app/tests/services/collections/providers/test_openai_provider.py +++ b/backend/app/tests/services/collections/providers/test_openai_provider.py @@ -296,15 +296,25 @@ def test_upload_files_empty_docs_is_noop() -> None: client.files.create.assert_not_called() -def test_upload_files_file_name_matches_doc_fname() -> None: - """The file tuple passed to OpenAI must carry the original filename.""" +@pytest.mark.parametrize( + ("fname", "expected"), + [ + ("report.pdf", "report.pdf"), + ("Report.PDF", "Report.pdf"), + ("README", "README"), + ], +) +def test_upload_files_file_name_lowercases_only_extension( + fname: str, expected: str +) -> None: + """The file tuple passed to OpenAI keeps the stem's case, lowercases the extension.""" client = MagicMock() client.files.create.return_value = MagicMock(id="file-abc") provider = OpenAIProvider(client=client) storage = _make_storage(b"data") doc = _make_doc(file_size_kb=1.0) - doc.fname = "report.pdf" + doc.fname = fname session_p, crud_p = _patch_session_and_crud() with session_p as MockSession, crud_p: @@ -313,8 +323,8 @@ def test_upload_files_file_name_matches_doc_fname() -> None: provider.upload_files(storage, [doc], project_id=1) _, kwargs = client.files.create.call_args - fname, _ = kwargs["file"] - assert fname == "report.pdf" + uploaded_fname, _ = kwargs["file"] + assert uploaded_fname == expected def test_upload_files_raises_on_db_update_failure() -> None: diff --git a/backend/app/tests/services/llm/test_guardrails.py b/backend/app/tests/services/llm/test_guardrails.py index 22004d179..b1ccabcdb 100644 --- a/backend/app/tests/services/llm/test_guardrails.py +++ b/backend/app/tests/services/llm/test_guardrails.py @@ -20,8 +20,10 @@ Validator, ) from app.services.llm.guardrails import ( + GuardrailsOutcome, list_validators_config, run_guardrails_validation, + summarize_validator_results, ) from app.tests.utils.utils import get_project @@ -58,9 +60,11 @@ def test_run_guardrails_validation_success(mock_client_cls) -> None: assert kwargs["json"]["input"] == TEST_TEXT assert kwargs["json"]["validators"] == TEST_CONFIG assert kwargs["json"]["request_id"] == str(TEST_JOB_ID) - assert kwargs["json"]["project_id"] == TEST_PROJECT_ID - assert kwargs["json"]["organization_id"] == TEST_ORGANIZATION_ID - assert kwargs["params"]["suppress_pass_logs"] == "true" + assert "project_id" not in kwargs["json"] + assert "organization_id" not in kwargs["json"] + assert kwargs["headers"]["X-PROJECT-ID"] == str(TEST_PROJECT_ID) + assert kwargs["headers"]["X-ORGANIZATION-ID"] == str(TEST_ORGANIZATION_ID) + assert kwargs["params"]["suppress_pass_logs"] == "false" assert kwargs["headers"]["Authorization"].startswith("Bearer ") assert kwargs["headers"]["Content-Type"] == "application/json" @@ -68,8 +72,9 @@ def test_run_guardrails_validation_success(mock_client_cls) -> None: @patch("app.services.llm.guardrails.httpx.Client") def test_run_guardrails_validation_http_error_bypasses(mock_client_cls) -> None: mock_response = MagicMock() + mock_response.status_code = 500 mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( - "bad", request=None, response=None + "bad", request=MagicMock(), response=mock_response ) mock_client = MagicMock() @@ -89,6 +94,55 @@ def test_run_guardrails_validation_http_error_bypasses(mock_client_cls) -> None: assert result["data"]["safe_text"] == TEST_TEXT +@pytest.mark.parametrize("status_code", [401, 403, 422]) +@patch("app.services.llm.guardrails.httpx.Client") +def test_run_guardrails_validation_auth_error_fails_closed( + mock_client_cls, status_code +) -> None: + mock_response = MagicMock() + mock_response.status_code = status_code + mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( + "unauthorized", request=MagicMock(), response=mock_response + ) + + mock_client = MagicMock() + mock_client.post.return_value = mock_response + mock_client_cls.return_value.__enter__.return_value = mock_client + + result = run_guardrails_validation( + TEST_TEXT, + TEST_CONFIG, + TEST_JOB_ID, + TEST_PROJECT_ID, + TEST_ORGANIZATION_ID, + ) + + assert result["success"] is False + assert result.get("bypassed") is False + assert "rejected the request" in result["error"] + + +@patch("app.services.llm.guardrails.httpx.Client") +def test_list_validators_config_auth_error_raises(mock_client_cls) -> None: + mock_response = MagicMock() + mock_response.status_code = 403 + mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( + "forbidden", request=MagicMock(), response=mock_response + ) + + mock_client = MagicMock() + mock_client.get.return_value = mock_response + mock_client_cls.return_value.__enter__.return_value = mock_client + + with pytest.raises(ValueError, match=r"rejected \(HTTP 403\)"): + list_validators_config( + input_validator_configs=[Validator(validator_config_id=uuid.uuid4())], + output_validator_configs=[], + organization_id=1, + project_id=1, + ) + + @patch("app.services.llm.guardrails.httpx.Client") def test_run_guardrails_validation_uses_settings(mock_client_cls) -> None: mock_response = MagicMock() @@ -254,6 +308,8 @@ def test_list_validators_config_fetches_input_and_output_by_refs( assert second_call_kwargs["params"]["ids"] == [ str(v.validator_config_id) for v in output_validator_configs ] + assert first_call_kwargs["headers"]["X-ORGANIZATION-ID"] == "1" + assert first_call_kwargs["headers"]["X-PROJECT-ID"] == "1" @patch("app.services.llm.guardrails.httpx.Client") @@ -273,7 +329,7 @@ def test_list_validators_config_empty_short_circuits_without_http( @patch("app.services.llm.guardrails.httpx.Client") -def test_list_validators_config_omits_none_query_params(mock_client_cls) -> None: +def test_list_validators_config_omits_none_tenant(mock_client_cls) -> None: input_validator_configs = [Validator(validator_config_id=uuid.uuid4())] mock_response = MagicMock() @@ -297,6 +353,8 @@ def test_list_validators_config_omits_none_query_params(mock_client_cls) -> None ] assert "organization_id" not in kwargs["params"] assert "project_id" not in kwargs["params"] + assert "X-ORGANIZATION-ID" not in kwargs["headers"] + assert "X-PROJECT-ID" not in kwargs["headers"] @patch("app.services.llm.guardrails.httpx.Client") @@ -318,6 +376,57 @@ def test_list_validators_config_network_error_fails_open(mock_client_cls) -> Non assert output_guardrails == [] +def test_summarize_validator_results_extracts_per_validator_fields() -> None: + outcome = GuardrailsOutcome( + safe_text="My credit card is [REDACTED]", + error=None, + bypassed=False, + rephrase_needed=False, + raw={ + "success": True, + "data": { + "safe_text": "My credit card is [REDACTED]", + "validator_results": [ + { + "name": "PIIRemover", + "type": "pii_remover", + "stage": "input", + "order": 1, + "outcome": "FAIL", + "error": "PII detected in the text.", + "input_text": "My credit card is 4111 1111 1111 1111", + "output_text": "My credit card is [REDACTED]", + } + ], + }, + }, + ) + + summaries = summarize_validator_results(outcome) + + assert summaries == [ + { + "name": "PIIRemover", + "outcome": "FAIL", + "error": "PII detected in the text.", + "input_text": "My credit card is 4111 1111 1111 1111", + "output_text": "My credit card is [REDACTED]", + } + ] + + +def test_summarize_validator_results_empty_when_raw_has_no_validator_results() -> None: + outcome = GuardrailsOutcome( + safe_text="hello", + error=None, + bypassed=True, + rephrase_needed=False, + raw={}, + ) + + assert summarize_validator_results(outcome) == [] + + _SAFE_TEXT = "Please rephrase: content not allowed." _CONFIG_BLOB = ConfigBlob( completion=NativeCompletionConfig( diff --git a/backend/app/tests/services/llm/test_jobs.py b/backend/app/tests/services/llm/test_jobs.py index 9ce03f2ca..fd8bdf898 100644 --- a/backend/app/tests/services/llm/test_jobs.py +++ b/backend/app/tests/services/llm/test_jobs.py @@ -1344,6 +1344,85 @@ def test_guardrails_sanitize_input_before_provider( assert result["success"] + def test_guardrails_metadata_reports_per_validator_text_for_input_guardrail( + self, db, job_env, job_for_execution + ): + """metadata.input_guardrail should surface the original text, what + was sent to the LLM, and each validator's own before/after text -- + not just the raw kaapi-guardrails response wrapper. + """ + env = job_env + env["provider"].execute.return_value = (env["mock_llm_response"], None) + + unsafe_input = "My credit card is 4111 1111 1111 1111" + sanitized_input = "My credit card is [REDACTED]" + + with ( + patch( + "app.services.llm.guardrails.run_guardrails_validation" + ) as mock_guardrails, + patch( + "app.services.llm.guardrails.list_validators_config" + ) as mock_fetch_configs, + ): + mock_guardrails.return_value = { + "success": True, + "bypassed": False, + "data": { + "safe_text": sanitized_input, + "rephrase_needed": False, + "validator_results": [ + { + "name": "PIIRemover", + "type": "pii_remover", + "stage": "input", + "order": 1, + "outcome": "FAIL", + "error": "PII detected in the text.", + "input_text": unsafe_input, + "output_text": sanitized_input, + } + ], + }, + } + mock_fetch_configs.return_value = ( + [{"type": "pii_remover", "stage": "input"}], + [], + ) + + request_data = { + "query": {"input": unsafe_input}, + "config": { + "blob": { + "completion": { + "provider": "openai-native", + "type": "text", + "params": {"model": "gpt-4o"}, + }, + "input_guardrails": [ + {"validator_config_id": VALIDATOR_CONFIG_ID_1} + ], + "output_guardrails": [], + } + }, + "include_provider_raw_response": False, + "callback_url": None, + } + result = self._execute_job(job_for_execution, db, request_data) + + assert result["success"] + + input_guardrail_metadata = result["metadata"]["input_guardrail"] + assert input_guardrail_metadata["user_input"] == unsafe_input + assert input_guardrail_metadata["input_sent_to_llm"] == sanitized_input + + validators = input_guardrail_metadata["validators"] + assert len(validators) == 1 + assert validators[0]["name"] == "PIIRemover" + assert validators[0]["outcome"] == "FAIL" + assert validators[0]["input_text"] == unsafe_input + assert validators[0]["output_text"] == sanitized_input + def test_guardrails_skip_input_validation_for_audio_input( self, db, job_env, job_for_execution ): @@ -1443,6 +1522,85 @@ def test_guardrails_sanitize_output_after_provider( assert "REDACTED" in result["data"]["response"]["output"]["content"]["value"] + def test_guardrails_metadata_reports_per_validator_text_for_output_guardrail( + self, db, job_env, job_for_execution + ): + """metadata.output_guardrail should surface the raw LLM output + (pre-guardrail), the final response text, and each validator's own + before/after text -- not just the raw kaapi-guardrails response + wrapper. + """ + env = job_env + + raw_llm_output = "Aadhar no 123-45-6789" + sanitized_output = "Aadhar [REDACTED]" + env["mock_llm_response"].response.output.content.value = raw_llm_output + env["provider"].execute.return_value = (env["mock_llm_response"], None) + + with ( + patch( + "app.services.llm.guardrails.run_guardrails_validation" + ) as mock_guardrails, + patch( + "app.services.llm.guardrails.list_validators_config" + ) as mock_fetch_configs, + ): + mock_guardrails.return_value = { + "success": True, + "bypassed": False, + "data": { + "safe_text": sanitized_output, + "rephrase_needed": False, + "validator_results": [ + { + "name": "PIIRemover", + "type": "pii_remover", + "stage": "output", + "order": 1, + "outcome": "FAIL", + "error": "PII detected in the text.", + "input_text": raw_llm_output, + "output_text": sanitized_output, + } + ], + }, + } + mock_fetch_configs.return_value = ( + [], + [{"type": "pii_remover", "stage": "output"}], + ) + + request_data = { + "query": {"input": "hello"}, + "config": { + "blob": { + "completion": { + "provider": "openai-native", + "type": "text", + "params": {"model": "gpt-4o"}, + }, + "input_guardrails": [], + "output_guardrails": [ + {"validator_config_id": VALIDATOR_CONFIG_ID_2} + ], + } + }, + } + result = self._execute_job(job_for_execution, db, request_data) + + assert result["success"] + + output_guardrail_metadata = result["metadata"]["output_guardrail"] + assert output_guardrail_metadata["llm_output_pre_guardrail"] == raw_llm_output + assert output_guardrail_metadata["response_to_user"] == sanitized_output + + validators = output_guardrail_metadata["validators"] + assert len(validators) == 1 + assert validators[0]["name"] == "PIIRemover" + assert validators[0]["outcome"] == "FAIL" + assert validators[0]["input_text"] == raw_llm_output + assert validators[0]["output_text"] == sanitized_output + def test_guardrails_output_validation_sends_input_output_pair( self, db, job_env, job_for_execution ): diff --git a/docs/wiki/modules/knowledge-base.md b/docs/wiki/modules/knowledge-base.md index 4cefd5719..179b0c69d 100644 --- a/docs/wiki/modules/knowledge-base.md +++ b/docs/wiki/modules/knowledge-base.md @@ -35,6 +35,7 @@ All paths relative to `backend/app/`. ## Gotchas - `signed_url` behaviour is chosen per request by the `download` query param (default false = inline, as before). `put()` stores the upload's `Content-Type`, so a bare presigned URL for a PDF opens in a tab (CSV/XLSX datasets happen to download, which is why only KB docs looked broken). `get_signed_url(..., filename=...)` adds `ResponseContentDisposition: attachment`; `_signed_url()` in `services/documents/helpers.py` passes `fname` only when `download` is set, and the flag threads through `build_document_schema(s)` / `build_job_schema(s)` from `api/routes/documents.py`, `doc_transformation_job.py` and `collections.py`. An attachment URL will not render in an `