feat: add governed Scheduling DSAR coverage

README.md

@@ -61,6 +61,18 @@ Participant availability is sensitive operational data. Scheduling must record:
 Availability responses should be removable or redacted after the poll decision
 unless a configured process requires longer evidence retention.
 
+Scheduling publishes `privacy.dsar.scheduling` for Core's governed
+data-subject-request workflow. It projects tenant-scoped participant, request,
+candidate-slot, and notification-envelope metadata while omitting Poll and
+invitation identifiers, public-link and proof material, Calendar identifiers,
+free/busy detail, notification content and errors, password hashes, opaque
+metadata, and unrelated participants. Poll owns the actual availability choices
+and response-retirement evidence; Calendar owns event and hold state. Terminal,
+responded, or notified records are retained or manually reviewed. Only a truly
+unengaged participant can be anonymized automatically, after tenant, identity,
+request status, invitation, response, enrollment, and notification state are
+revalidated under lock.
+
 ## Candidate Capabilities
 
 - `scheduling.polls`

src/govoplan_scheduling/backend/dsar_provider.py

@@ -0,0 +1,891 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from datetime import datetime, timezone
+
+from sqlalchemy import func, or_
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+from govoplan_scheduling.backend.db.models import (
+    SchedulingCandidateSlot,
+    SchedulingNotification,
+    SchedulingParticipant,
+    SchedulingRequest,
+)
+
+
+SCHEDULING_DSAR_CAPABILITY = dsar_capability_name("scheduling")
+_MAX_RECORDS = 5_000
+_TERMINAL_REQUEST_STATUSES = frozenset(
+    {"decided", "handed_off", "cancelled", "archived"}
+)
+_TERMINAL_NOTIFICATION_STATUSES = frozenset({"sent", "failed", "skipped", "cancelled"})
+
+
+class SchedulingDsarProvider:
+    provider_id = "scheduling"
+    module_id = "scheduling"
+
+    def search_subject(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+    ) -> Sequence[DsarRecordRef]:
+        db = _session(session)
+        references = _scheduling_references(subject)
+        membership_ids = _membership_ids(subject)
+        respondent_ids = _respondent_ids(subject)
+        account_ids = _account_ids(subject)
+        email = _subject_email(subject)
+        if not any(
+            (
+                references,
+                membership_ids,
+                respondent_ids,
+                account_ids,
+                email,
+            )
+        ):
+            return ()
+
+        participants = _matching_participants(
+            db,
+            tenant_id=tenant_id,
+            participant_id=references.get("participant"),
+            respondent_ids=respondent_ids,
+            account_ids=account_ids,
+            email=email,
+        )
+        participant_ids = {row.id for row in participants}
+        request_ids = {row.request_id for row in participants}
+        if references.get("request"):
+            request_ids.add(references["request"])
+        requests = _matching_requests(
+            db,
+            tenant_id=tenant_id,
+            request_ids=request_ids,
+            organizer_ids=membership_ids,
+        )
+        requests_by_id = {row.id: row for row in requests}
+        request_ids = set(requests_by_id)
+        notifications = _matching_notifications(
+            db,
+            tenant_id=tenant_id,
+            request_ids=request_ids,
+            participant_ids=participant_ids,
+            respondent_ids=respondent_ids,
+            email=email,
+        )
+        notifications_by_participant: dict[str, list[SchedulingNotification]] = {}
+        for notification in notifications:
+            if notification.participant_id:
+                notifications_by_participant.setdefault(
+                    notification.participant_id, []
+                ).append(notification)
+
+        records: list[DsarRecordRef] = []
+
+        def append(record: DsarRecordRef) -> None:
+            if len(records) >= _MAX_RECORDS:
+                raise ValueError(
+                    "Scheduling DSAR match limit exceeded; narrow the subject selectors."
+                )
+            records.append(record)
+
+        for request in requests:
+            participant_context = request.id in {row.request_id for row in participants}
+            organizer_context = bool(
+                request.organizer_user_id
+                and request.organizer_user_id in membership_ids
+            )
+            immutable = _request_is_evidence(request)
+            append(
+                _record(
+                    "scheduling_request",
+                    request.id,
+                    "scheduling_request",
+                    request.title,
+                    {
+                        "match_fields": (
+                            ["organizer_user_id"] if organizer_context else []
+                        ),
+                        "participant_context": participant_context,
+                        "title": request.title,
+                        "description": request.description,
+                        "location": request.location,
+                        "timezone": request.timezone,
+                        "status": request.status,
+                        "deadline_at": _iso(request.deadline_at),
+                        "allow_external_participants": request.allow_external_participants,
+                        "allow_participant_updates": request.allow_participant_updates,
+                        "result_visibility": request.result_visibility,
+                        "participant_visibility": request.participant_visibility,
+                        "notify_on_answers": request.notify_on_answers,
+                        "single_choice": request.single_choice,
+                        "max_participants_per_option": request.max_participants_per_option,
+                        "allow_maybe": request.allow_maybe,
+                        "allow_comments": request.allow_comments,
+                        "participant_email_required": request.participant_email_required,
+                        "calendar_integration_enabled": request.calendar_integration_enabled,
+                        "calendar_freebusy_enabled": request.calendar_freebusy_enabled,
+                        "calendar_hold_enabled": request.calendar_hold_enabled,
+                        "create_calendar_event_on_decision": request.create_calendar_event_on_decision,
+                        "handed_off_at": _iso(request.handed_off_at),
+                        "cancelled_at": _iso(request.cancelled_at),
+                        "deleted_at": _iso(request.deleted_at),
+                    },
+                    observed_at=request.updated_at,
+                    immutable=immutable,
+                    retention_reason=(
+                        "Decided, handed-off, cancelled, archived, or deleted scheduling state is institutional decision and coordination evidence."
+                        if immutable
+                        else None
+                    ),
+                    source_path=f"/scheduling?request={request.id}",
+                )
+            )
+
+        for slot in _candidate_slots(
+            db,
+            tenant_id=tenant_id,
+            request_ids=request_ids,
+        ):
+            request = requests_by_id[slot.request_id]
+            immutable = _request_is_evidence(request) or slot.deleted_at is not None
+            append(
+                _record(
+                    "scheduling_candidate_slot",
+                    slot.id,
+                    "scheduling_candidate_context",
+                    slot.label,
+                    {
+                        "request_id": slot.request_id,
+                        "label": slot.label,
+                        "description": slot.description,
+                        "start_at": _iso(slot.start_at),
+                        "end_at": _iso(slot.end_at),
+                        "timezone": slot.timezone,
+                        "location": slot.location,
+                        "position": slot.position,
+                        "freebusy_checked_at": _iso(slot.freebusy_checked_at),
+                        "freebusy_status": slot.freebusy_status,
+                        "deleted_at": _iso(slot.deleted_at),
+                    },
+                    observed_at=slot.updated_at,
+                    immutable=immutable,
+                    retention_reason=(
+                        "Candidate timing retained with terminal Scheduling decision evidence."
+                        if immutable
+                        else None
+                    ),
+                    source_path=f"/scheduling?request={slot.request_id}",
+                )
+            )
+
+        for participant in participants:
+            request = requests_by_id.get(participant.request_id)
+            if request is None:
+                continue
+            matching_fields = _participant_matching_fields(
+                participant,
+                participant_id=references.get("participant"),
+                respondent_ids=respondent_ids,
+                account_ids=account_ids,
+                email=email,
+            )
+            immutable = _request_is_evidence(request) or _participant_is_evidence(
+                participant,
+                notifications_by_participant.get(participant.id, ()),
+            )
+            erasable_without_coordination = not immutable and _participant_is_unengaged(
+                participant,
+                notifications_by_participant.get(participant.id, ()),
+            )
+            append(
+                _record(
+                    "scheduling_participant",
+                    participant.id,
+                    "scheduling_participation",
+                    participant.display_name or "Scheduling participant",
+                    {
+                        "match_fields": matching_fields,
+                        "request_id": participant.request_id,
+                        "display_name": participant.display_name,
+                        "email": _normalized_email(participant.email),
+                        "participant_type": participant.participant_type,
+                        "required": participant.required,
+                        "status": participant.status,
+                        "account_bound_at": _iso(participant.account_bound_at),
+                        "last_invited_at": _iso(participant.last_invited_at),
+                        "responded_at": _iso(participant.responded_at),
+                        "response_comment": _bounded_text(participant.response_comment),
+                        "deleted_at": _iso(participant.deleted_at),
+                        "erasable_without_coordination": erasable_without_coordination,
+                    },
+                    observed_at=participant.updated_at,
+                    immutable=immutable,
+                    retention_reason=(
+                        "Responded, notified, removed, or terminal participant state is retained with Poll and Scheduling decision evidence."
+                        if immutable
+                        else None
+                    ),
+                    source_path=f"/scheduling?request={participant.request_id}",
+                )
+            )
+
+        for notification in notifications:
+            immutable = bool(
+                notification.sent_at
+                or notification.status in _TERMINAL_NOTIFICATION_STATUSES
+            )
+            append(
+                _record(
+                    "scheduling_notification",
+                    notification.id,
+                    "scheduling_notification_evidence",
+                    f"Scheduling {notification.event_kind} notification",
+                    {
+                        "match_fields": _notification_matching_fields(
+                            notification,
+                            participant_ids=participant_ids,
+                            respondent_ids=respondent_ids,
+                            email=email,
+                        ),
+                        "request_id": notification.request_id,
+                        "participant_id": (
+                            notification.participant_id
+                            if notification.participant_id in participant_ids
+                            else None
+                        ),
+                        "event_kind": notification.event_kind,
+                        "channel": notification.channel,
+                        "recipient": _matching_recipient(
+                            notification.recipient,
+                            respondent_ids=respondent_ids,
+                            email=email,
+                        ),
+                        "status": notification.status,
+                        "sent_at": _iso(notification.sent_at),
+                    },
+                    observed_at=notification.updated_at,
+                    immutable=immutable,
+                    retention_reason=(
+                        "Terminal Scheduling notification state is delivery and participant-access evidence."
+                        if immutable
+                        else None
+                    ),
+                )
+            )
+        return tuple(records)
+
+    def plan_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        records: Sequence[DsarRecordRef],
+    ) -> Sequence[DsarErasureActionRef]:
+        del session
+        actions: list[DsarErasureActionRef] = []
+        for record in records:
+            if (
+                record.provider_id != self.provider_id
+                or record.module_id != self.module_id
+            ):
+                raise ValueError("Scheduling DSAR received a foreign provider record.")
+            actions.append(
+                _action(
+                    f"scheduling:{'retain' if record.immutable_evidence else 'review'}:{record.resource_type}:{record.resource_id}",
+                    "retain" if record.immutable_evidence else "manual_review",
+                    record,
+                    f"{'Retain' if record.immutable_evidence else 'Review'} {record.title}",
+                    record.retention_reason
+                    or "Scheduling data can overlap Poll responses, Calendar effects, shared participants, and institutional decisions; review it through the owning lifecycle controls.",
+                    executable=False,
+                )
+            )
+            if (
+                record.resource_type == "scheduling_participant"
+                and record.data.get("erasable_without_coordination") is True
+                and record.data.get("match_fields")
+            ):
+                actions.append(
+                    _action(
+                        f"scheduling:anonymize:scheduling_participant:{record.resource_id}",
+                        "anonymize",
+                        record,
+                        "Anonymize unengaged Scheduling participant",
+                        "A participant without invitation, response, self-enrollment, notification, or terminal decision evidence can be removed without changing Poll or Calendar state.",
+                        executable=True,
+                        irreversible=True,
+                        metadata={"tenant_id": tenant_id},
+                    )
+                )
+        action_ids = [action.action_id for action in actions]
+        if len(action_ids) != len(set(action_ids)):
+            raise ValueError("Scheduling DSAR produced duplicate action ids.")
+        return tuple(actions)
+
+    def execute_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        actions: Sequence[DsarErasureActionRef],
+        request_id: str,
+    ) -> Sequence[DsarExecutionResultRef]:
+        db = _session(session)
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            if (
+                action.provider_id != self.provider_id
+                or action.module_id != self.module_id
+                or action.metadata.get("tenant_id") != tenant_id
+                or not action.action_id.startswith(
+                    "scheduling:anonymize:scheduling_participant:"
+                )
+            ):
+                results.append(
+                    _blocked(action, "The Scheduling DSAR action is stale or invalid.")
+                )
+                continue
+            results.append(
+                _anonymize_unengaged_participant(
+                    db,
+                    tenant_id=tenant_id,
+                    subject=subject,
+                    action=action,
+                    request_id=request_id,
+                )
+            )
+        db.flush()
+        return tuple(results)
+
+
+def _matching_participants(
+    session: Session,
+    *,
+    tenant_id: str,
+    participant_id: str | None,
+    respondent_ids: set[str],
+    account_ids: set[str],
+    email: str | None,
+) -> list[SchedulingParticipant]:
+    conditions = []
+    if participant_id:
+        conditions.append(SchedulingParticipant.id == participant_id)
+    if respondent_ids:
+        conditions.append(SchedulingParticipant.respondent_id.in_(respondent_ids))
+    if account_ids:
+        conditions.append(SchedulingParticipant.bound_account_id.in_(account_ids))
+    if email:
+        conditions.append(func.lower(SchedulingParticipant.email) == email)
+    if not conditions:
+        return []
+    candidates = _bounded_rows(
+        session.query(SchedulingParticipant)
+        .filter(SchedulingParticipant.tenant_id == tenant_id, or_(*conditions))
+        .order_by(SchedulingParticipant.id)
+    )
+    return [
+        row
+        for row in candidates
+        if _participant_matching_fields(
+            row,
+            participant_id=participant_id,
+            respondent_ids=respondent_ids,
+            account_ids=account_ids,
+            email=email,
+        )
+    ]
+
+
+def _matching_requests(
+    session: Session,
+    *,
+    tenant_id: str,
+    request_ids: set[str],
+    organizer_ids: set[str],
+) -> list[SchedulingRequest]:
+    conditions = []
+    if request_ids:
+        conditions.append(SchedulingRequest.id.in_(request_ids))
+    if organizer_ids:
+        conditions.append(SchedulingRequest.organizer_user_id.in_(organizer_ids))
+    if not conditions:
+        return []
+    return _bounded_rows(
+        session.query(SchedulingRequest)
+        .filter(SchedulingRequest.tenant_id == tenant_id, or_(*conditions))
+        .order_by(SchedulingRequest.id)
+    )
+
+
+def _candidate_slots(
+    session: Session,
+    *,
+    tenant_id: str,
+    request_ids: set[str],
+) -> list[SchedulingCandidateSlot]:
+    if not request_ids:
+        return []
+    return _bounded_rows(
+        session.query(SchedulingCandidateSlot)
+        .filter(
+            SchedulingCandidateSlot.tenant_id == tenant_id,
+            SchedulingCandidateSlot.request_id.in_(request_ids),
+        )
+        .order_by(SchedulingCandidateSlot.request_id, SchedulingCandidateSlot.position)
+    )
+
+
+def _matching_notifications(
+    session: Session,
+    *,
+    tenant_id: str,
+    request_ids: set[str],
+    participant_ids: set[str],
+    respondent_ids: set[str],
+    email: str | None,
+) -> list[SchedulingNotification]:
+    if not request_ids:
+        return []
+    conditions = []
+    if participant_ids:
+        conditions.append(SchedulingNotification.participant_id.in_(participant_ids))
+    if respondent_ids:
+        conditions.append(SchedulingNotification.recipient.in_(respondent_ids))
+    if email:
+        conditions.append(func.lower(SchedulingNotification.recipient) == email)
+    if not conditions:
+        return []
+    candidates = _bounded_rows(
+        session.query(SchedulingNotification)
+        .filter(
+            SchedulingNotification.tenant_id == tenant_id,
+            SchedulingNotification.request_id.in_(request_ids),
+            or_(*conditions),
+        )
+        .order_by(SchedulingNotification.id)
+    )
+    return [
+        row
+        for row in candidates
+        if _notification_matching_fields(
+            row,
+            participant_ids=participant_ids,
+            respondent_ids=respondent_ids,
+            email=email,
+        )
+    ]
+
+
+def _participant_matching_fields(
+    row: SchedulingParticipant,
+    *,
+    participant_id: str | None,
+    respondent_ids: set[str],
+    account_ids: set[str],
+    email: str | None,
+) -> list[str]:
+    fields = []
+    if participant_id and row.id == participant_id:
+        fields.append("id")
+    if row.respondent_id and row.respondent_id in respondent_ids:
+        fields.append("respondent_id")
+    if row.bound_account_id and row.bound_account_id in account_ids:
+        fields.append("bound_account_id")
+    if email and _normalized_email(row.email) == email:
+        fields.append("email")
+    return fields
+
+
+def _notification_matching_fields(
+    row: SchedulingNotification,
+    *,
+    participant_ids: set[str],
+    respondent_ids: set[str],
+    email: str | None,
+) -> list[str]:
+    fields = []
+    if row.participant_id and row.participant_id in participant_ids:
+        fields.append("participant_id")
+    recipient = str(row.recipient or "").strip()
+    if recipient in respondent_ids:
+        fields.append("recipient")
+    if email and _normalized_email(recipient) == email:
+        fields.append("recipient")
+    return list(dict.fromkeys(fields))
+
+
+def _matching_recipient(
+    value: object,
+    *,
+    respondent_ids: set[str],
+    email: str | None,
+) -> str | None:
+    candidate = str(value or "").strip()
+    if candidate in respondent_ids:
+        return candidate
+    if email and _normalized_email(candidate) == email:
+        return email
+    return None
+
+
+def _request_is_evidence(row: SchedulingRequest) -> bool:
+    return bool(
+        row.status in _TERMINAL_REQUEST_STATUSES
+        or row.handed_off_at
+        or row.cancelled_at
+        or row.deleted_at
+    )
+
+
+def _participant_is_evidence(
+    row: SchedulingParticipant,
+    notifications: Sequence[SchedulingNotification],
+) -> bool:
+    return bool(
+        row.poll_invitation_id
+        or row.responded_at
+        or row.response_comment
+        or row.self_enrollment_link_id
+        or row.self_enrollment_proof_hash
+        or row.status in {"responded", "removed"}
+        or notifications
+    )
+
+
+def _participant_is_unengaged(
+    row: SchedulingParticipant,
+    notifications: Sequence[SchedulingNotification],
+) -> bool:
+    return bool(
+        row.deleted_at is None
+        and row.status in {"draft", "invited"}
+        and row.poll_invitation_id is None
+        and row.responded_at is None
+        and not row.response_comment
+        and row.self_enrollment_link_id is None
+        and row.self_enrollment_proof_hash is None
+        and not notifications
+    )
+
+
+def _anonymize_unengaged_participant(
+    session: Session,
+    *,
+    tenant_id: str,
+    subject: DsarSubjectRef,
+    action: DsarErasureActionRef,
+    request_id: str,
+) -> DsarExecutionResultRef:
+    row = (
+        session.query(SchedulingParticipant)
+        .filter(
+            SchedulingParticipant.id == action.resource_id,
+            SchedulingParticipant.tenant_id == tenant_id,
+        )
+        .with_for_update()
+        .one_or_none()
+    )
+    if row is None:
+        return _result(
+            action,
+            "unchanged",
+            "The Scheduling participant was already absent.",
+            {"request_id": request_id},
+        )
+    if _participant_is_anonymized(row):
+        return _result(
+            action,
+            "unchanged",
+            "The Scheduling participant was already anonymized.",
+            {"request_id": request_id},
+        )
+    if not _participant_matches_subject(row, subject):
+        return _blocked(
+            action, "The Scheduling participant identity changed after planning."
+        )
+    request = (
+        session.query(SchedulingRequest)
+        .filter(
+            SchedulingRequest.id == row.request_id,
+            SchedulingRequest.tenant_id == tenant_id,
+        )
+        .with_for_update()
+        .one_or_none()
+    )
+    if request is None or _request_is_evidence(request):
+        return _blocked(
+            action,
+            "The Scheduling request is absent or became retained decision evidence.",
+        )
+    notifications = _participant_notifications(session, row)
+    if not _participant_is_unengaged(row, notifications):
+        return _blocked(
+            action,
+            "The participant gained invitation, response, enrollment, or notification evidence after planning.",
+        )
+
+    now = datetime.now(timezone.utc)
+    row.respondent_id = None
+    row.display_name = None
+    row.email = None
+    row.status = "removed"
+    row.poll_invitation_id = None
+    row.participation_gateway = None
+    row.self_enrollment_link_id = None
+    row.self_enrollment_proof_hash = None
+    row.bound_account_id = None
+    row.account_bound_at = None
+    row.last_invited_at = None
+    row.responded_at = None
+    row.response_comment = None
+    row.deleted_at = now
+    row.metadata_ = None
+    return _result(
+        action,
+        "executed",
+        "The unengaged Scheduling participant was anonymized and retired.",
+        {"request_id": request_id},
+    )
+
+
+def _participant_notifications(
+    session: Session,
+    row: SchedulingParticipant,
+) -> list[SchedulingNotification]:
+    conditions = [SchedulingNotification.participant_id == row.id]
+    email = _normalized_email(row.email)
+    if email:
+        conditions.append(func.lower(SchedulingNotification.recipient) == email)
+    return _bounded_rows(
+        session.query(SchedulingNotification)
+        .filter(
+            SchedulingNotification.tenant_id == row.tenant_id,
+            SchedulingNotification.request_id == row.request_id,
+            or_(*conditions),
+        )
+        .order_by(SchedulingNotification.id)
+    )
+
+
+def _participant_matches_subject(
+    row: SchedulingParticipant,
+    subject: DsarSubjectRef,
+) -> bool:
+    return bool(
+        _participant_matching_fields(
+            row,
+            participant_id=_scheduling_references(subject).get("participant"),
+            respondent_ids=_respondent_ids(subject),
+            account_ids=_account_ids(subject),
+            email=_subject_email(subject),
+        )
+    )
+
+
+def _participant_is_anonymized(row: SchedulingParticipant) -> bool:
+    return bool(
+        row.deleted_at is not None
+        and row.status == "removed"
+        and row.respondent_id is None
+        and row.display_name is None
+        and row.email is None
+        and row.poll_invitation_id is None
+        and row.self_enrollment_link_id is None
+        and row.self_enrollment_proof_hash is None
+        and row.bound_account_id is None
+        and row.response_comment is None
+        and row.metadata_ is None
+    )
+
+
+def _scheduling_references(subject: DsarSubjectRef) -> dict[str, str]:
+    aliases = {
+        "scheduling.request": "request",
+        "scheduling.request_id": "request",
+        "scheduling.participant": "participant",
+        "scheduling.participant_id": "participant",
+    }
+    references: dict[str, str] = {}
+    for key, target in aliases.items():
+        value = str(subject.external_references.get(key) or "").strip()
+        if value and target not in references:
+            references[target] = value
+    return references
+
+
+def _membership_ids(subject: DsarSubjectRef) -> set[str]:
+    values = [subject.membership_id]
+    values.extend(
+        subject.external_references.get(key)
+        for key in (
+            "scheduling.user",
+            "scheduling.membership",
+            "access.membership",
+            "membership_id",
+        )
+    )
+    return _identifiers(values)
+
+
+def _respondent_ids(subject: DsarSubjectRef) -> set[str]:
+    values = [subject.membership_id, subject.identity_id, subject.account_id]
+    values.extend(
+        subject.external_references.get(key)
+        for key in (
+            "scheduling.respondent",
+            "poll.respondent",
+            "access.membership",
+            "membership_id",
+            "identity_id",
+            "account_id",
+        )
+    )
+    return _identifiers(values)
+
+
+def _account_ids(subject: DsarSubjectRef) -> set[str]:
+    values = [subject.account_id]
+    values.extend(
+        subject.external_references.get(key)
+        for key in ("scheduling.account", "access.account", "account_id")
+    )
+    return _identifiers(values)
+
+
+def _identifiers(values: Sequence[object]) -> set[str]:
+    return {text for value in values if (text := str(value or "").strip())}
+
+
+def _subject_email(subject: DsarSubjectRef) -> str | None:
+    candidates = [subject.email]
+    candidates.extend(
+        subject.external_references.get(key)
+        for key in ("scheduling.email", "scheduling.participant_email")
+    )
+    normalized = {
+        email for value in candidates if (email := _normalized_email(value)) is not None
+    }
+    return normalized.pop() if len(normalized) == 1 else None
+
+
+def _normalized_email(value: object) -> str | None:
+    if not isinstance(value, str):
+        return None
+    normalized = value.strip().casefold()
+    return normalized or None
+
+
+def _bounded_text(value: str | None) -> str | None:
+    return value[:2_000] if value else None
+
+
+def _record(
+    resource_type: str,
+    resource_id: str,
+    category: str,
+    title: str,
+    data: Mapping[str, object],
+    *,
+    observed_at: datetime | None = None,
+    immutable: bool = False,
+    retention_reason: str | None = None,
+    source_path: str | None = None,
+) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="scheduling",
+        module_id="scheduling",
+        resource_type=resource_type,
+        resource_id=resource_id,
+        category=category,
+        title=title,
+        data=data,
+        observed_at=observed_at,
+        immutable_evidence=immutable,
+        retention_reason=retention_reason,
+        source_path=source_path,
+    )
+
+
+def _action(
+    action_id: str,
+    kind: str,
+    record: DsarRecordRef,
+    title: str,
+    rationale: str,
+    *,
+    executable: bool,
+    irreversible: bool = False,
+    metadata: Mapping[str, object] | None = None,
+) -> DsarErasureActionRef:
+    return DsarErasureActionRef(
+        action_id=action_id,
+        provider_id="scheduling",
+        module_id="scheduling",
+        kind=kind,  # type: ignore[arg-type]
+        resource_type=record.resource_type,
+        resource_id=record.resource_id,
+        title=title,
+        rationale=rationale,
+        executable=executable,
+        irreversible=irreversible,
+        metadata=metadata or {},
+    )
+
+
+def _result(
+    action: DsarErasureActionRef,
+    status: str,
+    summary: str,
+    evidence: Mapping[str, object] | None = None,
+) -> DsarExecutionResultRef:
+    return DsarExecutionResultRef(
+        action_id=action.action_id,
+        status=status,  # type: ignore[arg-type]
+        summary=summary,
+        evidence=evidence or {},
+    )
+
+
+def _blocked(action: DsarErasureActionRef, summary: str) -> DsarExecutionResultRef:
+    return _result(action, "blocked", summary)
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Scheduling DSAR provider requires a SQLAlchemy session.")
+    return value
+
+
+def _bounded_rows(query: object) -> list[object]:
+    rows = query.limit(_MAX_RECORDS + 1).all()  # type: ignore[attr-defined]
+    if len(rows) > _MAX_RECORDS:
+        raise ValueError(
+            "Scheduling DSAR match limit exceeded; narrow the subject selectors."
+        )
+    return rows
+
+
+def _iso(value: datetime | None) -> str | None:
+    if value is None:
+        return None
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=timezone.utc)
+    return value.isoformat()
+
+
+__all__ = ["SCHEDULING_DSAR_CAPABILITY", "SchedulingDsarProvider"]

src/govoplan_scheduling/backend/manifest.py

@@ -6,8 +6,10 @@ from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPA
 from govoplan_core.core.calendar import CAPABILITY_CALENDAR_SCHEDULING
 from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard
 from govoplan_core.core.modules import (
+    CapabilityDocumentation,
     DocumentationTopic,
     DocumentationCondition,
+    DocumentationLink,
     FrontendModule,
     FrontendRoute,
     MigrationSpec,
@@ -36,10 +38,11 @@ from govoplan_core.core.policy import CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_P
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_scheduling.backend.db import models as scheduling_models  # noqa: F401 - populate Scheduling ORM metadata
+from govoplan_scheduling.backend.dsar_provider import SCHEDULING_DSAR_CAPABILITY
 
 MODULE_ID = "scheduling"
 MODULE_NAME = "Scheduling"
-MODULE_VERSION = "0.1.18"
+MODULE_VERSION = "0.1.19"
 READ_SCOPE = "scheduling:schedule:read"
 WRITE_SCOPE = "scheduling:schedule:write"
 ADMIN_SCOPE = "scheduling:schedule:admin"
@@ -102,6 +105,63 @@ DOCUMENTATION = (
         related_modules=("poll", "evaluation", "calendar", "appointments", "mail", "notifications", "portal"),
         metadata={"seed": True},
     ),
+    DocumentationTopic(
+        id="scheduling.privacy.data-subject-requests",
+        title="Review Scheduling data in a data-subject request",
+        summary="Collect tenant-scoped participation and coordination metadata while leaving Poll responses and Calendar effects with their owners.",
+        body=(
+            "Scheduling's DSAR provider searches the effective tenant by normalized participant email, membership, identity or bound-account references, and namespaced Scheduling request or participant references. "
+            "It isolates the matching participant, request and candidate-slot context, and matching notification envelopes. It does not export Poll or invitation identifiers, participation-gateway state, reusable enrollment links or proof hashes, anonymous-password hashes, Calendar event or hold identifiers, free/busy conflict detail, notification payloads or errors, token material, opaque metadata, or unrelated participants. Poll remains authoritative for actual availability choices and response-retirement evidence; Calendar remains authoritative for event, hold, and synchronization state. "
+            "Decided, handed-off, cancelled, archived, responded, notified, or removed state is retained with a reason. Active shared scheduling content requires coordinated manual review. The provider can anonymize and retire only a participant who has no invitation, response, enrollment, notification, or terminal decision evidence, and revalidates all of those conditions under tenant-bound row locks before acting."
+        ),
+        layer="configured",
+        documentation_types=("admin",),
+        audience=("privacy_officer", "scheduling_manager", "records_manager", "operator"),
+        order=5,
+        conditions=(
+            DocumentationCondition(
+                required_modules=("scheduling", "access"),
+                any_scopes=(
+                    "access:privacy:read",
+                    "access:privacy:manage",
+                    "access:privacy:erase",
+                ),
+            ),
+        ),
+        links=(
+            DocumentationLink(
+                label="Data-subject requests",
+                href="/admin?section=tenant-data-subject-requests",
+                kind="runtime",
+            ),
+            DocumentationLink(
+                label="Scheduling module guide",
+                href="govoplan-scheduling/README.md",
+                kind="repository",
+            ),
+        ),
+        related_modules=("access", "audit", "poll", "calendar", "notifications"),
+        metadata={
+            "kind": "workflow",
+            "route": "/admin?section=tenant-data-subject-requests",
+            "screen": "Data-subject requests",
+            "help_contexts": ["admin.privacy.data-subject-requests"],
+            "prerequisites": [
+                "The privacy request and Scheduling selectors have been independently authorized and corroborated.",
+                "The reviewer can coordinate with Poll and Calendar owners when a response or event is involved.",
+            ],
+            "steps": [
+                "Run the Scheduling provider search and review participant, request, slot, and notification dispositions.",
+                "Run the Poll provider for actual availability choices and the Calendar provider for event or hold state.",
+                "Retain terminal decision and delivery evidence with its reason.",
+                "Execute anonymization only for an approved participant classified as unengaged after revalidation.",
+            ],
+            "limitations": [
+                "Poll choices and invitation evidence are not copied into Scheduling's export.",
+                "Participant erasure remains manual whenever shared responses, delivery, self-enrollment, Calendar effects, or terminal decisions exist.",
+            ],
+        },
+    ),
     DocumentationTopic(
         id="scheduling.find-and-decide-meeting-time",
         title="Find and decide a meeting time",
@@ -265,6 +325,13 @@ def _scheduling_router(context: ModuleContext):
     return router
 
 
+def _scheduling_dsar_provider(context: ModuleContext) -> object:
+    del context
+    from govoplan_scheduling.backend.dsar_provider import SchedulingDsarProvider
+
+    return SchedulingDsarProvider()
+
+
 def _public_tenant_resolver(request: object, session: object) -> str | None:
     path_params = getattr(request, "path_params", {})
     request_id = str(path_params.get("request_id") or "").strip()
@@ -344,6 +411,7 @@ manifest = ModuleManifest(
     provides_interfaces=(
         ModuleInterfaceProvider(name="scheduling.candidate_slots", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="scheduling.decision_handoff", version=MODULE_VERSION),
+        ModuleInterfaceProvider(name=SCHEDULING_DSAR_CAPABILITY, version="0.1.0"),
     ),
     requires_interfaces=(
         ModuleInterfaceRequirement(name="poll.option_ordering", version_min="0.1.11", version_max_exclusive="0.2.0"),
@@ -434,6 +502,18 @@ manifest = ModuleManifest(
         ),
     ),
     documentation=DOCUMENTATION,
+    capability_factories={
+        SCHEDULING_DSAR_CAPABILITY: _scheduling_dsar_provider,
+    },
+    capability_documentation={
+        SCHEDULING_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Scheduling data-subject request provider",
+            summary="Finds isolated Scheduling participation and coordination metadata and classifies governed erasure actions.",
+            contract_version="0.1.0",
+            documentation_types=("admin",),
+            audience=("privacy_officer", "scheduling_manager", "records_manager"),
+        ),
+    },
     architecture=declared_module_architecture(
         layer="communication_participation",
         kind="domain",

tests/test_dsar_provider.py

@@ -0,0 +1,580 @@
+from __future__ import annotations
+
+import unittest
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_access.backend.db.models import Account, Group, User
+from govoplan_core.core.change_sequence import ChangeSequenceEntry
+from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef
+from govoplan_core.db.base import Base
+from govoplan_core.privacy.dsar_workflow import (
+    DataSubjectRequest,
+    create_data_subject_request,
+    execute_data_subject_erasure,
+    plan_data_subject_erasure,
+    search_data_subject_request,
+)
+from govoplan_scheduling.backend.db.models import (
+    SchedulingCandidateSlot,
+    SchedulingNotification,
+    SchedulingParticipant,
+    SchedulingPublicEnrollmentLink,
+    SchedulingRequest,
+)
+from govoplan_scheduling.backend.dsar_provider import (
+    SCHEDULING_DSAR_CAPABILITY,
+    SchedulingDsarProvider,
+)
+from govoplan_scheduling.backend.manifest import manifest
+
+
+class _Registry:
+    def __init__(
+        self,
+        provider: SchedulingDsarProvider,
+        *,
+        scheduling_active: bool = True,
+    ) -> None:
+        self.provider = provider
+        self.scheduling_active = scheduling_active
+
+    def capability_names(self):
+        return (SCHEDULING_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        self._assert_capability(name)
+        return "scheduling"
+
+    def tenant_entitlement_resolver(self):
+        scheduling_active = self.scheduling_active
+
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type(
+                    "State",
+                    (),
+                    {
+                        "effective_modules": (
+                            ("scheduling",) if scheduling_active else ()
+                        )
+                    },
+                )()
+
+        return _Resolver()
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del session, kwargs
+        self._assert_capability(name)
+        return self.provider
+
+    def manifests(self):
+        return (type("Manifest", (), {"id": "scheduling"})(),)
+
+    @staticmethod
+    def _assert_capability(name: str) -> None:
+        if name != SCHEDULING_DSAR_CAPABILITY:
+            raise KeyError(name)
+
+
+class SchedulingDsarProviderTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:", future=True)
+        Base.metadata.create_all(
+            bind=self.engine,
+            tables=[
+                Account.__table__,
+                User.__table__,
+                Group.__table__,
+                ChangeSequenceEntry.__table__,
+                DataSubjectRequest.__table__,
+                SchedulingRequest.__table__,
+                SchedulingPublicEnrollmentLink.__table__,
+                SchedulingCandidateSlot.__table__,
+                SchedulingParticipant.__table__,
+                SchedulingNotification.__table__,
+            ],
+        )
+        self.session = sessionmaker(bind=self.engine, future=True)()
+        now = datetime.now(timezone.utc)
+        account = Account(
+            id="account-1",
+            email="subject@example.test",
+            normalized_email="subject@example.test",
+            display_name="Subject",
+        )
+        other_account = Account(
+            id="account-2",
+            email="other@example.test",
+            normalized_email="other@example.test",
+            display_name="Other",
+        )
+        self.user = User(
+            id="membership-1",
+            tenant_id="tenant-1",
+            account_id=account.id,
+            email="subject@example.test",
+            display_name="Subject",
+        )
+        other_user = User(
+            id="membership-2",
+            tenant_id="tenant-1",
+            account_id=other_account.id,
+            email="other@example.test",
+            display_name="Other",
+        )
+        self.request = SchedulingRequest(
+            id="request-active",
+            tenant_id="tenant-1",
+            title="Choose an appointment",
+            description="Scheduling context visible to the participant",
+            location="Town hall",
+            status="collecting",
+            poll_id="poll-id-do-not-export",
+            organizer_user_id=other_user.id,
+            deadline_at=now + timedelta(days=3),
+            anonymous_password_protection_enabled=True,
+            anonymous_password_hash="password-hash-do-not-export",
+            calendar_integration_enabled=True,
+            calendar_id="calendar-id-do-not-export",
+            calendar_hold_enabled=True,
+            calendar_event_id="calendar-event-id-do-not-export",
+            metadata_={"secret": "request-metadata-do-not-export"},
+        )
+        slot = SchedulingCandidateSlot(
+            id="slot-subject",
+            tenant_id="tenant-1",
+            request_id=self.request.id,
+            poll_option_id="poll-option-id-do-not-export",
+            label="Tuesday morning",
+            description="First option",
+            start_at=now + timedelta(days=1),
+            end_at=now + timedelta(days=1, hours=1),
+            timezone="Europe/Berlin",
+            location="Town hall",
+            position=0,
+            freebusy_checked_at=now,
+            freebusy_status="busy",
+            freebusy_conflicts=[{"person": "Unrelated conflict person do not export"}],
+            tentative_hold_event_id="hold-event-id-do-not-export",
+            metadata_={"secret": "slot-metadata-do-not-export"},
+        )
+        self.engaged = SchedulingParticipant(
+            id="participant-engaged",
+            tenant_id="tenant-1",
+            request_id=self.request.id,
+            respondent_id=self.user.id,
+            display_name="Subject Person",
+            email="Subject@Example.Test",
+            participant_type="internal",
+            required=True,
+            status="responded",
+            poll_invitation_id="poll-invitation-id-do-not-export",
+            participation_gateway="public-gateway-do-not-export",
+            self_enrollment_proof_hash="proof-hash-do-not-export",
+            bound_account_id=account.id,
+            account_bound_at=now,
+            last_invited_at=now,
+            responded_at=now,
+            response_comment="Subject response comment",
+            metadata_={"secret": "participant-metadata-do-not-export"},
+        )
+        self.unengaged = SchedulingParticipant(
+            id="participant-unengaged",
+            tenant_id="tenant-1",
+            request_id=self.request.id,
+            respondent_id=self.user.id,
+            display_name="Subject duplicate draft",
+            email=None,
+            participant_type="external",
+            required=False,
+            status="draft",
+            metadata_={"directory": "internal-directory-data-do-not-export"},
+        )
+        unrelated = SchedulingParticipant(
+            id="participant-other",
+            tenant_id="tenant-1",
+            request_id=self.request.id,
+            respondent_id=other_user.id,
+            display_name="Unrelated Person",
+            email="other@example.test",
+            status="responded",
+            poll_invitation_id="other-invitation-do-not-export",
+            responded_at=now,
+            response_comment="Unrelated response do not export",
+        )
+        notification = SchedulingNotification(
+            id="notification-subject",
+            tenant_id="tenant-1",
+            request_id=self.request.id,
+            participant_id=self.engaged.id,
+            event_kind="invitation",
+            channel="mail",
+            recipient="subject@example.test",
+            status="sent",
+            payload={
+                "private": "notification-payload-do-not-export",
+                "token": "notification-token-do-not-export",
+            },
+            error="notification-error-do-not-export",
+            sent_at=now,
+            metadata_={"secret": "notification-metadata-do-not-export"},
+        )
+        unrelated_notification = SchedulingNotification(
+            id="notification-other",
+            tenant_id="tenant-1",
+            request_id=self.request.id,
+            participant_id=unrelated.id,
+            event_kind="decision",
+            channel="mail",
+            recipient="other@example.test",
+            status="sent",
+            payload={"private": "other-notification-do-not-export"},
+            sent_at=now,
+        )
+        organizer_request = SchedulingRequest(
+            id="request-organized",
+            tenant_id="tenant-1",
+            title="Subject organized meeting",
+            status="draft",
+            poll_id="organizer-poll-do-not-export",
+            organizer_user_id=self.user.id,
+        )
+        organizer_slot = SchedulingCandidateSlot(
+            id="slot-organized",
+            tenant_id="tenant-1",
+            request_id=organizer_request.id,
+            label="Organizer option",
+            start_at=now + timedelta(days=2),
+            end_at=now + timedelta(days=2, hours=1),
+        )
+        organizer_other_participant = SchedulingParticipant(
+            id="participant-organizer-other",
+            tenant_id="tenant-1",
+            request_id=organizer_request.id,
+            display_name="Organizer unrelated invitee do not export",
+            email="organizer-other@example.test",
+            status="draft",
+        )
+        unrelated_request = SchedulingRequest(
+            id="request-other",
+            tenant_id="tenant-1",
+            title="Unrelated request do not export",
+            status="collecting",
+            poll_id="unrelated-poll-do-not-export",
+            organizer_user_id=other_user.id,
+        )
+        tenant_two_request = SchedulingRequest(
+            id="request-tenant-2",
+            tenant_id="tenant-2",
+            title="Tenant two request do not export",
+            status="collecting",
+            poll_id="tenant-two-poll-do-not-export",
+        )
+        tenant_two_participant = SchedulingParticipant(
+            id="participant-tenant-2",
+            tenant_id="tenant-2",
+            request_id=tenant_two_request.id,
+            display_name="Tenant two subject",
+            email="subject@example.test",
+            status="draft",
+        )
+        self.session.add_all(
+            [
+                account,
+                other_account,
+                self.user,
+                other_user,
+                self.request,
+                slot,
+                self.engaged,
+                self.unengaged,
+                unrelated,
+                notification,
+                unrelated_notification,
+                organizer_request,
+                organizer_slot,
+                organizer_other_participant,
+                unrelated_request,
+                tenant_two_request,
+                tenant_two_participant,
+            ]
+        )
+        self.session.commit()
+        self.provider = SchedulingDsarProvider()
+        self.subject = DsarSubjectRef(
+            account_id=account.id,
+            membership_id=self.user.id,
+            email="subject@example.test",
+        )
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_manifest_publishes_protocol_conforming_provider(self) -> None:
+        provided_names = {item.name for item in manifest.provides_interfaces}
+        self.assertIn(SCHEDULING_DSAR_CAPABILITY, provided_names)
+        provider = manifest.capability_factories[SCHEDULING_DSAR_CAPABILITY](None)
+        self.assertIsInstance(provider, DsarProvider)
+        self.assertIn(
+            "scheduling.privacy.data-subject-requests",
+            {topic.id for topic in manifest.documentation},
+        )
+
+    def test_search_is_tenant_scoped_minimized_and_participant_specific(self) -> None:
+        records = self._records()
+        resource_types = {record.resource_type for record in records}
+        self.assertTrue(
+            {
+                "scheduling_request",
+                "scheduling_candidate_slot",
+                "scheduling_participant",
+                "scheduling_notification",
+            }.issubset(resource_types)
+        )
+        self.assertEqual(
+            {"participant-engaged", "participant-unengaged"},
+            {
+                record.resource_id
+                for record in records
+                if record.resource_type == "scheduling_participant"
+            },
+        )
+        engaged = next(
+            record for record in records if record.resource_id == "participant-engaged"
+        )
+        self.assertEqual("Subject response comment", engaged.data["response_comment"])
+
+        serialized = repr([record.to_dict() for record in records])
+        for hidden in (
+            "participant-other",
+            "Unrelated Person",
+            "other@example.test",
+            "Unrelated response do not export",
+            "notification-other",
+            "other-notification-do-not-export",
+            "participant-organizer-other",
+            "Organizer unrelated invitee do not export",
+            "request-other",
+            "Unrelated request do not export",
+            "request-tenant-2",
+            "Tenant two request do not export",
+            "participant-tenant-2",
+            "poll-id-do-not-export",
+            "password-hash-do-not-export",
+            "calendar-id-do-not-export",
+            "calendar-event-id-do-not-export",
+            "request-metadata-do-not-export",
+            "poll-option-id-do-not-export",
+            "Unrelated conflict person do not export",
+            "hold-event-id-do-not-export",
+            "slot-metadata-do-not-export",
+            "poll-invitation-id-do-not-export",
+            "public-gateway-do-not-export",
+            "proof-hash-do-not-export",
+            "participant-metadata-do-not-export",
+            "internal-directory-data-do-not-export",
+            "notification-payload-do-not-export",
+            "notification-token-do-not-export",
+            "notification-error-do-not-export",
+            "notification-metadata-do-not-export",
+            "organizer-poll-do-not-export",
+        ):
+            self.assertNotIn(hidden, serialized)
+
+    def test_conflicting_email_references_fail_closed_for_participant_data(
+        self,
+    ) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                email="subject@example.test",
+                external_references={"scheduling.email": "other@example.test"},
+            ),
+        )
+
+        self.assertEqual((), records)
+
+    def test_plan_retains_evidence_and_only_anonymizes_unengaged_participant(
+        self,
+    ) -> None:
+        actions = self.provider.plan_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self.subject,
+            records=self._records(),
+        )
+
+        self.assertTrue({"retain", "manual_review"}.issubset({a.kind for a in actions}))
+        self.assertTrue(
+            any(
+                action.action_id
+                == "scheduling:retain:scheduling_participant:participant-engaged"
+                for action in actions
+            )
+        )
+        self.assertEqual(
+            {"scheduling:anonymize:scheduling_participant:participant-unengaged"},
+            {action.action_id for action in actions if action.executable},
+        )
+
+    def test_execution_is_revalidated_tenant_bound_and_idempotent(self) -> None:
+        action = self._anonymize_action()
+        wrong_tenant = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-2",
+            subject=self.subject,
+            actions=(action,),
+            request_id="dsar-wrong-tenant",
+        )
+        self.assertEqual("blocked", wrong_tenant[0].status)
+
+        first = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self.subject,
+            actions=(action,),
+            request_id="dsar-scheduling-1",
+        )
+        self.assertEqual("executed", first[0].status)
+        self.session.flush()
+        self.assertEqual("removed", self.unengaged.status)
+        self.assertIsNotNone(self.unengaged.deleted_at)
+        self.assertIsNone(self.unengaged.display_name)
+        self.assertIsNone(self.unengaged.email)
+        self.assertIsNone(self.unengaged.respondent_id)
+        self.assertIsNone(self.unengaged.metadata_)
+
+        repeated = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self.subject,
+            actions=(action,),
+            request_id="dsar-scheduling-1",
+        )
+        self.assertEqual("unchanged", repeated[0].status)
+
+    def test_execution_blocks_when_evidence_appears_after_planning(self) -> None:
+        action = self._anonymize_action()
+        self.session.add(
+            SchedulingNotification(
+                id="notification-late",
+                tenant_id="tenant-1",
+                request_id=self.request.id,
+                participant_id=self.unengaged.id,
+                event_kind="invitation",
+                recipient=self.unengaged.email,
+                status="pending",
+                payload={},
+            )
+        )
+        self.session.flush()
+
+        result = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self.subject,
+            actions=(action,),
+            request_id="dsar-stale",
+        )
+
+        self.assertEqual("blocked", result[0].status)
+        self.assertIsNone(self.unengaged.deleted_at)
+
+    def test_core_workflow_discovers_active_provider_and_skips_it_when_disabled(
+        self,
+    ) -> None:
+        request = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-SCHEDULING-1",
+            request_kind="access_and_erasure",
+            subject=self.subject,
+            purpose="Respond to an authorized privacy request.",
+            legal_basis="Article 15 and 17 GDPR",
+            due_at=None,
+            requested_by_account_id="privacy-officer",
+        )
+        self.session.commit()
+        registry = _Registry(self.provider)
+
+        search_data_subject_request(
+            self.session,
+            registry=registry,
+            row=request,
+            expected_revision=1,
+        )
+        self.assertEqual("searched", request.status)
+        self.assertEqual(["scheduling"], request.coverage["covered_modules"])
+        plan_data_subject_erasure(
+            self.session,
+            registry=registry,
+            row=request,
+            expected_revision=2,
+        )
+        executable_ids = [
+            action["action_id"]
+            for action in request.erasure_plan["actions"]
+            if action["executable"]
+        ]
+        execute_data_subject_erasure(
+            self.session,
+            registry=registry,
+            row=request,
+            expected_revision=3,
+            action_ids=executable_ids,
+        )
+        self.assertEqual("completed", request.status)
+
+        disabled = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-SCHEDULING-DISABLED",
+            request_kind="access",
+            subject=self.subject,
+            purpose="Verify disabled-module coverage.",
+            legal_basis="Article 15 GDPR",
+            due_at=None,
+            requested_by_account_id="privacy-officer",
+        )
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider, scheduling_active=False),
+            row=disabled,
+            expected_revision=1,
+        )
+
+        self.assertEqual(0, disabled.search_result["record_count"])
+        self.assertEqual(
+            [SCHEDULING_DSAR_CAPABILITY],
+            disabled.coverage["inactive_provider_capabilities"],
+        )
+
+    def _records(self):
+        return self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self.subject,
+        )
+
+    def _anonymize_action(self):
+        return next(
+            action
+            for action in self.provider.plan_erasure(
+                self.session,
+                tenant_id="tenant-1",
+                subject=self.subject,
+                records=self._records(),
+            )
+            if action.executable
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()