feat(scheduling): retire replaced participant responses

README.md

@@ -127,7 +127,8 @@ poll-backed scheduling requests:
 - request lifecycle APIs: draft, collecting, closed, decided, handed off, cancelled
 - result summaries sourced from Poll response aggregation
 - optional Calendar free/busy checks, tentative holds, and final event creation
-- notification outbox jobs for invitations, reminders, decisions, and cancellations
+- notification outbox jobs for invitations, reminders, decisions,
+  cancellations, and participant access changes
 - a first Scheduling WebUI package with request creation, slot matrix, Calendar
   actions, decisions, and notification-job creation
 
@@ -225,6 +226,22 @@ Scheduling administrator can use these actions. If notification delivery is
 not installed, `send` fails before rotating the current link and the organizer
 can use `copy` for separate distribution.
 
+Participant edits carry a semantic revision so a stale organizer form cannot
+overwrite an invitation or response change. Corrections that retain a stable
+account or directory identity update the existing participant. A display-name
+correction keeps its invitation; changing a delivery email revokes the stale
+link but keeps a response tied to the unchanged account identity.
+
+Changing the canonical identity creates a new participant instead of assigning
+the former participant's response to another person. In the same transaction,
+Scheduling revokes the old invitation and Poll soft-deletes every matching
+live response. Poll retains the answers and a bounded retirement record for
+audit, while result summaries and capacity checks immediately exclude them.
+Removing an invited or responding participant follows the same retirement
+path. Scheduling records privacy-safe audit facts and queues a removal or
+replacement notice to the former recipient without placing email addresses or
+response contents in the audit event.
+
 ## FieldLabel omission register
 
 Scheduling uses the shared `FieldLabel` for form controls that need explanatory

src/govoplan_scheduling/backend/router.py

@@ -67,7 +67,7 @@ from govoplan_scheduling.backend.service import (
     submit_scheduling_availability,
     submit_public_scheduling_participation,
     update_scheduling_candidate_slot,
-    update_scheduling_request_with_invitation_tokens,
+    update_scheduling_request_with_change_log,
 )
 
 
@@ -431,7 +431,7 @@ def api_update_scheduling_request(
         request_id=request_id,
     )
     try:
-        request, _invitation_tokens = update_scheduling_request_with_invitation_tokens(
+        request, participant_mutations = update_scheduling_request_with_change_log(
             session,
             tenant_id=principal.tenant_id,
             request_id=request_id,
@@ -439,6 +439,21 @@ def api_update_scheduling_request(
         )
     except SchedulingError as exc:
         raise _scheduling_http_error(exc) from exc
+    for mutation in participant_mutations:
+        _audit_invitation_action(
+            session,
+            principal=principal,
+            request_id=request_id,
+            participant_id=mutation.participant_id,
+            action=mutation.action,
+            details={
+                "replacement_participant_id": mutation.replacement_participant_id,
+                "changed_fields": list(mutation.changed_fields),
+                "invitation_revoked": mutation.invitation_revoked,
+                "retired_response_count": mutation.retired_response_count,
+                "notification_id": mutation.notification_id,
+            },
+        )
     response = _request_response(request, principal=principal)
     session.commit()
     return response

src/govoplan_scheduling/backend/schemas.py

@@ -132,6 +132,20 @@ class SchedulingCandidateSlotReconcileInput(SchedulingCandidateSlotInput):
 
 class SchedulingParticipantReconcileInput(SchedulingParticipantInput):
     id: str | None = Field(default=None, max_length=36)
+    revision: str | None = Field(
+        default=None,
+        min_length=64,
+        max_length=64,
+        pattern=r"^[0-9a-f]{64}$",
+    )
+
+    @model_validator(mode="after")
+    def validate_existing_revision(self) -> "SchedulingParticipantReconcileInput":
+        if self.id is not None and self.revision is None:
+            raise ValueError("revision is required for an existing scheduling participant")
+        if self.id is None and self.revision is not None:
+            raise ValueError("revision can only be supplied for an existing scheduling participant")
+        return self
 
 
 class SchedulingRequestCreateRequest(BaseModel):
@@ -252,6 +266,7 @@ class SchedulingCandidateSlotResponse(BaseModel):
 
 class SchedulingParticipantResponse(BaseModel):
     id: str
+    revision: str | None = None
     is_current_participant: bool = False
     respondent_id: str | None = None
     display_name: str | None = None

src/govoplan_scheduling/backend/service.py

@@ -24,8 +24,11 @@ from govoplan_core.core.poll import (
     PollOptionUpdateCommand,
     PollOptionRequest,
     PollResponseRef,
+    PollResponseRetirementCommand,
+    PollResponseRetirementProvider,
     PollSchedulingProvider,
     PollUpdateCommand,
+    poll_response_retirement_provider,
     poll_scheduling_provider,
 )
 from govoplan_core.core.poll_participation import (
@@ -100,13 +103,31 @@ class SchedulingInvitationActionResult:
     replayed: bool = False
 
 
+@dataclass(frozen=True, slots=True)
+class SchedulingParticipantMutation:
+    action: str
+    participant_id: str
+    replacement_participant_id: str | None = None
+    changed_fields: tuple[str, ...] = ()
+    invitation_revoked: bool = False
+    retired_response_count: int = 0
+    notification_id: str | None = None
+
+
 WORKFLOW_DRAFT = "draft"
 WORKFLOW_COLLECTING = "collecting_availability"
 WORKFLOW_CLOSED = "availability_closed"
 WORKFLOW_DECIDED = "slot_decided"
 WORKFLOW_HANDED_OFF = "handed_off"
 WORKFLOW_CANCELLED = "cancelled"
-NOTIFICATION_KINDS = {"invitation", "reminder", "decision", "cancellation"}
+NOTIFICATION_KINDS = {
+    "invitation",
+    "reminder",
+    "decision",
+    "cancellation",
+    "participant_removed",
+    "participant_replaced",
+}
 PUBLIC_PARTICIPATION_POLICY_UNAVAILABLE_REASON = (
     "Poll signed-participation policy enforcement is unavailable; restricted "
     "public links are disabled until the Poll gateway contract is installed."
@@ -157,6 +178,29 @@ def scheduling_slot_revision(slot: SchedulingCandidateSlot) -> str:
     return hashlib.sha256(encoded).hexdigest()
 
 
+def scheduling_participant_revision(participant: SchedulingParticipant) -> str:
+    """Hash the participant state that makes an identity edit security-sensitive."""
+
+    snapshot = {
+        "respondent_id": participant.respondent_id,
+        "display_name": participant.display_name,
+        "email": participant.email,
+        "participant_type": participant.participant_type,
+        "required": participant.required,
+        "status": participant.status,
+        "poll_invitation_id": participant.poll_invitation_id,
+        "participation_gateway": participant.participation_gateway,
+        "metadata": participant.metadata_ or {},
+    }
+    encoded = json.dumps(
+        _jsonable(snapshot),
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
 def _now() -> datetime:
     return utcnow()
 
@@ -343,6 +387,10 @@ def _poll_participation_provider() -> PollParticipationGatewayProvider | None:
     return poll_participation_gateway_provider(get_registry())
 
 
+def _poll_response_retirement_provider() -> PollResponseRetirementProvider | None:
+    return poll_response_retirement_provider(get_registry())
+
+
 def _public_participation_gateway(request_id: str) -> PollResponseGatewayRef:
     return PollResponseGatewayRef(
         module_id="scheduling",
@@ -801,7 +849,14 @@ def _notification_dispatch_request(
                 else f"/poll/public/{invitation_token}"
             )
             if notification.event_kind == "invitation" and invitation_token
-            else f"/scheduling?request_id={request.id}"
+            else (
+                None
+                if notification.event_kind in {
+                    "participant_removed",
+                    "participant_replaced",
+                }
+                else f"/scheduling?request_id={request.id}"
+            )
         ),
         payload=_jsonable(payload),
         metadata={
@@ -818,6 +873,8 @@ def _notification_subject(request: SchedulingRequest, event_kind: str) -> str:
         "reminder": "Scheduling reminder",
         "decision": "Scheduling decision",
         "cancellation": "Scheduling cancelled",
+        "participant_removed": "Scheduling access removed",
+        "participant_replaced": "Scheduling access changed",
     }
     return f"{labels.get(event_kind, 'Scheduling update')}: {request.title}"
 
@@ -847,6 +904,16 @@ def _notification_body_intro(request: SchedulingRequest, event_kind: str) -> str
         return f"A final slot was selected for {request.title}."
     if event_kind == "cancellation":
         return f"The scheduling request {request.title} was cancelled."
+    if event_kind == "participant_replaced":
+        return (
+            "Your participant identity for this scheduling request was "
+            "replaced and its previous response link has been revoked."
+        )
+    if event_kind == "participant_removed":
+        return (
+            "You were removed from this scheduling request and your previous "
+            "response link has been revoked."
+        )
     return f"There is an update for {request.title}."
 
 
@@ -2054,7 +2121,11 @@ def _update_scheduling_request_with_invitation_tokens(
     tenant_id: str,
     request_id: str,
     payload: SchedulingRequestUpdateRequest,
-) -> tuple[SchedulingRequest, dict[str, str]]:
+) -> tuple[
+    SchedulingRequest,
+    dict[str, str],
+    tuple[SchedulingParticipantMutation, ...],
+]:
     request = _lock_scheduling_request(
         session,
         tenant_id=tenant_id,
@@ -2158,6 +2229,7 @@ def _update_scheduling_request_with_invitation_tokens(
     if payload.metadata is not None:
         request.metadata_ = payload.metadata
     invitation_tokens: dict[str, str] = {}
+    participant_mutations: tuple[SchedulingParticipantMutation, ...] = ()
     slots_changed = False
     if payload.slots is not None:
         slots_changed = _reconcile_scheduling_slots(
@@ -2166,7 +2238,7 @@ def _update_scheduling_request_with_invitation_tokens(
             supplied_slots=payload.slots,
         )
     if payload.participants is not None:
-        invitation_tokens = _reconcile_scheduling_participants(
+        invitation_tokens, participant_mutations = _reconcile_scheduling_participants(
             session,
             request=request,
             supplied_participants=payload.participants,
@@ -2225,7 +2297,7 @@ def _update_scheduling_request_with_invitation_tokens(
             reset_missing=True,
         )
     session.flush()
-    return request, invitation_tokens
+    return request, invitation_tokens, participant_mutations
 
 
 def update_scheduling_request(
@@ -2235,11 +2307,13 @@ def update_scheduling_request(
     request_id: str,
     payload: SchedulingRequestUpdateRequest,
 ) -> SchedulingRequest:
-    request, _invitation_tokens = _update_scheduling_request_with_invitation_tokens(
+    request, _invitation_tokens, _participant_mutations = (
-        session,
+        _update_scheduling_request_with_invitation_tokens(
-        tenant_id=tenant_id,
+            session,
-        request_id=request_id,
+            tenant_id=tenant_id,
-        payload=payload,
+            request_id=request_id,
+            payload=payload,
+        )
     )
     return request
 
@@ -2253,12 +2327,38 @@ def update_scheduling_request_with_invitation_tokens(
 ) -> tuple[SchedulingRequest, dict[str, str]]:
     """Compatibility wrapper; explicit invitation actions own raw tokens."""
 
-    return _update_scheduling_request_with_invitation_tokens(
+    request, invitation_tokens, _participant_mutations = (
-        session,
+        _update_scheduling_request_with_invitation_tokens(
-        tenant_id=tenant_id,
+            session,
-        request_id=request_id,
+            tenant_id=tenant_id,
-        payload=payload,
+            request_id=request_id,
+            payload=payload,
+        )
     )
+    return request, invitation_tokens
+
+
+def update_scheduling_request_with_change_log(
+    session: Session,
+    *,
+    tenant_id: str,
+    request_id: str,
+    payload: SchedulingRequestUpdateRequest,
+) -> tuple[
+    SchedulingRequest,
+    tuple[SchedulingParticipantMutation, ...],
+]:
+    """Update one request and expose privacy-safe participant audit facts."""
+
+    request, _invitation_tokens, participant_mutations = (
+        _update_scheduling_request_with_invitation_tokens(
+            session,
+            tenant_id=tenant_id,
+            request_id=request_id,
+            payload=payload,
+        )
+    )
+    return request, participant_mutations
 
 
 @dataclass(frozen=True, slots=True)
@@ -2598,13 +2698,115 @@ def _normalized_participant_identity(value: str | None) -> str | None:
 def _participant_identity_values(
     participant: SchedulingParticipant | SchedulingParticipantReconcileInput,
 ) -> tuple[str, ...]:
+    metadata = (
+        participant.metadata_
+        if isinstance(participant, SchedulingParticipant)
+        else participant.metadata
+    ) or {}
+    directory = metadata.get("directory_selection")
+    directory_identity = None
+    if isinstance(directory, dict):
+        reference_id = directory.get("reference_id")
+        kind = directory.get("kind")
+        source_module = directory.get("source_module")
+        if isinstance(reference_id, str) and reference_id.strip():
+            directory_identity = ":".join(
+                (
+                    "directory",
+                    str(source_module or "unknown").strip().casefold(),
+                    str(kind or "person").strip().casefold(),
+                    reference_id.strip(),
+                )
+            )
     identities = (
         _normalized_participant_identity(participant.respondent_id),
         _normalized_participant_identity(participant.email),
+        directory_identity,
     )
     return tuple(dict.fromkeys(value for value in identities if value is not None))
 
 
+def _canonical_participant_identity(
+    participant: SchedulingParticipant | SchedulingParticipantReconcileInput,
+) -> tuple[str, str] | None:
+    respondent_id = _normalized_participant_identity(participant.respondent_id)
+    if respondent_id and not respondent_id.startswith("scheduling-participant:"):
+        return ("respondent", respondent_id)
+    directory_identity = next(
+        (
+            identity
+            for identity in _participant_identity_values(participant)
+            if identity.startswith("directory:")
+        ),
+        None,
+    )
+    if directory_identity is not None:
+        return ("directory", directory_identity)
+    email = _normalized_participant_identity(participant.email)
+    if email is not None:
+        return ("email", email)
+    return None
+
+
+def _participant_identity_changed(
+    participant: SchedulingParticipant,
+    supplied: SchedulingParticipantReconcileInput,
+) -> bool:
+    return (
+        participant.participant_type != supplied.participant_type
+        or _canonical_participant_identity(participant)
+        != _canonical_participant_identity(supplied)
+    )
+
+
+def _participant_changed_fields(
+    participant: SchedulingParticipant,
+    supplied: SchedulingParticipantReconcileInput,
+) -> tuple[str, ...]:
+    comparisons = {
+        "respondent_id": (
+            _normalized_participant_identity(participant.respondent_id),
+            _normalized_participant_identity(supplied.respondent_id),
+        ),
+        "display_name": (participant.display_name, supplied.display_name),
+        "email": (
+            _normalized_participant_identity(participant.email),
+            _normalized_participant_identity(supplied.email),
+        ),
+        "participant_type": (
+            participant.participant_type,
+            supplied.participant_type,
+        ),
+        "required": (participant.required, supplied.required),
+        "metadata": (participant.metadata_ or {}, dict(supplied.metadata)),
+    }
+    return tuple(
+        field
+        for field, (current, requested) in comparisons.items()
+        if current != requested
+    )
+
+
+def _participant_response_identities(
+    participant: SchedulingParticipant,
+) -> tuple[str, ...]:
+    stored_response_identity = (participant.metadata_ or {}).get(
+        "poll_response_respondent_id"
+    )
+    candidates = (
+        participant.respondent_id,
+        participant.email,
+        stored_response_identity if isinstance(stored_response_identity, str) else None,
+    )
+    return tuple(
+        dict.fromkeys(
+            normalized
+            for value in candidates
+            if (normalized := _normalized_participant_identity(value)) is not None
+        )
+    )
+
+
 def _validate_participant_reconciliation(
     request: SchedulingRequest,
     current_by_id: dict[str, SchedulingParticipant],
@@ -2633,20 +2835,15 @@ def _validate_participant_reconciliation(
                 )
     for item in supplied_participants:
         if item.id is None:
+            if (item.respondent_id or "").startswith("scheduling-participant:"):
+                raise SchedulingError(
+                    "Server-owned scheduling participant identities cannot be supplied for a new participant"
+                )
             continue
         participant = current_by_id[item.id]
-        if participant.poll_invitation_id is None:
+        if item.revision != scheduling_participant_revision(participant):
-            continue
+            raise SchedulingConflictError(
-        identity_changed = (
+                "Scheduling participant changed while it was being edited; reload and try again"
-            _normalized_participant_identity(item.respondent_id)
-            != _normalized_participant_identity(participant.respondent_id)
-            or _normalized_participant_identity(item.email)
-            != _normalized_participant_identity(participant.email)
-            or item.display_name != participant.display_name
-        )
-        if identity_changed:
-            raise SchedulingError(
-                "An invited participant's identity, name, and email cannot be changed; remove and add the participant instead"
             )
 
 
@@ -2910,8 +3107,8 @@ def _reconcile_scheduling_participants(
     *,
     request: SchedulingRequest,
     supplied_participants: list[SchedulingParticipantReconcileInput],
-) -> dict[str, str]:
+) -> tuple[dict[str, str], tuple[SchedulingParticipantMutation, ...]]:
-    """Apply a complete participant collection and revoke removed links."""
+    """Reconcile participants without reassigning responses to new identities."""
 
     current_participants = _active_participants(request)
     current_by_id = {participant.id: participant for participant in current_participants}
@@ -2920,56 +3117,183 @@ def _reconcile_scheduling_participants(
         current_by_id,
         supplied_participants,
     )
-    supplied_ids = {
+    replacement_inputs = {
-        item.id
+        cast(str, item.id): item
         for item in supplied_participants
         if item.id is not None
+        and _participant_identity_changed(current_by_id[item.id], item)
+    }
+    retained_ids = {
+        item.id
+        for item in supplied_participants
+        if item.id is not None and item.id not in replacement_inputs
     }
     removed_participants = [
         participant
         for participant in current_participants
-        if participant.id not in supplied_ids
+        if participant.id not in retained_ids
     ]
+    participants_with_contact_link_changes = [
+        current_by_id[cast(str, item.id)]
+        for item in supplied_participants
+        if item.id is not None
+        and item.id not in replacement_inputs
+        and current_by_id[item.id].poll_invitation_id is not None
+        and _normalized_participant_identity(current_by_id[item.id].email)
+        != _normalized_participant_identity(item.email)
+    ]
+    if (removed_participants or participants_with_contact_link_changes) and request.poll_id is None:
+        raise SchedulingError("Scheduling request has no backing poll")
+    participation_provider = _poll_participation_provider()
     if (
-        any(participant.poll_invitation_id for participant in removed_participants)
+        any(
-        and _poll_participation_provider() is None
+            participant.poll_invitation_id
+            for participant in (
+                *removed_participants,
+                *participants_with_contact_link_changes,
+            )
+        )
+        and participation_provider is None
     ):
         raise SchedulingError(
             "Poll participation invitation revocation capability is unavailable"
         )
+    retirement_candidates = [
+        participant
+        for participant in removed_participants
+        if _participant_response_identities(participant)
+        or participant.poll_invitation_id is not None
+    ]
+    retirement_provider = _poll_response_retirement_provider()
+    if retirement_candidates and retirement_provider is None:
+        raise SchedulingError(
+            "Poll response retirement capability is unavailable; participants with possible responses cannot be removed safely"
+        )
 
+    mutations: list[SchedulingParticipantMutation] = []
     for item in supplied_participants:
-        if item.id is None:
+        if item.id is None or item.id in replacement_inputs:
             continue
         participant = current_by_id[item.id]
-        if participant.poll_invitation_id is None:
+        changed_fields = _participant_changed_fields(participant, item)
-            participant.respondent_id = item.respondent_id
+        if not changed_fields:
-            participant.display_name = item.display_name
+            continue
-            participant.email = item.email
+        invitation_revoked = False
+        if (
+            "email" in changed_fields
+            and participant.poll_invitation_id is not None
+        ):
+            if participation_provider is None:  # Guarded above.
+                raise SchedulingError(
+                    "Poll participation invitation revocation capability is unavailable"
+                )
+            try:
+                participation_provider.revoke_invitation(
+                    session,
+                    tenant_id=request.tenant_id,
+                    poll_id=cast(str, request.poll_id),
+                    invitation_id=participant.poll_invitation_id,
+                )
+            except PollCapabilityError as exc:
+                raise SchedulingError(str(exc)) from exc
+            participant.poll_invitation_id = None
+            participant.participation_gateway = None
+            if participant.status == "invited":
+                participant.status = "draft"
+            invitation_revoked = True
+        participant.respondent_id = item.respondent_id
+        participant.display_name = item.display_name
+        participant.email = item.email
         participant.participant_type = item.participant_type
         participant.required = item.required
         participant.metadata_ = dict(item.metadata)
+        mutations.append(
+            SchedulingParticipantMutation(
+                action=(
+                    "scheduling.participant_contact_updated"
+                    if "email" in changed_fields
+                    else "scheduling.participant_updated"
+                ),
+                participant_id=participant.id,
+                changed_fields=changed_fields,
+                invitation_revoked=invitation_revoked,
+            )
+        )
 
     invitation_tokens: dict[str, str] = {}
+    replacement_participants: dict[str, SchedulingParticipant] = {}
     for item in supplied_participants:
-        if item.id is not None:
+        if item.id is not None and item.id not in replacement_inputs:
             continue
+        replaced_participant = (
+            current_by_id[item.id]
+            if item.id is not None
+            else None
+        )
+        respondent_id = item.respondent_id
+        if (
+            replaced_participant is not None
+            and (respondent_id or "").startswith("scheduling-participant:")
+        ):
+            respondent_id = None
+        metadata = dict(item.metadata)
+        if replaced_participant is not None:
+            metadata["identity_replacement"] = {
+                "replaces_participant_id": replaced_participant.id,
+            }
         participant = SchedulingParticipant(
             tenant_id=request.tenant_id,
             request=request,
-            respondent_id=item.respondent_id,
+            respondent_id=respondent_id,
             display_name=item.display_name,
             email=item.email,
             participant_type=item.participant_type,
             required=item.required,
             status="draft",
-            metadata_=dict(item.metadata),
+            metadata_=metadata,
         )
         session.add(participant)
         session.flush()
+        if replaced_participant is not None:
+            replacement_participants[replaced_participant.id] = participant
 
-    participation_provider = _poll_participation_provider()
     for participant in removed_participants:
+        replacement = replacement_participants.get(participant.id)
+        retirement_count = 0
+        response_identities = _participant_response_identities(participant)
+        if response_identities or participant.poll_invitation_id is not None:
+            if retirement_provider is None:  # Guarded above.
+                raise SchedulingError(
+                    "Poll response retirement capability is unavailable"
+                )
+            try:
+                retirement = retirement_provider.retire_responses(
+                    session,
+                    tenant_id=request.tenant_id,
+                    poll_id=cast(str, request.poll_id),
+                    command=PollResponseRetirementCommand(
+                        respondent_ids=response_identities,
+                        invitation_id=participant.poll_invitation_id,
+                        reason=(
+                            "scheduling_participant_replaced"
+                            if replacement is not None
+                            else "scheduling_participant_removed"
+                        ),
+                        idempotency_key=(
+                            f"scheduling:{request.id}:participant:{participant.id}:removed"
+                        ),
+                        metadata={
+                            "source_module": "scheduling",
+                            "source_resource_type": "participant",
+                            "source_resource_id": participant.id,
+                            "scheduling_request_id": request.id,
+                        },
+                    ),
+                )
+            except PollCapabilityError as exc:
+                raise SchedulingError(str(exc)) from exc
+            retirement_count = retirement.newly_retired_count
+        invitation_revoked = False
         if participant.poll_invitation_id is not None:
             if participation_provider is None:  # Guarded above.
                 raise SchedulingError(
@@ -2984,10 +3308,73 @@ def _reconcile_scheduling_participants(
                 )
             except PollCapabilityError as exc:
                 raise SchedulingError(str(exc)) from exc
+            invitation_revoked = True
+        notification_id = None
+        if (
+            replacement is not None
+            or participant.poll_invitation_id is not None
+            or participant.status == "responded"
+            or retirement_count > 0
+        ):
+            notifications = create_scheduling_notification_jobs(
+                session,
+                tenant_id=request.tenant_id,
+                request_id=request.id,
+                event_kind=(
+                    "participant_replaced"
+                    if replacement is not None
+                    else "participant_removed"
+                ),
+                metadata={
+                    "reason": (
+                        "participant_identity_replaced"
+                        if replacement is not None
+                        else "participant_removed"
+                    ),
+                    "replacement_participant_id": (
+                        replacement.id if replacement is not None else None
+                    ),
+                },
+                participant_ids=frozenset({participant.id}),
+            )
+            notification_id = notifications[0].id
+        retired_at = _now()
+        participant.metadata_ = {
+            **(participant.metadata_ or {}),
+            "participant_retirement": {
+                "reason": (
+                    "identity_replaced"
+                    if replacement is not None
+                    else "removed"
+                ),
+                "retired_at": retired_at.isoformat(),
+                "retired_response_count": retirement_count,
+                "replacement_participant_id": (
+                    replacement.id if replacement is not None else None
+                ),
+            },
+        }
         participant.status = "removed"
-        participant.deleted_at = _now()
+        participant.deleted_at = retired_at
+        mutations.append(
+            SchedulingParticipantMutation(
+                action=(
+                    "scheduling.participant_identity_replaced"
+                    if replacement is not None
+                    else "scheduling.participant_removed"
+                ),
+                participant_id=participant.id,
+                replacement_participant_id=(
+                    replacement.id if replacement is not None else None
+                ),
+                changed_fields=("identity",) if replacement is not None else (),
+                invitation_revoked=invitation_revoked,
+                retired_response_count=retirement_count,
+                notification_id=notification_id,
+            )
+        )
 
-    return invitation_tokens
+    return invitation_tokens, tuple(mutations)
 
 
 def update_scheduling_candidate_slot(
@@ -3300,6 +3687,11 @@ def scheduling_participant_response(
     expose_email = include_management_fields or is_current_participant
     return {
         "id": participant.id,
+        "revision": (
+            scheduling_participant_revision(participant)
+            if include_management_fields
+            else None
+        ),
         "is_current_participant": is_current_participant,
         "respondent_id": (
             participant.respondent_id

tests/test_response_editing.py

@@ -12,6 +12,7 @@ from sqlalchemy.orm import Session, sessionmaker
 
 from govoplan_core.auth import ApiPrincipal
 from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.change_sequence import ChangeSequenceEntry
 from govoplan_core.core.modules import ModuleContext
 from govoplan_core.core.poll import PollCapabilityError
 from govoplan_core.core.registry import PlatformRegistry
@@ -63,6 +64,7 @@ from govoplan_scheduling.backend.service import (
     get_public_scheduling_participation,
     issue_scheduling_participant_invitation,
     scheduling_request_summary,
+    scheduling_participant_revision,
     scheduling_slot_revision,
     submit_scheduling_availability,
     submit_public_scheduling_participation,
@@ -86,6 +88,7 @@ class SchedulingResponseEditingTests(unittest.TestCase):
                 PollInvitation.__table__,
                 PollParticipationSubmission.__table__,
                 PollLifecycleTransition.__table__,
+                ChangeSequenceEntry.__table__,
                 SchedulingRequest.__table__,
                 SchedulingCandidateSlot.__table__,
                 SchedulingParticipant.__table__,
@@ -107,6 +110,7 @@ class SchedulingResponseEditingTests(unittest.TestCase):
                 PollParticipationSubmission.__table__,
                 PollInvitation.__table__,
                 PollLifecycleTransition.__table__,
+                ChangeSequenceEntry.__table__,
                 PollResponse.__table__,
                 PollOption.__table__,
                 Poll.__table__,
@@ -1054,6 +1058,7 @@ class SchedulingResponseEditingTests(unittest.TestCase):
                 participants=[
                     SchedulingParticipantReconcileInput(
                         id=alice.id,
+                        revision=scheduling_participant_revision(alice),
                         respondent_id=alice.respondent_id,
                         display_name=alice.display_name,
                         email=alice.email,
@@ -1125,6 +1130,7 @@ class SchedulingResponseEditingTests(unittest.TestCase):
                 participants=[
                     SchedulingParticipantReconcileInput(
                         id=alice.id,
+                        revision=scheduling_participant_revision(alice),
                         respondent_id=alice.respondent_id,
                         display_name=alice.display_name,
                         email=alice.email,
@@ -1277,6 +1283,237 @@ class SchedulingResponseEditingTests(unittest.TestCase):
             {first_slot.id: "available", second_slot.id: "maybe"},
         )
 
+    def test_identity_replacement_revokes_access_retires_response_and_notifies(self) -> None:
+        request = self._request()
+        self._submit_both(request)
+        original = request.participants[0]
+        invitation_id = original.poll_invitation_id
+        self.assertIsNotNone(invitation_id)
+        active_response = (
+            self.session.query(PollResponse)
+            .filter(
+                PollResponse.poll_id == request.poll_id,
+                PollResponse.deleted_at.is_(None),
+            )
+            .one()
+        )
+        original_answers = [dict(answer) for answer in active_response.answers]
+
+        with patch("govoplan_scheduling.backend.router.audit_event") as audit:
+            updated = api_update_scheduling_request(
+                request.id,
+                SchedulingRequestUpdateRequest(
+                    participants=[
+                        SchedulingParticipantReconcileInput(
+                            id=original.id,
+                            revision=scheduling_participant_revision(original),
+                            respondent_id=original.respondent_id,
+                            display_name="Alice Replacement",
+                            email="replacement@example.test",
+                            participant_type=original.participant_type,
+                            required=original.required,
+                            metadata=original.metadata_ or {},
+                        )
+                    ]
+                ),
+                session=self.session,
+                principal=self._principal(
+                    "organizer-1",
+                    email=None,
+                    scopes={WRITE_SCOPE},
+                ),
+            )
+
+        self.assertEqual(len(updated.participants), 1)
+        replacement = updated.participants[0]
+        self.assertNotEqual(replacement.id, original.id)
+        self.assertEqual(replacement.email, "replacement@example.test")
+        self.assertIsNone(replacement.poll_invitation_id)
+        self.assertEqual(original.status, "removed")
+        self.assertIsNotNone(original.deleted_at)
+        self.assertEqual(
+            original.metadata_["participant_retirement"][
+                "replacement_participant_id"
+            ],
+            replacement.id,
+        )
+        invitation = (
+            self.session.query(PollInvitation)
+            .filter(PollInvitation.id == invitation_id)
+            .one()
+        )
+        self.assertIsNotNone(invitation.revoked_at)
+        self.assertIsNotNone(active_response.deleted_at)
+        self.assertEqual(active_response.answers, original_answers)
+        self.assertEqual(
+            active_response.metadata_["response_retirement"]["reason"],
+            "scheduling_participant_replaced",
+        )
+        self.assertEqual(
+            scheduling_request_summary(
+                self.session,
+                tenant_id=request.tenant_id,
+                request_id=request.id,
+            )["response_count"],
+            0,
+        )
+        notice = (
+            self.session.query(SchedulingNotification)
+            .filter(
+                SchedulingNotification.participant_id == original.id,
+                SchedulingNotification.event_kind == "participant_replaced",
+            )
+            .one()
+        )
+        self.assertEqual(notice.recipient, "alice@example.test")
+        audit_calls = [call.kwargs for call in audit.call_args_list]
+        replacement_audit = next(
+            item
+            for item in audit_calls
+            if item["action"] == "scheduling.participant_identity_replaced"
+        )
+        self.assertEqual(
+            replacement_audit["details"]["replacement_participant_id"],
+            replacement.id,
+        )
+        self.assertEqual(replacement_audit["details"]["retired_response_count"], 1)
+        self.assertNotIn("alice@example.test", repr(audit_calls))
+        self.assertNotIn("replacement@example.test", repr(audit_calls))
+
+    def test_stable_account_corrections_keep_identity_and_only_revoke_stale_link(self) -> None:
+        request = self._request(
+            participants=[
+                SchedulingParticipantInput(
+                    respondent_id="alice-account",
+                    display_name="Ailce",
+                    email="alice@example.test",
+                    participant_type="internal",
+                )
+            ]
+        )
+        self._submit_both(request)
+        participant = request.participants[0]
+        invitation_id = participant.poll_invitation_id
+        self.assertIsNotNone(invitation_id)
+
+        renamed = api_update_scheduling_request(
+            request.id,
+            SchedulingRequestUpdateRequest(
+                participants=[
+                    SchedulingParticipantReconcileInput(
+                        id=participant.id,
+                        revision=scheduling_participant_revision(participant),
+                        respondent_id=participant.respondent_id,
+                        display_name="Alice",
+                        email=participant.email,
+                        participant_type=participant.participant_type,
+                        required=participant.required,
+                        metadata=participant.metadata_ or {},
+                    )
+                ]
+            ),
+            session=self.session,
+            principal=self._principal(
+                "organizer-1",
+                email=None,
+                scopes={WRITE_SCOPE},
+            ),
+        )
+
+        self.assertEqual(renamed.participants[0].id, participant.id)
+        self.assertEqual(participant.display_name, "Alice")
+        self.assertEqual(participant.poll_invitation_id, invitation_id)
+        self.assertIsNone(
+            self.session.query(PollInvitation)
+            .filter(PollInvitation.id == invitation_id)
+            .one()
+            .revoked_at
+        )
+
+        with patch("govoplan_scheduling.backend.router.audit_event") as audit:
+            corrected = api_update_scheduling_request(
+                request.id,
+                SchedulingRequestUpdateRequest(
+                    participants=[
+                        SchedulingParticipantReconcileInput(
+                            id=participant.id,
+                            revision=scheduling_participant_revision(participant),
+                            respondent_id=participant.respondent_id,
+                            display_name=participant.display_name,
+                            email="alice.corrected@example.test",
+                            participant_type=participant.participant_type,
+                            required=participant.required,
+                            metadata=participant.metadata_ or {},
+                        )
+                    ]
+                ),
+                session=self.session,
+                principal=self._principal(
+                    "organizer-1",
+                    email=None,
+                    scopes={WRITE_SCOPE},
+                ),
+            )
+
+        self.assertEqual(corrected.participants[0].id, participant.id)
+        self.assertEqual(participant.email, "alice.corrected@example.test")
+        self.assertIsNone(participant.poll_invitation_id)
+        self.assertIsNotNone(
+            self.session.query(PollInvitation)
+            .filter(PollInvitation.id == invitation_id)
+            .one()
+            .revoked_at
+        )
+        self.assertEqual(
+            self.session.query(PollResponse)
+            .filter(
+                PollResponse.poll_id == request.poll_id,
+                PollResponse.deleted_at.is_(None),
+            )
+            .count(),
+            1,
+        )
+        contact_audit = next(
+            call.kwargs
+            for call in audit.call_args_list
+            if call.kwargs["action"] == "scheduling.participant_contact_updated"
+        )
+        self.assertTrue(contact_audit["details"]["invitation_revoked"])
+
+    def test_stale_participant_revision_is_rejected_without_revoking_access(self) -> None:
+        request = self._request()
+        participant = request.participants[0]
+        invitation_id = participant.poll_invitation_id
+
+        with self.assertRaises(HTTPException) as raised:
+            api_update_scheduling_request(
+                request.id,
+                SchedulingRequestUpdateRequest(
+                    participants=[
+                        SchedulingParticipantReconcileInput(
+                            id=participant.id,
+                            revision="0" * 64,
+                            respondent_id=participant.respondent_id,
+                            display_name="Stale update",
+                            email=participant.email,
+                            participant_type=participant.participant_type,
+                            required=participant.required,
+                            metadata=participant.metadata_ or {},
+                        )
+                    ]
+                ),
+                session=self.session,
+                principal=self._principal(
+                    "organizer-1",
+                    email=None,
+                    scopes={WRITE_SCOPE},
+                ),
+            )
+
+        self.assertEqual(raised.exception.status_code, 409)
+        self.assertEqual(participant.poll_invitation_id, invitation_id)
+        self.assertIsNone(participant.deleted_at)
+
     def test_draft_edit_does_not_issue_link_for_added_participant(self) -> None:
         request = self._request(
             status="draft",