📝(changelog) document summary privacy improvements

Record signed URL protection, secret-aware validation errors and safer
logging and Sentry capture under Unreleased.

Document the retention of redacted frame locals and application DEBUG
diagnostics.

CHANGELOG.md

@@ -12,6 +12,13 @@ and this project adheres to
 
 - ⬆️(summary) update ffmpeg to 8.1.3-r0
 
+### Fixed
+
+- 🔒️(summary) protect signed URLs in models, source logs and analytics metadata
+- 🔒️(summary) mask secret validation inputs and sanitize upstream and task errors
+- 🔒️(summary) scrub sensitive Sentry event and breadcrumb content while retaining
+  redacted frame locals and application DEBUG diagnostics
+
 ## [1.36.3] - 2026-10-23
 
 ### Fixed

src/summary/summary/api/route/tasks_v2.py

@@ -5,6 +5,7 @@ from datetime import datetime, timezone
 
 from celery.result import AsyncResult
 from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import ValidationError
 
 from summary.core.analytics import get_analytics
 from summary.core.celery_worker import (
@@ -13,7 +14,9 @@ from summary.core.celery_worker import (
     summarize_v2_task,
 )
 from summary.core.config import AuthorizedTenant, get_settings
+from summary.core.exceptions import CleanedValidationError
 from summary.core.models import SummarizeTaskApiRequest, TranscribeTaskApiRequest
+from summary.core.privacy import dump_for_transport
 from summary.core.security import verify_tenant_api_key_v2
 from summary.core.shared_models import (
     SummarizeWebhookFailurePayload,
@@ -52,7 +55,7 @@ async def create_transcribe_task_v2(
     task = process_audio_transcribe_v2_task.apply_async(
         args=[
             {
-                **request.model_dump(),
+                **dump_for_transport(request),
                 "tenant_id": request_tenant.id,
                 "received_at": datetime.now(timezone.utc),
             }
@@ -125,7 +128,11 @@ async def get_transcribe_job_status(
 
     if task.status == "SUCCESS":
         result = task.result
-        return TranscribeWebhookSuccessPayload.model_validate(result).model_dump()
+        try:
+            payload = TranscribeWebhookSuccessPayload.model_validate(result)
+        except ValidationError as exc:
+            raise CleanedValidationError(exc, secret_input=True) from None
+        return dump_for_transport(payload)
 
     if task.status == "FAILURE":
         return TranscribeWebhookFailurePayload(
@@ -155,7 +162,11 @@ async def get_summarize_job_status(
 
     if task.status == "SUCCESS":
         result = task.result
-        return SummarizeWebhookSuccessPayload.model_validate(result).model_dump()
+        try:
+            payload = SummarizeWebhookSuccessPayload.model_validate(result)
+        except ValidationError as exc:
+            raise CleanedValidationError(exc, secret_input=True) from None
+        return dump_for_transport(payload)
 
     if task.status == "FAILURE":
         return SummarizeWebhookFailurePayload(

src/summary/summary/core/analytics.py

@@ -5,7 +5,6 @@ import socket
 import time
 from collections import Counter
 from functools import cached_property, lru_cache
-from urllib.parse import urlsplit, urlunsplit
 
 import redis
 from celery.utils.log import get_task_logger
@@ -13,6 +12,7 @@ from posthog import Posthog
 
 from summary.core.config import get_settings
 from summary.core.models import SummarizeTaskJob, TranscribeTaskJob
+from summary.core.privacy import redact_url
 
 logger = get_task_logger(__name__)
 settings = get_settings()
@@ -54,8 +54,8 @@ class Analytics:
             self._client.capture(
                 event_name, distinct_id=distinct_id, properties=properties
             )
-        except Exception as e:
-            raise AnalyticsException("Failed to capture analytics event") from e
+        except Exception:
+            raise AnalyticsException("Failed to capture analytics event") from None
 
     def is_feature_enabled(self, feature_name: str, distinct_id: str = None) -> bool:
         """Check if a feature flag is enabled for a user."""
@@ -64,8 +64,8 @@ class Analytics:
 
         try:
             return self._client.feature_enabled(feature_name, distinct_id)
-        except Exception as e:
-            logger.error("Error checking feature flag %s: %s", feature_name, e)
+        except Exception:
+            logger.error("Error checking feature flag %s", feature_name)
             return False
 
 
@@ -133,8 +133,7 @@ class MetadataManager:
         }
 
         if isinstance(task_payload, TranscribeTaskJob):
-            parts = urlsplit(task_payload.cloud_storage_url)
-            clean_url = urlunsplit((parts.scheme, parts.netloc, parts.path, "", ""))
+            clean_url = redact_url(task_payload.cloud_storage_url.get_secret_value())
             initial_metadata["source_url"] = clean_url
             initial_metadata["asr_model"] = settings.whisperx_asr_model
         elif isinstance(task_payload, SummarizeTaskJob):

src/summary/summary/core/celery_worker.py

@@ -3,20 +3,23 @@
 # ruff: noqa: PLR0913
 
 import json
+import logging
 import time
 from datetime import datetime, timezone
 from typing import Any
 from urllib.parse import urljoin
 
 import requests
-from celery import Celery, signals
+from celery import Celery, Task, signals
 from celery.utils.log import get_task_logger
 from openai.types.audio import Transcription
+from pydantic import SecretStr, ValidationError
 from requests import exceptions
 
 from summary.core.analytics import MetadataManager, get_analytics
 from summary.core.config import get_settings
 from summary.core.docs_service import create_document_in_lasuite_docs
+from summary.core.exceptions import CleanedValidationError
 from summary.core.file_service import (
     CorruptedAudioFile,
     FileService,
@@ -31,6 +34,7 @@ from summary.core.models import (
     SummarizeTaskJob,
     TranscribeTaskJob,
 )
+from summary.core.privacy import dump_for_transport, redact_url
 from summary.core.prompt import (
     FORMAT_NEXT_STEPS,
     FORMAT_PLAN,
@@ -63,8 +67,24 @@ metadata_manager = MetadataManager()
 
 logger = get_task_logger(__name__)
 
+
+def _reveal(secret: SecretStr | None) -> str | None:
+    """Return the underlying value of an optional secret."""
+    return secret.get_secret_value() if secret is not None else None
+
+
+class PrivateTask(Task):
+    """Hide task arguments in Celery logs and monitoring events."""
+
+    def apply_async(self, args=None, kwargs=None, **options):
+        """Publish with redacted display representations."""
+        options.update(argsrepr="[Filtered]", kwargsrepr="[Filtered]")
+        return super().apply_async(args=args, kwargs=kwargs, **options)
+
+
 celery = Celery(
     __name__,
+    task_cls=PrivateTask,
     broker=settings.celery_broker_url,
     backend=settings.celery_result_backend,
     broker_connection_retry_on_startup=True,
@@ -76,10 +96,28 @@ celery = Celery(
 celery.config_from_object("summary.core.celery_config")
 
 
+class CeleryLogPrivacyFilter(logging.Filter):
+    """Prevent Celery trace logs from serializing task results or exceptions."""
+
+    def filter(self, record):
+        """Keep task identity and state without upstream values."""
+        if record.levelno < logging.INFO:
+            return True
+        data = getattr(record, "data", None)
+        if isinstance(data, dict):
+            record.msg = "Task %s[%s] finished or retried (details omitted)"
+            record.args = (data.get("name", "unknown"), data.get("id", "unknown"))
+            record.data = {}
+            record.exc_info = None
+            record.exc_text = None
+        return True
+
+
 @signals.celeryd_init.connect
 def init_celery_sentry(**_kwargs):
     """Initialize Sentry in the Celery worker."""
     init_sentry()
+    logging.getLogger("celery.app.trace").addFilter(CeleryLogPrivacyFilter())
 
 
 file_service = FileService()
@@ -162,9 +200,9 @@ def transcribe_audio(
             try:
                 res.raise_for_status()
             except requests.exceptions.HTTPError:
-                logger.exception("WhisperX transcription failed")
+                logger.error("WhisperX transcription failed")
                 # We reraise the error so that it can be retried by celery
-                raise
+                raise requests.exceptions.HTTPError("WhisperX request failed") from None
 
             transcription_json: dict[str, Any] = res.json()
             # We remove the "usage" key from the transcription_json dictionary
@@ -175,13 +213,16 @@ def transcribe_audio(
             # We force the use of the Transcription model here
             # to avoid changing too much code for now.
             # Note that it should be WhisperXResponse instead.
-            transcription = Transcription.model_validate(
-                # We add a dummy "text" to make the model validate,
-                # Some API responses lack the "text" key.
-                {"text": "", **transcription_json},
-                extra="allow",
-                strict=False,
-            )
+            try:
+                transcription = Transcription.model_validate(
+                    # We add a dummy "text" to make the model validate,
+                    # Some API responses lack the "text" key.
+                    {"text": "", **transcription_json},
+                    extra="allow",
+                    strict=False,
+                )
+            except ValidationError as exc:
+                raise CleanedValidationError(exc, secret_input=True) from None
 
             # Logging
             transcription_duration = round(time.time() - transcription_start_time, 2)
@@ -199,9 +240,9 @@ def transcribe_audio(
         if raises:
             raise e
         redacted_cloud_storage_url = (
-            cloud_storage_url.split("?", 1)[0] if cloud_storage_url else None
+            redact_url(cloud_storage_url) if cloud_storage_url else None
         )
-        logger.exception(
+        logger.error(
             ("Unexpected error while preparing file %s "),
             redacted_cloud_storage_url,
         )
@@ -230,7 +271,7 @@ def resolve_speaker_identities_and_apply_to(
     logger.debug("Running resolve_speaker_identities")
     try:
         metadata = file_service.read_cloud_storage_json(
-            recording_metadata.cloud_storage_url
+            recording_metadata.cloud_storage_url.get_secret_value()
         )
         speaker_mapping = resolve_speaker_identities(
             metadata,
@@ -241,21 +282,18 @@ def resolve_speaker_identities_and_apply_to(
         new_transcription = speaker_mapping.apply_to(transcription.model_dump())
         return WhisperXResponse.model_validate(new_transcription)
 
-    except FileServiceException as exc:
+    except FileServiceException:
         logger.error(
-            "Error reading metadata for task %s; skipping speaker assignment."
-            " Error: %s",
+            "Error reading metadata for task %s; skipping speaker assignment.",
             task_id,
-            exc,
         )
         return transcription
 
-    except Exception as exc:
-        logger.exception(
+    except Exception:
+        logger.error(
             "resolve_speaker_identities failed for task %s; skipping"
-            " speaker assignment. Error: %s",
+            " speaker assignment.",
             task_id,
-            exc,
         )
         return transcription
 
@@ -302,6 +340,17 @@ def format_actions(llm_output: dict) -> str:
     return ""
 
 
+def parse_llm_json(value: str) -> dict:
+    """Parse an LLM response without exposing it in decoding errors."""
+    try:
+        result = json.loads(value)
+        if not isinstance(result, dict):
+            raise ValueError
+        return result
+    except (TypeError, ValueError):
+        raise LLMException("Invalid LLM response") from None
+
+
 def summarize_transcription_internals(
     *, distinct_id: str, transcript: str, session_id: str
 ) -> str:
@@ -342,13 +391,13 @@ def summarize_transcription_internals(
     )
     logger.info("Plan generated")
 
-    res = json.loads(parts)
+    res = parse_llm_json(parts)
     parts = res.get("titles", [])
-    logger.info("Parts to summarize: %s", parts)
+    logger.info("Parts to summarize: %s", len(parts))
     parts_summarized = []
-    for part in parts:
+    for i, part in enumerate(parts):
         prompt_user_part = PROMPT_USER_PART.format(part=part, transcript=transcript)
-        logger.info("Summarizing part: %s", part)
+        logger.info("Summarizing part %s", i + 1)
         parts_summarized.append(
             llm_service.call(PROMPT_SYSTEM_PART, prompt_user_part, name="part")
         )
@@ -364,7 +413,7 @@ def summarize_transcription_internals(
         response_format=FORMAT_NEXT_STEPS,
     )
 
-    next_steps = format_actions(json.loads(next_steps))
+    next_steps = format_actions(parse_llm_json(next_steps))
 
     logger.info("Next steps generated")
 
@@ -437,9 +486,11 @@ def call_webhook_v2_task(
     tenant_id: str,
 ):
     """Calls a webhook asynchrously (retry handled by celery)."""
-    call_webhook_v2(
-        payload=webhook_payload_adapter.validate_python(payload), tenant_id=tenant_id
-    )
+    try:
+        validated_payload = webhook_payload_adapter.validate_python(payload)
+    except ValidationError as exc:
+        raise CleanedValidationError(exc, secret_input=True) from None
+    call_webhook_v2(payload=validated_payload, tenant_id=tenant_id)
 
 
 @celery.task(
@@ -465,7 +516,10 @@ def process_audio_transcribe_v2_task(
         self: Celery task instance (passed on with bind=True)
         payload: Serialized dictionary of TranscribeSummarizeTaskCreationV2
     """
-    payload = TranscribeTaskJob.model_validate(payload)
+    try:
+        payload = TranscribeTaskJob.model_validate(payload)
+    except ValidationError as exc:
+        raise CleanedValidationError(exc, secret_input=True) from None
     logger.info(
         "Transcribing for object received | Owner: %s",
         payload.user_sub,
@@ -477,7 +531,7 @@ def process_audio_transcribe_v2_task(
         transcription_res = WhisperXResponse(
             **transcribe_audio(  # type: ignore
                 task_id=job_id,
-                cloud_storage_url=payload.cloud_storage_url,
+                cloud_storage_url=payload.cloud_storage_url.get_secret_value(),
                 language=payload.language,
                 raises=True,
             ).model_dump()
@@ -514,8 +568,8 @@ def process_audio_transcribe_v2_task(
             transcription_res.model_dump(),
             payload.context_language,
             payload.language,
-            payload.push_to_docs_config.download_link,
-            payload.push_to_docs_config.form_link,
+            _reveal(payload.push_to_docs_config.download_link),
+            _reveal(payload.push_to_docs_config.form_link),
         )
 
         create_document_in_lasuite_docs(
@@ -556,18 +610,22 @@ def process_audio_transcribe_v2_task(
         transcription_data_url=file_service.get_transcript_signed_url(job_id),
     )
     call_webhook_v2_task.apply_async(
-        args=[success_payload.model_dump(), payload.tenant_id]
+        args=[dump_for_transport(success_payload), payload.tenant_id]
     )
     metadata_manager.capture(job_id, settings.posthog_transcript_success)
 
-    return success_payload.model_dump()
+    return dump_for_transport(success_payload)
 
 
 @signals.task_prerun.connect(sender=process_audio_transcribe_v2_task)
 def task_started_transcript(task_id=None, task=None, args=None, **kwargs):
     """Signal handler called before task execution begins."""
     if args:
-        metadata_manager.create(task_id, TranscribeTaskJob.model_validate(args[0]))
+        try:
+            payload = TranscribeTaskJob.model_validate(args[0])
+        except ValidationError as exc:
+            raise CleanedValidationError(exc, secret_input=True) from None
+        metadata_manager.create(task_id, payload)
 
 
 @signals.task_retry.connect(sender=process_audio_transcribe_v2_task)
@@ -627,7 +685,10 @@ def summarize_v2_task(
     1. Run summary internals
     2. Sends the final summary via webhook.
     """
-    payload = SummarizeTaskJob.model_validate(payload)
+    try:
+        payload = SummarizeTaskJob.model_validate(payload)
+    except ValidationError as exc:
+        raise CleanedValidationError(exc, secret_input=True) from None
     summary = summarize_transcription_internals(
         distinct_id=payload.user_sub,
         transcript=payload.content,
@@ -652,18 +713,22 @@ def summarize_v2_task(
         summary_data_url=file_service.get_summary_signed_url(job_id),
     )
     call_webhook_v2_task.apply_async(
-        args=[success_payload.model_dump(), payload.tenant_id]
+        args=[dump_for_transport(success_payload), payload.tenant_id]
     )
     metadata_manager.capture(job_id, settings.posthog_summary_success)
 
-    return success_payload.model_dump()
+    return dump_for_transport(success_payload)
 
 
 @signals.task_prerun.connect(sender=summarize_v2_task)
 def task_started_summary(task_id=None, task=None, args=None, **kwargs):
     """Signal handler called before task execution begins."""
     if args:
-        metadata_manager.create(task_id, SummarizeTaskJob.model_validate(args[0]))
+        try:
+            payload = SummarizeTaskJob.model_validate(args[0])
+        except ValidationError as exc:
+            raise CleanedValidationError(exc, secret_input=True) from None
+        metadata_manager.create(task_id, payload)
 
 
 @signals.task_retry.connect(sender=summarize_v2_task)

src/summary/summary/core/docs_service.py

@@ -4,7 +4,7 @@ import json
 import logging
 from urllib.parse import urljoin
 
-from requests import Session
+from requests import RequestException, Session
 from requests.adapters import HTTPAdapter
 from urllib3.util import Retry
 
@@ -43,6 +43,8 @@ def _post_with_retries(*, url, data, api_key: str | None = None):
         response = session.post(url, json=data, timeout=(20, 3 * 60))
         response.raise_for_status()
         return response
+    except RequestException:
+        raise RequestException("Docs delivery failed") from None
     finally:
         session.close()
 

src/summary/summary/core/exceptions.py

@@ -0,0 +1,62 @@
+"""Validation errors suitable for logging and transport."""
+
+from typing import get_args
+
+from pydantic import SecretStr, ValidationError
+from pydantic_core import ErrorType, PydanticCustomError
+
+_BUILTIN_ERROR_TYPES = frozenset(get_args(ErrorType))
+
+
+def clean_validation_errors(errors: list[dict]) -> list[dict]:
+    """Mask SecretStr values, including those nested in validation input."""
+
+    def clean(value):
+        if isinstance(value, SecretStr):
+            return "[redacted]"
+        if isinstance(value, Exception):
+            # Do not retain validator exception objects, whose args may hold secrets.
+            return str(value)
+        if isinstance(value, dict):
+            return {key: clean(item) for key, item in value.items()}
+        if isinstance(value, list):
+            return [clean(item) for item in value]
+        if isinstance(value, tuple):
+            return tuple(clean(item) for item in value)
+        return value
+
+    return clean(errors)
+
+
+class CleanedValidationError(ValidationError):
+    """A new Pydantic exception with secret inputs replaced by a marker."""
+
+    def __init__(self, error, *args, secret_input=False, **kwargs):
+        """State is initialized by Pydantic during __new__."""
+
+    def __new__(cls, error, *args, secret_input=False, **kwargs):
+        """Accept an original error, or Pydantic's native construction arguments."""
+        if not isinstance(error, ValidationError):
+            # Pydantic uses this path when constructing or unpickling the error.
+            return super().__new__(cls, error, *args, **kwargs)
+
+        details = clean_validation_errors(error.errors(include_url=False))
+        for detail in details:
+            if secret_input:
+                # An adapter may have received the unwrapped value of a SecretStr.
+                detail["input"] = "[redacted]"
+        return cls._from_details(error.title, details)
+
+    @classmethod
+    def _from_details(cls, title, details):
+        """Reconstruct builtin and custom Pydantic error codes."""
+        for detail in details:
+            if detail["type"] not in _BUILTIN_ERROR_TYPES:
+                detail["type"] = PydanticCustomError(
+                    detail["type"], detail["msg"], detail.get("ctx")
+                )
+        return cls.from_exception_data(title, details, hide_input=True)
+
+    def __reduce__(self):
+        """Preserve custom error codes when Celery pickles the exception."""
+        return self._from_details, (self.title, self.errors(include_url=False))

src/summary/summary/core/file_service.py

@@ -17,6 +17,7 @@ import requests
 from botocore.config import Config
 
 from summary.core.config import get_settings
+from summary.core.privacy import redact_url
 from summary.core.shared_models import WhisperXResponse
 
 settings = get_settings()
@@ -314,7 +315,7 @@ class FileService:
         """Download file from a cloud storage URL to local temporary file."""
         logger.info(
             "Download recording from URL | cloud_storage_url: %s",
-            cloud_storage_url,
+            redact_url(cloud_storage_url),
         )
 
         if not cloud_storage_url:
@@ -352,10 +353,10 @@ class FileService:
                     logger.debug("Recording local file path: %s", local_path)
                     return local_path
 
-        except requests.RequestException as e:
+        except requests.RequestException:
             raise FileServiceException(
                 "Unexpected error while downloading object from cloud_storage_url."
-            ) from e
+            ) from None
 
     def _validate_duration(self, duration_seconds: float) -> None:
         """Validate audio file duration against configured maximum."""
@@ -376,7 +377,7 @@ class FileService:
 
     def read_cloud_storage_json(self, cloud_storage_url: str) -> dict:
         """Read and parse a JSON file from a url."""
-        logger.info("Reading JSON: %s", cloud_storage_url)
+        logger.info("Reading JSON: %s", redact_url(cloud_storage_url))
         res = requests.get(cloud_storage_url, timeout=(10, 100))
         res.raise_for_status()
         return res.json()

src/summary/summary/core/llm_service.py

@@ -140,6 +140,6 @@ class LLMService:
             response = self._client.chat.completions.create(**params)
             return response.choices[0].message.content
 
-        except Exception as e:
-            logger.exception("LLM call failed: %s", e)
-            raise LLMException(f"LLM call failed: {e}") from e
+        except Exception:
+            logger.error("LLM call failed")
+            raise LLMException("LLM call failed") from None

src/summary/summary/core/models.py

@@ -2,10 +2,17 @@
 
 from datetime import datetime
 
-from pydantic import AwareDatetime, BaseModel, EmailStr, Field, field_validator
+from pydantic import (
+    AwareDatetime,
+    BaseModel,
+    ConfigDict,
+    EmailStr,
+    Field,
+    field_validator,
+)
 
 from summary.core.config import get_settings
-from summary.core.types import Url
+from summary.core.types import SecretUrl
 
 settings = get_settings()
 
@@ -13,6 +20,8 @@ settings = get_settings()
 class SharedV2TaskCreation(BaseModel):
     """Model that holds basic information for task creation."""
 
+    model_config = ConfigDict(hide_input_in_errors=True)
+
     user_sub: str = Field(title="User Sub", description="The user's sub.")
     user_email: EmailStr | None = Field(
         default=None,
@@ -24,7 +33,9 @@ class SharedV2TaskCreation(BaseModel):
 class RecordingMetadata(BaseModel):
     """Model for recording metadata."""
 
-    cloud_storage_url: Url = Field(
+    model_config = ConfigDict(hide_input_in_errors=True)
+
+    cloud_storage_url: SecretUrl = Field(
         title="Cloud Storage URL",
         description="The URL of the metadata file for speaker assignment.",
     )
@@ -35,6 +46,8 @@ class RecordingMetadata(BaseModel):
 class PushToDocsBaseConfig(BaseModel):
     """Model containing information for pushing transcript and summaries to docs."""
 
+    model_config = ConfigDict(hide_input_in_errors=True)
+
     user_email: EmailStr = Field(
         title="User Email", description="The user's email, future owner of the docs."
     )
@@ -44,12 +57,12 @@ class PushToDocsBaseConfig(BaseModel):
 class PushToDocsTranscriptConfig(PushToDocsBaseConfig):
     """Model for push to docs information for transcripts."""
 
-    download_link: Url | None = Field(
+    download_link: SecretUrl | None = Field(
         default=None,
         title="Download Link",
         description="The link to download the recording.",
     )
-    form_link: Url | None = Field(
+    form_link: SecretUrl | None = Field(
         default=None,
         title="Form Link",
         description="The link to fill out a form for the recording.",
@@ -69,7 +82,7 @@ class PushToDocsSummaryConfig(PushToDocsBaseConfig):
 class TranscribeTaskApiRequest(SharedV2TaskCreation):
     """Model for creating a transcribe task (used for API request)."""
 
-    cloud_storage_url: Url = Field(
+    cloud_storage_url: SecretUrl = Field(
         title="Cloud storage URL",
         description="The URL of the audio file to transcribe.",
     )

src/summary/summary/core/privacy.py

@@ -0,0 +1,36 @@
+"""Privacy helpers for observability."""
+
+from urllib.parse import urlsplit, urlunsplit
+
+from pydantic import BaseModel, SecretStr
+
+
+def dump_for_transport(model: BaseModel) -> dict:
+    """Reveal secrets explicitly for task messages, HTTP bodies and API responses."""
+
+    def reveal(value):
+        if isinstance(value, SecretStr):
+            return value.get_secret_value()
+        if isinstance(value, dict):
+            return {key: reveal(item) for key, item in value.items()}
+        if isinstance(value, list):
+            return [reveal(item) for item in value]
+        return value
+
+    return reveal(model.model_dump())
+
+
+def redact_url(url):
+    """Keep the source identifiable without logging query or user credentials."""
+    if not isinstance(url, str):
+        return "[redacted]"
+    # Remove query and fragment before parsing, including for malformed URLs.
+    base, has_query, _ = url.partition("?")
+    base = base.partition("#")[0]
+    try:
+        parts = urlsplit(base)
+    except ValueError:
+        return "[redacted]"
+    authority = parts.netloc.rsplit("@", 1)[-1]
+    redacted = urlunsplit(parts._replace(netloc=authority))
+    return redacted + ("?[redacted]" if has_query else "")

src/summary/summary/core/sentry.py

@@ -1,9 +1,12 @@
 """Sentry configuration."""
 
+import re
+
 import sentry_sdk
 from sentry_sdk.scrubber import DEFAULT_DENYLIST, DEFAULT_PII_DENYLIST, EventScrubber
 
 from summary.core.config import get_settings
+from summary.core.privacy import redact_url
 
 # Exact names (case-insensitive) of variables and dict keys to redact.
 SENSITIVE_DATA_DENYLIST = [
@@ -90,5 +93,45 @@ def init_sentry() -> None:
         # Task creation requests carry the content to summarize.
         max_request_body_size="never",
         include_local_variables=True,
+        before_send=before_send,
+        before_send_transaction=before_send,
+        before_breadcrumb=before_breadcrumb,
         event_scrubber=build_event_scrubber(),
     )
+
+
+def _scrub_strings(value):
+    """Remove URL credentials, queries and fragments throughout an event."""
+    if isinstance(value, str):
+        return re.sub(
+            r"https?://[^\s\"'<>]+",
+            lambda m: redact_url(m.group()),
+            value,
+            flags=re.IGNORECASE,
+        )
+    if isinstance(value, dict):
+        return {key: _scrub_strings(item) for key, item in value.items()}
+    if isinstance(value, list):
+        return [_scrub_strings(item) for item in value]
+    return value
+
+
+def before_send(event, hint):
+    """Keep exception types and stacks, but discard untrusted message text."""
+    del hint
+    event.pop("message", None)
+    event.pop("logentry", None)
+    request = event.get("request", {})
+    request.pop("data", None)
+    request.pop("query_string", None)
+    for exception in event.get("exception", {}).get("values", []):
+        exception["value"] = "[Filtered]"
+    return _scrub_strings(event)
+
+
+def before_breadcrumb(crumb, hint):
+    """Drop arbitrary breadcrumb text and scrub URLs in structured data."""
+    del hint
+    crumb.pop("message", None)
+    crumb.pop("data", None)
+    return _scrub_strings(crumb)

src/summary/summary/core/shared_models.py

@@ -2,7 +2,9 @@
 
 from typing import Annotated, Literal, Union
 
-from pydantic import BaseModel, Field, TypeAdapter
+from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
+
+from summary.core.types import SecretUrl
 
 
 class WordSegment(BaseModel):
@@ -65,6 +67,8 @@ class WhisperXResponse(BaseModel):
 class BaseWebhook(BaseModel):
     """Base webhook payload."""
 
+    model_config = ConfigDict(hide_input_in_errors=True)
+
     job_id: str = Field(
         title="Job ID",
         description="The ID of the job document in the receiver system.",
@@ -76,7 +80,7 @@ class TranscribeWebhookSuccessPayload(BaseWebhook):
 
     type: Literal["transcript"] = Field(default="transcript")
     status: Literal["success"] = Field(default="success")
-    transcription_data_url: str = Field(
+    transcription_data_url: SecretUrl = Field(
         title="Transcript", description="URL to the raw transcription data."
     )
 
@@ -121,7 +125,7 @@ class SummarizeWebhookSuccessPayload(BaseWebhook):
 
     type: Literal["summary"] = Field(default="summary")
     status: Literal["success"] = Field(default="success")
-    summary_data_url: str = Field(
+    summary_data_url: SecretUrl = Field(
         title="Summary", description="URL to the raw summary data."
     )
 

src/summary/summary/core/types.py

@@ -2,9 +2,31 @@
 
 from typing import Annotated
 
-from pydantic import BeforeValidator, HttpUrl, TypeAdapter
+from pydantic import (
+    AfterValidator,
+    BeforeValidator,
+    HttpUrl,
+    SecretStr,
+    TypeAdapter,
+    ValidationError,
+)
+
+from summary.core.exceptions import CleanedValidationError
 
 http_url_adapter = TypeAdapter(HttpUrl)
 Url = Annotated[
     str, BeforeValidator(lambda value: str(http_url_adapter.validate_python(value)))
 ]
+
+
+def _validate_secret_url(value: SecretStr) -> SecretStr:
+    """Validate and normalize a URL without echoing it in validator messages."""
+    try:
+        return SecretStr(
+            str(http_url_adapter.validate_python(value.get_secret_value()))
+        )
+    except ValidationError as exc:
+        raise CleanedValidationError(exc, secret_input=True) from None
+
+
+SecretUrl = Annotated[SecretStr, AfterValidator(_validate_secret_url)]

src/summary/summary/core/webhook_service.py

@@ -6,6 +6,7 @@ import logging
 import requests
 
 from summary.core.config import get_settings
+from summary.core.privacy import dump_for_transport
 from summary.core.shared_models import (
     WebhookPayloads,
 )
@@ -29,16 +30,19 @@ def call_webhook_v2(
     logger.debug("Submitting to %s", tenant.webhook_url)
     logger.debug("Request payload: %s", payload.model_dump_json(indent=2))
 
-    response = requests.post(
-        tenant.webhook_url,
-        json=payload.model_dump(),
-        headers={
-            "Authorization": f"Bearer {tenant.webhook_api_key.get_secret_value()}",
-            "User-Agent": settings.app_external_user_agent,
-        },
-        timeout=(10, 20),
-    )
-    response.raise_for_status()
+    try:
+        response = requests.post(
+            tenant.webhook_url,
+            json=dump_for_transport(payload),
+            headers={
+                "Authorization": f"Bearer {tenant.webhook_api_key.get_secret_value()}",
+                "User-Agent": settings.app_external_user_agent,
+            },
+            timeout=(10, 20),
+        )
+        response.raise_for_status()
+    except requests.RequestException:
+        raise requests.RequestException("Webhook delivery failed") from None
 
     try:
         response_data = response.json()

src/summary/summary/main.py

@@ -1,11 +1,15 @@
 """Application."""
 
 from dockerflow.fastapi import router as dockerflow_router
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
 
 from summary.api.main import api_router_v2
 from summary.core import checks  # noqa: F401 -- registers the Dockerflow checks
 from summary.core.config import get_settings
+from summary.core.exceptions import clean_validation_errors
 from summary.core.sentry import init_sentry
 
 settings = get_settings()
@@ -19,3 +23,13 @@ app = FastAPI(
 
 app.include_router(api_router_v2, prefix=settings.app_api_v2_str)
 app.include_router(dockerflow_router)
+
+
+@app.exception_handler(RequestValidationError)
+async def validation_error_handler(request: Request, exc: RequestValidationError):
+    """Keep validation details while masking SecretStr input values."""
+    del request
+    return JSONResponse(
+        status_code=422,
+        content=jsonable_encoder({"detail": clean_validation_errors(exc.errors())}),
+    )

src/summary/tests/api/test_api_tasks_v2.py

@@ -3,6 +3,10 @@
 from datetime import datetime
 from unittest.mock import ANY, MagicMock, patch
 
+import pytest
+
+from summary.core.exceptions import CleanedValidationError
+
 
 class TestTasksV2:
     """Tests for the /v2/async-jobs-jobs endpoints."""
@@ -174,3 +178,163 @@ class TestTasksV2:
 
         assert response.status_code == 403
         assert response.json() == {"detail": "Forbidden"}
+
+
+def test_validation_preserves_ordinary_input(client):
+    """Validation errors keep input values that are not SecretStr."""
+    canary = "private-meeting-content"
+    response = client.post(
+        "/api/v2/async-jobs/summarize",
+        headers={"Authorization": "Bearer test-api-token"},
+        json={"user_sub": "user", "content": {"unexpected": canary}},
+    )
+    assert response.status_code == 422
+    assert canary in response.text
+    assert response.json()["detail"] == [
+        {
+            "type": "string_type",
+            "loc": ["body", "content"],
+            "msg": "Input should be a valid string",
+            "input": {"unexpected": canary},
+        }
+    ]
+
+
+def test_validation_preserves_context(client):
+    """Validator errors retain location, message and serializable context."""
+    canary = "private-invalid-url-signature"
+    response = client.post(
+        "/api/v2/async-jobs/transcribe",
+        headers={"Authorization": "Bearer test-api-token"},
+        json={
+            "user_sub": "user",
+            "language": "en",
+            "cloud_storage_url": f"invalid-url?signature={canary}",
+        },
+    )
+    assert response.status_code == 422
+    assert canary not in response.text
+    assert response.json()["detail"] == [
+        {
+            "type": "url_parsing",
+            "loc": ["body", "cloud_storage_url"],
+            "msg": "Input should be a valid URL, relative URL without a base",
+            "input": "[redacted]",
+            "ctx": {"error": "relative URL without a base"},
+        }
+    ]
+
+
+@patch(
+    "summary.api.route.tasks_v2.process_audio_transcribe_v2_task.apply_async",
+    return_value=MagicMock(id="transcribe-job"),
+)
+def test_signed_urls_are_unwrapped_for_task_submission(mock_apply_async, client):
+    """API input secrets reach Celery intact, including nested metadata URLs."""
+    signed_url = "https://storage.example.com/file?signature=deterministic-signature"
+    response = client.post(
+        "/api/v2/async-jobs/transcribe",
+        headers={"Authorization": "Bearer test-api-token"},
+        json={
+            "user_sub": "user",
+            "language": "en",
+            "cloud_storage_url": signed_url,
+            "metadata": {
+                "cloud_storage_url": signed_url,
+                "started_at": "2026-01-01T10:00:00Z",
+                "ended_at": "2026-01-01T11:00:00Z",
+            },
+        },
+    )
+    assert response.status_code == 200
+    payload = mock_apply_async.call_args.kwargs["args"][0]
+    assert payload["cloud_storage_url"] == signed_url
+    assert payload["metadata"]["cloud_storage_url"] == signed_url
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        ("transcribe", "transcript", "transcription_data_url"),
+        ("summarize", "summary", "summary_data_url"),
+    ],
+)
+@patch("summary.api.route.tasks_v2.celery")
+@patch("summary.api.route.tasks_v2.AsyncResult")
+def test_success_status_returns_usable_signed_url(
+    mock_async_result, mock_celery, client, case
+):
+    """Success responses explicitly reveal result URLs for authorized clients."""
+    job_type, payload_type, url_field = case
+    signed_url = "https://storage.example.com/result?signature=deterministic-signature"
+    mock_celery.backend.client.exists.return_value = True
+    mock_async_result.return_value = MagicMock(
+        status="SUCCESS",
+        args=[{"tenant_id": "test-tenant"}],
+        result={"job_id": "job", url_field: signed_url},
+    )
+    response = client.get(
+        f"/api/v2/async-jobs/{job_type}/job",
+        headers={"Authorization": "Bearer test-api-token"},
+    )
+    assert response.status_code == 200
+    assert response.json() == {
+        "job_id": "job",
+        "type": payload_type,
+        "status": "success",
+        url_field: signed_url,
+    }
+
+
+@pytest.mark.parametrize(
+    "case",
+    [
+        ("transcribe", "transcription_data_url"),
+        ("summarize", "summary_data_url"),
+    ],
+)
+@patch("summary.api.route.tasks_v2.celery")
+@patch("summary.api.route.tasks_v2.AsyncResult")
+def test_invalid_success_result_hides_revealed_url(
+    mock_async_result, mock_celery, client, case
+):
+    """Missing result fields cannot retain a raw signed URL in error details."""
+    job_type, url_field = case
+    canary = "private-result-signature"
+    mock_celery.backend.client.exists.return_value = True
+    mock_async_result.return_value = MagicMock(
+        status="SUCCESS",
+        args=[{"tenant_id": "test-tenant"}],
+        result={url_field: f"https://storage.example.com/result?signature={canary}"},
+    )
+    with pytest.raises(CleanedValidationError) as exc:
+        client.get(
+            f"/api/v2/async-jobs/{job_type}/job",
+            headers={"Authorization": "Bearer test-api-token"},
+        )
+    assert exc.value.errors()[0]["type"] == "missing"
+    assert exc.value.errors()[0]["loc"] == ("job_id",)
+    assert exc.value.errors()[0]["input"] == "[redacted]"
+    assert canary not in str(exc.value)
+    assert canary not in exc.value.json()
+
+
+@pytest.mark.parametrize("nested_error", [False, True])
+def test_missing_field_errors_preserve_raw_json_strings(client, nested_error):
+    """Raw JSON input is retained when it has not been converted to SecretStr."""
+    canary = "private-validation-signature"
+    signed_url = f"https://storage.example.com/recording?signature={canary}"
+    payload = {"language": "en", "cloud_storage_url": signed_url}
+    if nested_error:
+        payload["user_sub"] = "user"
+        payload["metadata"] = {"cloud_storage_url": signed_url}
+    response = client.post(
+        "/api/v2/async-jobs/transcribe",
+        headers={"Authorization": "Bearer test-api-token"},
+        json=payload,
+    )
+    assert response.status_code == 422
+    assert canary in response.text
+    assert all(error["type"] == "missing" for error in response.json()["detail"])
+    expected_input = payload["metadata"] if nested_error else payload
+    assert all(error["input"] == expected_input for error in response.json()["detail"])

src/summary/tests/conftest.py

@@ -22,6 +22,12 @@ def get_settings_override():
     )
 
 
+@pytest.fixture()
+def settings_override():
+    """Provide isolated settings containing the test tenant."""
+    return get_settings_override()
+
+
 @pytest.fixture()
 def client():
     """Provide a FastAPI TestClient for tests."""

src/summary/tests/models/test_secret_urls.py

@@ -0,0 +1,171 @@
+"""Secret URLs stay masked until explicitly serialized for transport."""
+
+from unittest.mock import Mock
+
+import pytest
+from kombu.serialization import dumps, loads
+from pydantic import SecretStr, ValidationError
+
+from summary.core import celery_worker
+from summary.core.models import (
+    PushToDocsTranscriptConfig,
+    RecordingMetadata,
+    TranscribeTaskApiRequest,
+    TranscribeTaskJob,
+)
+from summary.core.privacy import dump_for_transport
+from summary.core.shared_models import (
+    SummarizeWebhookSuccessPayload,
+    TranscribeWebhookSuccessPayload,
+    WhisperXResponse,
+)
+
+CANARY = "private-signed-url-canary"
+SIGNED_URL = f"https://storage.example.com/recording.mp4?signature={CANARY}"
+METADATA = {
+    "cloud_storage_url": SIGNED_URL,
+    "started_at": "2026-01-01T10:00:00Z",
+    "ended_at": "2026-01-01T11:00:00Z",
+}
+DOCS_CONFIG = {
+    "user_email": "user@example.com",
+    "title": "Meeting",
+    "download_link": SIGNED_URL,
+    "form_link": SIGNED_URL,
+}
+MODEL_CASES = [
+    (RecordingMetadata, METADATA, "cloud_storage_url"),
+    (
+        TranscribeTaskApiRequest,
+        {"user_sub": "user", "language": "en", "cloud_storage_url": SIGNED_URL},
+        "cloud_storage_url",
+    ),
+    (PushToDocsTranscriptConfig, DOCS_CONFIG, "download_link"),
+    (PushToDocsTranscriptConfig, DOCS_CONFIG, "form_link"),
+    (
+        TranscribeWebhookSuccessPayload,
+        {"job_id": "job", "transcription_data_url": SIGNED_URL},
+        "transcription_data_url",
+    ),
+    (
+        SummarizeWebhookSuccessPayload,
+        {"job_id": "job", "summary_data_url": SIGNED_URL},
+        "summary_data_url",
+    ),
+]
+
+
+@pytest.mark.parametrize(("model", "data", "field"), MODEL_CASES)
+def test_signed_urls_are_masked_by_default(model, data, field):
+    """Reprs, ordinary dumps and JSON output mask all signed URL fields."""
+    value = model.model_validate(data)
+    secret = getattr(value, field)
+    assert isinstance(secret, SecretStr)
+    assert secret.get_secret_value() == SIGNED_URL
+    assert CANARY not in repr(value)
+    assert CANARY not in str(value.model_dump())
+    assert CANARY not in value.model_dump_json()
+    assert dump_for_transport(value)[field] == SIGNED_URL
+    assert getattr(model.model_validate(value.model_dump()), field) == secret
+
+
+@pytest.mark.parametrize(("model", "data", "field"), MODEL_CASES)
+def test_missing_fields_do_not_echo_signed_urls(model, data, field):
+    """Root validation errors omit raw input even before secrets are wrapped."""
+    required_field = (
+        "job_id"
+        if "job_id" in data
+        else next(
+            key for key in data if key not in {field, "download_link", "form_link"}
+        )
+    )
+    invalid_data = {key: value for key, value in data.items() if key != required_field}
+    with pytest.raises(ValidationError) as exc:
+        model.model_validate(invalid_data)
+    assert CANARY not in str(exc.value)
+
+
+@pytest.mark.parametrize(("model", "data", "field"), MODEL_CASES[:4])
+def test_invalid_signed_urls_have_sanitized_validation_messages(model, data, field):
+    """URL validation still rejects invalid URLs without quoting their value."""
+    with pytest.raises(ValidationError) as exc:
+        model.model_validate({**data, field: f"not-a-url?signature={CANARY}"})
+    assert CANARY not in str(exc.value)
+    assert CANARY not in str(
+        exc.value.errors(include_input=False, include_context=False)
+    )
+    assert exc.value.errors()[0]["type"] == "url_parsing"
+    assert exc.value.errors()[0]["input"] == "[redacted]"
+
+
+def test_nested_task_urls_survive_broker_serialization():
+    """Nested secrets round-trip through Celery's JSON serializer as real URLs."""
+    payload = TranscribeTaskJob(
+        user_sub="user",
+        language="en",
+        cloud_storage_url=SIGNED_URL,
+        tenant_id="tenant",
+        received_at="2026-01-01T10:00:00Z",
+        metadata=METADATA,
+        push_to_docs_config=DOCS_CONFIG,
+    )
+    assert CANARY not in repr(payload)
+    assert CANARY not in payload.model_dump_json()
+    content_type, encoding, body = dumps(dump_for_transport(payload), serializer="json")
+    restored = TranscribeTaskJob.model_validate(loads(body, content_type, encoding))
+    assert restored.cloud_storage_url.get_secret_value() == SIGNED_URL
+    assert restored.metadata.cloud_storage_url.get_secret_value() == SIGNED_URL
+    assert restored.push_to_docs_config.download_link.get_secret_value() == SIGNED_URL
+    assert restored.push_to_docs_config.form_link.get_secret_value() == SIGNED_URL
+
+
+def test_transcription_worker_uses_real_urls(monkeypatch):
+    """The worker unwraps input URLs and publishes usable signed result URLs."""
+    transcription = WhisperXResponse(segments=())
+    transcribe = Mock(return_value=transcription)
+    monkeypatch.setattr(celery_worker, "transcribe_audio", transcribe)
+    monkeypatch.setattr(celery_worker, "_should_push_to_docs", Mock(return_value=True))
+    monkeypatch.setattr(
+        celery_worker, "_should_auto_create_summary", Mock(return_value=False)
+    )
+    formatter = Mock(return_value="Formatted transcript")
+    monkeypatch.setattr(celery_worker, "format_transcript", formatter)
+    monkeypatch.setattr(celery_worker, "create_document_in_lasuite_docs", Mock())
+    storage = Mock()
+    storage.get_transcript_signed_url.return_value = SIGNED_URL
+    monkeypatch.setattr(celery_worker, "file_service", storage)
+    monkeypatch.setattr(celery_worker, "metadata_manager", Mock())
+    publish = Mock()
+    monkeypatch.setattr(celery_worker.call_webhook_v2_task, "apply_async", publish)
+    task = celery_worker.process_audio_transcribe_v2_task
+    monkeypatch.setattr(task.request, "id", "job")
+
+    result = task.run(
+        {
+            "user_sub": "user",
+            "language": "en",
+            "cloud_storage_url": SIGNED_URL,
+            "tenant_id": "tenant",
+            "received_at": "2026-01-01T10:00:00Z",
+            "push_to_docs_config": DOCS_CONFIG,
+        }
+    )
+    assert transcribe.call_args.kwargs["cloud_storage_url"] == SIGNED_URL
+    assert formatter.call_args.args[3:] == (SIGNED_URL, SIGNED_URL)
+    assert result["transcription_data_url"] == SIGNED_URL
+    assert publish.call_args.kwargs["args"][0] == result
+
+
+def test_recording_metadata_url_is_unwrapped(monkeypatch):
+    """Speaker assignment fetches metadata using the real URL."""
+    storage = Mock()
+    monkeypatch.setattr(celery_worker, "file_service", storage)
+    resolver = Mock()
+    resolver.return_value.apply_to.return_value = {"segments": []}
+    monkeypatch.setattr(celery_worker, "resolve_speaker_identities", resolver)
+    celery_worker.resolve_speaker_identities_and_apply_to(
+        transcription=WhisperXResponse(segments=()),
+        recording_metadata=RecordingMetadata.model_validate(METADATA),
+        task_id="job",
+    )
+    storage.read_cloud_storage_json.assert_called_once_with(SIGNED_URL)

src/summary/tests/unit/test_privacy.py

@@ -0,0 +1,209 @@
+"""Regression tests for observability privacy boundaries."""
+
+import logging
+from unittest.mock import Mock
+
+import pytest
+import requests
+
+from summary.core import analytics, celery_worker, webhook_service
+from summary.core.llm_service import LLMException, LLMService
+from summary.core.privacy import redact_url
+from summary.core.sentry import before_breadcrumb, before_send
+from summary.core.shared_models import SummarizeWebhookSuccessPayload
+
+CANARY = "sensitive-canary"
+URL = f"https://user:password@example.com/file?signature={CANARY}#fragment"
+
+
+def test_url_redaction():
+    """Credentials and signed query strings are removed."""
+    assert redact_url(URL) == "https://example.com/file?[redacted]"
+
+
+def test_sentry_scrubs_events_and_breadcrumbs():
+    """Untrusted exception messages and breadcrumb data never leave in events."""
+    event = {
+        "request": {"url": URL, "data": CANARY, "query_string": CANARY},
+        "exception": {
+            "values": [
+                {
+                    "type": "ValueError",
+                    "value": CANARY,
+                    "stacktrace": {"frames": [{"vars": {"job_id": "job-1"}}]},
+                }
+            ]
+        },
+        "extra": {"url": URL},
+    }
+    result = before_send(event, {})
+    assert CANARY not in str(result)
+    assert result["exception"]["values"][0]["type"] == "ValueError"
+    assert result["exception"]["values"][0]["stacktrace"]["frames"][0]["vars"] == {
+        "job_id": "job-1"
+    }
+    assert before_breadcrumb({"message": CANARY, "data": {"url": URL}}, {}) == {}
+
+
+@pytest.mark.parametrize("scheme", ["http", "HTTP", "hTtP", "https", "HTTPS", "hTtPs"])
+def test_sentry_url_scrubbing_is_case_insensitive(scheme):
+    """Scheme casing cannot bypass redaction of URLs nested in event strings."""
+    event = {
+        "extra": {
+            "nested": [
+                {
+                    "detail": (
+                        f"prefix <{scheme}://user:password@example.com/Path"
+                        f"?signature={CANARY}#fragment> suffix"
+                    )
+                }
+            ]
+        }
+    }
+    result = before_send(event, {})
+    assert result["extra"]["nested"][0]["detail"] == (
+        f"prefix <{scheme.lower()}://example.com/Path?[redacted]> suffix"
+    )
+    assert CANARY not in str(result)
+
+
+def test_llm_exception_does_not_include_upstream_message(caplog):
+    """LLM failures retain their category without logging upstream text."""
+    observability = Mock(is_enabled=False)
+    observability.get_openai_client.return_value.chat.completions.create.side_effect = (
+        ValueError(CANARY)
+    )
+    with caplog.at_level(logging.DEBUG), pytest.raises(LLMException) as exc:
+        LLMService(observability).call("system", CANARY, "test")
+    assert CANARY not in str(exc.value)
+    assert CANARY not in caplog.text
+    assert exc.value.__suppress_context__
+
+
+@pytest.mark.parametrize("log_level", [logging.DEBUG, logging.INFO])
+def test_webhook_content_is_only_logged_at_debug(
+    monkeypatch, caplog, settings_override, log_level
+):
+    """DEBUG retains response content; INFO and above only contain metadata."""
+    monkeypatch.setattr(webhook_service, "settings", settings_override)
+    response = Mock(status_code=200)
+    response.json.return_value = {"id": "document-123", "content": CANARY}
+    monkeypatch.setattr(webhook_service.requests, "post", Mock(return_value=response))
+    payload = SummarizeWebhookSuccessPayload(job_id="job", summary_data_url=URL)
+    with caplog.at_level(log_level):
+        webhook_service.call_webhook_v2(tenant_id="test-tenant", payload=payload)
+    assert (
+        webhook_service.requests.post.call_args.kwargs["json"]["summary_data_url"]
+        == URL
+    )
+    assert all(
+        CANARY not in record.getMessage()
+        for record in caplog.records
+        if record.levelno >= logging.INFO
+    )
+    if log_level == logging.DEBUG:
+        assert any(
+            record.levelno == logging.DEBUG and CANARY in record.getMessage()
+            for record in caplog.records
+        )
+    else:
+        assert CANARY not in caplog.text
+    assert "Document document-123 submitted (HTTP 200)" in caplog.text
+
+
+def test_webhook_failure_keeps_retry_category(monkeypatch, settings_override):
+    """Webhook exceptions remain retryable without upstream text."""
+    monkeypatch.setattr(webhook_service, "settings", settings_override)
+    monkeypatch.setattr(
+        webhook_service.requests,
+        "post",
+        Mock(side_effect=requests.RequestException(URL)),
+    )
+    payload = SummarizeWebhookSuccessPayload(job_id="job", summary_data_url=URL)
+    with pytest.raises(requests.RequestException) as exc:
+        webhook_service.call_webhook_v2(tenant_id="test-tenant", payload=payload)
+    assert CANARY not in str(exc.value)
+    assert exc.value.__suppress_context__
+
+
+def test_analytics_failure_does_not_chain_payload(monkeypatch):
+    """Analytics errors do not expose arbitrary client exception text."""
+    client = analytics.Analytics()
+    client._client = Mock()
+    client._client.capture.side_effect = ValueError(CANARY)
+    with pytest.raises(analytics.AnalyticsException) as exc:
+        client.capture("event", "user")
+    assert CANARY not in str(exc.value)
+    assert exc.value.__suppress_context__
+
+
+@pytest.mark.parametrize("log_level", [logging.DEBUG, logging.INFO, logging.ERROR])
+def test_celery_log_filter_respects_log_level(log_level):
+    """Celery trace values remain available in DEBUG only."""
+    record = logging.LogRecord(
+        "celery.app.trace", log_level, "", 0, "%s", (CANARY,), None
+    )
+    record.data = {"name": "task", "id": "job", "return_value": URL}
+    assert celery_worker.CeleryLogPrivacyFilter().filter(record)
+    if log_level == logging.DEBUG:
+        assert CANARY in record.getMessage()
+        assert record.data["return_value"] == URL
+    else:
+        assert CANARY not in record.getMessage()
+        assert record.data == {}
+
+
+def test_task_arguments_have_private_display_representations(monkeypatch):
+    """The actual task arguments are delivered but their event repr is hidden."""
+    publish = Mock()
+    monkeypatch.setattr(celery_worker.Task, "apply_async", publish)
+    task = celery_worker.PrivateTask()
+    task.apply_async(args=[CANARY], kwargs={"url": URL})
+    assert publish.call_args.kwargs["args"] == [CANARY]
+    assert publish.call_args.kwargs["argsrepr"] == "[Filtered]"
+    assert publish.call_args.kwargs["kwargsrepr"] == "[Filtered]"
+
+
+def test_analytics_source_url_removes_signature():
+    """Task analytics keeps the source path without credentials or signature."""
+    from datetime import datetime, timezone  # noqa: PLC0415
+
+    from summary.core.models import TranscribeTaskJob  # noqa: PLC0415
+
+    manager = analytics.MetadataManager()
+    manager._is_disabled = False
+    manager.has_task_id = Mock(return_value=False)
+    manager._save_metadata = Mock()
+    payload = TranscribeTaskJob(
+        user_sub="user",
+        tenant_id="tenant",
+        cloud_storage_url=URL,
+        language="en",
+        received_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
+    )
+    manager.create("job", payload)
+    metadata = manager._save_metadata.call_args.args[1]
+    assert metadata["source_url"] == "https://example.com/file?[redacted]"
+    assert CANARY not in str(metadata)
+
+
+def test_invalid_llm_json_is_sanitized():
+    """Invalid structured LLM output retains a retryable LLM category."""
+    with pytest.raises(LLMException) as exc:
+        celery_worker.parse_llm_json(CANARY)
+    assert CANARY not in str(exc.value)
+    assert exc.value.__suppress_context__
+
+
+@pytest.mark.parametrize(
+    ("url", "expected"),
+    [
+        (None, "[redacted]"),
+        ("https://[invalid?secret=value", "[redacted]"),
+        ("https://example.com/path#fragment", "https://example.com/path"),
+        ("https://example.com/path?", "https://example.com/path?[redacted]"),
+    ],
+)
+def test_redact_url_edge_cases(url, expected):
+    """Malformed and non-string URLs do not expose credentials or query data."""
+    assert redact_url(url) == expected

src/summary/tests/unit/test_sentry.py

@@ -1,27 +1,32 @@
 """Tests for the Sentry configuration.
 
 Each test raises an error from code handling meeting content and inspects the
-event Sentry would send: the content must be redacted, while harmless local
-variables are kept for debugging.
+event Sentry would send: sensitive content is redacted and useful locals remain.
 """
 
 import json
+import logging
 from collections.abc import Callable, Iterator
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from threading import Thread
 from unittest.mock import Mock
 
 import httpx
 import openai
 import pytest
+import requests
 import sentry_sdk
 from botocore.exceptions import ClientError
 from botocore.stub import Stubber
+from sentry_sdk.integrations.logging import LoggingIntegration
+from sentry_sdk.integrations.stdlib import StdlibIntegration
 from sentry_sdk.transport import Transport
 
-from summary.core import file_service
+from summary.core import file_service, webhook_service
 from summary.core import sentry as sentry_module
-from summary.core.file_service import FileService
+from summary.core.file_service import FileService, FileServiceException
 from summary.core.llm_service import LLMException, LLMService
-from summary.core.shared_models import WhisperXResponse
+from summary.core.shared_models import SummarizeWebhookSuccessPayload, WhisperXResponse
 
 CANARY = "CANARY-MEETING-CONTENT"
 
@@ -51,8 +56,14 @@ def sentry_events() -> Iterator[SentryEvents]:
         transport=transport,
         send_default_pii=False,
         include_local_variables=True,
+        before_send=sentry_module.before_send,
+        before_breadcrumb=sentry_module.before_breadcrumb,
         event_scrubber=sentry_module.build_event_scrubber(),
         default_integrations=False,
+        integrations=[
+            LoggingIntegration(level=logging.DEBUG, event_level=None),
+            StdlibIntegration(),
+        ],
     )
 
     def flush() -> list[str]:
@@ -63,6 +74,125 @@ def sentry_events() -> Iterator[SentryEvents]:
     sentry_sdk.init()  # Disable Sentry for the following tests
 
 
+@pytest.fixture
+def missing_recording_url() -> Iterator[str]:
+    """Serve a deterministic 404 through the real instrumented HTTP client."""
+
+    class MissingRecordingHandler(BaseHTTPRequestHandler):
+        def do_GET(self):
+            self.send_response(404)
+            self.end_headers()
+
+        def log_message(self, format, *args):
+            pass
+
+    with ThreadingHTTPServer(("127.0.0.1", 0), MissingRecordingHandler) as server:
+        thread = Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            yield (
+                f"http://127.0.0.1:{server.server_port}/recordings/recording-123.mp4"
+                f"?X-Amz-Credential={CANARY}-credential"
+                f"&X-Amz-Signature={CANARY}-signature"
+            )
+        finally:
+            server.shutdown()
+            thread.join()
+
+
+def test_sentry_download_404_excludes_signed_url(
+    sentry_events: SentryEvents, missing_recording_url: str, caplog
+) -> None:
+    """A failed download excludes signed URLs from logs, locals and breadcrumbs."""
+    with caplog.at_level(logging.INFO):
+        try:
+            FileService()._download_from_cloud_storage_url(missing_recording_url)
+        except FileServiceException as exc:
+            assert exc.__cause__ is None
+            assert exc.__suppress_context__
+            sentry_sdk.capture_exception()
+        else:
+            pytest.fail("the download should have failed")
+
+    [event] = sentry_events()
+    assert CANARY not in event
+    assert CANARY not in caplog.text
+    assert "/recordings/recording-123.mp4?[redacted]" in caplog.text
+    parsed = json.loads(event)
+    assert [item["type"] for item in parsed["exception"]["values"]] == [
+        "FileServiceException"
+    ]
+    download_vars = next(
+        frame_vars for frame_vars in _local_vars(event) if "extension" in frame_vars
+    )
+    assert download_vars["extension"] == "'.mp4'"
+    assert download_vars["cloud_storage_url"] == "[Filtered]"
+    breadcrumbs = parsed["breadcrumbs"]["values"]
+    assert any(crumb.get("type") == "http" for crumb in breadcrumbs)
+    assert any(
+        crumb.get("category") == "summary.core.file_service" for crumb in breadcrumbs
+    )
+    assert all("message" not in crumb and "data" not in crumb for crumb in breadcrumbs)
+
+
+def test_sentry_chained_http_error_excludes_signed_url(
+    sentry_events: SentryEvents, missing_recording_url: str
+) -> None:
+    """Sentry also filters signed URLs in explicitly chained HTTP errors."""
+    try:
+        try:
+            response = requests.get(missing_recording_url, timeout=5)
+            response.raise_for_status()
+        except requests.HTTPError as exc:
+            raise FileServiceException("Download failed") from exc
+    except FileServiceException:
+        sentry_sdk.capture_exception()
+
+    [event] = sentry_events()
+    assert CANARY not in event
+    exceptions = json.loads(event)["exception"]["values"]
+    assert [item["type"] for item in exceptions] == [
+        "HTTPError",
+        "FileServiceException",
+    ]
+    assert all(item["value"] == "[Filtered]" for item in exceptions)
+
+
+def test_sentry_excludes_webhook_debug_content(
+    sentry_events: SentryEvents, monkeypatch, settings_override, caplog
+) -> None:
+    """Raw DEBUG responses stay in application logs and out of Sentry events."""
+    monkeypatch.setattr(webhook_service, "settings", settings_override)
+    signed_url = f"https://storage.example.com/result?signature={CANARY}"
+    response = Mock(status_code=200)
+    response.json.return_value = {
+        "id": "document-123",
+        "content": CANARY,
+        "url": signed_url,
+    }
+    monkeypatch.setattr(webhook_service.requests, "post", Mock(return_value=response))
+    with caplog.at_level(logging.DEBUG):
+        webhook_service.call_webhook_v2(
+            tenant_id="test-tenant",
+            payload=SummarizeWebhookSuccessPayload(
+                job_id="job", summary_data_url=signed_url
+            ),
+        )
+        try:
+            raise RuntimeError("A later failure")
+        except RuntimeError:
+            sentry_sdk.capture_exception()
+
+    assert CANARY in caplog.text
+    [event] = sentry_events()
+    assert CANARY not in event
+    assert any(
+        crumb.get("category") == "summary.core.webhook_service"
+        and crumb.get("level") == "debug"
+        for crumb in json.loads(event)["breadcrumbs"]["values"]
+    )
+
+
 def _transcript() -> WhisperXResponse:
     return WhisperXResponse.model_validate(
         {
@@ -121,7 +251,7 @@ def s3_stubber(monkeypatch: pytest.MonkeyPatch) -> Iterator[Stubber]:
 def test_sentry_store_transcript_failure_redacts_transcript(
     sentry_events: SentryEvents, s3_stubber: Stubber
 ) -> None:
-    """A failed S3 upload does not send the transcript, but keeps the job id."""
+    """A failed S3 upload redacts the transcript while retaining useful locals."""
     s3_stubber.add_client_error("put_object", service_error_code="InvalidDigest")
 
     try:

src/summary/tests/unit/test_validation_error_formatting.py

@@ -0,0 +1,26 @@
+"""Formatted validation errors hide input values, including after pickling."""
+
+import pickle
+import traceback
+
+from pydantic import ValidationError
+
+from summary.core.exceptions import CleanedValidationError
+
+CANARY = "private-validation-canary"
+
+
+def test_formatted_errors_hide_ordinary_input_after_pickle_roundtrip():
+    """Formatting hides raw input even when diagnostics intentionally retain it."""
+    original = ValidationError.from_exception_data(
+        "TestModel",
+        [{"type": "int_parsing", "loc": ("field",), "input": CANARY}],
+    )
+    cleaned = CleanedValidationError(original)
+    restored = pickle.loads(pickle.dumps(cleaned))  # noqa: S301 -- own test exception
+    for error in (cleaned, restored):
+        assert error.errors()[0]["input"] == CANARY
+        assert CANARY not in str(error)
+        assert CANARY not in repr(error)
+        assert CANARY not in "".join(traceback.format_exception(error))
+        assert error.errors()[0]["type"] == "int_parsing"

src/summary/tests/unit/test_validation_errors.py

@@ -0,0 +1,223 @@
+"""Validation cleanup preserves diagnostics without retaining SecretStr values."""
+
+import asyncio
+import json
+import pickle
+import traceback
+from unittest.mock import MagicMock, Mock
+
+import pytest
+from celery.exceptions import Retry
+from celery.utils.serialization import get_pickleable_exception
+from fastapi import Request
+from fastapi.exceptions import RequestValidationError
+from pydantic import SecretStr, ValidationError
+from pydantic_core import PydanticCustomError
+
+from summary.core import celery_worker
+from summary.core.exceptions import CleanedValidationError, clean_validation_errors
+from summary.main import validation_error_handler
+
+CANARY = "private-validation-canary"
+
+
+@pytest.mark.parametrize("custom", [False, True])
+def test_cleaned_validation_error_preserves_details_and_pickles(custom):
+    """Builtin and custom validation errors retain their type, location and message."""
+    error_type = (
+        PydanticCustomError(
+            "custom_validation", "Invalid value: {reason}", {"reason": "test"}
+        )
+        if custom
+        else "int_type"
+    )
+    original = ValidationError.from_exception_data(
+        "TestModel",
+        [
+            {"type": error_type, "loc": ("secret",), "input": SecretStr(CANARY)},
+            {"type": "int_parsing", "loc": ("ordinary",), "input": "not-an-int"},
+        ],
+    )
+    cleaned = CleanedValidationError(original)
+    original_details = original.errors()
+    details = cleaned.errors()
+    assert isinstance(cleaned, ValidationError)
+    assert cleaned.title == original.title
+    assert cleaned.error_count() == original.error_count()
+    assert details[0] == {**original_details[0], "input": "[redacted]"}
+    assert details[1] == original_details[1]
+    assert isinstance(original.errors()[0]["input"], SecretStr)
+    assert CANARY not in str(cleaned)
+    assert CANARY not in cleaned.json()
+    serialized = pickle.dumps(cleaned)
+    assert CANARY.encode() not in serialized
+    restored = pickle.loads(serialized)  # noqa: S301 -- own in-memory test exception
+    assert isinstance(restored, CleanedValidationError)
+    assert restored.errors() == details
+    assert isinstance(get_pickleable_exception(cleaned), CleanedValidationError)
+
+
+def test_cleanup_masks_nested_secrets_and_preserves_plain_input():
+    """Missing-field inputs keep ordinary values while masking nested secrets."""
+    original = ValidationError.from_exception_data(
+        "NestedModel",
+        [
+            {
+                "type": "missing",
+                "loc": ("required",),
+                "input": {
+                    "ordinary": "keep me",
+                    "nested": [SecretStr(CANARY), {"url": SecretStr(CANARY)}],
+                    "tuple": (SecretStr(CANARY), 42),
+                },
+            }
+        ],
+    )
+    cleaned = CleanedValidationError(original)
+    assert cleaned.errors()[0]["input"] == {
+        "ordinary": "keep me",
+        "nested": ["[redacted]", {"url": "[redacted]"}],
+        "tuple": ("[redacted]", 42),
+    }
+    assert CANARY not in cleaned.json()
+    assert CANARY.encode() not in pickle.dumps(cleaned)
+
+
+def test_cleanup_does_not_retain_validator_exception_objects():
+    """Validator context keeps its readable message without secret-bearing args."""
+    original = ValidationError.from_exception_data(
+        "TestModel",
+        [
+            {
+                "type": "value_error",
+                "loc": (),
+                "input": SecretStr(CANARY),
+                "ctx": {"error": ValueError(SecretStr(CANARY))},
+            }
+        ],
+    )
+    cleaned = CleanedValidationError(original)
+    assert cleaned.errors()[0]["ctx"] == {"error": "**********"}
+    assert CANARY.encode() not in pickle.dumps(cleaned)
+
+
+def test_clean_validation_errors_keeps_all_other_fields():
+    """The API cleanup preserves diagnostics and nonsecret input."""
+    details = [
+        {
+            "type": "test",
+            "loc": ("body", "field"),
+            "msg": "Invalid value",
+            "url": "https://example.com/errors/test",
+            "ctx": {"limit": 3},
+            "input": "ordinary input",
+        }
+    ]
+    assert clean_validation_errors(details) == details
+    assert (
+        json.loads(json.dumps(clean_validation_errors(details)))[0]["input"]
+        == "ordinary input"
+    )
+
+
+@pytest.mark.parametrize(
+    "handler_name",
+    [
+        "call_webhook_v2_task",
+        "process_audio_transcribe_v2_task",
+        "summarize_v2_task",
+        "task_started_transcript",
+        "task_started_summary",
+    ],
+)
+@pytest.mark.parametrize("secret_wrapped", [False, True])
+def test_task_validation_uses_cleaned_errors(handler_name, secret_wrapped):
+    """Task entry points and prerun signals consistently suppress the original error."""
+    handler = getattr(celery_worker, handler_name)
+    signed_url = f"https://storage.example.com/file?signature={CANARY}"
+    value = SecretStr(signed_url) if secret_wrapped else signed_url
+    payload = {"cloud_storage_url": value, "content": value}
+    with pytest.raises(CleanedValidationError) as exc:
+        if handler_name.startswith("task_started"):
+            handler(task_id="job", args=[payload])
+        elif handler_name == "call_webhook_v2_task":
+            handler.run(
+                payload={
+                    "type": "summary",
+                    "status": "success",
+                    "summary_data_url": value,
+                },
+                tenant_id="tenant",
+            )
+        elif handler_name == "summarize_v2_task":
+            # Exercise validation without triggering the summary task's retries.
+            handler._orig_run(payload=payload)
+        else:
+            handler.run(payload=payload)
+    assert exc.value.__suppress_context__
+    assert CANARY not in "".join(traceback.format_exception(exc.value))
+    assert CANARY not in exc.value.json()
+    assert CANARY.encode() not in pickle.dumps(exc.value)
+
+
+def test_upstream_transcription_validation_hides_raw_input(monkeypatch):
+    """Invalid ASR responses do not remain in structured validation errors."""
+    storage = MagicMock()
+    storage.prepare_audio_file.return_value.__enter__.return_value = (
+        Mock(),
+        {"duration": 1},
+    )
+    monkeypatch.setattr(celery_worker, "file_service", storage)
+    monkeypatch.setattr(celery_worker, "metadata_manager", Mock())
+    response = Mock(status_code=200)
+    response.json.return_value = {"text": {"content": CANARY}}
+    monkeypatch.setattr(celery_worker.requests, "post", Mock(return_value=response))
+    with pytest.raises(CleanedValidationError) as exc:
+        celery_worker.transcribe_audio(
+            task_id="job", language="en", cloud_storage_url="https://example.com/audio"
+        )
+    assert exc.value.errors()[0]["input"] == "[redacted]"
+    assert CANARY not in exc.value.json()
+    assert CANARY.encode() not in pickle.dumps(exc.value)
+
+
+def test_api_error_handler_uses_secret_aware_cleanup():
+    """API errors mask typed secrets while retaining other values and context."""
+    errors = [
+        {
+            "type": "test",
+            "loc": ("body", "field"),
+            "msg": "Invalid value",
+            "input": {"url": SecretStr(CANARY), "ordinary": "keep me"},
+            "ctx": {"limit": 3},
+            "url": "https://example.com/errors/test",
+        },
+    ]
+    response = asyncio.run(
+        validation_error_handler(
+            Request({"type": "http"}), RequestValidationError(errors)
+        )
+    )
+    assert response.status_code == 422
+    assert CANARY.encode() not in response.body
+    assert json.loads(response.body)["detail"] == [
+        {
+            "type": "test",
+            "loc": ["body", "field"],
+            "msg": "Invalid value",
+            "input": {"url": "[redacted]", "ordinary": "keep me"},
+            "ctx": {"limit": 3},
+            "url": "https://example.com/errors/test",
+        },
+    ]
+
+
+def test_summary_task_retries_with_the_cleaned_exception(monkeypatch):
+    """The summary task still retries validation failures using the safe exception."""
+    retry = Mock(side_effect=Retry())
+    monkeypatch.setattr(celery_worker.summarize_v2_task, "retry", retry)
+    with pytest.raises(Retry):
+        celery_worker.summarize_v2_task.run(payload={"received_at": SecretStr(CANARY)})
+    error = retry.call_args.kwargs["exc"]
+    assert isinstance(error, CleanedValidationError)
+    assert CANARY.encode() not in pickle.dumps(error)