Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 40 additions & 0 deletions backend/app/alembic/versions/083_add_llm_call_metadata.py
Original file line number Diff line number Diff line change
@@ -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")
12 changes: 11 additions & 1 deletion backend/app/api/routes/collections.py
Original file line number Diff line number Diff line change
Expand Up @@ -249,6 +249,13 @@ def collection_info(
include_url: bool = Query(
False, description="Include a signed URL to access the document"
),
download: bool = Query(
False,
description=(
"When true the signed URL forces a download under the document's "
"original filename; when false it opens inline in the browser"
),
),
limit: int
| None = Query(
None,
Expand All @@ -275,7 +282,10 @@ def collection_info(
)

collection_with_docs.documents = build_document_schemas(
documents=documents, storage=storage, include_url=include_url
documents=documents,
storage=storage,
include_url=include_url,
download=download,
)

return APIResponse[CollectionWithDocsPublic].success_response(collection_with_docs)
16 changes: 16 additions & 0 deletions backend/app/api/routes/doc_transformation_job.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,13 @@ def get_transformation_job(
include_url: bool = Query(
False, description="Include a signed URL for the transformed document"
),
download: bool = Query(
False,
description=(
"When true the signed URL forces a download under the document's "
"original filename; when false it opens inline in the browser"
),
),
):
job_crud = DocTransformationJobCrud(session, current_user.project_.id)
doc_crud = DocumentCrud(session, current_user.project_.id)
Expand All @@ -48,6 +55,7 @@ def get_transformation_job(
doc_crud=doc_crud,
include_url=include_url,
storage=storage,
download=download,
)
return APIResponse.success_response(job_schema)

Expand All @@ -70,6 +78,13 @@ def get_multiple_transformation_jobs(
include_url: bool = Query(
False, description="Include a signed URL for each transformed document"
),
download: bool = Query(
False,
description=(
"When true the signed URL forces a download under the document's "
"original filename; when false it opens inline in the browser"
),
),
):
job_crud = DocTransformationJobCrud(session, project_id=current_user.project_.id)
doc_crud = DocumentCrud(session, project_id=current_user.project_.id)
Expand All @@ -89,6 +104,7 @@ def get_multiple_transformation_jobs(
doc_crud=doc_crud,
include_url=include_url,
storage=storage,
download=download,
)

return APIResponse.success_response(
Expand Down
16 changes: 16 additions & 0 deletions backend/app/api/routes/documents.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,13 @@ def list_docs(
include_url: bool = Query(
False, description="Include a signed URL to access each document"
),
download: bool = Query(
False,
description=(
"When true the signed URL forces a download under the document's "
"original filename; when false it opens inline in the browser"
),
),
) -> APIResponse[list[Union[DocumentPublic, TransformedDocumentPublic]]]:
crud = DocumentCrud(session, current_user.project_.id)
documents, has_more = crud.read_many(skip, limit)
Expand All @@ -96,6 +103,7 @@ def list_docs(
documents=documents,
include_url=include_url,
storage=storage,
download=download,
)
return APIResponse[
list[Union[DocumentPublic, TransformedDocumentPublic]]
Expand Down Expand Up @@ -268,6 +276,13 @@ def doc_info(
include_url: bool = Query(
False, description="Include a signed URL to access the document"
),
download: bool = Query(
False,
description=(
"When true the signed URL forces a download under the document's "
"original filename; when false it opens inline in the browser"
),
),
) -> APIResponse[Union[DocumentPublic, TransformedDocumentPublic]]:
crud = DocumentCrud(session, current_user.project_.id)
document = crud.read_one(doc_id)
Expand All @@ -282,6 +297,7 @@ def doc_info(
document=document,
include_url=include_url,
storage=storage,
download=download,
)

return APIResponse[
Expand Down
4 changes: 3 additions & 1 deletion backend/app/api/routes/llm.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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)
3 changes: 2 additions & 1 deletion backend/app/celery/celery_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -48,6 +48,7 @@ def _initialize_worker_observability() -> None:
# 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(
Expand Down
28 changes: 24 additions & 4 deletions backend/app/core/cloud/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import functools as ft
from pathlib import Path
from dataclasses import dataclass, asdict
from urllib.parse import ParseResult, urlparse, urlunparse
from urllib.parse import ParseResult, quote, urlparse, urlunparse

from abc import ABC, abstractmethod
from typing import Any
Expand All @@ -32,6 +32,18 @@ def _mask(value: str | None) -> str:
logger = logging.getLogger(__name__)


def _attachment_disposition(filename: str) -> str:
"""
Build a Content-Disposition header that forces a download.
"""
ascii_fallback = (
filename.encode("ascii", "replace").decode("ascii").replace('"', "")
)
return (
f"attachment; filename=\"{ascii_fallback}\"; filename*=UTF-8''{quote(filename)}"
)


class CloudStorageError(Exception):
pass

Expand Down Expand Up @@ -151,7 +163,9 @@ def get_file_size_kb(self, url: str) -> float:
pass

@abstractmethod
def get_signed_url(self, url: str, expires_in: int = 3600) -> str:
def get_signed_url(
self, url: str, expires_in: int = 3600, filename: str | None = None
) -> str:
"""Generate a signed URL with an optional expiry"""
pass

Expand Down Expand Up @@ -255,21 +269,27 @@ def get_file_size_kb(self, url: str) -> float:
# Maximum allowed expiry for signed URLs (24 hours)
MAX_SIGNED_URL_EXPIRY = 86400

def get_signed_url(self, url: str, expires_in: int = 3600) -> str:
def get_signed_url(
self, url: str, expires_in: int = 3600, filename: str | None = None
) -> str:
"""
Generate a signed S3 URL for the given file.
:param url: S3 url (e.g., s3://bucket/key)
:param expires_in: Expiry time in seconds (default: 1 hour, max: 24 hours)
:param filename: When set, the URL forces a download under this name
:return: Signed URL as string
"""
# Cap expiry at maximum allowed value to prevent excessively long-lived URLs
expires_in = min(expires_in, self.MAX_SIGNED_URL_EXPIRY)

name = SimpleStorageName.from_url(url)
try:
params: dict[str, str] = {"Bucket": name.Bucket, "Key": name.Key}
if filename:
params["ResponseContentDisposition"] = _attachment_disposition(filename)
signed_url = self.aws.client.generate_presigned_url(
"get_object",
Params={"Bucket": name.Bucket, "Key": name.Key},
Params=params,
ExpiresIn=expires_in,
)
logger.info(
Expand Down
63 changes: 63 additions & 0 deletions backend/app/core/sentry_filters.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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
Loading
Loading