diff --git a/.env.example b/.env.example index 5a041fe26..e8da9041c 100644 --- a/.env.example +++ b/.env.example @@ -110,3 +110,5 @@ SMTP_PASSWORD= EMAILS_FROM_EMAIL= EMAILS_FROM_NAME=Kaapi FRONTEND_HOST= + +LLM_CALL_RETENTION_HOURS= diff --git a/backend/app/api/routes/cron.py b/backend/app/api/routes/cron.py index 6acade45c..ddcf3239b 100644 --- a/backend/app/api/routes/cron.py +++ b/backend/app/api/routes/cron.py @@ -8,7 +8,9 @@ from app.api.permissions import Permission, require_permission from app.core.config import settings from app.crud.evaluations import process_all_pending_evaluations +from app.models.llm.response import LlmCallRedactionResult from app.services.job_monitoring import monitor_pending_jobs +from app.services.llm.retention import redact_aged_llm_calls from app.crud.stats import StatRow, get_daily_stats from app.services.stats import format_sections, post_to_discord @@ -23,15 +25,10 @@ "value": settings.CRON_INTERVAL_MINUTES, "unit": "minute", }, - # Timezone for the schedule (only affects crontab-style schedules). "timezone": "UTC", - # Grace period (minutes) before a late check-in is marked as missed. "checkin_margin": 2, - # Max runtime (minutes) before an in-progress run is marked as timed out. "max_runtime": 2 * settings.CRON_INTERVAL_MINUTES, - # Consecutive failures/missed/timeouts required to open a Sentry issue. "failure_issue_threshold": 2, - # Consecutive successful check-ins required to auto-resolve the issue. "recovery_threshold": 1, } @@ -59,6 +56,16 @@ } +LLM_CALL_RETENTION_CRON_MONITOR_CONFIG: MonitorConfig = { + "schedule": {"type": "crontab", "value": "0 9 * * *"}, + "timezone": "Asia/Kolkata", + "checkin_margin": 5, + "max_runtime": 30, + "failure_issue_threshold": 1, + "recovery_threshold": 1, +} + + @router.get( "/cron/evaluations", include_in_schema=False, @@ -158,6 +165,35 @@ def daily_stats_cron_job(session: SessionDep) -> dict[str, list[StatRow]]: raise +@router.get( + "/cron/llm-call-retention", + include_in_schema=False, + dependencies=[Depends(require_permission(Permission.SUPERUSER))], +) +@sentry_sdk.monitor( + monitor_slug="llm-call-retention-cron-job", + monitor_config=LLM_CALL_RETENTION_CRON_MONITOR_CONFIG, +) +def llm_call_retention_cron_job(session: SessionDep) -> LlmCallRedactionResult: + logger.info("[llm_call_retention_cron_job] Cron job invoked") + + try: + result = redact_aged_llm_calls(session=session) + logger.info( + f"[llm_call_retention_cron_job] Completed: " + f"rows_redacted={result.rows_redacted}, " + f"cutoff={result.cutoff}" + ) + return result + except Exception as e: + logger.error( + f"[llm_call_retention_cron_job] Error executing cron job: {e}", + exc_info=True, + ) + sentry_sdk.capture_exception(e) + raise + + @router.get( "/cron/pending-jobs", include_in_schema=False, diff --git a/backend/app/core/cloud/storage.py b/backend/app/core/cloud/storage.py index 9dade4f54..9e7c98a58 100644 --- a/backend/app/core/cloud/storage.py +++ b/backend/app/core/cloud/storage.py @@ -177,34 +177,44 @@ def from_url(cls, url: str): PENDING_PREFIX = "pending" PENDING_TTL_DAYS = 1 +# LLM call audio (STT input, TTS output). Its own literal prefix so one lifecycle +# rule can expire it in step with the llm_call retention cron, which redacts the +# only DB pointer to each object. +LLM_AUDIO_PREFIX = "llm-audio" + class CloudStorage(ABC): def __init__(self, project_id: int, storage_path: UUID): self.project_id = project_id self.storage_path = str(storage_path) - def url_for(self, file_path: Path, is_pending: bool = False) -> SimpleStorageName: + def url_for( + self, file_path: Path, root_prefix: str | None = None + ) -> SimpleStorageName: """Resolve a project-relative path into a fully qualified storage name. The single place a key is built — callers never join storage_path themselves. Args: file_path: Path relative to the project's storage root. - is_pending: Place the key under ``PENDING_PREFIX`` — the holding area for - uploads that have not been registered yet. An S3 lifecycle rule deletes - everything there after ``PENDING_TTL_DAYS`` (1 day), so pass True only - for objects that are meant to disappear if registration never happens. - Registered documents use the default (False) and are never expired. + root_prefix: Literal prefix placed *ahead of* storage_path, so a single + S3 lifecycle rule matches it across every project. ``PENDING_PREFIX`` + is the holding area for unregistered uploads, deleted after + ``PENDING_TTL_DAYS``; ``LLM_AUDIO_PREFIX`` holds LLM call audio, + deleted after the llm_call retention window. Leave as None for + objects that are never expired. """ if file_path.is_absolute(): raise ValueError("file_path must be relative to the project's storage root") roots = ( - (PENDING_PREFIX, self.storage_path) if is_pending else (self.storage_path,) + (root_prefix, self.storage_path) if root_prefix else (self.storage_path,) ) return SimpleStorageName(Path(*roots, file_path).as_posix()) @abstractmethod - def put(self, source: UploadFile, filepath: Path) -> SimpleStorageName: + def put( + self, source: UploadFile, filepath: Path, root_prefix: str | None = None + ) -> SimpleStorageName: """Upload a file to storage""" pass @@ -269,8 +279,10 @@ def __init__(self, project_id: int, storage_path: UUID): super().__init__(project_id, storage_path) self.aws = AmazonCloudStorageClient() - def put(self, source: UploadFile, file_path: Path) -> SimpleStorageName: - destination = self.url_for(file_path) + def put( + self, source: UploadFile, file_path: Path, root_prefix: str | None = None + ) -> SimpleStorageName: + destination = self.url_for(file_path, root_prefix=root_prefix) kwargs = asdict(destination) try: @@ -424,7 +436,7 @@ def create_upload_ticket( """ expires_in = min(expires_in, self.MAX_SIGNED_URL_EXPIRY) - name = self.url_for(file_path, is_pending=True) + name = self.url_for(file_path, root_prefix=PENDING_PREFIX) encoded = quote(filename) meta_field = f"x-amz-meta-{FILENAME_METADATA_KEY}" try: diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 7bf84205b..44691bdf3 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -2,6 +2,7 @@ import os import secrets import warnings +from datetime import timedelta from typing import Any, Literal, Self from pydantic import ( @@ -204,10 +205,13 @@ def AWS_S3_BUCKET(self) -> str: EVAL_FAST_STALL_THRESHOLD_MINUTES: int = 15 PENDING_JOB_QUERY_TIMEOUT_MS: int = 1000 - # AI-assisted prompt improvement settings. - # See docs/srd-ai-prompt-improvement.md for the full design rationale. - # Platform-owned Anthropic key shared by every org/project for this feature, - # so prompt improvement works without per-project credentials. + LLM_CALL_RETENTION_HOURS: int = 168 + + @computed_field # type: ignore[prop-decorator] + @property + def LLM_CALL_RETENTION_TIMEDELTA(self) -> timedelta: + return timedelta(hours=self.LLM_CALL_RETENTION_HOURS) + ANTHROPIC_API_KEY: str = "" PROMPT_IMPROVEMENT_MODEL: str = "claude-opus-4-8" diff --git a/backend/app/core/storage_utils.py b/backend/app/core/storage_utils.py index 27a67fccc..15f9c3d78 100644 --- a/backend/app/core/storage_utils.py +++ b/backend/app/core/storage_utils.py @@ -10,6 +10,7 @@ import mimetypes from collections.abc import Mapping, Sequence from datetime import datetime +from enum import StrEnum from io import BytesIO from pathlib import Path from typing import Literal @@ -18,7 +19,7 @@ from starlette.datastructures import Headers, UploadFile -from app.core.cloud.storage import CloudStorage, CloudStorageError +from app.core.cloud.storage import LLM_AUDIO_PREFIX, CloudStorage, CloudStorageError logger = logging.getLogger(__name__) @@ -48,6 +49,7 @@ def upload_to_object_store( filename: str, subdirectory: str, content_type: str = "application/octet-stream", + root_prefix: str | None = None, ) -> str | None: """ Upload content to object store. @@ -60,6 +62,8 @@ def upload_to_object_store( filename: Name of the file subdirectory: Subdirectory path in object store (e.g., "datasets", "stt_datasets") content_type: MIME type of the content (default: "application/octet-stream") + root_prefix: Literal prefix ahead of the project's storage path, see + ``CloudStorage.url_for`` Returns: Object store URL as string if successful, None if failed @@ -84,7 +88,9 @@ def upload_to_object_store( headers=headers, ) - destination = storage.put(source=upload_file, file_path=file_path) + destination = storage.put( + source=upload_file, file_path=file_path, root_prefix=root_prefix + ) object_store_url = str(destination) logger.info( @@ -223,21 +229,30 @@ def load_json_from_object_store(storage: CloudStorage, url: str) -> list | dict } +class LlmAudioKindEnum(StrEnum): + STT = "stt" + TTS = "tts" + + def upload_audio_bytes_to_s3( storage: CloudStorage, audio_bytes: bytes, call_id: UUID, mime_type: str | None, - prefix: str, + kind: LlmAudioKindEnum, ) -> str | None: - """Upload decoded audio bytes to S3 and return the s3:// URI. + """Upload LLM call audio to S3 and return the s3:// URI. + + The key is ``llm-audio///.``: the literal + ``LLM_AUDIO_PREFIX`` lets one lifecycle rule expire it, and tenant ids stay out + of the key, which reaches clients inside presigned URLs. Args: storage: CloudStorage instance audio_bytes: Raw audio bytes call_id: LLM call UUID used as the filename stem mime_type: MIME type of the audio (determines file extension) - prefix: S3 subdirectory, e.g. "llm/tts/audio" or "llm/stt/audio" + kind: Whether this is STT input or TTS output Returns: s3:// URI if successful, None on failure @@ -245,7 +260,12 @@ def upload_audio_bytes_to_s3( ext = _MIME_TO_EXT.get(mime_type or "", "wav") filename = f"{call_id}.{ext}" return upload_to_object_store( - storage, audio_bytes, filename, prefix, mime_type or "audio/wav" + storage, + audio_bytes, + filename, + kind.value, + mime_type or "audio/wav", + root_prefix=LLM_AUDIO_PREFIX, ) diff --git a/backend/app/crud/llm.py b/backend/app/crud/llm.py index f96b598a1..4bf3dbe20 100644 --- a/backend/app/crud/llm.py +++ b/backend/app/crud/llm.py @@ -1,9 +1,11 @@ import logging import base64 import json +from datetime import datetime from uuid import UUID from typing import Any, Literal +from sqlalchemy import text from sqlmodel import Session, select from app.core.util import now @@ -345,6 +347,44 @@ def get_llm_call_by_id( return session.exec(statement).first() +REDACTED_SENTINEL = "[redacted]" + +REDACT_LLM_CALLS_SQL = text( + """ + UPDATE llm_call + SET input = :redacted_sentinel, + content = CASE + WHEN jsonb_typeof(content) = 'object' + THEN jsonb_set(content, '{content,value}', to_jsonb(CAST(:redacted_sentinel AS text)), false) + ELSE content + END + WHERE updated_at <= :cutoff + AND ( + input <> :redacted_sentinel + OR ( + jsonb_typeof(content) = 'object' + AND content #>> '{content,value}' <> :redacted_sentinel + ) + ) + """ +) + + +def redact_llm_calls(*, session: Session, cutoff: datetime) -> int: + result = session.connection().execute( + REDACT_LLM_CALLS_SQL, + {"cutoff": cutoff, "redacted_sentinel": REDACTED_SENTINEL}, + ) + session.commit() + + logger.info( + f"[redact_llm_calls] Redacted rows | rows: {result.rowcount} | " + f"cutoff: {cutoff.isoformat()}" + ) + + return result.rowcount + + def get_llm_calls_by_job_id( session: Session, job_id: UUID, project_id: int ) -> list[LlmCall]: diff --git a/backend/app/models/llm/response.py b/backend/app/models/llm/response.py index 439a6adfc..afc6905ed 100644 --- a/backend/app/models/llm/response.py +++ b/backend/app/models/llm/response.py @@ -19,6 +19,13 @@ class Usage(SQLModel): reasoning_tokens: int | None = None +class LlmCallRedactionResult(SQLModel): + """Outcome of a single llm_call retention redaction run.""" + + rows_redacted: int + cutoff: datetime + + class TextOutput(SQLModel): type: Literal["text"] = "text" content: TextContent diff --git a/backend/app/services/documents/registration.py b/backend/app/services/documents/registration.py index 16cb3c5ca..dd3617fe3 100644 --- a/backend/app/services/documents/registration.py +++ b/backend/app/services/documents/registration.py @@ -8,7 +8,7 @@ from sqlmodel import Session from app.core.cloud import get_cloud_storage -from app.core.cloud.storage import ObjectNotFoundError +from app.core.cloud.storage import PENDING_PREFIX, ObjectNotFoundError from app.crud import DocumentCrud from app.models import Document from app.services.doctransform.registry import get_file_format @@ -47,7 +47,9 @@ def register_uploaded_document( raise HTTPException(status_code=409, detail=DUPLICATE_DOCUMENT_DETAIL) storage = get_cloud_storage(session=session, project_id=project_id) - pending_url = str(storage.url_for(Path(str(document_id)), is_pending=True)) + pending_url = str( + storage.url_for(Path(str(document_id)), root_prefix=PENDING_PREFIX) + ) # Copy first, then read the frozen final key: the pending object stays writable # through its ticket, so measuring it directly would race the copy. diff --git a/backend/app/services/llm/jobs.py b/backend/app/services/llm/jobs.py index 22a6dee95..86386b336 100644 --- a/backend/app/services/llm/jobs.py +++ b/backend/app/services/llm/jobs.py @@ -61,7 +61,7 @@ TextInput, ) from app.core.cloud.storage import get_cloud_storage -from app.core.storage_utils import upload_audio_bytes_to_s3 +from app.core.storage_utils import LlmAudioKindEnum, upload_audio_bytes_to_s3 from app.models.llm.response import ( AudioOutput, LLMCallResponse, @@ -1082,13 +1082,12 @@ def _upload_stt_input(session: Session, query: QueryParams, ctx: _CallContext) - stt_bytes = base64.b64decode(query.input.content.value) storage = get_cloud_storage(session, ctx.project_id) - subfolder_path = f"orgs/{ctx.organization_id}/{ctx.project_id}/audio/stt" s3_url = upload_audio_bytes_to_s3( storage, stt_bytes, ctx.llm_call_id, query.input.content.mime_type, - subfolder_path, + LlmAudioKindEnum.STT, ) if s3_url: stt_input_record = json.dumps( @@ -1121,13 +1120,12 @@ def _upload_tts_output(output: AudioOutput, ctx: _CallContext) -> dict | None: with Session(engine) as s3_session: storage = get_cloud_storage(s3_session, ctx.project_id) tts_bytes = base64.b64decode(output.content.value) - subfolder_path = f"orgs/{ctx.organization_id}/{ctx.project_id}/audio/tts" s3_url = upload_audio_bytes_to_s3( storage, tts_bytes, ctx.llm_call_id, output.content.mime_type, - subfolder_path, + LlmAudioKindEnum.TTS, ) if s3_url: output.content.uri = s3_url diff --git a/backend/app/services/llm/retention.py b/backend/app/services/llm/retention.py new file mode 100644 index 000000000..688eee95f --- /dev/null +++ b/backend/app/services/llm/retention.py @@ -0,0 +1,27 @@ +import logging + +from sqlmodel import Session + +from app.core.config import settings +from app.core.util import now +from app.crud.llm import redact_llm_calls +from app.models.llm.response import LlmCallRedactionResult + +logger = logging.getLogger(__name__) + + +def redact_aged_llm_calls(*, session: Session) -> LlmCallRedactionResult: + cutoff = now() - settings.LLM_CALL_RETENTION_TIMEDELTA + + logger.info( + f"[redact_aged_llm_calls] Starting redaction | cutoff: {cutoff.isoformat()}" + ) + + rows_redacted = redact_llm_calls(session=session, cutoff=cutoff) + + logger.info( + f"[redact_aged_llm_calls] Completed | rows_redacted: {rows_redacted} | " + f"cutoff: {cutoff.isoformat()}" + ) + + return LlmCallRedactionResult(rows_redacted=rows_redacted, cutoff=cutoff) diff --git a/backend/app/tests/api/routes/test_cron.py b/backend/app/tests/api/routes/test_cron.py index 399ed6eee..9868eec77 100644 --- a/backend/app/tests/api/routes/test_cron.py +++ b/backend/app/tests/api/routes/test_cron.py @@ -1,3 +1,4 @@ +from datetime import datetime from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -5,6 +6,7 @@ from app.api.routes import cron from app.core.config import settings +from app.models.llm.response import LlmCallRedactionResult from app.tests.utils.auth import TestAuthContext @@ -326,6 +328,61 @@ def test_daily_stats_cron_job_captures_and_reraises_on_error() -> None: sentry.capture_exception.assert_called_once() +def test_llm_call_retention_cron_job_success( + client: TestClient, + superuser_api_key: TestAuthContext, +) -> None: + """Returns the redaction summary produced by the retention service.""" + result = LlmCallRedactionResult( + rows_redacted=4137, + cutoff=datetime(2026, 2, 22, 12, 0, 0), + ) + with patch( + "app.api.routes.cron.redact_aged_llm_calls", + return_value=result, + ) as redact: + response = client.get( + f"{settings.API_V1_STR}/cron/llm-call-retention", + headers={"X-API-KEY": superuser_api_key.key}, + ) + + assert response.status_code == 200 + assert response.json() == { + "rows_redacted": 4137, + "cutoff": "2026-02-22T12:00:00", + } + redact.assert_called_once() + + +def test_llm_call_retention_cron_job_requires_superuser( + client: TestClient, + user_api_key: TestAuthContext, +) -> None: + """Non-superuser cannot access the llm_call retention cron endpoint.""" + response = client.get( + f"{settings.API_V1_STR}/cron/llm-call-retention", + headers={"X-API-KEY": user_api_key.key}, + ) + + assert response.status_code == 403 + assert "Insufficient permissions" in response.json()["error"] + + +def test_llm_call_retention_cron_job_captures_and_reraises_on_error() -> None: + """On failure the job reports to Sentry and re-raises.""" + with ( + patch( + "app.api.routes.cron.redact_aged_llm_calls", + side_effect=RuntimeError("boom"), + ), + patch("app.api.routes.cron.sentry_sdk") as sentry, + ): + with pytest.raises(RuntimeError): + cron.llm_call_retention_cron_job(session=MagicMock()) + + sentry.capture_exception.assert_called_once() + + def test_evaluation_cron_job_not_in_schema( client: TestClient, ) -> None: @@ -340,6 +397,7 @@ def test_evaluation_cron_job_not_in_schema( assert f"{settings.API_V1_STR}/cron/evaluations" not in paths assert f"{settings.API_V1_STR}/cron/pending-jobs" not in paths assert f"{settings.API_V1_STR}/cron/daily-stats" not in paths + assert f"{settings.API_V1_STR}/cron/llm-call-retention" not in paths def test_cron_intervals_match_to_prevent_sentry_monitor_drift() -> None: diff --git a/backend/app/tests/core/cloud/test_storage.py b/backend/app/tests/core/cloud/test_storage.py index f5d7f7a54..ab88bb836 100644 --- a/backend/app/tests/core/cloud/test_storage.py +++ b/backend/app/tests/core/cloud/test_storage.py @@ -12,6 +12,8 @@ from app.core.cloud.storage import ( GCS_SCOPES, + LLM_AUDIO_PREFIX, + PENDING_PREFIX, AmazonCloudStorage, AmazonCloudStorageClient, CloudStorageError, @@ -50,11 +52,18 @@ def test_joins_the_projects_storage_path(self) -> None: def test_pending_prefix_leads_the_key(self) -> None: storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) - name = storage.url_for(Path("report.pdf"), is_pending=True) + name = storage.url_for(Path("report.pdf"), root_prefix=PENDING_PREFIX) # Ahead of storage_path, not after it: one literal-prefix rule must match every project. assert name.Key == f"pending/{storage.storage_path}/report.pdf" + def test_llm_audio_prefix_leads_the_key(self) -> None: + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + + name = storage.url_for(Path("tts/call.wav"), root_prefix=LLM_AUDIO_PREFIX) + + assert name.Key == f"llm-audio/{storage.storage_path}/tts/call.wav" + def test_absolute_path_is_rejected(self) -> None: storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) @@ -139,7 +148,7 @@ def test_returns_size_and_decoded_filename(self) -> None: aws = AmazonCloudStorageClient() aws.create() storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) - name = storage.url_for(Path(str(uuid4())), is_pending=True) + name = storage.url_for(Path(str(uuid4())), root_prefix=PENDING_PREFIX) aws.client.put_object( Bucket=name.Bucket, Key=name.Key, @@ -180,7 +189,7 @@ def test_copies_the_source_bytes_to_the_destination_key(self) -> None: aws = AmazonCloudStorageClient() aws.create() storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) - source = storage.url_for(Path("waiting.pdf"), is_pending=True) + source = storage.url_for(Path("waiting.pdf"), root_prefix=PENDING_PREFIX) aws.client.put_object( Bucket=source.Bucket, Key=source.Key, Body=b"pending bytes" ) @@ -197,7 +206,7 @@ def test_copies_the_source_bytes_to_the_destination_key(self) -> None: def test_missing_source_raises_object_not_found(self) -> None: AmazonCloudStorageClient().create() storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) - source = str(storage.url_for(Path("absent.pdf"), is_pending=True)) + source = str(storage.url_for(Path("absent.pdf"), root_prefix=PENDING_PREFIX)) with pytest.raises(ObjectNotFoundError): storage.copy(source, Path("final.pdf")) diff --git a/backend/app/tests/core/test_storage_utils.py b/backend/app/tests/core/test_storage_utils.py index 9ebe9e130..0e56ef8c4 100644 --- a/backend/app/tests/core/test_storage_utils.py +++ b/backend/app/tests/core/test_storage_utils.py @@ -4,15 +4,18 @@ from datetime import datetime from io import BytesIO from unittest.mock import MagicMock, patch +from uuid import uuid4 import pytest -from app.core.cloud.storage import CloudStorageError +from app.core.cloud.storage import AmazonCloudStorage, CloudStorageError from app.core.storage_utils import ( + LlmAudioKindEnum, generate_timestamped_filename, get_mime_from_url, load_json_from_object_store, upload_jsonl_to_object_store, + upload_audio_bytes_to_s3, upload_to_object_store, ) @@ -152,6 +155,32 @@ def test_upload_returns_none_on_generic_error(self, mock_storage): assert result is None +class TestUploadAudioBytesToS3: + @pytest.mark.parametrize("kind", list(LlmAudioKindEnum)) + def test_key_sits_under_llm_audio_prefix_without_tenant_ids( + self, kind: LlmAudioKindEnum + ) -> None: + storage = AmazonCloudStorage(project_id=42, storage_path=uuid4()) + storage.aws = MagicMock() + call_id = uuid4() + + url = upload_audio_bytes_to_s3(storage, b"RIFF", call_id, "audio/mpeg", kind) + + assert url is not None + key = url.split("/", 3)[3] + assert key == f"llm-audio/{storage.storage_path}/{kind.value}/{call_id}.mp3" + + def test_unknown_mime_type_falls_back_to_wav(self) -> None: + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + storage.aws = MagicMock() + + url = upload_audio_bytes_to_s3( + storage, b"RIFF", uuid4(), None, LlmAudioKindEnum.TTS + ) + + assert url is not None and url.endswith(".wav") + + class TestUploadJsonlToObjectStore: """Test cases for upload_jsonl_to_object_store function.""" diff --git a/backend/app/tests/crud/test_llm_retention.py b/backend/app/tests/crud/test_llm_retention.py new file mode 100644 index 000000000..2683a45d9 --- /dev/null +++ b/backend/app/tests/crud/test_llm_retention.py @@ -0,0 +1,161 @@ +from datetime import datetime, timedelta + +import pytest +from sqlalchemy import text +from sqlmodel import Session + +from app.models import Job +from app.crud.llm import REDACTED_SENTINEL, redact_llm_calls +from app.models.llm import LlmCall +from app.tests.utils.llm import create_aged_llm_call, create_llm_job + +CUTOFF = datetime(2026, 1, 8, 0, 0, 0) +AGED = CUTOFF - timedelta(days=1) +RECENT = CUTOFF + timedelta(minutes=1) + +TEXT_CONTENT = {"type": "text", "content": {"format": "text", "value": "hello"}} + + +@pytest.fixture +def job(db: Session) -> Job: + # Other llm_call rows (seed data, sibling fixtures) would also match the + # cutoff and skew the rowcount assertions, so start from an empty table. + db.exec(text("DELETE FROM llm_call")) + db.commit() + return create_llm_job(db) + + +def _reload(db: Session, llm_call: LlmCall) -> LlmCall: + db.expire_all() + return db.get(LlmCall, llm_call.id) + + +def test_redacts_input_and_content_value_only(db: Session, job: Job) -> None: + llm_call = create_aged_llm_call( + db, + updated_at=AGED, + job_id=job.id, + input="my social security number is 123-45-6789", + content=TEXT_CONTENT, + ) + + redacted = redact_llm_calls(session=db, cutoff=CUTOFF) + + assert redacted == 1 + row = _reload(db, llm_call) + assert row.input == REDACTED_SENTINEL + assert row.content["content"]["value"] == REDACTED_SENTINEL + assert row.content["content"]["format"] == "text" + assert row.content["type"] == "text" + + +def test_leaves_rows_newer_than_cutoff_untouched(db: Session, job: Job) -> None: + llm_call = create_aged_llm_call( + db, + updated_at=RECENT, + job_id=job.id, + input="still within the retention window", + content=TEXT_CONTENT, + ) + + redacted = redact_llm_calls(session=db, cutoff=CUTOFF) + + assert redacted == 0 + row = _reload(db, llm_call) + assert row.input == "still within the retention window" + assert row.content["content"]["value"] == "hello" + + +def test_row_exactly_at_cutoff_is_redacted(db: Session, job: Job) -> None: + llm_call = create_aged_llm_call( + db, + updated_at=CUTOFF, + job_id=job.id, + content=TEXT_CONTENT, + ) + + assert redact_llm_calls(session=db, cutoff=CUTOFF) == 1 + assert _reload(db, llm_call).input == REDACTED_SENTINEL + + +def test_second_run_skips_already_redacted_rows(db: Session, job: Job) -> None: + create_aged_llm_call(db, updated_at=AGED, job_id=job.id, content=TEXT_CONTENT) + + assert redact_llm_calls(session=db, cutoff=CUTOFF) == 1 + assert redact_llm_calls(session=db, cutoff=CUTOFF) == 0 + + +def test_row_with_null_content_is_redacted_without_error(db: Session, job: Job) -> None: + llm_call = create_aged_llm_call( + db, + updated_at=AGED, + job_id=job.id, + input="no response was ever recorded", + content=None, + ) + # A call that never got a response stores the JSONB scalar 'null', not SQL + # NULL; jsonb_set errors on scalars, so this must take a guarded path. + assert ( + db.exec( + text( + "SELECT jsonb_typeof(content) FROM llm_call WHERE id = :id" + ).bindparams(id=llm_call.id) + ).scalar_one() + == "null" + ) + + assert redact_llm_calls(session=db, cutoff=CUTOFF) == 1 + row = _reload(db, llm_call) + assert row.input == REDACTED_SENTINEL + assert row.content is None + + +def test_returns_zero_when_no_rows_match(db: Session, job: Job) -> None: + assert redact_llm_calls(session=db, cutoff=CUTOFF) == 0 + + +def test_redacts_only_aged_rows_in_a_mixed_table(db: Session, job: Job) -> None: + aged = create_aged_llm_call( + db, updated_at=AGED, job_id=job.id, input="aged", content=TEXT_CONTENT + ) + recent = create_aged_llm_call( + db, updated_at=RECENT, job_id=job.id, input="recent", content=TEXT_CONTENT + ) + + assert redact_llm_calls(session=db, cutoff=CUTOFF) == 1 + assert _reload(db, aged).input == REDACTED_SENTINEL + assert _reload(db, recent).input == "recent" + + +def test_redacts_content_value_of_row_whose_input_is_already_sentinel( + db: Session, job: Job +) -> None: + llm_call = create_aged_llm_call( + db, + updated_at=AGED, + job_id=job.id, + input=REDACTED_SENTINEL, + content={"type": "audio", "content": {"format": "uri", "value": "s3://a.wav"}}, + ) + + assert redact_llm_calls(session=db, cutoff=CUTOFF) == 1 + assert _reload(db, llm_call).content["content"]["value"] == REDACTED_SENTINEL + + +def test_second_run_skips_redacted_row_with_null_content(db: Session, job: Job) -> None: + create_aged_llm_call(db, updated_at=AGED, job_id=job.id, content=None) + + assert redact_llm_calls(session=db, cutoff=CUTOFF) == 1 + assert redact_llm_calls(session=db, cutoff=CUTOFF) == 0 + + +def test_second_run_skips_redacted_row_without_content_value( + db: Session, job: Job +) -> None: + llm_call = create_aged_llm_call( + db, updated_at=AGED, job_id=job.id, content={"type": "text"} + ) + + assert redact_llm_calls(session=db, cutoff=CUTOFF) == 1 + assert redact_llm_calls(session=db, cutoff=CUTOFF) == 0 + assert _reload(db, llm_call).content == {"type": "text"} diff --git a/backend/app/tests/services/llm/test_retention.py b/backend/app/tests/services/llm/test_retention.py new file mode 100644 index 000000000..5a6c9b58f --- /dev/null +++ b/backend/app/tests/services/llm/test_retention.py @@ -0,0 +1,75 @@ +from datetime import datetime, timedelta + +import pytest +from sqlmodel import Session + +from app.core.config import settings +from app.services.llm import retention +from app.services.llm.retention import redact_aged_llm_calls + + +def _stub_redact( + monkeypatch: pytest.MonkeyPatch, calls: list[dict], returns: int = 0 +) -> None: + def fake_redact(**kwargs) -> int: + calls.append(kwargs) + return returns + + monkeypatch.setattr(retention, "redact_llm_calls", fake_redact) + + +@pytest.fixture +def redact_calls(monkeypatch: pytest.MonkeyPatch) -> list[dict]: + calls: list[dict] = [] + _stub_redact(monkeypatch, calls) + return calls + + +def test_reports_rows_redacted_from_a_single_crud_call( + db: Session, monkeypatch: pytest.MonkeyPatch +) -> None: + calls: list[dict] = [] + _stub_redact(monkeypatch, calls, returns=4137) + + result = redact_aged_llm_calls(session=db) + + assert result.rows_redacted == 4137 + assert len(calls) == 1 + + +def test_no_matching_rows(db: Session, redact_calls: list[dict]) -> None: + result = redact_aged_llm_calls(session=db) + + assert result.rows_redacted == 0 + assert len(redact_calls) == 1 + + +def test_cutoff_is_naive_utc(db: Session, redact_calls: list[dict]) -> None: + # llm_call.updated_at is a naive TIMESTAMP; a tz-aware cutoff would raise on compare. + result = redact_aged_llm_calls(session=db) + + cutoff = redact_calls[0]["cutoff"] + assert cutoff.tzinfo is None + assert result.cutoff == cutoff + + +def test_cutoff_trails_now_by_the_configured_rolling_window( + db: Session, redact_calls: list[dict], monkeypatch: pytest.MonkeyPatch +) -> None: + frozen = datetime(2026, 3, 1, 12, 0, 0) + monkeypatch.setattr(retention, "now", lambda: frozen) + monkeypatch.setattr(settings, "LLM_CALL_RETENTION_HOURS", 168) + + redact_aged_llm_calls(session=db) + + assert redact_calls[0]["cutoff"] == datetime(2026, 2, 22, 12, 0, 0) + + +def test_passes_session_to_crud(db: Session, redact_calls: list[dict]) -> None: + redact_aged_llm_calls(session=db) + + assert redact_calls[0]["session"] is db + + +def test_default_rolling_window_is_one_week() -> None: + assert settings.LLM_CALL_RETENTION_TIMEDELTA == timedelta(days=7) diff --git a/backend/app/tests/utils/llm.py b/backend/app/tests/utils/llm.py index f16a883df..5a933c6d3 100644 --- a/backend/app/tests/utils/llm.py +++ b/backend/app/tests/utils/llm.py @@ -1,11 +1,13 @@ +from datetime import datetime from typing import Any +from uuid import UUID from sqlmodel import Session from app.crud import JobCrud from app.crud.llm import create_llm_call, update_llm_call_response from app.models import Job, JobType -from app.models.llm import LLMCallRequest +from app.models.llm import LlmCall, LLMCallRequest from app.models.llm.request import ( ConfigBlob, LLMCallConfig, @@ -19,7 +21,42 @@ TextOutput, Usage, ) -from app.tests.utils.utils import get_project +from app.tests.utils.utils import get_organization, get_project + + +def create_aged_llm_call( + db: Session, + *, + updated_at: datetime, + job_id: UUID, + input: str = "what is the capital of France?", + content: dict[str, Any] | None = None, +) -> LlmCall: + """Persist an LlmCall with an explicit `updated_at`, for retention tests.""" + project = get_project(db, "Dalgo") + organization = get_organization(db) + + llm_call = LlmCall( + job_id=job_id, + project_id=project.id, + organization_id=organization.id, + input=input, + input_type="text", + output_type="text", + provider="openai", + model="gpt-4o", + content=content, + ) + db.add(llm_call) + db.commit() + + # updated_at has a server/model default, so it is overwritten after insert. + llm_call.updated_at = updated_at + db.add(llm_call) + db.commit() + db.refresh(llm_call) + + return llm_call def create_llm_job(db: Session) -> Job: diff --git a/docs/wiki/cross-cutting/observability.md b/docs/wiki/cross-cutting/observability.md index b60d5c93b..c2915a731 100644 --- a/docs/wiki/cross-cutting/observability.md +++ b/docs/wiki/cross-cutting/observability.md @@ -18,7 +18,7 @@ OTel-first, Sentry as sole in-process sink (`instrumenter="otel"`). Init at `mai - Trace propagation: `CeleryIntegration(propagate_traces=True)` links API to worker as one trace; poll-loop re-enqueues pass `SENTRY_NO_PROPAGATE_HEADERS` (`celery/tasks/job_execution.py`) to start a fresh trace. - Error capture: generic handler in `core/exception_handlers.py` calls `capture_exception`. - Tenant impact (both in `core/telemetry/context.py`, called from `api/deps.py`): `set_request_log_context` puts org/project in the log context + Sentry tags; `bind_sentry_user` binds user/org/project via `sentry_sdk.set_user` so issues report users/orgs affected. The user binding is scope-only — ids in the log context would stamp per-user cardinality on every INFO record `enable_logs` ships. Per-request identity is the `correlation_id` tag (`core/middleware.py`), not the user binding. -- Crons: `@sentry_sdk.monitor` on `/cron/*` endpoints (`api/routes/cron.py`). +- Crons: `@sentry_sdk.monitor` on `/cron/*` endpoints (`api/routes/cron.py`), e.g. `llm-call-retention-cron-job` (daily 09:00 IST). - Runbook (alerts/dashboards/debugging): `features/sentry-utilization/SENTRY-RUNBOOK.md`. ## Telemetry / misc diff --git a/docs/wiki/modules/llm-call.md b/docs/wiki/modules/llm-call.md index 5df13faf0..353baec7f 100644 --- a/docs/wiki/modules/llm-call.md +++ b/docs/wiki/modules/llm-call.md @@ -9,6 +9,7 @@ All paths relative to `backend/app/`. - `api/routes/llm.py` — `POST /llm/call` - `api/routes/llm_chain.py` — chains - `api/routes/llm_sts.py` — speech-to-speech +- `api/routes/cron.py` — `GET /cron/llm-call-retention` (superuser, hidden from the schema) — daily retention redaction, see Gotchas - `api/routes/config/config.py`, `api/routes/config/version.py` — saved config CRUD + versions - `api/routes/guardrails.py` — `POST /guardrails` (async job) + `GET /guardrails/{job_id}` (poll), plus thin proxies over the internal kaapi-guardrails management API: - `GET /guardrails` — validator catalogue @@ -39,6 +40,7 @@ All paths relative to `backend/app/`. - `mappers.py` is the **only** param mapper in the codebase; the assessment fork was merged back into it. Callers: `crud/evaluations/{batch,fast,judge}.py`, `crud/assessment/batch.py`, `services/assessment/api/batch.py`, `services/assessment/prefilter/request_builder.py`. Structured output is keyed `output_schema` for every provider (OpenAI `text.format`, Anthropic `output_config.format`, Gemini `output_schema`), and Anthropic reads `effort` into `output_config.effort`. Every param that only assessment set (`top_p`, `max_output_tokens`, `thinking_level`, `thinking`, `output_schema`) defaults to `None` on `TextLLMParams` and is dropped by `ParamSerialization._dump_compact`, so a config that does not set it maps exactly as before. Anthropic never receives `temperature`/`top_p`: the Messages API returns 400 for a non-default sampling value on every Claude model Kaapi serves, so the mapper drops both, warning only when the value was not Anthropic's own default of 1.0. `normalize_llm_text` no longer lives here; it moved to `services/assessment/validators.py`, the only place that calls it. - `services/guardrails/` — validator execution - `crud/llm.py`, `crud/llm_chain.py`, `crud/config/` — persistence +- `services/llm/retention.py::redact_aged_llm_calls` → `crud/llm.py::redact_llm_calls` — retention redaction of aged `llm_call` rows ## Async - Job execution via Celery: `celery/tasks/job_execution.py`; `job` table tracks state. @@ -53,3 +55,5 @@ All paths relative to `backend/app/`. - Feature needs an LLM config? Spec `LLMCallConfig` whole (never a bespoke params + prompt pair), and prefer an optional per-request field over a per-project binding table — saved references already give durable versioned config via `config`/`config_version`. - `PromptTemplate.template` is a plain prompt string, and its `{{input}}` interpolation is **not** llm-chain-only: `execute_llm_call` does `template.replace("{{input}}", value)` unconditionally for every caller, before input guardrails run. A template that omits the placeholder therefore sends the template alone and silently drops the user input — which is why fast evaluation gates on it (`422 config_template_missing_input`). A feature that assembles its own inputs must leave `prompt_template` unset rather than assume it is ignored. - `transform_kaapi_config_to_native` adds `include=["file_search_call.results"]` on the OpenAI branch when the mapped params carry a `file_search` tool **and** the caller passed `include_file_search_results=True`, so a knowledge-base call gets its retrieved chunks back. It lives in the transform and not in `map_kaapi_to_openai_params` because that mapper's other callers build Batch API bodies, where `include` is invalid. It is opt-in, and `execute_llm_call` drives it off `include_provider_raw_response`: the hits are only reachable through the raw provider response, so for a caller that did not ask for one they are extra payload on every request. Evaluation is the caller that asks; ordinary `/llm/call` and `/llm/chain` traffic is unaffected unless the client sets the flag. +- `llm_call` retention: rows whose `updated_at` is older than `LLM_CALL_RETENTION_HOURS` (default 168) get `input` and `content.content.value` overwritten with `[redacted]` by one raw UPDATE. The row itself and its metadata (usage, provider, ids) are kept. The UPDATE does not bump `updated_at`, so re-runs skip rows only through the sentinel check in the WHERE clause. +- STT input and TTS output audio live at `llm-audio//{stt|tts}/.` (`storage_utils.upload_audio_bytes_to_s3`, `LLM_AUDIO_PREFIX` in `core/cloud/storage.py`). The literal prefix sits ahead of `storage_path` so one S3 lifecycle rule can expire it across every project; that rule is configured in AWS, not in this repo. Keys carry no org or project ids, because the TTS key reaches clients inside presigned URLs. Objects written before this change sit under `/orgs///audio/` and are not covered by the rule. diff --git a/docs/wiki/modules/platform.md b/docs/wiki/modules/platform.md index 55e9e954f..f116f0618 100644 --- a/docs/wiki/modules/platform.md +++ b/docs/wiki/modules/platform.md @@ -13,7 +13,7 @@ All paths relative to `backend/app/`. | Credentials | `api/routes/credentials.py` | `credential` (`models/credentials.py`) | `crud/credentials.py`; provider keys per org/project; envelope encryption (KMS-wrapped data key + AES-GCM), prefix-versioned ciphertexts | | Model config | `api/routes/model_config.py` | `model_config` (`models/model_config.py`) | `crud/model_config.py`; cost: `estimate_model_cost` (single lookup) vs `list_active_model_pricing` + `compute_cost_from_pricing` (one batched lookup — use in loops/aggregations, e.g. `services/analytics/aggregation.py`) | | Bucket providers | — | reuses `credential` (`google-gcp`) | `services/buckets/` — global registry + resolver (`providers/`), GCS V4 signed + bulk-signed URLs with a 24h expiry cap (`providers/gcs.py`, cap in `providers/base.py`), attachment path selection + URL resolution (`attachments.py`) | -| Cron | `api/routes/cron.py` | — | triggers batch polling (`crud/evaluations/cron.py`, `crud/assessment/cron.py` — the latter polls `method == RUN` assessments only; API-client BATCH runs are driven by their own Celery self-re-enqueue) | +| Cron | `api/routes/cron.py` | — | triggers batch polling (`crud/evaluations/cron.py`, `crud/assessment/cron.py` — the latter polls `method == RUN` assessments only; API-client BATCH runs are driven by their own Celery self-re-enqueue); `llm_call` retention redaction (`services/llm/retention.py`, see [llm-call.md](llm-call.md)) | | Jobs | — | `job` (`models/job.py`), `batch_job` (`models/batch_job.py`) | `crud/jobs.py`, `crud/job/`, `services/job_monitoring.py` | ## Credential contracts diff --git a/scripts/python/invoke-cron.py b/scripts/python/invoke-cron.py index e589c65aa..2156b00ad 100644 --- a/scripts/python/invoke-cron.py +++ b/scripts/python/invoke-cron.py @@ -19,6 +19,7 @@ "/api/v1/cron/evaluations", "/api/v1/cron/pending-jobs", "/api/v1/cron/daily-stats", + "/api/v1/cron/llm-call-retention", ] REQUEST_TIMEOUT = 30 # Timeout for requests in seconds