fix(poll): harden lifecycle policy boundaries

README.md

@@ -63,6 +63,36 @@ Poll stores the shared poll/options/responses/result summary. Scheduling owns
 the domain workflow around rooms, calendars, free/busy checks, reminders,
 appointment creation, and optional Workflow integration.
 
+## Poll lifecycle
+
+Poll configures its lifecycle through the separate, pure transition engine in
+`backend/transitions.py`. The default policy is:
+
+- `draft` → `open` or `archived`
+- `open` → `draft`, `closed`, or `archived`
+- `closed` → `open`, `decided`, or `archived`
+- `decided` → `open`, `decided` (an explicit re-decision), or `archived`
+- `archived` → the status from which the Poll was archived
+
+Reopening and archiving preserve responses, close history, and decision
+metadata. Every applied transition has a Poll-owned lifecycle audit record.
+Re-deciding always appends a record and supersedes the current decision. An
+exact retry carrying the same `Idempotency-Key` is a no-op and returns the
+original transition record; reusing that key for a different action or option
+is rejected. Without an idempotency identity, a repeated transition is invalid
+except for the deliberately auditable `decided` → `decided` action.
+
+Poll API representations expose every lifecycle action with its availability
+and, when unavailable, the policy reason. Management clients can use the
+`GET /poll/polls/{poll_id}/transitions` endpoint to read the durable history
+instead of duplicating the matrix in their UI.
+
+Polls created through the v0.1.9 contract start in `draft` or `open`; later
+states are entered through the transition engine. Older archived rows may not
+contain the status from which they were archived. Their prior state cannot be
+inferred safely, so unarchive restores them to the least-permissive `draft`
+state and marks that legacy fallback explicitly in the audit metadata.
+
 ## Development Install
 
 ```bash

pyproject.toml

@@ -4,14 +4,14 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-poll"
-version = "0.1.8"
+version = "0.1.9"
 description = "GovOPlaN lightweight poll and availability decision module seed."
 readme = "README.md"
 requires-python = ">=3.12"
 license = { file = "LICENSE" }
 authors = [{ name = "GovOPlaN" }]
 dependencies = [
-  "govoplan-core>=0.1.8",
+  "govoplan-core>=0.1.9",
 ]
 
 [tool.setuptools.packages.find]

src/govoplan_poll/__init__.py

@@ -2,4 +2,4 @@
 
 __all__ = ["__version__"]
 
-__version__ = "0.1.8"
+__version__ = "0.1.9"

src/govoplan_poll/backend/capabilities.py

@@ -3,11 +3,13 @@ from __future__ import annotations
 from collections.abc import Mapping, Sequence
 
 from govoplan_core.core.poll import (
+    PollAnswerRef,
     PollCapabilityError,
     PollCreateCommand,
     PollInvitationCommand,
     PollInvitationRef,
     PollOptionRef,
+    PollOptionUpdateCommand,
     PollRef,
     PollResponseRef,
     PollSchedulingProvider,
@@ -29,10 +31,12 @@ from govoplan_poll.backend.service import (
     create_poll_invitation,
     decide_poll,
     get_poll,
+    get_poll_response_for_respondents,
     list_poll_responses,
     open_poll,
     poll_result_summary_by_id,
     submit_poll_response,
+    update_poll_option,
 )
 
 
@@ -49,6 +53,27 @@ def _response_invitation_id(response: object) -> str | None:
     return value if isinstance(value, str) else None
 
 
+def _response_ref(response: object) -> PollResponseRef:
+    answers: list[PollAnswerRef] = []
+    for answer in response.answers or []:
+        if not isinstance(answer, Mapping):
+            continue
+        answers.append(
+            PollAnswerRef(
+                option_id=answer.get("option_id"),
+                option_key=answer.get("option_key"),
+                value=answer.get("value"),
+                rank=answer.get("rank"),
+            )
+        )
+    return PollResponseRef(
+        invitation_id=_response_invitation_id(response),
+        submitted_at=response.submitted_at,
+        respondent_id=response.respondent_id,
+        answers=tuple(answers),
+    )
+
+
 class SqlPollSchedulingProvider(PollSchedulingProvider):
     def create_poll(
         self,
@@ -151,11 +176,7 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
             )
         except PollError as exc:
             raise PollCapabilityError(str(exc)) from exc
-        return PollResponseRef(
-            invitation_id=_response_invitation_id(response),
-            submitted_at=response.submitted_at,
-            respondent_id=response.respondent_id,
-        )
+        return _response_ref(response)
 
     def update_poll(
         self,
@@ -179,6 +200,30 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
         session.flush()
         return _poll_ref(poll)
 
+    def update_option(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        poll_id: str,
+        option_id: str,
+        command: PollOptionUpdateCommand,
+    ) -> PollOptionRef:
+        try:
+            option = update_poll_option(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                option_id=option_id,
+                label=command.label,
+                description=command.description,
+                value=dict(command.value) if command.value is not None else None,
+                metadata=dict(command.metadata),
+            )
+        except PollError as exc:
+            raise PollCapabilityError(str(exc)) from exc
+        return PollOptionRef(id=option.id, position=option.position)
+
     def open_poll(self, session: object, *, tenant_id: str, poll_id: str) -> PollRef:
         return self._transition(open_poll, session, tenant_id=tenant_id, poll_id=poll_id)
 
@@ -245,14 +290,28 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
             responses = list_poll_responses(session, tenant_id=tenant_id, poll_id=poll_id)
         except PollError as exc:
             raise PollCapabilityError(str(exc)) from exc
-        return tuple(
-            PollResponseRef(
-                invitation_id=_response_invitation_id(response),
-                submitted_at=response.submitted_at,
-                respondent_id=response.respondent_id,
-            )
-            for response in responses
+        return tuple(_response_ref(response) for response in responses)
+
+    def get_response(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        poll_id: str,
+        respondent_ids: Sequence[str],
+        invitation_id: str | None = None,
+    ) -> PollResponseRef | None:
+        try:
+            response = get_poll_response_for_respondents(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                respondent_ids=tuple(respondent_ids),
+                invitation_id=invitation_id,
             )
+        except PollError as exc:
+            raise PollCapabilityError(str(exc)) from exc
+        return _response_ref(response) if response is not None else None
 
     @staticmethod
     def _transition(callback, session: object, *, tenant_id: str, poll_id: str) -> PollRef:

src/govoplan_poll/backend/db/__init__.py

@@ -1,5 +1,5 @@
 from __future__ import annotations
 
-from govoplan_poll.backend.db.models import Poll, PollInvitation, PollOption, PollResponse
+from govoplan_poll.backend.db.models import Poll, PollInvitation, PollLifecycleTransition, PollOption, PollResponse
 
-__all__ = ["Poll", "PollInvitation", "PollOption", "PollResponse"]
+__all__ = ["Poll", "PollInvitation", "PollLifecycleTransition", "PollOption", "PollResponse"]

src/govoplan_poll/backend/db/models.py

@@ -53,6 +53,7 @@ class Poll(Base, TimestampMixin):
     closed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     decided_option_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
+    archived_from_status: Mapped[str | None] = mapped_column(String(30), nullable=True)
     created_by_user_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
     deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
     metadata_: Mapped[dict[str, Any] | None] = mapped_column("metadata", JSON, nullable=True)
@@ -60,6 +61,11 @@ class Poll(Base, TimestampMixin):
     options: Mapped[list["PollOption"]] = relationship(back_populates="poll", cascade="all, delete-orphan", order_by="PollOption.position")
     responses: Mapped[list["PollResponse"]] = relationship(back_populates="poll", cascade="all, delete-orphan")
     invitations: Mapped[list["PollInvitation"]] = relationship(back_populates="poll", cascade="all, delete-orphan")
+    lifecycle_transitions: Mapped[list["PollLifecycleTransition"]] = relationship(
+        back_populates="poll",
+        cascade="all, delete-orphan",
+        order_by="PollLifecycleTransition.created_at",
+    )
 
 
 class PollOption(Base, TimestampMixin):
@@ -127,4 +133,28 @@ class PollInvitation(Base, TimestampMixin):
     poll: Mapped[Poll] = relationship(back_populates="invitations")
 
 
-__all__ = ["Poll", "PollInvitation", "PollOption", "PollResponse", "new_uuid"]
+class PollLifecycleTransition(Base, TimestampMixin):
+    __tablename__ = "poll_lifecycle_transitions"
+    __table_args__ = (
+        UniqueConstraint("poll_id", "idempotency_key", name="uq_poll_lifecycle_transition_idempotency"),
+        Index("ix_poll_lifecycle_transition_poll_created", "poll_id", "created_at"),
+        Index("ix_poll_lifecycle_transition_tenant_poll", "tenant_id", "poll_id"),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    poll_id: Mapped[str] = mapped_column(ForeignKey("poll_polls.id", ondelete="CASCADE"), nullable=False, index=True)
+    action: Mapped[str] = mapped_column(String(30), nullable=False, index=True)
+    from_status: Mapped[str] = mapped_column(String(30), nullable=False)
+    to_status: Mapped[str] = mapped_column(String(30), nullable=False)
+    decision_option_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
+    previous_decision_option_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
+    idempotency_key: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    actor_user_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
+    actor_api_key_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
+    metadata_: Mapped[dict[str, Any] | None] = mapped_column("metadata", JSON, nullable=True)
+
+    poll: Mapped[Poll] = relationship(back_populates="lifecycle_transitions")
+
+
+__all__ = ["Poll", "PollInvitation", "PollLifecycleTransition", "PollOption", "PollResponse", "new_uuid"]

src/govoplan_poll/backend/manifest.py

@@ -19,7 +19,7 @@ from govoplan_poll.backend.db import models as poll_models  # noqa: F401 - popul
 
 MODULE_ID = "poll"
 MODULE_NAME = "Poll"
-MODULE_VERSION = "0.1.8"
+MODULE_VERSION = "0.1.9"
 READ_SCOPE = "poll:poll:read"
 WRITE_SCOPE = "poll:poll:write"
 ADMIN_SCOPE = "poll:poll:admin"
@@ -115,11 +115,11 @@ manifest = ModuleManifest(
     optional_dependencies=("access", "calendar", "campaigns", "portal", "mail", "notifications", "forms_runtime"),
     optional_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
     provides_interfaces=(
-        ModuleInterfaceProvider(name="poll.option_selection", version="0.1.8"),
-        ModuleInterfaceProvider(name="poll.availability_matrix", version="0.1.8"),
-        ModuleInterfaceProvider(name="poll.response_collection", version="0.1.8"),
-        ModuleInterfaceProvider(name="poll.workflow_context", version="0.1.8"),
-        ModuleInterfaceProvider(name="poll.signed_participation", version="0.1.8"),
+        ModuleInterfaceProvider(name="poll.option_selection", version="0.1.9"),
+        ModuleInterfaceProvider(name="poll.availability_matrix", version="0.1.9"),
+        ModuleInterfaceProvider(name="poll.response_collection", version="0.1.9"),
+        ModuleInterfaceProvider(name="poll.workflow_context", version="0.1.9"),
+        ModuleInterfaceProvider(name="poll.signed_participation", version="0.1.9"),
     ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
@@ -134,6 +134,7 @@ manifest = ModuleManifest(
         retirement_provider=drop_table_retirement_provider(
             poll_models.Poll,
             poll_models.PollInvitation,
+            poll_models.PollLifecycleTransition,
             poll_models.PollOption,
             poll_models.PollResponse,
             label="Poll",
@@ -144,6 +145,7 @@ manifest = ModuleManifest(
         persistent_table_uninstall_guard(
             poll_models.Poll,
             poll_models.PollInvitation,
+            poll_models.PollLifecycleTransition,
             poll_models.PollOption,
             poll_models.PollResponse,
             label="Poll",

src/govoplan_poll/backend/migrations/versions/3b4c5d6e7f8a_v019_poll_lifecycle.py

@@ -0,0 +1,67 @@
+"""v0.1.9 configurable Poll lifecycle history
+
+Revision ID: 3b4c5d6e7f8a
+Revises: 2a3b4c5d6e7f
+Create Date: 2026-07-20 00:00:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "3b4c5d6e7f8a"
+down_revision = "2a3b4c5d6e7f"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("poll_polls", sa.Column("archived_from_status", sa.String(length=30), nullable=True))
+    op.create_table(
+        "poll_lifecycle_transitions",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("poll_id", sa.String(length=36), nullable=False),
+        sa.Column("action", sa.String(length=30), nullable=False),
+        sa.Column("from_status", sa.String(length=30), nullable=False),
+        sa.Column("to_status", sa.String(length=30), nullable=False),
+        sa.Column("decision_option_id", sa.String(length=36), nullable=True),
+        sa.Column("previous_decision_option_id", sa.String(length=36), nullable=True),
+        sa.Column("idempotency_key", sa.String(length=255), nullable=True),
+        sa.Column("actor_user_id", sa.String(length=36), nullable=True),
+        sa.Column("actor_api_key_id", sa.String(length=36), nullable=True),
+        sa.Column("metadata", sa.JSON(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["poll_id"],
+            ["poll_polls.id"],
+            name=op.f("fk_poll_lifecycle_transitions_poll_id_poll_polls"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_poll_lifecycle_transitions")),
+        sa.UniqueConstraint("poll_id", "idempotency_key", name="uq_poll_lifecycle_transition_idempotency"),
+    )
+    op.create_index(op.f("ix_poll_lifecycle_transitions_action"), "poll_lifecycle_transitions", ["action"], unique=False)
+    op.create_index(op.f("ix_poll_lifecycle_transitions_actor_api_key_id"), "poll_lifecycle_transitions", ["actor_api_key_id"], unique=False)
+    op.create_index(op.f("ix_poll_lifecycle_transitions_actor_user_id"), "poll_lifecycle_transitions", ["actor_user_id"], unique=False)
+    op.create_index(op.f("ix_poll_lifecycle_transitions_poll_id"), "poll_lifecycle_transitions", ["poll_id"], unique=False)
+    op.create_index(op.f("ix_poll_lifecycle_transitions_tenant_id"), "poll_lifecycle_transitions", ["tenant_id"], unique=False)
+    op.create_index(
+        "ix_poll_lifecycle_transition_poll_created",
+        "poll_lifecycle_transitions",
+        ["poll_id", "created_at"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_poll_lifecycle_transition_tenant_poll",
+        "poll_lifecycle_transitions",
+        ["tenant_id", "poll_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("poll_lifecycle_transitions")
+    op.drop_column("poll_polls", "archived_from_status")

src/govoplan_poll/backend/router.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Header, HTTPException, Query, status
 from sqlalchemy.orm import Session
 
 from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
@@ -12,6 +14,8 @@ from govoplan_poll.backend.schemas import (
     PollInvitationCreateRequest,
     PollInvitationListResponse,
     PollInvitationResponse,
+    PollLifecycleResponse,
+    PollLifecycleTransitionResponse,
     PollListResponse,
     PollResponse,
     PollResponseItem,
@@ -19,21 +23,22 @@ from govoplan_poll.backend.schemas import (
     PollResultSummaryResponse,
     PollStatusResponse,
     PollSubmitResponseRequest,
+    PollTransitionRequest,
+    PollTransitionAvailabilityResponse,
     PollUpdateRequest,
 )
 from govoplan_poll.backend.service import (
     PollError,
-    close_poll,
     create_poll_invitation,
     create_poll,
-    decide_poll,
     get_poll_by_invitation_token,
     get_visible_poll,
     list_poll_responses,
     list_poll_invitations,
+    list_poll_lifecycle_transitions,
     list_visible_polls,
-    open_poll,
     poll_invitation_response,
+    poll_lifecycle_transition_response,
     poll_response,
     poll_response_item,
     poll_result_summary_by_id,
@@ -41,6 +46,7 @@ from govoplan_poll.backend.service import (
     revoke_poll_invitation,
     submit_poll_response,
     submit_poll_response_with_token,
+    transition_poll,
     update_poll,
 )
 
@@ -67,6 +73,45 @@ def _poll_response(poll) -> PollResponse:
     return PollResponse.model_validate(poll_response(poll))
 
 
+def _transition_status_response(result) -> PollStatusResponse:
+    return PollStatusResponse(
+        poll=_poll_response(result.poll),
+        transition=PollLifecycleTransitionResponse.model_validate(
+            poll_lifecycle_transition_response(result.transition)
+        ),
+        replayed=result.replayed,
+    )
+
+
+def _apply_transition(
+    session: Session,
+    principal: ApiPrincipal,
+    *,
+    poll_id: str,
+    action: str,
+    option_id: str | None = None,
+    option_key: str | None = None,
+    idempotency_key: str | None = None,
+) -> PollStatusResponse:
+    try:
+        result = transition_poll(
+            session,
+            tenant_id=principal.tenant_id,
+            poll_id=poll_id,
+            action=action,
+            option_id=option_id,
+            option_key=option_key,
+            idempotency_key=idempotency_key,
+            actor_user_id=getattr(principal.user, "id", None),
+            actor_api_key_id=getattr(principal.api_key, "id", None) if principal.api_key else None,
+        )
+    except PollError as exc:
+        raise _poll_http_error(exc) from exc
+    response = _transition_status_response(result)
+    session.commit()
+    return response
+
+
 def _principal_actor_ids(principal: ApiPrincipal) -> tuple[str, ...]:
     candidates = (
         principal.account_id,
@@ -175,50 +220,161 @@ def api_update_poll(
 @router.post("/polls/{poll_id}/open", response_model=PollStatusResponse)
 def api_open_poll(
     poll_id: str,
+    idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
 ) -> PollStatusResponse:
     _require_scope(principal, WRITE_SCOPE)
-    try:
-        poll = open_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id)
-    except PollError as exc:
-        raise _poll_http_error(exc) from exc
-    response = PollStatusResponse(poll=_poll_response(poll))
-    session.commit()
-    return response
+    return _apply_transition(
+        session,
+        principal,
+        poll_id=poll_id,
+        action="open",
+        idempotency_key=idempotency_key,
+    )
+
+
+@router.post("/polls/{poll_id}/draft", response_model=PollStatusResponse)
+def api_draft_poll(
+    poll_id: str,
+    idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> PollStatusResponse:
+    _require_scope(principal, WRITE_SCOPE)
+    return _apply_transition(
+        session,
+        principal,
+        poll_id=poll_id,
+        action="draft",
+        idempotency_key=idempotency_key,
+    )
 
 
 @router.post("/polls/{poll_id}/close", response_model=PollStatusResponse)
 def api_close_poll(
     poll_id: str,
+    idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
 ) -> PollStatusResponse:
     _require_scope(principal, WRITE_SCOPE)
-    try:
-        poll = close_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id)
-    except PollError as exc:
-        raise _poll_http_error(exc) from exc
-    response = PollStatusResponse(poll=_poll_response(poll))
-    session.commit()
-    return response
+    return _apply_transition(
+        session,
+        principal,
+        poll_id=poll_id,
+        action="close",
+        idempotency_key=idempotency_key,
+    )
 
 
 @router.post("/polls/{poll_id}/decide", response_model=PollStatusResponse)
 def api_decide_poll(
     poll_id: str,
     payload: PollDecisionRequest,
+    idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
 ) -> PollStatusResponse:
+    _require_scope(principal, WRITE_SCOPE)
+    return _apply_transition(
+        session,
+        principal,
+        poll_id=poll_id,
+        action="decide",
+        option_id=payload.option_id,
+        option_key=payload.option_key,
+        idempotency_key=idempotency_key,
+    )
+
+
+@router.post("/polls/{poll_id}/archive", response_model=PollStatusResponse)
+def api_archive_poll(
+    poll_id: str,
+    idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> PollStatusResponse:
+    _require_scope(principal, WRITE_SCOPE)
+    return _apply_transition(
+        session,
+        principal,
+        poll_id=poll_id,
+        action="archive",
+        idempotency_key=idempotency_key,
+    )
+
+
+@router.post("/polls/{poll_id}/unarchive", response_model=PollStatusResponse)
+def api_unarchive_poll(
+    poll_id: str,
+    idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> PollStatusResponse:
+    _require_scope(principal, WRITE_SCOPE)
+    return _apply_transition(
+        session,
+        principal,
+        poll_id=poll_id,
+        action="unarchive",
+        idempotency_key=idempotency_key,
+    )
+
+
+@router.post("/polls/{poll_id}/transitions", response_model=PollStatusResponse)
+def api_transition_poll(
+    poll_id: str,
+    payload: PollTransitionRequest,
+    idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> PollStatusResponse:
+    _require_scope(principal, WRITE_SCOPE)
+    return _apply_transition(
+        session,
+        principal,
+        poll_id=poll_id,
+        action=payload.action,
+        option_id=payload.option_id,
+        option_key=payload.option_key,
+        idempotency_key=idempotency_key,
+    )
+
+
+@router.get("/polls/{poll_id}/transitions", response_model=PollLifecycleResponse)
+def api_poll_lifecycle(
+    poll_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> PollLifecycleResponse:
     _require_scope(principal, WRITE_SCOPE)
     try:
-        poll = decide_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload)
+        poll = get_visible_poll(
+            session,
+            tenant_id=principal.tenant_id,
+            poll_id=poll_id,
+            actor_ids=_principal_actor_ids(principal),
+            can_manage=True,
+        )
+        history = list_poll_lifecycle_transitions(
+            session,
+            tenant_id=principal.tenant_id,
+            poll_id=poll_id,
+        )
     except PollError as exc:
         raise _poll_http_error(exc) from exc
-    response = PollStatusResponse(poll=_poll_response(poll))
-    session.commit()
-    return response
+    projected = _poll_response(poll)
+    return PollLifecycleResponse(
+        poll_id=poll.id,
+        status=poll.status,
+        archived_from_status=poll.archived_from_status,
+        actions=[PollTransitionAvailabilityResponse.model_validate(item) for item in projected.lifecycle_actions],
+        history=[
+            PollLifecycleTransitionResponse.model_validate(poll_lifecycle_transition_response(item))
+            for item in history
+        ],
+    )
 
 
 @router.post("/polls/{poll_id}/responses", response_model=PollResponseItem, status_code=status.HTTP_201_CREATED)

src/govoplan_poll/backend/schemas.py

@@ -8,6 +8,8 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 PollKind = Literal["single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ranked_choice", "availability"]
 PollStatus = Literal["draft", "open", "closed", "decided", "archived"]
+PollInitialStatus = Literal["draft", "open"]
+PollTransitionAction = Literal["open", "draft", "close", "decide", "archive", "unarchive"]
 PollVisibility = Literal["private", "tenant", "public", "unlisted"]
 PollResultVisibility = Literal["organizer", "after_response", "after_close", "public"]
 AvailabilityValue = Literal["available", "maybe", "unavailable"]
@@ -30,7 +32,7 @@ class PollCreateRequest(BaseModel):
     slug: str | None = Field(default=None, max_length=120)
     description: str | None = None
     kind: PollKind = "single_choice"
-    status: PollStatus = "draft"
+    status: PollInitialStatus = "draft"
     visibility: PollVisibility = "private"
     result_visibility: PollResultVisibility = "after_close"
     context_module: str | None = Field(default=None, max_length=100)
@@ -85,6 +87,13 @@ class PollOptionResponse(BaseModel):
     metadata: dict[str, Any] = Field(default_factory=dict)
 
 
+class PollTransitionAvailabilityResponse(BaseModel):
+    action: PollTransitionAction
+    target_status: PollStatus | None = None
+    available: bool
+    reason: str | None = None
+
+
 class PollResponseItem(BaseModel):
     id: str
     respondent_id: str | None = None
@@ -120,11 +129,13 @@ class PollResponse(BaseModel):
     closed_at: datetime | None = None
     decided_at: datetime | None = None
     decided_option_id: str | None = None
+    archived_from_status: str | None = None
     created_by_user_id: str | None = None
     created_at: datetime
     updated_at: datetime
     metadata: dict[str, Any] = Field(default_factory=dict)
     options: list[PollOptionResponse] = Field(default_factory=list)
+    lifecycle_actions: list[PollTransitionAvailabilityResponse] = Field(default_factory=list)
 
 
 class PollListResponse(BaseModel):
@@ -175,8 +186,40 @@ class PollDecisionRequest(BaseModel):
     option_key: str | None = None
 
 
+class PollTransitionRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    action: PollTransitionAction
+    option_id: str | None = None
+    option_key: str | None = None
+
+
+class PollLifecycleTransitionResponse(BaseModel):
+    id: str
+    action: PollTransitionAction
+    from_status: PollStatus
+    to_status: PollStatus
+    decision_option_id: str | None = None
+    previous_decision_option_id: str | None = None
+    idempotency_key: str | None = None
+    actor_user_id: str | None = None
+    actor_api_key_id: str | None = None
+    created_at: datetime
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
 class PollStatusResponse(BaseModel):
     poll: PollResponse
+    transition: PollLifecycleTransitionResponse | None = None
+    replayed: bool = False
+
+
+class PollLifecycleResponse(BaseModel):
+    poll_id: str
+    status: PollStatus
+    archived_from_status: str | None = None
+    actions: list[PollTransitionAvailabilityResponse] = Field(default_factory=list)
+    history: list[PollLifecycleTransitionResponse] = Field(default_factory=list)
 
 
 class PollResponseListResponse(BaseModel):

src/govoplan_poll/backend/service.py

@@ -3,13 +3,14 @@ from __future__ import annotations
 import hashlib
 import re
 import secrets
+from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import Any
 
 from sqlalchemy.orm import Session
 
 from govoplan_core.db.base import utcnow
-from govoplan_poll.backend.db.models import Poll, PollInvitation, PollOption, PollResponse
+from govoplan_poll.backend.db.models import Poll, PollInvitation, PollLifecycleTransition, PollOption, PollResponse
 from govoplan_poll.backend.schemas import (
     PollCreateRequest,
     PollDecisionRequest,
@@ -18,6 +19,7 @@ from govoplan_poll.backend.schemas import (
     PollSubmitResponseRequest,
     PollUpdateRequest,
 )
+from govoplan_poll.backend.transitions import DEFAULT_POLL_TRANSITION_ENGINE, PollTransitionEngine
 
 
 class PollError(ValueError):
@@ -25,7 +27,7 @@ class PollError(ValueError):
 
 
 POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ranked_choice", "availability"}
-POLL_STATUSES = {"draft", "open", "closed", "decided", "archived"}
+POLL_INITIAL_STATUSES = {"draft", "open"}
 CHOICE_POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ranked_choice"}
 AVAILABILITY_VALUES = {"available", "maybe", "unavailable"}
 YES_NO_OPTIONS = (
@@ -39,6 +41,13 @@ YES_NO_MAYBE_OPTIONS = (
 )
 
 
+@dataclass(frozen=True)
+class PollTransitionResult:
+    poll: Poll
+    transition: PollLifecycleTransition
+    replayed: bool = False
+
+
 def slugify(value: str, *, fallback: str = "poll") -> str:
     slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
     return slug or fallback
@@ -141,8 +150,8 @@ def _validate_choice_bounds(kind: str, min_choices: int, max_choices: int | None
 def _ensure_valid_poll_payload(payload: PollCreateRequest) -> tuple[list[PollOptionInput], int, int | None]:
     if payload.kind not in POLL_KINDS:
         raise PollError(f"Unsupported poll kind: {payload.kind}")
-    if payload.status not in POLL_STATUSES:
-        raise PollError(f"Unsupported poll status: {payload.status}")
+    if payload.status not in POLL_INITIAL_STATUSES:
+        raise PollError("Poll initial status must be draft or open")
     if payload.opens_at is not None and payload.closes_at is not None and payload.closes_at <= payload.opens_at:
         raise PollError("closes_at must be after opens_at")
     options = _normalize_options(payload.kind, payload.options)
@@ -297,7 +306,9 @@ def poll_results_are_visible(
     if poll.result_visibility == "public":
         return True
     if poll.result_visibility == "after_close":
-        if poll.status in {"closed", "decided", "archived"}:
+        if poll.status in {"closed", "decided"}:
+            return True
+        if poll.status == "archived" and poll.archived_from_status in {"closed", "decided"}:
             return True
         return poll.closes_at is not None and response_datetime(poll.closes_at) <= _now()
     if poll.result_visibility == "after_response" and ids:
@@ -370,38 +381,260 @@ def update_poll(session: Session, *, tenant_id: str, poll_id: str, payload: Poll
     return poll
 
 
-def open_poll(session: Session, *, tenant_id: str, poll_id: str) -> Poll:
-    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
-    if poll.status == "archived":
-        raise PollError("Archived polls cannot be opened")
-    if poll.closes_at is not None and response_datetime(poll.closes_at) <= _now():
-        raise PollError("Poll close time is already in the past")
-    poll.status = "open"
-    poll.closed_at = None
-    session.flush()
+def _lock_poll_for_transition(session: Session, *, tenant_id: str, poll_id: str) -> Poll:
+    poll = (
+        session.query(Poll)
+        .filter(Poll.tenant_id == tenant_id, Poll.id == poll_id, Poll.deleted_at.is_(None))
+        .populate_existing()
+        .with_for_update()
+        .first()
+    )
+    if poll is None:
+        raise PollError("Poll not found")
     return poll
 
 
-def close_poll(session: Session, *, tenant_id: str, poll_id: str) -> Poll:
-    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
-    if poll.status == "archived":
-        raise PollError("Archived polls cannot be closed")
-    poll.status = "closed"
-    poll.closed_at = _now()
-    session.flush()
-    return poll
+def _normalize_idempotency_key(value: str | None) -> str | None:
+    if value is None:
+        return None
+    normalized = value.strip()
+    if not normalized:
+        raise PollError("Idempotency key cannot be empty")
+    if len(normalized) > 255:
+        raise PollError("Idempotency key cannot be longer than 255 characters")
+    return normalized
 
 
-def decide_poll(session: Session, *, tenant_id: str, poll_id: str, payload: PollDecisionRequest) -> Poll:
-    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
-    option = _option_by_id_or_key(poll, option_id=payload.option_id, option_key=payload.option_key)
-    poll.status = "decided"
-    poll.decided_option_id = option.id
-    poll.decided_at = _now()
+def _transition_for_idempotency_key(
+    session: Session,
+    *,
+    poll_id: str,
+    idempotency_key: str | None,
+) -> PollLifecycleTransition | None:
+    if idempotency_key is None:
+        return None
+    return (
+        session.query(PollLifecycleTransition)
+        .filter(
+            PollLifecycleTransition.poll_id == poll_id,
+            PollLifecycleTransition.idempotency_key == idempotency_key,
+        )
+        .one_or_none()
+    )
+
+
+def transition_poll(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    action: str,
+    option_id: str | None = None,
+    option_key: str | None = None,
+    idempotency_key: str | None = None,
+    actor_user_id: str | None = None,
+    actor_api_key_id: str | None = None,
+    transition_engine: PollTransitionEngine = DEFAULT_POLL_TRANSITION_ENGINE,
+) -> PollTransitionResult:
+    """Apply one policy-controlled transition and append its durable audit record."""
+
+    poll = _lock_poll_for_transition(session, tenant_id=tenant_id, poll_id=poll_id)
+    normalized_key = _normalize_idempotency_key(idempotency_key)
+    decision_option = None
+    if action == "decide":
+        decision_option = _option_by_id_or_key(poll, option_id=option_id, option_key=option_key)
+    elif option_id is not None or option_key is not None:
+        raise PollError("Only a decide transition accepts a poll option")
+
+    existing = _transition_for_idempotency_key(
+        session,
+        poll_id=poll.id,
+        idempotency_key=normalized_key,
+    )
+    if existing is not None:
+        requested_option_id = decision_option.id if decision_option is not None else None
+        if existing.action != action or existing.decision_option_id != requested_option_id:
+            raise PollError("Idempotency key was already used for a different poll transition")
+        return PollTransitionResult(poll=poll, transition=existing, replayed=True)
+
+    try:
+        plan = transition_engine.plan(
+            current_status=poll.status,
+            action=action,
+            archived_from_status=poll.archived_from_status,
+        )
+    except ValueError as exc:
+        raise PollError(str(exc)) from exc
+
+    previous_decision_option_id = poll.decided_option_id
+    now = _now()
+    used_legacy_unarchive_fallback = action == "unarchive" and poll.archived_from_status is None
+    cleared_expired_closes_at: datetime | None = None
+    if action == "open" and poll.closes_at is not None:
+        normalized_closes_at = response_datetime(poll.closes_at)
+        if normalized_closes_at is not None and normalized_closes_at <= now:
+            cleared_expired_closes_at = normalized_closes_at
+            poll.closes_at = None
+
+    if action == "archive":
+        poll.archived_from_status = plan.from_status
+    elif action == "unarchive":
+        poll.archived_from_status = None
+    elif action == "close":
+        poll.closed_at = now
+    elif action == "decide":
+        assert decision_option is not None
+        poll.decided_option_id = decision_option.id
+        poll.decided_at = now
         if poll.closed_at is None:
-        poll.closed_at = poll.decided_at
+            poll.closed_at = now
+
+    poll.status = plan.to_status
+    transition_metadata: dict[str, Any] = {}
+    if action == "archive":
+        transition_metadata["archived_from_status"] = plan.from_status
+    elif action == "unarchive":
+        transition_metadata["restored_status"] = plan.to_status
+        if used_legacy_unarchive_fallback:
+            transition_metadata["legacy_restore_fallback"] = {
+                "reason": "missing_archived_from_status",
+                "restored_status": plan.to_status,
+            }
+    if cleared_expired_closes_at is not None:
+        transition_metadata["cleared_expired_closes_at"] = cleared_expired_closes_at.isoformat()
+    transition = PollLifecycleTransition(
+        tenant_id=tenant_id,
+        poll_id=poll.id,
+        action=action,
+        from_status=plan.from_status,
+        to_status=plan.to_status,
+        decision_option_id=decision_option.id if decision_option is not None else None,
+        previous_decision_option_id=previous_decision_option_id,
+        idempotency_key=normalized_key,
+        actor_user_id=actor_user_id,
+        actor_api_key_id=actor_api_key_id,
+        metadata_=transition_metadata,
+    )
+    session.add(transition)
     session.flush()
-    return poll
+    return PollTransitionResult(poll=poll, transition=transition)
+
+
+def open_poll(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    idempotency_key: str | None = None,
+) -> Poll:
+    return transition_poll(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+        action="open",
+        idempotency_key=idempotency_key,
+    ).poll
+
+
+def draft_poll(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    idempotency_key: str | None = None,
+) -> Poll:
+    return transition_poll(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+        action="draft",
+        idempotency_key=idempotency_key,
+    ).poll
+
+
+def close_poll(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    idempotency_key: str | None = None,
+) -> Poll:
+    return transition_poll(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+        action="close",
+        idempotency_key=idempotency_key,
+    ).poll
+
+
+def decide_poll(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    payload: PollDecisionRequest,
+    idempotency_key: str | None = None,
+) -> Poll:
+    return transition_poll(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+        action="decide",
+        option_id=payload.option_id,
+        option_key=payload.option_key,
+        idempotency_key=idempotency_key,
+    ).poll
+
+
+def archive_poll(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    idempotency_key: str | None = None,
+) -> Poll:
+    return transition_poll(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+        action="archive",
+        idempotency_key=idempotency_key,
+    ).poll
+
+
+def unarchive_poll(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    idempotency_key: str | None = None,
+) -> Poll:
+    return transition_poll(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+        action="unarchive",
+        idempotency_key=idempotency_key,
+    ).poll
+
+
+def list_poll_lifecycle_transitions(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+) -> list[PollLifecycleTransition]:
+    get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+    return (
+        session.query(PollLifecycleTransition)
+        .filter(
+            PollLifecycleTransition.tenant_id == tenant_id,
+            PollLifecycleTransition.poll_id == poll_id,
+        )
+        .order_by(PollLifecycleTransition.created_at.asc(), PollLifecycleTransition.id.asc())
+        .all()
+    )
 
 
 def _assert_poll_accepts_responses(poll: Poll, *, now: datetime | None = None) -> None:
@@ -518,6 +751,125 @@ def _lock_poll_for_response(session: Session, *, tenant_id: str, poll_id: str) -
     return poll
 
 
+def get_poll_response_for_respondents(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    respondent_ids: tuple[str, ...],
+    invitation_id: str | None = None,
+) -> PollResponse | None:
+    """Resolve a response from server-trusted respondent identities."""
+
+    get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+    ids = tuple(dict.fromkeys(value for value in respondent_ids if value))
+    if ids:
+        response = (
+            session.query(PollResponse)
+            .filter(
+                PollResponse.tenant_id == tenant_id,
+                PollResponse.poll_id == poll_id,
+                PollResponse.respondent_id.in_(ids),
+                PollResponse.deleted_at.is_(None),
+            )
+            .order_by(PollResponse.submitted_at.desc(), PollResponse.id.desc())
+            .first()
+        )
+        if response is not None:
+            return response
+    if invitation_id is None:
+        return None
+    return next(
+        (
+            response
+            for response in reversed(list_poll_responses(session, tenant_id=tenant_id, poll_id=poll_id))
+            if (response.metadata_ or {}).get("invitation_id") == invitation_id
+        ),
+        None,
+    )
+
+
+def update_poll_option(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    option_id: str,
+    label: str,
+    description: str | None,
+    value: dict[str, Any] | None,
+    metadata: dict[str, Any],
+) -> PollOption:
+    """Update one option and invalidate only answers bound to that option.
+
+    The Poll row is locked first, matching response submission.  The owning
+    API commits the option and response JSON updates in the same transaction.
+    """
+
+    poll = _lock_poll_for_response(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+    )
+    if poll.status not in {"draft", "open"}:
+        raise PollError("Only draft or open poll options can be edited")
+    option = (
+        session.query(PollOption)
+        .filter(
+            PollOption.tenant_id == tenant_id,
+            PollOption.poll_id == poll.id,
+            PollOption.id == option_id,
+            PollOption.deleted_at.is_(None),
+        )
+        .populate_existing()
+        .with_for_update()
+        .one_or_none()
+    )
+    if option is None:
+        raise PollError("Unknown poll option")
+    normalized_value = dict(value) if value is not None else None
+    normalized_metadata = dict(metadata)
+    changed = (
+        option.label != label
+        or option.description != description
+        or option.value != normalized_value
+        or (option.metadata_ or {}) != normalized_metadata
+    )
+    if not changed:
+        return option
+
+    responses = (
+        session.query(PollResponse)
+        .filter(
+            PollResponse.tenant_id == tenant_id,
+            PollResponse.poll_id == poll.id,
+            PollResponse.deleted_at.is_(None),
+        )
+        .order_by(PollResponse.id.asc())
+        .populate_existing()
+        .with_for_update()
+        .all()
+    )
+    if responses and not poll.allow_response_update:
+        raise PollError("Poll options cannot be edited after responses when response updates are disabled")
+    option.label = label
+    option.description = description
+    option.value = normalized_value
+    option.metadata_ = normalized_metadata
+    for response in responses:
+        retained_answers = [
+            answer
+            for answer in (response.answers or [])
+            if not isinstance(answer, dict) or answer.get("option_id") != option.id
+        ]
+        if retained_answers != (response.answers or []):
+            response.answers = retained_answers
+            if not retained_answers:
+                response.deleted_at = _now()
+    session.flush()
+    return option
+
+
 def submit_poll_response(session: Session, *, tenant_id: str, poll_id: str, payload: PollSubmitResponseRequest) -> PollResponse:
     poll = _lock_poll_for_response(
         session,
@@ -564,7 +916,31 @@ def poll_option_response(option: PollOption) -> dict[str, Any]:
     }
 
 
-def poll_response(poll: Poll) -> dict[str, Any]:
+def poll_lifecycle_transition_response(transition: PollLifecycleTransition) -> dict[str, Any]:
+    return {
+        "id": transition.id,
+        "action": transition.action,
+        "from_status": transition.from_status,
+        "to_status": transition.to_status,
+        "decision_option_id": transition.decision_option_id,
+        "previous_decision_option_id": transition.previous_decision_option_id,
+        "idempotency_key": transition.idempotency_key,
+        "actor_user_id": transition.actor_user_id,
+        "actor_api_key_id": transition.actor_api_key_id,
+        "created_at": response_datetime(transition.created_at),
+        "metadata": transition.metadata_ or {},
+    }
+
+
+def poll_response(
+    poll: Poll,
+    *,
+    transition_engine: PollTransitionEngine = DEFAULT_POLL_TRANSITION_ENGINE,
+) -> dict[str, Any]:
+    lifecycle_actions = transition_engine.available_actions(
+        current_status=poll.status,
+        archived_from_status=poll.archived_from_status,
+    )
     return {
         "id": poll.id,
         "tenant_id": poll.tenant_id,
@@ -589,11 +965,21 @@ def poll_response(poll: Poll) -> dict[str, Any]:
         "closed_at": response_datetime(poll.closed_at),
         "decided_at": response_datetime(poll.decided_at),
         "decided_option_id": poll.decided_option_id,
+        "archived_from_status": poll.archived_from_status,
         "created_by_user_id": poll.created_by_user_id,
         "created_at": response_datetime(poll.created_at),
         "updated_at": response_datetime(poll.updated_at),
         "metadata": poll.metadata_ or {},
         "options": [poll_option_response(option) for option in _active_options(poll)],
+        "lifecycle_actions": [
+            {
+                "action": item.action,
+                "target_status": item.target_status,
+                "available": item.available,
+                "reason": item.reason,
+            }
+            for item in lifecycle_actions
+        ],
     }
 
 

src/govoplan_poll/backend/transitions.py

@@ -0,0 +1,196 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Mapping
+
+
+POLL_STATUSES = frozenset({"draft", "open", "closed", "decided", "archived"})
+POLL_TRANSITION_ACTIONS = ("open", "draft", "close", "decide", "archive", "unarchive")
+
+
+@dataclass(frozen=True)
+class PollTransitionRule:
+    """One configurable lifecycle action.
+
+    Archive restoration is deliberately represented as a dynamic target. The
+    engine remains a pure policy component; persistence and domain side effects
+    stay in the Poll service.
+    """
+
+    action: str
+    source_statuses: frozenset[str]
+    target_status: str | None
+    restore_archived_status: bool = False
+    missing_restore_target_status: str | None = None
+
+
+@dataclass(frozen=True)
+class PollTransitionPlan:
+    action: str
+    from_status: str
+    to_status: str
+
+
+@dataclass(frozen=True)
+class PollTransitionAvailability:
+    action: str
+    target_status: str | None
+    available: bool
+    reason: str | None = None
+
+
+class PollTransitionPolicy:
+    """Validated, replaceable configuration for a Poll lifecycle."""
+
+    def __init__(self, rules: Mapping[str, PollTransitionRule]) -> None:
+        normalized = dict(rules)
+        if set(normalized) != set(POLL_TRANSITION_ACTIONS):
+            missing = sorted(set(POLL_TRANSITION_ACTIONS) - set(normalized))
+            extra = sorted(set(normalized) - set(POLL_TRANSITION_ACTIONS))
+            raise ValueError(f"Poll transition policy has missing actions {missing} and extra actions {extra}")
+        for action, rule in normalized.items():
+            if rule.action != action:
+                raise ValueError(f"Poll transition rule key {action!r} does not match action {rule.action!r}")
+            if not rule.source_statuses <= POLL_STATUSES:
+                raise ValueError(f"Poll transition rule {action!r} has unknown source statuses")
+            if rule.restore_archived_status:
+                if rule.target_status is not None:
+                    raise ValueError(f"Restoring transition {action!r} cannot have a fixed target")
+                if rule.missing_restore_target_status not in (POLL_STATUSES - {"archived"}) | {None}:
+                    raise ValueError(f"Restoring transition {action!r} has an invalid missing-origin target")
+            elif rule.target_status not in POLL_STATUSES:
+                raise ValueError(f"Poll transition rule {action!r} has an unknown target status")
+            elif rule.missing_restore_target_status is not None:
+                raise ValueError(f"Non-restoring transition {action!r} cannot have a missing-origin target")
+        self._rules = MappingProxyType(normalized)
+
+    @property
+    def rules(self) -> Mapping[str, PollTransitionRule]:
+        return self._rules
+
+
+DEFAULT_POLL_TRANSITION_POLICY = PollTransitionPolicy(
+    {
+        "open": PollTransitionRule(
+            action="open",
+            source_statuses=frozenset({"draft", "closed", "decided"}),
+            target_status="open",
+        ),
+        "draft": PollTransitionRule(
+            action="draft",
+            source_statuses=frozenset({"open"}),
+            target_status="draft",
+        ),
+        "close": PollTransitionRule(
+            action="close",
+            source_statuses=frozenset({"open"}),
+            target_status="closed",
+        ),
+        "decide": PollTransitionRule(
+            action="decide",
+            source_statuses=frozenset({"closed", "decided"}),
+            target_status="decided",
+        ),
+        "archive": PollTransitionRule(
+            action="archive",
+            source_statuses=frozenset({"draft", "open", "closed", "decided"}),
+            target_status="archived",
+        ),
+        "unarchive": PollTransitionRule(
+            action="unarchive",
+            source_statuses=frozenset({"archived"}),
+            target_status=None,
+            restore_archived_status=True,
+            missing_restore_target_status="draft",
+        ),
+    }
+)
+
+
+class PollTransitionEngine:
+    """Pure lifecycle planner backed by an injected transition policy."""
+
+    def __init__(self, policy: PollTransitionPolicy = DEFAULT_POLL_TRANSITION_POLICY) -> None:
+        self.policy = policy
+
+    def plan(
+        self,
+        *,
+        current_status: str,
+        action: str,
+        archived_from_status: str | None = None,
+    ) -> PollTransitionPlan:
+        if current_status not in POLL_STATUSES:
+            raise ValueError(f"Unknown poll status: {current_status}")
+        rule = self.policy.rules.get(action)
+        if rule is None:
+            raise ValueError(f"Unknown poll transition action: {action}")
+        if current_status not in rule.source_statuses:
+            raise ValueError(f"Poll transition {action!r} is not allowed from status {current_status!r}")
+
+        target_status = rule.target_status
+        if rule.restore_archived_status:
+            if archived_from_status is None and rule.missing_restore_target_status is not None:
+                target_status = rule.missing_restore_target_status
+            elif archived_from_status not in POLL_STATUSES - {"archived"}:
+                raise ValueError("Archived poll has no valid status to restore")
+            else:
+                target_status = archived_from_status
+        assert target_status is not None
+        return PollTransitionPlan(action=action, from_status=current_status, to_status=target_status)
+
+    def available_actions(
+        self,
+        *,
+        current_status: str,
+        archived_from_status: str | None = None,
+    ) -> tuple[PollTransitionAvailability, ...]:
+        actions: list[PollTransitionAvailability] = []
+        for action in POLL_TRANSITION_ACTIONS:
+            rule = self.policy.rules[action]
+            target_status = (
+                archived_from_status or rule.missing_restore_target_status
+                if rule.restore_archived_status
+                else rule.target_status
+            )
+            try:
+                self.plan(
+                    current_status=current_status,
+                    action=action,
+                    archived_from_status=archived_from_status,
+                )
+            except ValueError as exc:
+                actions.append(
+                    PollTransitionAvailability(
+                        action=action,
+                        target_status=target_status,
+                        available=False,
+                        reason=str(exc),
+                    )
+                )
+            else:
+                actions.append(
+                    PollTransitionAvailability(
+                        action=action,
+                        target_status=target_status,
+                        available=True,
+                    )
+                )
+        return tuple(actions)
+
+
+DEFAULT_POLL_TRANSITION_ENGINE = PollTransitionEngine()
+
+
+__all__ = [
+    "DEFAULT_POLL_TRANSITION_ENGINE",
+    "DEFAULT_POLL_TRANSITION_POLICY",
+    "POLL_STATUSES",
+    "POLL_TRANSITION_ACTIONS",
+    "PollTransitionAvailability",
+    "PollTransitionEngine",
+    "PollTransitionPlan",
+    "PollTransitionPolicy",
+    "PollTransitionRule",
+]

tests/test_authorization.py

@@ -10,7 +10,7 @@ from sqlalchemy.orm import Session, sessionmaker
 from govoplan_core.auth import ApiPrincipal
 from govoplan_core.core.access import PrincipalRef
 from govoplan_core.db.base import Base
-from govoplan_poll.backend.db.models import Poll, PollInvitation, PollOption, PollResponse
+from govoplan_poll.backend.db.models import Poll, PollInvitation, PollLifecycleTransition, PollOption, PollResponse
 from govoplan_poll.backend.router import (
     api_create_poll,
     api_get_poll,
@@ -39,7 +39,13 @@ class PollAuthorizationTests(unittest.TestCase):
         self.engine = create_engine("sqlite:///:memory:")
         Base.metadata.create_all(
             self.engine,
-            tables=[Poll.__table__, PollOption.__table__, PollResponse.__table__, PollInvitation.__table__],
+            tables=[
+                Poll.__table__,
+                PollOption.__table__,
+                PollResponse.__table__,
+                PollInvitation.__table__,
+                PollLifecycleTransition.__table__,
+            ],
         )
         self.Session = sessionmaker(bind=self.engine)
         self.session: Session = self.Session()
@@ -48,7 +54,13 @@ class PollAuthorizationTests(unittest.TestCase):
         self.session.close()
         Base.metadata.drop_all(
             self.engine,
-            tables=[PollInvitation.__table__, PollResponse.__table__, PollOption.__table__, Poll.__table__],
+            tables=[
+                PollLifecycleTransition.__table__,
+                PollInvitation.__table__,
+                PollResponse.__table__,
+                PollOption.__table__,
+                Poll.__table__,
+            ],
         )
         self.engine.dispose()
 

tests/test_lifecycle.py

@@ -0,0 +1,467 @@
+from __future__ import annotations
+
+import unittest
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+
+from fastapi import HTTPException
+from pydantic import ValidationError
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.db.base import Base
+from govoplan_poll.backend.db.models import (
+    Poll,
+    PollInvitation,
+    PollLifecycleTransition,
+    PollOption,
+    PollResponse,
+)
+from govoplan_poll.backend.router import api_poll_lifecycle, api_transition_poll
+from govoplan_poll.backend.schemas import (
+    PollAnswerInput,
+    PollCreateRequest,
+    PollOptionInput,
+    PollSubmitResponseRequest,
+    PollTransitionRequest,
+)
+from govoplan_poll.backend.service import (
+    PollError,
+    create_poll,
+    list_poll_lifecycle_transitions,
+    poll_response,
+    poll_results_are_visible,
+    response_datetime,
+    submit_poll_response,
+    transition_poll,
+)
+from govoplan_poll.backend.transitions import (
+    DEFAULT_POLL_TRANSITION_POLICY,
+    PollTransitionEngine,
+    PollTransitionPolicy,
+    PollTransitionRule,
+)
+
+
+WRITE_SCOPE = "poll:poll:write"
+
+
+class PollLifecycleTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(
+            self.engine,
+            tables=[
+                Poll.__table__,
+                PollOption.__table__,
+                PollResponse.__table__,
+                PollInvitation.__table__,
+                PollLifecycleTransition.__table__,
+            ],
+        )
+        self.Session = sessionmaker(bind=self.engine)
+        self.session: Session = self.Session()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(
+            self.engine,
+            tables=[
+                PollLifecycleTransition.__table__,
+                PollInvitation.__table__,
+                PollResponse.__table__,
+                PollOption.__table__,
+                Poll.__table__,
+            ],
+        )
+        self.engine.dispose()
+
+    def _poll(self, *, status: str = "draft", title: str = "Decision") -> Poll:
+        return create_poll(
+            self.session,
+            tenant_id="tenant-1",
+            user_id="owner",
+            payload=PollCreateRequest(
+                title=title,
+                kind="single_choice",
+                status=status,
+                visibility="tenant",
+                options=[
+                    PollOptionInput(key="yes", label="Yes"),
+                    PollOptionInput(key="no", label="No"),
+                ],
+            ),
+        )
+
+    @staticmethod
+    def _principal() -> ApiPrincipal:
+        return ApiPrincipal(
+            principal=PrincipalRef(
+                account_id="owner",
+                membership_id="owner-membership",
+                tenant_id="tenant-1",
+                scopes=frozenset({WRITE_SCOPE}),
+                display_name="Owner",
+            ),
+            account=SimpleNamespace(id="owner"),
+            user=SimpleNamespace(id="owner-membership"),
+        )
+
+    def test_default_policy_exposes_every_allowed_and_rejected_transition(self) -> None:
+        expected = {
+            "draft": {"open", "archive"},
+            "open": {"draft", "close", "archive"},
+            "closed": {"open", "decide", "archive"},
+            "decided": {"open", "decide", "archive"},
+            "archived": {"unarchive"},
+        }
+        engine = PollTransitionEngine()
+
+        for status, allowed in expected.items():
+            archived_from_status = "closed" if status == "archived" else None
+            availability = engine.available_actions(
+                current_status=status,
+                archived_from_status=archived_from_status,
+            )
+            self.assertEqual({item.action for item in availability if item.available}, allowed)
+            for item in availability:
+                if item.available:
+                    self.assertIsNone(item.reason)
+                else:
+                    self.assertTrue(item.reason)
+
+        with self.assertRaisesRegex(ValueError, "not allowed"):
+            engine.plan(current_status="draft", action="close")
+
+    def test_injected_policy_governs_service_transition_and_projection(self) -> None:
+        rules = dict(DEFAULT_POLL_TRANSITION_POLICY.rules)
+        rules["close"] = PollTransitionRule(
+            action="close",
+            source_statuses=frozenset({"draft", "open"}),
+            target_status="closed",
+        )
+        custom_engine = PollTransitionEngine(PollTransitionPolicy(rules))
+        poll = self._poll()
+
+        projected = poll_response(poll, transition_engine=custom_engine)
+        close_action = next(item for item in projected["lifecycle_actions"] if item["action"] == "close")
+        result = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="close",
+            transition_engine=custom_engine,
+        )
+
+        self.assertTrue(close_action["available"])
+        self.assertEqual(result.poll.status, "closed")
+        self.assertEqual(result.transition.from_status, "draft")
+
+    def test_only_draft_or_open_are_valid_initial_states(self) -> None:
+        for initial_status in ("closed", "decided", "archived"):
+            with self.subTest(initial_status=initial_status):
+                with self.assertRaises(ValidationError):
+                    PollCreateRequest(title="Invalid initial state", kind="yes_no", status=initial_status)
+
+        bypassed_schema = PollCreateRequest(title="Invalid initial state", kind="yes_no").model_copy(
+            update={"status": "closed"}
+        )
+        with self.assertRaisesRegex(PollError, "initial status must be draft or open"):
+            create_poll(
+                self.session,
+                tenant_id="tenant-1",
+                user_id="owner",
+                payload=bypassed_schema,
+            )
+
+    def test_reopening_preserves_responses_close_history_and_previous_decision(self) -> None:
+        poll = self._poll()
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
+        response = submit_poll_response(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            payload=PollSubmitResponseRequest(
+                respondent_id="participant",
+                answers=[PollAnswerInput(option_key="yes")],
+            ),
+        )
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close")
+        first_closed_at = poll.closed_at
+        transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="decide",
+            option_key="yes",
+        )
+        decided_at = poll.decided_at
+        decided_option_id = poll.decided_option_id
+
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
+
+        self.assertEqual(poll.status, "open")
+        self.assertEqual(response_datetime(poll.closed_at), response_datetime(first_closed_at))
+        self.assertEqual(response_datetime(poll.decided_at), response_datetime(decided_at))
+        self.assertEqual(poll.decided_option_id, decided_option_id)
+        self.assertEqual(response.deleted_at, None)
+        self.assertEqual(response.answers[0]["option_key"], "yes")
+        history = list_poll_lifecycle_transitions(self.session, tenant_id="tenant-1", poll_id=poll.id)
+        self.assertEqual([item.action for item in history], ["open", "close", "decide", "open"])
+
+    def test_direct_redecision_is_audited_and_exact_retry_is_idempotent(self) -> None:
+        poll = self._poll()
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close")
+        first = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="decide",
+            option_key="yes",
+            idempotency_key="decision-1",
+        )
+        replay = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="decide",
+            option_key="yes",
+            idempotency_key="decision-1",
+        )
+        second = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="decide",
+            option_key="no",
+            idempotency_key="decision-2",
+        )
+
+        self.assertFalse(first.replayed)
+        self.assertTrue(replay.replayed)
+        self.assertEqual(replay.transition.id, first.transition.id)
+        self.assertFalse(second.replayed)
+        self.assertEqual(second.transition.previous_decision_option_id, first.transition.decision_option_id)
+        self.assertEqual(poll.decided_option_id, second.transition.decision_option_id)
+        self.assertEqual(
+            self.session.query(PollLifecycleTransition)
+            .filter(
+                PollLifecycleTransition.poll_id == poll.id,
+                PollLifecycleTransition.action == "decide",
+            )
+            .count(),
+            2,
+        )
+
+        with self.assertRaisesRegex(PollError, "different poll transition"):
+            transition_poll(
+                self.session,
+                tenant_id="tenant-1",
+                poll_id=poll.id,
+                action="decide",
+                option_key="yes",
+                idempotency_key="decision-2",
+            )
+
+    def test_repeated_exact_transition_without_an_identity_is_rejected(self) -> None:
+        poll = self._poll()
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
+
+        with self.assertRaisesRegex(PollError, "not allowed"):
+            transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
+
+    def test_archive_and_unarchive_restore_status_and_append_audit_history(self) -> None:
+        poll = self._poll()
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close")
+        response = PollResponse(
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            respondent_id="participant",
+            answers=[{"option_key": "yes"}],
+            submitted_at=poll.created_at,
+        )
+        self.session.add(response)
+        self.session.flush()
+
+        archived = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="archive",
+            idempotency_key="archive-1",
+        )
+        restored = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="unarchive",
+            idempotency_key="unarchive-1",
+        )
+
+        self.assertEqual(archived.transition.to_status, "archived")
+        self.assertEqual(restored.transition.to_status, "closed")
+        self.assertEqual(poll.status, "closed")
+        self.assertIsNone(poll.archived_from_status)
+        self.assertIsNone(response.deleted_at)
+        self.assertEqual(
+            [item.action for item in poll.lifecycle_transitions],
+            ["open", "close", "archive", "unarchive"],
+        )
+
+    def test_legacy_archive_without_origin_restores_to_draft_with_audit_marker(self) -> None:
+        poll = Poll(
+            tenant_id="tenant-1",
+            slug="legacy-archive",
+            title="Legacy archive",
+            kind="yes_no",
+            status="archived",
+            visibility="tenant",
+            result_visibility="after_close",
+            workflow_steps=[],
+            allow_anonymous=False,
+            allow_response_update=True,
+            min_choices=1,
+        )
+        self.session.add(poll)
+        self.session.flush()
+
+        restored = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="unarchive",
+        )
+
+        self.assertEqual(restored.poll.status, "draft")
+        self.assertEqual(restored.transition.to_status, "draft")
+        self.assertEqual(
+            restored.transition.metadata_["legacy_restore_fallback"],
+            {
+                "reason": "missing_archived_from_status",
+                "restored_status": "draft",
+            },
+        )
+
+    def test_reopening_clears_only_an_expired_deadline_and_preserves_close_history(self) -> None:
+        poll = self._poll()
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close")
+        closed_at = poll.closed_at
+        expired_deadline = datetime.now(timezone.utc) - timedelta(days=1)
+        poll.closes_at = expired_deadline
+
+        reopened = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="open",
+        )
+
+        self.assertEqual(reopened.poll.status, "open")
+        self.assertIsNone(reopened.poll.closes_at)
+        self.assertEqual(response_datetime(reopened.poll.closed_at), response_datetime(closed_at))
+        self.assertEqual(
+            reopened.transition.metadata_["cleared_expired_closes_at"],
+            expired_deadline.isoformat(),
+        )
+        submit_poll_response(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            payload=PollSubmitResponseRequest(
+                respondent_id="participant-after-reopen",
+                answers=[PollAnswerInput(option_key="yes")],
+            ),
+        )
+
+        future_deadline = datetime.now(timezone.utc) + timedelta(days=1)
+        future_poll = self._poll(title="Future deadline")
+        future_poll.closes_at = future_deadline
+        opened = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=future_poll.id,
+            action="open",
+        )
+        self.assertEqual(response_datetime(opened.poll.closes_at), future_deadline)
+        self.assertNotIn("cleared_expired_closes_at", opened.transition.metadata_)
+
+    def test_archiving_draft_or_open_poll_does_not_grant_after_close_visibility(self) -> None:
+        for initial_status in ("draft", "open"):
+            with self.subTest(initial_status=initial_status):
+                poll = self._poll(status=initial_status, title=f"Archive from {initial_status}")
+                transition_poll(
+                    self.session,
+                    tenant_id="tenant-1",
+                    poll_id=poll.id,
+                    action="archive",
+                )
+
+                self.assertFalse(
+                    poll_results_are_visible(
+                        self.session,
+                        poll=poll,
+                        actor_ids=("participant",),
+                    )
+                )
+
+        closed_poll = self._poll(title="Archive from closed")
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=closed_poll.id, action="open")
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=closed_poll.id, action="close")
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=closed_poll.id, action="archive")
+        self.assertTrue(
+            poll_results_are_visible(
+                self.session,
+                poll=closed_poll,
+                actor_ids=("participant",),
+            )
+        )
+
+    def test_generic_api_returns_action_availability_history_and_actor(self) -> None:
+        poll = self._poll()
+        principal = self._principal()
+
+        opened = api_transition_poll(
+            poll.id,
+            PollTransitionRequest(action="open"),
+            idempotency_key="open-api-1",
+            session=self.session,
+            principal=principal,
+        )
+        replay = api_transition_poll(
+            poll.id,
+            PollTransitionRequest(action="open"),
+            idempotency_key="open-api-1",
+            session=self.session,
+            principal=principal,
+        )
+        lifecycle = api_poll_lifecycle(poll.id, session=self.session, principal=principal)
+
+        self.assertEqual(opened.poll.status, "open")
+        self.assertFalse(opened.replayed)
+        self.assertTrue(replay.replayed)
+        self.assertEqual(len(lifecycle.history), 1)
+        self.assertEqual(lifecycle.history[0].actor_user_id, "owner-membership")
+        self.assertEqual(
+            {item.action for item in lifecycle.actions if item.available},
+            {"draft", "close", "archive"},
+        )
+
+        draft = self._poll(title="Draft cannot close")
+        with self.assertRaises(HTTPException) as rejected:
+            api_transition_poll(
+                draft.id,
+                PollTransitionRequest(action="close"),
+                idempotency_key=None,
+                session=self.session,
+                principal=principal,
+            )
+        self.assertEqual(rejected.exception.status_code, 400)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_response_editing.py

@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.core.poll import PollCapabilityError, PollOptionUpdateCommand, PollSchedulingProvider
+from govoplan_core.db.base import Base
+from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
+from govoplan_poll.backend.db.models import Poll, PollOption, PollResponse
+from govoplan_poll.backend.schemas import PollAnswerInput, PollCreateRequest, PollOptionInput, PollSubmitResponseRequest
+from govoplan_poll.backend.service import create_poll, submit_poll_response
+
+
+class PollResponseEditingTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(
+            self.engine,
+            tables=[Poll.__table__, PollOption.__table__, PollResponse.__table__],
+        )
+        self.Session = sessionmaker(bind=self.engine)
+        self.session: Session = self.Session()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(
+            self.engine,
+            tables=[PollResponse.__table__, PollOption.__table__, Poll.__table__],
+        )
+        self.engine.dispose()
+
+    def _poll(self, *, allow_response_update: bool = True) -> Poll:
+        return create_poll(
+            self.session,
+            tenant_id="tenant-1",
+            user_id="organizer-1",
+            payload=PollCreateRequest(
+                title="Availability",
+                kind="availability",
+                status="open",
+                min_choices=1,
+                max_choices=2,
+                allow_response_update=allow_response_update,
+                options=[
+                    PollOptionInput(
+                        key="slot-1",
+                        label="Monday",
+                        value={"slot_id": "slot-1"},
+                        metadata={"source": "scheduling"},
+                    ),
+                    PollOptionInput(
+                        key="slot-2",
+                        label="Tuesday",
+                        value={"slot_id": "slot-2"},
+                    ),
+                ],
+            ),
+        )
+
+    def _submit_both(self, poll: Poll, respondent_id: str) -> PollResponse:
+        return submit_poll_response(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            payload=PollSubmitResponseRequest(
+                respondent_id=respondent_id,
+                answers=[
+                    PollAnswerInput(option_id=poll.options[0].id, value="available"),
+                    PollAnswerInput(option_id=poll.options[1].id, value="maybe"),
+                ],
+            ),
+        )
+
+    def test_provider_projects_answers_and_option_change_invalidates_only_that_option(self) -> None:
+        poll = self._poll()
+        first = self._submit_both(poll, "person-1")
+        second = self._submit_both(poll, "person-2")
+        provider = SqlPollSchedulingProvider()
+
+        self.assertIsInstance(provider, PollSchedulingProvider)
+        projected = provider.list_responses(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+        )
+        self.assertEqual([answer.value for answer in projected[0].answers], ["available", "maybe"])
+        current = provider.get_response(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            respondent_ids=("person-1",),
+        )
+        self.assertIsNotNone(current)
+        self.assertEqual([answer.value for answer in current.answers], ["available", "maybe"])
+
+        option_ref = provider.update_option(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            option_id=poll.options[0].id,
+            command=PollOptionUpdateCommand(
+                label="Monday, revised",
+                value={"slot_id": "slot-1"},
+                metadata={"source": "scheduling"},
+            ),
+        )
+
+        self.assertEqual(option_ref.id, poll.options[0].id)
+        self.assertEqual(first.answers, [{"option_id": poll.options[1].id, "option_key": "slot-2", "value": "maybe"}])
+        self.assertEqual(second.answers, [{"option_id": poll.options[1].id, "option_key": "slot-2", "value": "maybe"}])
+
+        # Re-answer, then replay the exact option snapshot. No response data is
+        # invalidated because the option did not actually change.
+        first = self._submit_both(poll, "person-1")
+        provider.update_option(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            option_id=poll.options[0].id,
+            command=PollOptionUpdateCommand(
+                label="Monday, revised",
+                value={"slot_id": "slot-1"},
+                metadata={"source": "scheduling"},
+            ),
+        )
+        self.assertEqual(len(first.answers), 2)
+
+    def test_fully_invalidated_response_is_no_longer_active_or_counted(self) -> None:
+        poll = self._poll()
+        response = submit_poll_response(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            payload=PollSubmitResponseRequest(
+                respondent_id="person-1",
+                answers=[PollAnswerInput(option_id=poll.options[0].id, value="available")],
+            ),
+        )
+
+        SqlPollSchedulingProvider().update_option(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            option_id=poll.options[0].id,
+            command=PollOptionUpdateCommand(label="Monday, revised", value={"slot_id": "slot-1"}, metadata={"source": "scheduling"}),
+        )
+
+        self.assertEqual(response.answers, [])
+        self.assertIsNotNone(response.deleted_at)
+        self.assertEqual(
+            SqlPollSchedulingProvider().list_responses(self.session, tenant_id="tenant-1", poll_id=poll.id),
+            (),
+        )
+
+    def test_option_change_is_rejected_when_responses_cannot_be_updated(self) -> None:
+        poll = self._poll(allow_response_update=False)
+        response = self._submit_both(poll, "person-1")
+
+        with self.assertRaisesRegex(PollCapabilityError, "response updates are disabled"):
+            SqlPollSchedulingProvider().update_option(
+                self.session,
+                tenant_id="tenant-1",
+                poll_id=poll.id,
+                option_id=poll.options[0].id,
+                command=PollOptionUpdateCommand(
+                    label="Monday, revised",
+                    value={"slot_id": "slot-1"},
+                    metadata={"source": "scheduling"},
+                ),
+            )
+
+        self.assertEqual(poll.options[0].label, "Monday")
+        self.assertEqual(len(response.answers), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_service.py

@@ -9,7 +9,7 @@ from sqlalchemy.orm import Session, sessionmaker
 from govoplan_core.db.base import Base
 from govoplan_core.core.poll import PollResponseRef, PollResponseSubmissionProvider, PollSchedulingProvider
 from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
-from govoplan_poll.backend.db.models import Poll, PollInvitation, PollOption, PollResponse
+from govoplan_poll.backend.db.models import Poll, PollInvitation, PollLifecycleTransition, PollOption, PollResponse
 from govoplan_poll.backend.schemas import (
     PollAnswerInput,
     PollCreateRequest,
@@ -33,7 +33,13 @@ class PollServiceTests(unittest.TestCase):
         self.engine = create_engine("sqlite:///:memory:")
         Base.metadata.create_all(
             self.engine,
-            tables=[Poll.__table__, PollOption.__table__, PollResponse.__table__, PollInvitation.__table__],
+            tables=[
+                Poll.__table__,
+                PollOption.__table__,
+                PollResponse.__table__,
+                PollInvitation.__table__,
+                PollLifecycleTransition.__table__,
+            ],
         )
         self.Session = sessionmaker(bind=self.engine)
         self.session: Session = self.Session()
@@ -42,7 +48,13 @@ class PollServiceTests(unittest.TestCase):
         self.session.close()
         Base.metadata.drop_all(
             self.engine,
-            tables=[PollInvitation.__table__, PollResponse.__table__, PollOption.__table__, Poll.__table__],
+            tables=[
+                PollLifecycleTransition.__table__,
+                PollInvitation.__table__,
+                PollResponse.__table__,
+                PollOption.__table__,
+                Poll.__table__,
+            ],
         )
         self.engine.dispose()