feat(cron): Clean up LLM call rows and stt/tts audio filepath reconciliation

.env.example

@@ -110,3 +110,5 @@ SMTP_PASSWORD=
 EMAILS_FROM_EMAIL=
 EMAILS_FROM_NAME=Kaapi
 FRONTEND_HOST=
+
+LLM_CALL_RETENTION_HOURS=

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,

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:

backend/app/core/config.py

@@ -204,10 +204,8 @@ 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 = Field(default=168, ge=1)
+
     ANTHROPIC_API_KEY: str = ""
     PROMPT_IMPROVEMENT_MODEL: str = "claude-opus-4-8"
 

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,29 +229,43 @@ 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/<storage_path>/<kind>/<call_id>.<ext>``: 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
     """
     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,
     )
 
 

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,39 @@ 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()
+
+    return result.rowcount
+
+
 def get_llm_calls_by_job_id(
     session: Session, job_id: UUID, project_id: int
 ) -> list[LlmCall]:

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

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.

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