docs: define CalDAV secret retirement

README.md

@@ -48,6 +48,14 @@ environment and external-provider references are rejected. Trusted deployment
 code may resolve an `env:NAME` reference only through the separate deployment
 configuration helper.
 
+Deleting a source or its calendar immediately scrubs Calendar-owned ciphertext
+and provider references and emits non-secret audit evidence. If an external
+secret provider cannot confirm deletion, the operation fails closed without
+retiring the source or cancelling its queued work; retry is idempotent after a
+database rollback. Destructive module retirement first deletes and audits all
+active and legacy retained provider secrets and stops before table removal on
+provider failure.
+
 Inbound sync uses CalDAV `calendar-query` for full sync and `sync-collection`
 when a sync token exists. It imports all VEVENT components in a resource and
 soft-deletes local events when remote resources disappear. Two-way sources also

docs/CALENDAR_INTEGRATION_CONCEPT.md

@@ -69,6 +69,12 @@ bindings or unresolved delivery. Credential rotation does not discard committed
 desired state. Public event mutation cannot set sync-owned source hrefs, kinds,
 or ETags. Deleting a synchronized collection or retiring its source is a local
 unlink: it never deletes the remote collection or its remaining remote events.
+The unlink immediately scrubs Calendar-owned credential ciphertext and external
+provider references and audits the deletion. External provider failure blocks
+retirement before queued work is changed; a later retry tolerates a provider
+secret already removed by an earlier attempt whose database transaction rolled
+back. Provider errors and audit details never contain credential values or
+secret references.
 The current singular ownership model permits one active sync source per
 calendar; multi-source fan-in will require per-event source routing. Celery beat
 triggers recovery every minute, while root-transaction after-commit dispatch

src/govoplan_calendar/backend/manifest.py

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
+from dataclasses import replace
 from pathlib import Path
 
+from sqlalchemy import inspect
+
 from govoplan_calendar.backend.db import models as calendar_models  # noqa: F401 - populate Calendar ORM metadata
 from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
 from govoplan_core.core.calendar import (
@@ -24,6 +27,41 @@ from govoplan_core.core.modules import (
 from govoplan_core.db.base import Base
 
 
+_calendar_table_retirement_provider = drop_table_retirement_provider(
+    calendar_models.CalendarCollection,
+    calendar_models.CalendarEvent,
+    calendar_models.CalendarOutboxOperation,
+    calendar_models.CalendarSyncCredential,
+    calendar_models.CalendarSyncSource,
+    label="Calendar",
+)
+
+
+def _calendar_retirement_provider(session: object | None, module_id: str):
+    plan = _calendar_table_retirement_provider(session, module_id)
+    base_executor = plan.destroy_data_executor
+    if base_executor is None:
+        return plan
+
+    def executor(execute_session: object, execute_module_id: str) -> None:
+        if not hasattr(execute_session, "get_bind") or not hasattr(execute_session, "query"):
+            raise RuntimeError("No database session is available for Calendar credential retirement.")
+        if inspect(execute_session.get_bind()).has_table(calendar_models.CalendarSyncCredential.__tablename__):
+            from govoplan_calendar.backend.service import delete_calendar_credentials_for_retirement
+
+            delete_calendar_credentials_for_retirement(execute_session)
+        base_executor(execute_session, execute_module_id)
+
+    return replace(
+        plan,
+        destroy_data_warnings=(
+            *plan.destroy_data_warnings,
+            "Calendar-owned credentials are deleted immediately before tables are dropped; retirement fails if an external secret provider is unavailable.",
+        ),
+        destroy_data_executor=executor,
+    )
+
+
 def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
     module_id, resource, action = scope.split(":", 2)
     return PermissionDefinition(
@@ -163,14 +201,7 @@ manifest = ModuleManifest(
         metadata=Base.metadata,
         script_location=str(Path(__file__).with_name("migrations") / "versions"),
         retirement_supported=True,
-        retirement_provider=drop_table_retirement_provider(
-            calendar_models.CalendarCollection,
-            calendar_models.CalendarEvent,
-            calendar_models.CalendarOutboxOperation,
-            calendar_models.CalendarSyncCredential,
-            calendar_models.CalendarSyncSource,
-            label="Calendar",
-        ),
+        retirement_provider=_calendar_retirement_provider,
         retirement_notes="Destructive retirement drops calendar-owned database tables after the installer captures a database snapshot.",
     ),
     uninstall_guard_providers=(

src/govoplan_calendar/backend/router.py

@@ -298,7 +298,14 @@ def api_delete_calendar(
 ):
     _require_scope(principal, "calendar:calendar:admin")
     try:
-        delete_calendar(session, tenant_id=principal.tenant_id, calendar_id=calendar_id, payload=payload)
+        delete_calendar(
+            session,
+            tenant_id=principal.tenant_id,
+            calendar_id=calendar_id,
+            payload=payload,
+            user_id=principal.user.id,
+            api_key_id=principal.api_key_id,
+        )
         session.commit()
         return Response(status_code=status.HTTP_204_NO_CONTENT)
     except CalendarError as exc:
@@ -344,7 +351,14 @@ def api_update_sync_source(
 ):
     _require_scope(principal, "calendar:calendar:admin")
     try:
-        source = update_sync_source(session, tenant_id=principal.tenant_id, source_id=source_id, payload=payload)
+        source = update_sync_source(
+            session,
+            tenant_id=principal.tenant_id,
+            source_id=source_id,
+            payload=payload,
+            user_id=principal.user.id,
+            api_key_id=principal.api_key_id,
+        )
         session.commit()
         session.refresh(source)
         return _sync_source_response(source)
@@ -361,7 +375,13 @@ def api_delete_sync_source(
 ):
     _require_scope(principal, "calendar:calendar:admin")
     try:
-        delete_sync_source(session, tenant_id=principal.tenant_id, source_id=source_id)
+        delete_sync_source(
+            session,
+            tenant_id=principal.tenant_id,
+            source_id=source_id,
+            user_id=principal.user.id,
+            api_key_id=principal.api_key_id,
+        )
         session.commit()
         return Response(status_code=status.HTTP_204_NO_CONTENT)
     except CalendarError as exc:
@@ -479,7 +499,14 @@ def api_update_caldav_source(
 ):
     _require_scope(principal, "calendar:calendar:admin")
     try:
-        source = update_caldav_source(session, tenant_id=principal.tenant_id, source_id=source_id, payload=payload)
+        source = update_caldav_source(
+            session,
+            tenant_id=principal.tenant_id,
+            source_id=source_id,
+            payload=payload,
+            user_id=principal.user.id,
+            api_key_id=principal.api_key_id,
+        )
         session.commit()
         session.refresh(source)
         return _caldav_source_response(source)
@@ -496,7 +523,13 @@ def api_delete_caldav_source(
 ):
     _require_scope(principal, "calendar:calendar:admin")
     try:
-        delete_caldav_source(session, tenant_id=principal.tenant_id, source_id=source_id)
+        delete_caldav_source(
+            session,
+            tenant_id=principal.tenant_id,
+            source_id=source_id,
+            user_id=principal.user.id,
+            api_key_id=principal.api_key_id,
+        )
         session.commit()
         return Response(status_code=status.HTTP_204_NO_CONTENT)
     except CalendarError as exc:

src/govoplan_calendar/backend/service.py

@@ -23,6 +23,7 @@ from govoplan_core.security.outbound_http import (
     build_outbound_http_opener,
     validate_outbound_http_url,
 )
+from govoplan_core.audit.logging import audit_event
 
 from govoplan_calendar.backend.caldav import CalDAVClient, CalDAVError, CalDAVNotFound, CalDAVReportResult, CalDAVSyncUnsupported, ensure_collection_url
 from govoplan_calendar.backend.db.models import CalendarCollection, CalendarEvent, CalendarSyncCredential, CalendarSyncSource
@@ -710,7 +711,15 @@ def _move_calendar_before_delete(
     return calendar
 
 
-def delete_calendar(session: Session, *, tenant_id: str, calendar_id: str, payload: CalendarCollectionDeleteRequest | None = None) -> None:
+def delete_calendar(
+    session: Session,
+    *,
+    tenant_id: str,
+    calendar_id: str,
+    payload: CalendarCollectionDeleteRequest | None = None,
+    user_id: str | None = None,
+    api_key_id: str | None = None,
+) -> None:
     payload = payload or CalendarCollectionDeleteRequest()
     calendar = get_calendar(session, tenant_id=tenant_id, calendar_id=calendar_id)
     deleted_at = utcnow()
@@ -729,7 +738,15 @@ def delete_calendar(session: Session, *, tenant_id: str, calendar_id: str, paylo
     if calendar.is_default:
         calendar.is_default = False
     calendar.deleted_at = deleted_at
-    retire_sync_sources_for_calendar(session, tenant_id=tenant_id, calendar_id=calendar.id, deleted_at=deleted_at)
+    retire_sync_sources_for_calendar(
+        session,
+        tenant_id=tenant_id,
+        calendar_id=calendar.id,
+        deleted_at=deleted_at,
+        deletion_reason="calendar_deleted",
+        user_id=user_id,
+        api_key_id=api_key_id,
+    )
     for event in calendar.events:
         if payload.event_action == "delete" and event.deleted_at is None:
             event.deleted_at = deleted_at
@@ -1135,6 +1152,9 @@ def _update_sync_source_credential_and_schedule(
     tenant_id: str,
     source: CalendarSyncSource,
     payload: CalendarSyncSourceUpdateRequest,
+    previous_auth_type: str,
+    user_id: str | None,
+    api_key_id: str | None,
 ) -> None:
     credential_value = caldav_secret_from_payload(
         auth_type=source.auth_type,
@@ -1145,15 +1165,39 @@ def _update_sync_source_credential_and_schedule(
         source.credential_ref = store_caldav_credential(
             session,
             tenant_id=tenant_id,
-            user_id=None,
+            user_id=user_id,
             source=source,
             secret=credential_value,
+            api_key_id=api_key_id,
         )
+    elif payload.auth_type is not None and source.auth_type != previous_auth_type:
+        delete_caldav_credential(
+            session,
+            tenant_id=tenant_id,
+            source_id=source.id,
+            credential_ref=source.credential_ref,
+            deletion_reason=(
+                "authentication_disabled"
+                if source.auth_type == "none"
+                else "authentication_type_changed"
+            ),
+            user_id=user_id,
+            api_key_id=api_key_id,
+        )
+        source.credential_ref = None
     if payload.sync_enabled is not None or payload.sync_interval_seconds is not None:
         source.next_sync_at = utcnow() if source.sync_enabled else None
 
 
-def update_sync_source(session: Session, *, tenant_id: str, source_id: str, payload: CalendarSyncSourceUpdateRequest) -> CalendarSyncSource:
+def update_sync_source(
+    session: Session,
+    *,
+    tenant_id: str,
+    source_id: str,
+    payload: CalendarSyncSourceUpdateRequest,
+    user_id: str | None = None,
+    api_key_id: str | None = None,
+) -> CalendarSyncSource:
     source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id)
     source = _lock_sync_source_for_update(
         session,
@@ -1164,6 +1208,7 @@ def update_sync_source(session: Session, *, tenant_id: str, source_id: str, payl
         raise CalendarError(
             "Caller-supplied credential references are not accepted; provide a replacement password or bearer token"
         )
+    previous_auth_type = source.auth_type
     (
         normalized_collection_url,
         materially_reconfigured,
@@ -1194,30 +1239,87 @@ def update_sync_source(session: Session, *, tenant_id: str, source_id: str, payl
         tenant_id=tenant_id,
         source=source,
         payload=payload,
+        previous_auth_type=previous_auth_type,
+        user_id=user_id,
+        api_key_id=api_key_id,
     )
     mark_calendar_sync_source(calendar, source)
     session.flush()
     return source
 
 
-def update_caldav_source(session: Session, *, tenant_id: str, source_id: str, payload: CalendarCalDavSourceUpdateRequest) -> CalendarSyncSource:
+def update_caldav_source(
+    session: Session,
+    *,
+    tenant_id: str,
+    source_id: str,
+    payload: CalendarCalDavSourceUpdateRequest,
+    user_id: str | None = None,
+    api_key_id: str | None = None,
+) -> CalendarSyncSource:
     source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id, source_kind="caldav")
-    return update_sync_source(session, tenant_id=tenant_id, source_id=source.id, payload=payload)
+    return update_sync_source(
+        session,
+        tenant_id=tenant_id,
+        source_id=source.id,
+        payload=payload,
+        user_id=user_id,
+        api_key_id=api_key_id,
+    )
 
 
-def delete_sync_source(session: Session, *, tenant_id: str, source_id: str) -> None:
+def delete_sync_source(
+    session: Session,
+    *,
+    tenant_id: str,
+    source_id: str,
+    user_id: str | None = None,
+    api_key_id: str | None = None,
+) -> None:
     source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id)
-    retire_sync_source(session, tenant_id=tenant_id, source=source, deleted_at=utcnow())
+    retire_sync_source(
+        session,
+        tenant_id=tenant_id,
+        source=source,
+        deleted_at=utcnow(),
+        deletion_reason="sync_source_deleted",
+        user_id=user_id,
+        api_key_id=api_key_id,
+    )
     session.flush()
 
 
-def delete_caldav_source(session: Session, *, tenant_id: str, source_id: str) -> None:
+def delete_caldav_source(
+    session: Session,
+    *,
+    tenant_id: str,
+    source_id: str,
+    user_id: str | None = None,
+    api_key_id: str | None = None,
+) -> None:
     source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id, source_kind="caldav")
-    retire_sync_source(session, tenant_id=tenant_id, source=source, deleted_at=utcnow())
+    retire_sync_source(
+        session,
+        tenant_id=tenant_id,
+        source=source,
+        deleted_at=utcnow(),
+        deletion_reason="sync_source_deleted",
+        user_id=user_id,
+        api_key_id=api_key_id,
+    )
     session.flush()
 
 
-def retire_sync_sources_for_calendar(session: Session, *, tenant_id: str, calendar_id: str, deleted_at: datetime) -> None:
+def retire_sync_sources_for_calendar(
+    session: Session,
+    *,
+    tenant_id: str,
+    calendar_id: str,
+    deleted_at: datetime,
+    deletion_reason: str = "calendar_deleted",
+    user_id: str | None = None,
+    api_key_id: str | None = None,
+) -> None:
     if not hasattr(session, "query"):
         return
     sources = (
@@ -1230,7 +1332,15 @@ def retire_sync_sources_for_calendar(session: Session, *, tenant_id: str, calend
         .all()
     )
     for source in sources:
-        retire_sync_source(session, tenant_id=tenant_id, source=source, deleted_at=deleted_at)
+        retire_sync_source(
+            session,
+            tenant_id=tenant_id,
+            source=source,
+            deleted_at=deleted_at,
+            deletion_reason=deletion_reason,
+            user_id=user_id,
+            api_key_id=api_key_id,
+        )
 
 
 def retire_caldav_sources_for_calendar(session: Session, *, tenant_id: str, calendar_id: str, deleted_at: datetime) -> None:
@@ -1252,7 +1362,13 @@ def retire_stale_sync_sources_for_url(session: Session, *, tenant_id: str, sourc
     now = utcnow()
     for source in sources:
         if source.calendar is None or source.calendar.deleted_at is not None:
-            retire_sync_source(session, tenant_id=tenant_id, source=source, deleted_at=now)
+            retire_sync_source(
+                session,
+                tenant_id=tenant_id,
+                source=source,
+                deleted_at=now,
+                deletion_reason="stale_sync_source_retired",
+            )
     session.flush()
 
 
@@ -1281,7 +1397,16 @@ def active_caldav_source_for_url(session: Session, *, tenant_id: str, collection
     return active_sync_source_for_url(session, tenant_id=tenant_id, source_kind="caldav", collection_url=collection_url)
 
 
-def retire_sync_source(session: Session, *, tenant_id: str, source: CalendarSyncSource, deleted_at: datetime) -> None:
+def retire_sync_source(
+    session: Session,
+    *,
+    tenant_id: str,
+    source: CalendarSyncSource,
+    deleted_at: datetime,
+    deletion_reason: str = "sync_source_retired",
+    user_id: str | None = None,
+    api_key_id: str | None = None,
+) -> None:
     from govoplan_calendar.backend.outbox import (
         calendar_outbox_has_live_lease,
         cancel_calendar_outbox_for_source,
@@ -1302,18 +1427,26 @@ def retire_sync_source(session: Session, *, tenant_id: str, source: CalendarSync
             "CalDAV source cannot be retired while an outbound operation has an active lease"
         )
 
+    # Delete credentials first. An external secret provider cannot participate
+    # in the SQL transaction, so provider failure must leave the source and its
+    # queued work untouched. If the later SQL commit fails after provider
+    # success, the source fails closed and the idempotent delete can be retried.
+    delete_caldav_credential(
+        session,
+        tenant_id=tenant_id,
+        source_id=source.id,
+        credential_ref=source.credential_ref,
+        deletion_reason=deletion_reason,
+        user_id=user_id,
+        api_key_id=api_key_id,
+    )
+    source.credential_ref = None
     cancel_calendar_outbox_for_source(
         session,
         source_id=source.id,
         reason="CalDAV source was retired before queued external changes were delivered",
     )
     source.deleted_at = deleted_at
-    delete_caldav_credential(
-        session,
-        tenant_id=tenant_id,
-        source_id=source.id,
-        credential_ref=source.credential_ref,
-    )
 
 
 def retire_caldav_source(session: Session, *, tenant_id: str, source: CalendarSyncSource, deleted_at: datetime) -> None:
@@ -1343,6 +1476,7 @@ def store_caldav_credential(
     user_id: str | None,
     source: CalendarSyncSource,
     secret: str,
+    api_key_id: str | None = None,
 ) -> str:
     existing = internal_caldav_credential(
         session,
@@ -1350,6 +1484,7 @@ def store_caldav_credential(
         source_id=source.id,
         credential_ref=source.credential_ref,
     )
+    replacing_existing = existing is not None
     if existing is None:
         existing = CalendarSyncCredential(
             tenant_id=tenant_id,
@@ -1361,10 +1496,45 @@ def store_caldav_credential(
         session.add(existing)
         session.flush()
 
+    old_provider_ref = _credential_provider_ref(existing)
     provider = secret_provider()
     name = f"caldav:{source.id}:{source.auth_type}"
+    if old_provider_ref and provider is None:
+        raise CalendarError(
+            "Stored sync credential cannot be replaced while its secret provider is unavailable"
+        )
     if provider is not None:
-        provider_ref = str(provider.store_secret(scope=f"calendar:{tenant_id}", name=name, value=secret))
+        try:
+            provider_ref = str(
+                provider.store_secret(
+                    scope=f"calendar:{tenant_id}",
+                    name=name,
+                    value=secret,
+                )
+            ).strip()
+        except Exception:
+            raise CalendarError("Stored sync credential could not be written to its secret provider") from None
+        if not provider_ref:
+            raise CalendarError("Secret provider returned an empty credential reference")
+        if old_provider_ref and old_provider_ref != provider_ref:
+            try:
+                provider.delete_secret(old_provider_ref)
+            except Exception:
+                if not _provider_secret_is_absent(provider, old_provider_ref):
+                    compensation_failed = False
+                    try:
+                        provider.delete_secret(provider_ref)
+                    except Exception:
+                        compensation_failed = not _provider_secret_is_absent(provider, provider_ref)
+                    compensation_detail = (
+                        "; replacement cleanup also failed and requires provider-side reconciliation"
+                        if compensation_failed
+                        else ""
+                    )
+                    raise CalendarError(
+                        "Previous sync credential could not be deleted from its secret provider; "
+                        f"the replacement was not activated{compensation_detail}"
+                    ) from None
         existing.secret_encrypted = None
         existing.metadata_ = {"source_id": source.id, "provider_ref": provider_ref}
     else:
@@ -1373,6 +1543,24 @@ def store_caldav_credential(
     existing.credential_kind = credential_kind_for_auth_type(source.auth_type)
     existing.label = source.display_name or source.collection_url
     existing.deleted_at = None
+    if replacing_existing:
+        audit_event(
+            session,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            api_key_id=api_key_id,
+            action="calendar.sync_credential_rotated",
+            object_type="calendar_sync_credential",
+            object_id=existing.id,
+            details={
+                "sync_source_id": source.id,
+                "storage_backend": (
+                    "external_secret_provider"
+                    if provider is not None
+                    else "encrypted_database"
+                ),
+            },
+        )
     session.flush()
     return f"{CALDAV_INTERNAL_CREDENTIAL_PREFIX}{existing.id}"
 
@@ -1383,9 +1571,12 @@ def delete_caldav_credential(
     tenant_id: str,
     source_id: str,
     credential_ref: str | None,
-) -> None:
+    deletion_reason: str = "sync_source_retired",
+    user_id: str | None = None,
+    api_key_id: str | None = None,
+) -> bool:
     if not credential_ref:
-        return
+        return False
     credential = internal_caldav_credential(
         session,
         tenant_id=tenant_id,
@@ -1395,12 +1586,97 @@ def delete_caldav_credential(
     if credential is None:
         # Legacy external/env references have no locally provable ownership and
         # must never be passed to a provider delete operation.
-        return
+        return False
+    return _delete_calendar_owned_credential(
+        session,
+        credential=credential,
+        source_id=source_id,
+        deletion_reason=deletion_reason,
+        user_id=user_id,
+        api_key_id=api_key_id,
+    )
+
+
+def _delete_calendar_owned_credential(
+    session: Session,
+    *,
+    credential: CalendarSyncCredential,
+    source_id: str | None,
+    deletion_reason: str,
+    user_id: str | None = None,
+    api_key_id: str | None = None,
+) -> bool:
     provider_ref = _credential_provider_ref(credential)
+    if not credential.secret_encrypted and not provider_ref:
+        return False
     provider = secret_provider()
-    if provider is not None and provider_ref:
-        provider.delete_secret(provider_ref)
+    storage_backend = "encrypted_database"
+    if provider_ref:
+        if provider is None:
+            raise CalendarError(
+                "Stored sync credential cannot be deleted while its secret provider is unavailable; "
+                "the requested deletion was not completed"
+            )
+        try:
+            provider.delete_secret(provider_ref)
+        except Exception:
+            if not _provider_secret_is_absent(provider, provider_ref):
+                raise CalendarError(
+                    "Stored sync credential could not be deleted from its secret provider; "
+                    "the requested deletion was not completed"
+                ) from None
+        storage_backend = "external_secret_provider"
+    credential.secret_encrypted = None
+    credential.metadata_ = {"source_id": source_id} if source_id else {}
     credential.deleted_at = utcnow()
+    audit_event(
+        session,
+        tenant_id=credential.tenant_id,
+        user_id=user_id,
+        api_key_id=api_key_id,
+        action="calendar.sync_credential_deleted",
+        object_type="calendar_sync_credential",
+        object_id=credential.id,
+        details={
+            "sync_source_id": source_id,
+            "storage_backend": storage_backend,
+            "deletion_reason": deletion_reason,
+        },
+    )
+    return True
+
+
+def _provider_secret_is_absent(provider: Any, provider_ref: str) -> bool:
+    """Confirm an idempotent delete when a provider reports missing/error."""
+
+    try:
+        return provider.read_secret(provider_ref) is None
+    except Exception:
+        return False
+
+
+def delete_calendar_credentials_for_retirement(session: Session) -> int:
+    """Delete all Calendar-owned secrets before destructive module retirement.
+
+    This includes pre-hardening soft-deleted rows whose provider references or
+    ciphertext were retained. Provider deletion is intentionally fail-closed;
+    the caller must not drop Calendar tables if any external deletion fails.
+    """
+
+    credentials = session.query(CalendarSyncCredential).order_by(CalendarSyncCredential.id.asc()).all()
+    deleted = 0
+    for credential in credentials:
+        metadata = credential.metadata_ if isinstance(credential.metadata_, dict) else {}
+        source_id = metadata.get("source_id")
+        if _delete_calendar_owned_credential(
+            session,
+            credential=credential,
+            source_id=str(source_id) if source_id else None,
+            deletion_reason="module_data_retired",
+        ):
+            deleted += 1
+    session.flush()
+    return deleted
 
 
 def resolve_caldav_secret(

tests/test_caldav.py

@@ -4,12 +4,13 @@ import unittest
 from datetime import datetime, timezone
 from unittest.mock import patch
 
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, inspect
 from sqlalchemy.orm import sessionmaker
 
 from govoplan_access.backend.db import models as access_models  # noqa: F401 - populate users/accounts tables
 from govoplan_calendar.backend.caldav import CalDAVClient, CalDAVError, CalDAVNotFound, CalDAVObject, CalDAVPreconditionFailed, CalDAVReportResult, CalDAVWriteResult, parse_multistatus
 from govoplan_calendar.backend.db.models import CalendarEvent, CalendarOutboxOperation, CalendarSyncCredential
+from govoplan_calendar.backend.manifest import manifest
 from govoplan_calendar.backend.outbox import dispatch_calendar_outbox
 from govoplan_calendar.backend.schemas import (
     CalendarCalDavDiscoveryRequest,
@@ -27,6 +28,7 @@ from govoplan_calendar.backend.service import (
     create_calendar,
     create_caldav_source,
     create_event,
+    delete_caldav_source,
     delete_calendar,
     delete_event,
     discover_caldav_calendars,
@@ -365,6 +367,7 @@ class CalDAVSyncTests(unittest.TestCase):
 
             def delete_secret(self, secret_ref: str) -> None:
                 self.deleted.append(secret_ref)
+                self.values.pop(secret_ref, None)
 
         provider = Provider()
         session = self.session()
@@ -437,9 +440,23 @@ class CalDAVSyncTests(unittest.TestCase):
             delete_calendar(session, tenant_id="tenant-1", calendar_id=other_calendar.id)
             self.assertEqual(provider.deleted, [])
 
-            delete_calendar(session, tenant_id="tenant-1", calendar_id=calendar.id)
+            with patch("govoplan_calendar.backend.service.audit_event") as audit:
+                delete_calendar(session, tenant_id="tenant-1", calendar_id=calendar.id)
 
         self.assertEqual(provider.deleted, [provider_ref])
+        self.assertNotIn(provider_ref, provider.values)
+        self.assertIsNone(source.credential_ref)
+        self.assertIsNone(credential.secret_encrypted)
+        self.assertNotIn("provider_ref", credential.metadata_ or {})
+        audit.assert_called_once()
+        audit_call = audit.call_args.kwargs
+        self.assertEqual(audit_call["action"], "calendar.sync_credential_deleted")
+        self.assertEqual(audit_call["object_id"], credential.id)
+        self.assertEqual(audit_call["details"]["storage_backend"], "external_secret_provider")
+        self.assertEqual(audit_call["details"]["deletion_reason"], "calendar_deleted")
+        self.assertNotIn("credential_ref", audit_call["details"])
+        self.assertNotIn("provider_ref", audit_call["details"])
+        self.assertNotIn(provider_ref, repr(audit_call["details"]))
 
     def test_legacy_unowned_refs_are_neither_read_nor_deleted(self) -> None:
         class Provider:
@@ -516,9 +533,312 @@ class CalDAVSyncTests(unittest.TestCase):
 
         credential = session.query(CalendarSyncCredential).one()
         self.assertIsNotNone(source.deleted_at)
+        self.assertIsNone(source.credential_ref)
         self.assertIsNotNone(credential.deleted_at)
+        self.assertIsNone(credential.secret_encrypted)
+        self.assertEqual(credential.metadata_, {"source_id": source.id})
         self.assertEqual(list_caldav_sources(session, tenant_id="tenant-1"), [])
 
+    def test_external_credential_deletion_fails_closed_when_provider_is_unavailable(self) -> None:
+        class Provider:
+            def __init__(self) -> None:
+                self.values: dict[str, str] = {}
+
+            def store_secret(self, *, scope: str, name: str, value: str) -> str:
+                reference = f"vault:{scope}:{name}"
+                self.values[reference] = value
+                return reference
+
+            def read_secret(self, secret_ref: str) -> str | None:
+                return self.values.get(secret_ref)
+
+            def delete_secret(self, secret_ref: str) -> None:
+                self.values.pop(secret_ref, None)
+
+        provider = Provider()
+        session = self.session()
+        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
+        calendar = create_calendar(
+            session,
+            tenant_id="tenant-1",
+            user_id=None,
+            payload=CalendarCollectionCreateRequest(name="Remote"),
+        )
+        with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider):
+            source = create_caldav_source(
+                session,
+                tenant_id="tenant-1",
+                user_id=None,
+                payload=CalendarCalDavSourceCreateRequest(
+                    calendar_id=calendar.id,
+                    collection_url="https://dav.example.test/provider-down",
+                    auth_type="bearer",
+                    bearer_token="do-not-log-this",
+                ),
+            )
+            session.commit()
+            credential = session.query(CalendarSyncCredential).one()
+            provider_ref = str((credential.metadata_ or {})["provider_ref"])
+
+        with patch("govoplan_calendar.backend.service.secret_provider", return_value=None), self.assertRaisesRegex(
+            CalendarError,
+            "secret provider is unavailable",
+        ):
+            delete_caldav_source(session, tenant_id="tenant-1", source_id=source.id)
+        session.rollback()
+
+        self.assertIsNone(source.deleted_at)
+        self.assertEqual(source.credential_ref, f"{CALDAV_INTERNAL_CREDENTIAL_PREFIX}{credential.id}")
+        self.assertIsNone(credential.deleted_at)
+        self.assertEqual((credential.metadata_ or {})["provider_ref"], provider_ref)
+        self.assertIn(provider_ref, provider.values)
+
+        with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider), patch.object(
+            provider,
+            "delete_secret",
+            side_effect=RuntimeError(f"provider failed for {provider_ref}"),
+        ):
+            with self.assertRaises(CalendarError) as raised:
+                delete_caldav_source(session, tenant_id="tenant-1", source_id=source.id)
+        self.assertNotIn(provider_ref, str(raised.exception))
+        session.rollback()
+        self.assertIsNone(source.deleted_at)
+        self.assertIsNone(credential.deleted_at)
+
+    def test_disabling_auth_immediately_scrubs_and_audits_database_credential(self) -> None:
+        session = self.session()
+        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
+        calendar = create_calendar(
+            session,
+            tenant_id="tenant-1",
+            user_id=None,
+            payload=CalendarCollectionCreateRequest(name="Remote"),
+        )
+        source = create_caldav_source(
+            session,
+            tenant_id="tenant-1",
+            user_id=None,
+            payload=CalendarCalDavSourceCreateRequest(
+                calendar_id=calendar.id,
+                collection_url="https://dav.example.test/disable-auth",
+                auth_type="basic",
+                username="ada",
+                password="do-not-log-this",
+            ),
+        )
+        session.commit()
+        credential = session.query(CalendarSyncCredential).one()
+        self.assertIsNotNone(credential.secret_encrypted)
+
+        with patch("govoplan_calendar.backend.service.audit_event") as audit:
+            update_caldav_source(
+                session,
+                tenant_id="tenant-1",
+                source_id=source.id,
+                payload=CalendarCalDavSourceUpdateRequest(auth_type="none"),
+            )
+            session.commit()
+
+        self.assertIsNone(source.credential_ref)
+        self.assertIsNone(credential.secret_encrypted)
+        self.assertIsNotNone(credential.deleted_at)
+        audit.assert_called_once()
+        self.assertEqual(
+            audit.call_args.kwargs["details"]["deletion_reason"],
+            "authentication_disabled",
+        )
+        self.assertNotIn("do-not-log-this", repr(audit.call_args))
+
+    def test_external_credential_replacement_fails_closed_when_provider_is_unavailable(self) -> None:
+        class Provider:
+            def __init__(self) -> None:
+                self.values: dict[str, str] = {}
+
+            def store_secret(self, *, scope: str, name: str, value: str) -> str:
+                reference = f"vault:{scope}:{name}"
+                self.values[reference] = value
+                return reference
+
+            def read_secret(self, secret_ref: str) -> str | None:
+                return self.values.get(secret_ref)
+
+            def delete_secret(self, secret_ref: str) -> None:
+                self.values.pop(secret_ref, None)
+
+        provider = Provider()
+        session = self.session()
+        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
+        calendar = create_calendar(
+            session,
+            tenant_id="tenant-1",
+            user_id=None,
+            payload=CalendarCollectionCreateRequest(name="Remote"),
+        )
+        with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider):
+            source = create_caldav_source(
+                session,
+                tenant_id="tenant-1",
+                user_id=None,
+                payload=CalendarCalDavSourceCreateRequest(
+                    calendar_id=calendar.id,
+                    collection_url="https://dav.example.test/replacement-provider-down",
+                    auth_type="bearer",
+                    bearer_token="old-secret",
+                ),
+            )
+            session.commit()
+            credential = session.query(CalendarSyncCredential).one()
+            provider_ref = str((credential.metadata_ or {})["provider_ref"])
+
+        with patch("govoplan_calendar.backend.service.secret_provider", return_value=None), self.assertRaisesRegex(
+            CalendarError,
+            "cannot be replaced.*provider is unavailable",
+        ):
+            update_caldav_source(
+                session,
+                tenant_id="tenant-1",
+                source_id=source.id,
+                payload=CalendarCalDavSourceUpdateRequest(bearer_token="new-secret"),
+            )
+        session.rollback()
+
+        self.assertEqual(source.credential_ref, f"{CALDAV_INTERNAL_CREDENTIAL_PREFIX}{credential.id}")
+        self.assertEqual((credential.metadata_ or {})["provider_ref"], provider_ref)
+        self.assertEqual(provider.values[provider_ref], "old-secret")
+
+    def test_external_credential_delete_is_retryable_after_database_rollback(self) -> None:
+        class Provider:
+            def __init__(self) -> None:
+                self.values: dict[str, str] = {}
+                self.deletes: list[str] = []
+
+            def store_secret(self, *, scope: str, name: str, value: str) -> str:
+                reference = f"vault:{scope}:{name}"
+                self.values[reference] = value
+                return reference
+
+            def read_secret(self, secret_ref: str) -> str | None:
+                return self.values.get(secret_ref)
+
+            def delete_secret(self, secret_ref: str) -> None:
+                self.deletes.append(secret_ref)
+                if secret_ref not in self.values:
+                    raise KeyError("already deleted")
+                self.values.pop(secret_ref, None)
+
+        provider = Provider()
+        session = self.session()
+        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
+        calendar = create_calendar(
+            session,
+            tenant_id="tenant-1",
+            user_id=None,
+            payload=CalendarCollectionCreateRequest(name="Remote"),
+        )
+        with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider):
+            source = create_caldav_source(
+                session,
+                tenant_id="tenant-1",
+                user_id=None,
+                payload=CalendarCalDavSourceCreateRequest(
+                    calendar_id=calendar.id,
+                    collection_url="https://dav.example.test/retry-delete",
+                    auth_type="basic",
+                    username="ada",
+                    password="do-not-log-this",
+                ),
+            )
+            session.commit()
+            credential = session.query(CalendarSyncCredential).one()
+            provider_ref = str((credential.metadata_ or {})["provider_ref"])
+
+            with patch("govoplan_calendar.backend.service.audit_event"):
+                delete_caldav_source(session, tenant_id="tenant-1", source_id=source.id)
+            self.assertNotIn(provider_ref, provider.values)
+            self.assertIsNotNone(credential.deleted_at)
+            self.assertIsNone(source.credential_ref)
+            session.rollback()
+
+            self.assertIsNone(source.deleted_at)
+            session.refresh(credential)
+            self.assertIsNone(credential.deleted_at)
+            self.assertEqual((credential.metadata_ or {})["provider_ref"], provider_ref)
+            self.assertIsNone(resolve_caldav_credential_ref(
+                session,
+                tenant_id="tenant-1",
+                source_id=source.id,
+                credential_ref=source.credential_ref,
+            ))
+
+            with patch("govoplan_calendar.backend.service.audit_event") as audit:
+                delete_caldav_source(session, tenant_id="tenant-1", source_id=source.id)
+                session.commit()
+
+        self.assertEqual(provider.deletes, [provider_ref, provider_ref])
+        self.assertIsNotNone(source.deleted_at)
+        self.assertIsNone(source.credential_ref)
+        self.assertIsNotNone(credential.deleted_at)
+        self.assertEqual(audit.call_count, 1)
+
+    def test_destructive_module_retirement_deletes_external_credentials_before_tables(self) -> None:
+        class Provider:
+            def __init__(self) -> None:
+                self.values: dict[str, str] = {}
+                self.deleted: list[str] = []
+
+            def store_secret(self, *, scope: str, name: str, value: str) -> str:
+                reference = f"vault:{scope}:{name}"
+                self.values[reference] = value
+                return reference
+
+            def read_secret(self, secret_ref: str) -> str | None:
+                return self.values.get(secret_ref)
+
+            def delete_secret(self, secret_ref: str) -> None:
+                self.deleted.append(secret_ref)
+                self.values.pop(secret_ref, None)
+
+        provider = Provider()
+        session = self.session()
+        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
+        calendar = create_calendar(
+            session,
+            tenant_id="tenant-1",
+            user_id=None,
+            payload=CalendarCollectionCreateRequest(name="Remote"),
+        )
+        with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider):
+            create_caldav_source(
+                session,
+                tenant_id="tenant-1",
+                user_id=None,
+                payload=CalendarCalDavSourceCreateRequest(
+                    calendar_id=calendar.id,
+                    collection_url="https://dav.example.test/module-retirement",
+                    auth_type="bearer",
+                    bearer_token="do-not-log-this",
+                ),
+            )
+            session.commit()
+            credential = session.query(CalendarSyncCredential).one()
+            provider_ref = str((credential.metadata_ or {})["provider_ref"])
+            retirement_provider = manifest.migration_spec.retirement_provider
+            assert retirement_provider is not None
+            plan = retirement_provider(session, "calendar")
+            assert plan.destroy_data_executor is not None
+            with patch("govoplan_calendar.backend.service.audit_event") as audit:
+                plan.destroy_data_executor(session, "calendar")
+
+        self.assertEqual(provider.deleted, [provider_ref])
+        self.assertEqual(provider.values, {})
+        self.assertEqual(audit.call_count, 1)
+        self.assertEqual(
+            audit.call_args.kwargs["details"]["deletion_reason"],
+            "module_data_retired",
+        )
+        self.assertFalse(inspect(self.engine).has_table("calendar_sync_credentials"))
+        self.assertFalse(inspect(self.engine).has_table("calendar_sync_sources"))
+
     def test_create_source_retires_orphaned_source_for_deleted_calendar(self) -> None:
         session = self.session()
         session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))