feat(guardrails): Add intermediate information fields to guardrails execution in the callback payload

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")

backend/app/api/routes/guardrails.py

@@ -106,7 +106,7 @@ def _upstream_response(status_code: int, payload: Any) -> Response:
     return JSONResponse(status_code=status_code, content=payload, headers=headers)
 
 
-# ROUTE ORDERING: these fixed paths must stay above GET /guardrails/{job_id} — FastAPI matches in declaration order.
+# ROUTE ORDERING: these fixed paths must stay above GET /guardrails/{job_id}.
 
 
 @router.get(

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)

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
@@ -46,6 +46,7 @@ def _initialize_worker_observability() -> None:
             traces_sample_rate=1.0,
             max_request_body_size="never",
             enable_logs=True,
+            before_send=before_send_filter,
             before_send_transaction=before_send_transaction_filter,
             integrations=[
                 LoggingIntegration(

backend/app/celery/tasks/job_execution.py

@@ -70,9 +70,8 @@ def _run_with_otel_parent(
 ) -> T:  # noqa: UP047 (black doesn't support PEP 695 generics yet)
     """Attach the extracted parent context and execute `fn` under it.
 
-    opentelemetry-instrumentation-celery's CeleryGetter misses propagation
-    headers (they live under `.headers`, not top-level task attrs), so its
-    span is always unparented; we extract and attach the context ourselves.
+    Needed because otel's Celery instrumentation misses propagation headers
+    under `task.request.headers`, leaving `run/...` spans unparented.
     """
     parent_ctx = _extract_parent_context(task_instance)
     parent_span_ctx = trace.get_current_span(parent_ctx).get_span_context()

backend/app/core/sentry_filters.py

@@ -2,6 +2,24 @@
 from typing import Any
 
 
+_REDACTED = "[REDACTED]"
+
+# LLM job kwargs land in Sentry via sentry_sdk's CeleryIntegration; these keys
+# carry end-user text and must be redacted before that happens.
+_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 +103,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

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(

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
 
@@ -38,6 +38,7 @@
         # 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(

backend/app/models/llm/request.py

@@ -589,6 +589,10 @@ class LLMCallRequest(SQLModel):
         default=False,
         description="Whether to include the raw LLM provider response in the output",
     )
+    include_guardrail_metadata: bool = Field(
+        default=False,
+        description="Include per-validator guardrail metadata (input/output text, pass/fail) in the response",
+    )
     request_metadata: dict[str, Any] | None = Field(
         default=None,
         description=(
@@ -764,6 +768,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,

backend/app/services/llm/guardrails.py

@@ -56,11 +56,6 @@ def proxy_guardrails_request(
     # 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(
@@ -111,13 +106,15 @@ def proxy_guardrails_request(
         ) from None
 
 
+# 422 included: a missing/invalid tenant header is a backend bug, not a
+# transient outage, so it must not fall open like one.
+_AUTH_ERROR_STATUS_CODES = (401, 403, 422)
+
+
 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,
+    return (
+        isinstance(e, httpx.HTTPStatusError)
+        and e.response.status_code in _AUTH_ERROR_STATUS_CODES
     )
 
 
@@ -240,6 +237,26 @@ 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: Sequence[Validator | dict[str, Any]],
@@ -324,10 +341,6 @@ def run_guardrails_validation(
         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,