feat(campaigns): add accountable work assignments

tests/test_access_provider.py

@@ -28,6 +28,8 @@ from govoplan_campaign.backend.db.models import (
     CampaignMessageActionAttempt,
     CampaignShare,
     CampaignVersion,
+    CampaignWorkAssignment,
+    CampaignWorkAssignmentEvent,
     ImapAppendAttempt,
     PostboxDeliveryAttempt,
     PrintOutputAttempt,
@@ -47,6 +49,81 @@ GROUP_ID = "group-1"
 
 
 class CampaignAccessProviderTests(unittest.TestCase):
+    def test_work_assignment_provenance_keeps_accountability_separate_from_access(self) -> None:
+        session = _session()
+        self.addCleanup(_close_session, session)
+        _seed_access_subjects(session)
+        campaign = Campaign(
+            id="campaign-work",
+            tenant_id=TENANT_ID,
+            owner_user_id=OTHER_USER_ID,
+            external_id="work",
+            name="Work campaign",
+        )
+        checked_at = datetime.now(UTC)
+        assignment = CampaignWorkAssignment(
+            id="assignment-1",
+            tenant_id=TENANT_ID,
+            campaign_id=campaign.id,
+            purpose="Not disclosed by provenance",
+            status="open",
+            assignee_type="account",
+            assignee_id="account-1",
+            assignee_label_snapshot="Subject",
+            assignee_resolution_state="resolved",
+            resolution_provenance={"policy_code": "assignment_does_not_grant_access"},
+            resolution_checked_at=checked_at,
+            assigned_by_user_id=OTHER_USER_ID,
+            assigned_by_label_snapshot="Other user",
+        )
+        event = CampaignWorkAssignmentEvent(
+            id="assignment-event-1",
+            tenant_id=TENANT_ID,
+            campaign_id=campaign.id,
+            assignment_id=assignment.id,
+            event_kind="assigned",
+            actor_user_id=OTHER_USER_ID,
+            actor_label_snapshot="Other user",
+            status_snapshot="open",
+            assignee_type_snapshot="account",
+            assignee_id_snapshot="account-1",
+            assignee_label_snapshot="Subject",
+            resolution_state_snapshot="resolved",
+        )
+        session.add_all(
+            [
+                campaign,
+                assignment,
+                event,
+                CampaignShare(
+                    id="share-work",
+                    tenant_id=TENANT_ID,
+                    campaign_id=campaign.id,
+                    target_type="group",
+                    target_id=GROUP_ID,
+                    permission="read",
+                ),
+            ]
+        )
+        session.commit()
+
+        items = CampaignAccessService().explain_resource_provenance(
+            session,
+            _principal(
+                scopes={"campaigns:campaign:read", "campaigns:assignment:read"},
+                group_ids={GROUP_ID},
+            ),
+            resource_type="campaign_work_assignment",
+            resource_id=assignment.id,
+            action="campaigns:assignment:read",
+        )
+        work = next(item for item in items if item.source == "campaigns.work_assignment")
+        self.assertEqual("accountability_does_not_grant_access", work.details["authorization_mode"])
+        self.assertEqual(["campaigns:assignment:read"], work.details["permission_actions"])
+        self.assertFalse(work.details["purpose_disclosed"])
+        self.assertNotIn("Not disclosed", repr(items))
+        self.assertTrue(any(item.id == "share-work" for item in items))
+
     def test_collaboration_access_is_explained_independently_from_campaign_edit(self) -> None:
         session = _session()
         self.addCleanup(_close_session, session)
@@ -975,6 +1052,8 @@ def _session():
             Campaign.__table__,
             CampaignShare.__table__,
             CampaignCollaborationEntry.__table__,
+            CampaignWorkAssignment.__table__,
+            CampaignWorkAssignmentEvent.__table__,
             CampaignVersion.__table__,
             CampaignJob.__table__,
             CampaignIssue.__table__,

tests/test_campaign_work_assignments.py

@@ -0,0 +1,472 @@
+from __future__ import annotations
+
+import importlib
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+from fastapi import HTTPException
+from sqlalchemy import create_engine, inspect, text
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_access.backend.db.models import Account, Group, User
+from govoplan_campaign.backend.db.models import (
+    Campaign,
+    CampaignShare,
+    CampaignVersion,
+    CampaignWorkAssignment,
+    CampaignWorkAssignmentEvent,
+)
+from govoplan_campaign.backend.routes.assignments import (
+    create_campaign_work_assignment,
+    list_campaign_work_assignment_history,
+    reassign_campaign_work_assignment,
+    reconcile_campaign_work_assignments,
+    transition_campaign_work_assignment,
+)
+from govoplan_campaign.backend.schemas import (
+    CampaignWorkAssigneeInput,
+    CampaignWorkAssignmentCreateRequest,
+    CampaignWorkAssignmentReassignRequest,
+    CampaignWorkAssignmentTransitionRequest,
+)
+from govoplan_core.core.access import GroupRef, UserRef
+from govoplan_core.core.change_sequence import ChangeSequenceEntry
+from govoplan_core.core.organizations import OrganizationFunctionRef
+from govoplan_core.core.tasks import WorkItem
+from govoplan_core.db.base import Base
+
+
+TENANT_ID = "tenant-1"
+
+
+class _Principal:
+    tenant_id = TENANT_ID
+    api_key = None
+
+    def __init__(self, user_id: str, account_id: str, *scopes: str) -> None:
+        self.user = SimpleNamespace(
+            id=user_id,
+            display_name=f"User {user_id}",
+            email=f"{user_id}@example.test",
+        )
+        self.account_id = account_id
+        self.scopes = frozenset(scopes)
+
+    def has(self, scope: str) -> bool:
+        return scope in self.scopes or "tenant:*" in self.scopes
+
+
+class _Directory:
+    def __init__(self) -> None:
+        self.user_2_active = True
+
+    def users_for_tenant(self, tenant_id: str):
+        if tenant_id != TENANT_ID:
+            return ()
+        return (
+            UserRef(id="user-1", account_id="account-1", tenant_id=TENANT_ID, display_name="Owner"),
+            UserRef(
+                id="user-2",
+                account_id="account-2",
+                tenant_id=TENANT_ID,
+                display_name="Collaborator",
+                status="active" if self.user_2_active else "inactive",
+            ),
+            UserRef(id="user-3", account_id="account-3", tenant_id=TENANT_ID, display_name="Unrelated"),
+        )
+
+    def groups_for_tenant(self, tenant_id: str):
+        return (GroupRef(id="group-1", tenant_id=TENANT_ID, name="Campaign group"),) if tenant_id == TENANT_ID else ()
+
+    def groups_for_user(self, user_id: str, *, tenant_id: str):
+        if tenant_id == TENANT_ID and user_id == "user-2":
+            return (GroupRef(id="group-1", tenant_id=TENANT_ID, name="Campaign group"),)
+        return ()
+
+
+class _Tasks:
+    def __init__(self) -> None:
+        self.commands = []
+
+    def create_task(self, _session, _principal, *, command):
+        self.commands.append(command)
+        return WorkItem(
+            id=f"task-{len(self.commands)}",
+            provider_id="tasks",
+            owner_module="tasks",
+            tenant_id=command.tenant_id,
+            title=command.title,
+            assignments=command.assignments,
+            sources=command.sources,
+        )
+
+
+class _Notifications:
+    def __init__(self) -> None:
+        self.requests = []
+
+    def tenant_id_for_notification(self, _session, *, notification_id: str):
+        del notification_id
+        return TENANT_ID
+
+    def enqueue_notification(self, _session, request, *, enqueue_delivery: bool = True):
+        self.requests.append((request, enqueue_delivery))
+        return {"id": f"notification-{len(self.requests)}"}
+
+    def deliver_notification(self, _session, *, notification_id: str):
+        return {"id": notification_id}
+
+    def deliver_pending(self, _session, *, tenant_id=None, limit: int = 50):
+        return {"tenant_id": tenant_id, "limit": limit}
+
+
+class _FailingTasks(_Tasks):
+    def create_task(self, _session, _principal, *, command):
+        del command
+        raise RuntimeError("Tasks is temporarily unavailable")
+
+
+class _Registry:
+    def __init__(self, tasks: _Tasks | None = None, notifications: _Notifications | None = None) -> None:
+        self.tasks = tasks
+        self.notifications = notifications
+
+    def has_capability(self, name: str) -> bool:
+        return (
+            (name == "tasks.commands" and self.tasks is not None)
+            or (name == "notifications.dispatch" and self.notifications is not None)
+        )
+
+    def capability(self, name: str):
+        if name == "tasks.commands":
+            return self.tasks
+        if name == "notifications.dispatch":
+            return self.notifications
+        return None
+
+
+class _Organizations:
+    def get_function(self, function_id: str):
+        if function_id != "function-1":
+            return None
+        return OrganizationFunctionRef(
+            id=function_id,
+            tenant_id=TENANT_ID,
+            organization_unit_id="unit-1",
+            slug="campaign-review",
+            name="Campaign review function",
+        )
+
+
+class _Idm:
+    def organization_function_assignments_for_function(self, function_id: str, *, tenant_id=None, effective_at=None):
+        del effective_at
+        if function_id == "function-1" and tenant_id == TENANT_ID:
+            return (SimpleNamespace(account_id="account-2", status="active", valid_from=None, valid_until=None),)
+        return ()
+
+    def organization_function_incumbencies(self, function_ids, *, tenant_id, effective_at=None):
+        del effective_at
+        return {item: SimpleNamespace(assignments=self.organization_function_assignments_for_function(item, tenant_id=tenant_id)) for item in function_ids}
+
+
+@pytest.fixture()
+def session() -> Session:
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    Base.metadata.create_all(
+        engine,
+        tables=[
+            Account.__table__,
+            User.__table__,
+            Group.__table__,
+            ChangeSequenceEntry.__table__,
+            Campaign.__table__,
+            CampaignVersion.__table__,
+            CampaignShare.__table__,
+            CampaignWorkAssignment.__table__,
+            CampaignWorkAssignmentEvent.__table__,
+        ],
+    )
+    factory = sessionmaker(bind=engine, class_=Session, expire_on_commit=False)
+    database = factory()
+    database.add_all(
+        [
+            Account(id=f"account-{number}", email=f"user-{number}@example.test", normalized_email=f"user-{number}@example.test")
+            for number in range(1, 4)
+        ]
+    )
+    database.flush()
+    database.add_all(
+        [
+            User(id=f"user-{number}", tenant_id=TENANT_ID, account_id=f"account-{number}", email=f"user-{number}@example.test")
+            for number in range(1, 4)
+        ]
+    )
+    database.add(Group(id="group-1", tenant_id=TENANT_ID, slug="campaign-group", name="Campaign group"))
+    campaign = Campaign(
+        id="campaign-1",
+        tenant_id=TENANT_ID,
+        owner_user_id="user-1",
+        external_id="campaign-1",
+        name="Campaign One",
+        current_version_id="version-1",
+    )
+    database.add_all(
+        [
+            campaign,
+            CampaignVersion(id="version-1", campaign_id=campaign.id, version_number=1, raw_json={"version": "1.0"}),
+            CampaignShare(
+                id="share-1",
+                tenant_id=TENANT_ID,
+                campaign_id=campaign.id,
+                target_type="group",
+                target_id="group-1",
+                permission="read",
+            ),
+        ]
+    )
+    database.commit()
+    try:
+        yield database
+    finally:
+        database.close()
+        engine.dispose()
+
+
+def _manager() -> _Principal:
+    return _Principal(
+        "user-1",
+        "account-1",
+        "campaigns:campaign:read",
+        "campaigns:assignment:read",
+        "campaigns:assignment:manage",
+        "campaigns:assignment:complete",
+    )
+
+
+def _assignee() -> _Principal:
+    return _Principal(
+        "user-2",
+        "account-2",
+        "campaigns:campaign:read",
+        "campaigns:assignment:read",
+        "campaigns:assignment:complete",
+    )
+
+
+def _commit_audit(session: Session, *_args, **_kwargs) -> None:
+    session.commit()
+
+
+def test_assignment_is_authorization_neutral_and_mirrors_through_optional_tasks(session: Session) -> None:
+    directory = _Directory()
+    tasks = _Tasks()
+    notifications = _Notifications()
+    with (
+        patch("govoplan_campaign.backend.routes.assignments._access_directory", return_value=directory),
+        patch("govoplan_campaign.backend.routes.assignments.get_registry", return_value=_Registry(tasks, notifications)),
+        patch("govoplan_campaign.backend.routes.assignments.audit_from_principal", side_effect=_commit_audit),
+    ):
+        response = create_campaign_work_assignment(
+            "campaign-1",
+            CampaignWorkAssignmentCreateRequest(
+                purpose="Review the frozen recipient import",
+                assignee=CampaignWorkAssigneeInput(type="account", id="account-2"),
+                reference={"kind": "campaign_version", "id": "version-1"},
+            ),
+            session,
+            _manager(),
+        )
+
+    assert response.assignee_label_snapshot == "Collaborator"
+    assert response.assignee_resolution_state == "resolved"
+    assert response.resolution_provenance["policy_code"] == "assignment_does_not_grant_access"
+    assert response.task_mirror_id == "task-1"
+    assert tasks.commands[0].assignments[0].kind == "account"
+    assert tasks.commands[0].provenance["authorization_neutral"] is True
+    assert notifications.requests[0][0].recipient_id == "account-2"
+    assert notifications.requests[0][0].payload["purpose_disclosed"] is False
+    assert notifications.requests[0][1] is False
+    assert session.query(CampaignShare).count() == 1
+    assert [item.event_kind for item in session.query(CampaignWorkAssignmentEvent).all()] == ["assigned"]
+
+
+def test_assignment_rejects_target_without_existing_campaign_access(session: Session) -> None:
+    with patch("govoplan_campaign.backend.routes.assignments._access_directory", return_value=_Directory()):
+        with pytest.raises(HTTPException) as exc_info:
+            create_campaign_work_assignment(
+                "campaign-1",
+                CampaignWorkAssignmentCreateRequest(
+                    purpose="Should not grant access",
+                    assignee=CampaignWorkAssigneeInput(type="account", id="account-3"),
+                ),
+                session,
+                _manager(),
+            )
+
+    assert exc_info.value.status_code == 422
+    assert exc_info.value.detail["code"] == "campaign_assignment_assignee_inaccessible"
+    assert session.query(CampaignWorkAssignment).count() == 0
+    assert session.query(CampaignShare).count() == 1
+
+
+def test_optional_tasks_failure_is_recorded_without_blocking_campaign_work(session: Session) -> None:
+    with (
+        patch("govoplan_campaign.backend.routes.assignments._access_directory", return_value=_Directory()),
+        patch("govoplan_campaign.backend.routes.assignments.get_registry", return_value=_Registry(_FailingTasks())),
+        patch("govoplan_campaign.backend.routes.assignments.audit_from_principal", side_effect=_commit_audit),
+    ):
+        created = create_campaign_work_assignment(
+            "campaign-1",
+            CampaignWorkAssignmentCreateRequest(
+                purpose="Continue even without Tasks",
+                assignee=CampaignWorkAssigneeInput(type="account", id="account-2"),
+            ),
+            session,
+            _manager(),
+        )
+
+    assert created.task_mirror_status == "failed"
+    assert "RuntimeError" in (created.task_mirror_error or "")
+    assert session.get(CampaignWorkAssignment, created.id) is not None
+
+
+def test_assignee_can_start_and_complete_but_not_cancel(session: Session) -> None:
+    directory = _Directory()
+    with (
+        patch("govoplan_campaign.backend.routes.assignments._access_directory", return_value=directory),
+        patch("govoplan_campaign.backend.route_support._access_directory", return_value=directory),
+        patch("govoplan_campaign.backend.routes.assignments.audit_from_principal", side_effect=_commit_audit),
+    ):
+        created = create_campaign_work_assignment(
+            "campaign-1",
+            CampaignWorkAssignmentCreateRequest(
+                purpose="Review campaign",
+                assignee=CampaignWorkAssigneeInput(type="account", id="account-2"),
+                mirror_to_tasks=False,
+            ),
+            session,
+            _manager(),
+        )
+        started = transition_campaign_work_assignment(
+            "campaign-1",
+            created.id,
+            CampaignWorkAssignmentTransitionRequest(expected_revision=1, action="start"),
+            session,
+            _assignee(),
+        )
+        completed = transition_campaign_work_assignment(
+            "campaign-1",
+            created.id,
+            CampaignWorkAssignmentTransitionRequest(expected_revision=2, action="complete"),
+            session,
+            _assignee(),
+        )
+
+    assert started.status == "in_progress"
+    assert completed.status == "completed"
+    assert completed.resource_revision == 3
+    assert [item.event_kind for item in session.query(CampaignWorkAssignmentEvent).order_by(CampaignWorkAssignmentEvent.created_at, CampaignWorkAssignmentEvent.id)] == ["assigned", "started", "completed"]
+
+
+def test_reassignment_and_deactivation_reconciliation_preserve_history(session: Session) -> None:
+    directory = _Directory()
+    with (
+        patch("govoplan_campaign.backend.routes.assignments._access_directory", return_value=directory),
+        patch("govoplan_campaign.backend.routes.assignments.audit_from_principal", side_effect=_commit_audit),
+    ):
+        created = create_campaign_work_assignment(
+            "campaign-1",
+            CampaignWorkAssignmentCreateRequest(
+                purpose="Coordinate delivery",
+                assignee=CampaignWorkAssigneeInput(type="group", id="group-1"),
+                mirror_to_tasks=False,
+            ),
+            session,
+            _manager(),
+        )
+        reassigned = reassign_campaign_work_assignment(
+            "campaign-1",
+            created.id,
+            CampaignWorkAssignmentReassignRequest(
+                expected_revision=1,
+                assignee=CampaignWorkAssigneeInput(type="account", id="account-2"),
+                reason="Named accountability is now required.",
+                mirror_to_tasks=False,
+            ),
+            session,
+            _manager(),
+        )
+        directory.user_2_active = False
+        result = reconcile_campaign_work_assignments("campaign-1", 100, session, _manager())
+        history = list_campaign_work_assignment_history("campaign-1", created.id, 50, None, session, _manager())
+
+    assert reassigned.assignee_type == "account"
+    assert result.changed == 1
+    assert result.assignments[0].assignee_resolution_state == "unavailable"
+    assert [item.event_kind for item in reversed(history.items)] == ["assigned", "reassigned", "assignee_unavailable"]
+    assert history.items[1].details["assignee_id"] == "group-1"
+
+
+def test_organization_function_requires_authorized_current_incumbencies(session: Session) -> None:
+    with (
+        patch("govoplan_campaign.backend.routes.assignments._access_directory", return_value=_Directory()),
+        patch("govoplan_campaign.backend.routes.assignments.organization_directory", return_value=_Organizations()),
+        patch("govoplan_campaign.backend.routes.assignments._idm_function_directory", return_value=_Idm()),
+        patch("govoplan_campaign.backend.routes.assignments.audit_from_principal", side_effect=_commit_audit),
+    ):
+        created = create_campaign_work_assignment(
+            "campaign-1",
+            CampaignWorkAssignmentCreateRequest(
+                purpose="Approve the recipient segment",
+                assignee=CampaignWorkAssigneeInput(type="organization_function", id="function-1"),
+                mirror_to_tasks=False,
+            ),
+            session,
+            _manager(),
+        )
+
+    assert created.assignee_label_snapshot == "Campaign review function"
+    assert created.resolution_provenance["resolved_members"] == 1
+    assert created.resolution_provenance["all_current_incumbents_authorized"] is True
+
+
+def test_assignment_migration_is_repeatable_and_creates_history_indexes() -> None:
+    migration = importlib.import_module(
+        "govoplan_campaign.backend.migrations.versions."
+        "d8e9f0a1b2c3_v0121_campaign_work_assignments"
+    )
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    with engine.begin() as connection:
+        connection.execute(text("CREATE TABLE access_users (id VARCHAR(36) PRIMARY KEY)"))
+        connection.execute(text("CREATE TABLE campaigns (id VARCHAR(36) PRIMARY KEY)"))
+        connection.execute(text("CREATE TABLE campaign_versions (id VARCHAR(36) PRIMARY KEY, campaign_id VARCHAR(36) NOT NULL)"))
+        context = MigrationContext.configure(connection)
+        with patch.object(migration, "op", Operations(context)):
+            migration.upgrade()
+            migration.upgrade()
+
+        inspector = inspect(connection)
+        assignment_columns = {item["name"] for item in inspector.get_columns("campaign_work_assignments")}
+        assignment_indexes = {item["name"] for item in inspector.get_indexes("campaign_work_assignments")}
+        event_indexes = {item["name"] for item in inspector.get_indexes("campaign_work_assignment_events")}
+        assert {
+            "purpose",
+            "assignee_type",
+            "assignee_id",
+            "assignee_label_snapshot",
+            "assignee_resolution_state",
+            "resolution_provenance",
+            "task_mirror_status",
+            "resource_revision",
+        }.issubset(assignment_columns)
+        assert "ix_campaign_work_assignments_campaign_status" in assignment_indexes
+        assert "ix_campaign_work_assignment_events_history" in event_indexes
+
+        with patch.object(migration, "op", Operations(context)):
+            migration.downgrade()
+        assert not inspect(connection).has_table("campaign_work_assignment_events")
+        assert not inspect(connection).has_table("campaign_work_assignments")

tests/test_documentation.py

@@ -480,7 +480,14 @@ def test_static_campaign_handbook_has_unique_ids_help_contexts_and_no_planned_re
             "campaign.activity.composer",
             "campaign.activity.action.post",
             "campaign.activity.action.withdraw",
-            "campaign.activity.action.redact",
+                "campaign.activity.action.redact",
+                "campaign.work",
+                "campaign.work.create",
+                "campaign.work.action.start",
+                "campaign.work.action.complete",
+                "campaign.work.action.reassign",
+                "campaign.work.action.cancel",
+                "campaign.work.history",
         }
 
     assert len(ids) == len(set(ids))

tests/test_dsar_provider.py

@@ -19,6 +19,8 @@ from govoplan_campaign.backend.db.models import (
     CampaignSchedule,
     CampaignShare,
     CampaignVersion,
+    CampaignWorkAssignment,
+    CampaignWorkAssignmentEvent,
     ImapAppendAttempt,
     PostboxDeliveryAttempt,
     PrintOutputAttempt,
@@ -102,6 +104,8 @@ class CampaignDsarProviderTests(unittest.TestCase):
                 Campaign.__table__,
                 CampaignShare.__table__,
                 CampaignCollaborationEntry.__table__,
+                CampaignWorkAssignment.__table__,
+                CampaignWorkAssignmentEvent.__table__,
                 CampaignVersion.__table__,
                 CampaignJob.__table__,
                 CampaignIssue.__table__,
@@ -437,6 +441,7 @@ class CampaignDsarProviderTests(unittest.TestCase):
         self.session.commit()
         self.provider = CampaignDsarProvider()
         self.subject = DsarSubjectRef(
+            account_id=self.account.id,
             membership_id=self.user.id,
             email="subject@example.test",
         )
@@ -455,6 +460,54 @@ class CampaignDsarProviderTests(unittest.TestCase):
             {topic.id for topic in manifest.documentation},
         )
 
+    def test_search_includes_account_assignment_and_immutable_lifecycle_evidence(self) -> None:
+        now = datetime.now(timezone.utc)
+        assignment = CampaignWorkAssignment(
+            id="work-assignment-subject",
+            tenant_id="tenant-1",
+            campaign_id=self.campaign.id,
+            purpose="Review the individualized notice",
+            status="open",
+            assignee_type="account",
+            assignee_id=self.account.id,
+            assignee_label_snapshot="Subject",
+            assignee_resolution_state="resolved",
+            resolution_provenance={"policy_code": "assignment_does_not_grant_access"},
+            resolution_checked_at=now,
+            assigned_by_user_id=self.other_user.id,
+            assigned_by_label_snapshot="Other",
+        )
+        event = CampaignWorkAssignmentEvent(
+            id="work-assignment-event-subject",
+            tenant_id="tenant-1",
+            campaign_id=self.campaign.id,
+            assignment_id=assignment.id,
+            event_kind="assigned",
+            actor_user_id=self.other_user.id,
+            actor_label_snapshot="Other",
+            status_snapshot="open",
+            assignee_type_snapshot="account",
+            assignee_id_snapshot=self.account.id,
+            assignee_label_snapshot="Subject",
+            resolution_state_snapshot="resolved",
+        )
+        self.session.add_all((assignment, event))
+        self.session.commit()
+
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self.subject,
+        )
+        work = next(item for item in records if item.resource_type == "campaign_work_assignment")
+        history = next(item for item in records if item.resource_type == "campaign_work_assignment_event")
+
+        self.assertIn("assignee_id", work.data["match_fields"])
+        self.assertEqual("Review the individualized notice", work.data["purpose"])
+        self.assertFalse(work.immutable_evidence)
+        self.assertTrue(history.immutable_evidence)
+        self.assertIn("accountable institutional work history", history.retention_reason)
+
     def test_search_is_tenant_scoped_minimized_and_recipient_specific(self) -> None:
         records = self.provider.search_subject(
             self.session,

tests/test_route_registration.py

@@ -4,6 +4,7 @@ from collections import Counter
 
 from govoplan_campaign.backend.router import router
 from govoplan_campaign.backend.routes.attachments import router as attachments_router
+from govoplan_campaign.backend.routes.assignments import router as assignments_router
 from govoplan_campaign.backend.routes.campaigns import router as campaigns_router
 from govoplan_campaign.backend.routes.collaboration import router as collaboration_router
 from govoplan_campaign.backend.routes.delivery import router as delivery_router
@@ -27,6 +28,7 @@ def test_campaign_router_composes_every_workflow_operation_once() -> None:
     workflow_routers = (
         operations_router,
         campaigns_router,
+        assignments_router,
         collaboration_router,
         versions_router,
         jobs_router,
@@ -44,7 +46,7 @@ def test_campaign_router_composes_every_workflow_operation_once() -> None:
     actual = _operation_keys(router)
 
     assert actual == expected
-    assert len(actual) == 86
+    assert len(actual) == 93
     assert not [operation for operation, count in Counter(actual).items() if count > 1]
 
 
@@ -56,6 +58,7 @@ def test_key_routes_are_owned_by_their_focused_router() -> None:
         ),
         (campaigns_router, ("GET", "/campaigns/{campaign_id}/workspace")),
         (collaboration_router, ("POST", "/campaigns/{campaign_id}/collaboration")),
+        (assignments_router, ("POST", "/campaigns/{campaign_id}/assignments")),
         (versions_router, ("POST", "/campaigns/versions/{version_id}/build")),
         (jobs_router, ("GET", "/campaigns/{campaign_id}/jobs")),
         (reports_router, ("GET", "/campaigns/{campaign_id}/report")),