feat(campaigns): add governed collaboration thread

src/govoplan_campaign/backend/capabilities.py

@@ -28,6 +28,7 @@ from govoplan_campaign.backend.db.models import (
     AttachmentInstance,
     CampaignIssue,
     Campaign,
+    CampaignCollaborationEntry,
     CampaignJob,
     CampaignMessageAction,
     CampaignMessageActionAttempt,
@@ -41,7 +42,13 @@ from govoplan_campaign.backend.db.models import (
 )
 
 
-READ_ACTIONS = {"campaigns:campaign:read", "campaign:read"}
+READ_ACTIONS = {
+    "campaigns:campaign:read",
+    "campaign:read",
+    "campaigns:discussion:read",
+    "campaigns:discussion:post",
+    "campaigns:discussion:moderate",
+}
 CAMPAIGN_RESOURCE_TYPES = {
     "campaign",
     "campaign_object",
@@ -51,6 +58,10 @@ CAMPAIGN_VERSION_RESOURCE_TYPES = {
     "campaign_version",
     "campaigns:version",
 }
+CAMPAIGN_COLLABORATION_RESOURCE_TYPES = {
+    "campaign_collaboration_entry",
+    "campaigns:collaboration_entry",
+}
 CAMPAIGN_DELIVERY_JOB_RESOURCE_TYPES = {
     "campaign_delivery_job",
     "campaign_job",
@@ -558,6 +569,52 @@ class CampaignAccessService(CampaignAccessProvider):
             return tuple(items)
         if normalized_type in CAMPAIGN_RESOURCE_TYPES:
             campaign = session.get(Campaign, resource_id)  # type: ignore[attr-defined]
+        elif normalized_type in CAMPAIGN_COLLABORATION_RESOURCE_TYPES:
+            entry = session.get(CampaignCollaborationEntry, resource_id)  # type: ignore[attr-defined]
+            if entry is None:
+                return _missing_resource_provenance(
+                    principal,
+                    resource_type="campaign_collaboration_entry",
+                    resource_id=resource_id,
+                )
+            campaign = session.get(Campaign, entry.campaign_id)  # type: ignore[attr-defined]
+            normalized_action = action.strip().lower()
+            required_actions = (
+                ("campaigns:discussion:moderate",)
+                if normalized_action.endswith(":moderate")
+                else (
+                    ("campaigns:discussion:post",)
+                    if normalized_action.endswith(":post")
+                    else ("campaigns:discussion:read",)
+                )
+            )
+            child_item = _child_provenance(
+                principal,
+                resource_id=entry.id,
+                source="campaigns.collaboration_entry",
+                label="Campaign collaboration entry",
+                campaign=campaign,
+                version_id=entry.campaign_version_id,
+                details={
+                    "resource_type": "campaign_collaboration_entry",
+                    "visibility": entry.visibility,
+                    "reference_kind": entry.reference_kind,
+                    "reference_id": entry.reference_id,
+                    "state": (
+                        "redacted"
+                        if entry.redacted_at
+                        else "withdrawn" if entry.withdrawn_at else "posted"
+                    ),
+                    "content_disclosed": False,
+                    "actor_identity_disclosed": False,
+                    "permission_classes": {
+                        "read": ["campaigns:discussion:read"],
+                        "post": ["campaigns:discussion:post"],
+                        "moderate": ["campaigns:discussion:moderate"],
+                    },
+                },
+                required_actions=required_actions,
+            )
         elif normalized_type in CAMPAIGN_SHARE_RESOURCE_TYPES:
             share = session.get(CampaignShare, resource_id)  # type: ignore[attr-defined]
             if share is None:

src/govoplan_campaign/backend/db/models.py

@@ -168,6 +168,61 @@ class CampaignShare(Base, TimestampMixin):
     revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
 
 
+class CampaignCollaborationEntry(Base, TimestampMixin):
+    __tablename__ = "campaign_collaboration_entries"
+    __table_args__ = (
+        Index(
+            "ix_campaign_collaboration_entries_thread",
+            "tenant_id",
+            "campaign_id",
+            "created_at",
+            "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)
+    campaign_id: Mapped[str] = mapped_column(
+        ForeignKey("campaigns.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    campaign_version_id: Mapped[str | None] = mapped_column(
+        ForeignKey("campaign_versions.id", ondelete="RESTRICT"),
+        nullable=True,
+        index=True,
+    )
+    reference_kind: Mapped[str | None] = mapped_column(String(40), nullable=True, index=True)
+    reference_id: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    reference_label: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    actor_user_id: Mapped[str | None] = mapped_column(
+        ForeignKey("access_users.id", ondelete="SET NULL"),
+        nullable=True,
+        index=True,
+    )
+    actor_label_snapshot: Mapped[str] = mapped_column(String(255), nullable=False)
+    visibility: Mapped[str] = mapped_column(
+        String(30),
+        default="collaborators",
+        nullable=False,
+        index=True,
+    )
+    content: Mapped[str | None] = mapped_column(Text, nullable=True)
+    content_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
+    mention_user_ids: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False)
+    withdrawn_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+    withdrawn_by_user_id: Mapped[str | None] = mapped_column(
+        ForeignKey("access_users.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    redacted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+    redacted_by_user_id: Mapped[str | None] = mapped_column(
+        ForeignKey("access_users.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    tombstone_reason: Mapped[str | None] = mapped_column(String(500), nullable=True)
+
+
 class CampaignSchedule(Base, TimestampMixin):
     __tablename__ = "campaign_schedules"
     __table_args__ = (

src/govoplan_campaign/backend/documentation.py

@@ -25,6 +25,9 @@ _CAMPAIGN_USER_SCOPES = (
     "campaigns:campaign:send",
     "campaigns:campaign:retry",
     "campaigns:campaign:reconcile",
+    "campaigns:discussion:read",
+    "campaigns:discussion:post",
+    "campaigns:discussion:moderate",
     "campaigns:recipient:read",
     "campaigns:recipient:write",
     "campaigns:recipient:import",
@@ -195,6 +198,50 @@ CAMPAIGN_USER_DOCUMENTATION = (
             }
         },
     ),
+    _workflow_topic(
+        topic_id="campaigns.workflow.collaborate-on-campaign",
+        title="Discuss campaign work without changing its evidence",
+        summary="Use the governed collaboration thread for human discussion linked to stable Campaign evidence.",
+        body="Collaboration is governed independently from Campaign editing. A discussion reader still needs access to the parent Campaign; posting and moderation use separate permissions. Posted text is append-only. Authors can withdraw their own entry and moderators can redact an entry, but both actions leave the actor, timestamp, reference, evidence hash, tombstone, and audit record. A comment may reference a Campaign version, recipient import batch, attachment rule, delivery job, or report. The reference never edits the historical version. Mentions create an in-app notification only when Notifications is available and only for active users who already have Campaign access. Human discussion is not system state and never replaces Audit evidence.",
+        order=33,
+        audience=("campaign_manager", "campaign_reviewer", "campaign_sender"),
+        required_scopes=("campaigns:campaign:read", "campaigns:discussion:read"),
+        route="/campaigns/{campaign_id}/activity",
+        screen="Campaign collaboration",
+        help_contexts=(
+            "campaign.activity",
+            "campaign.activity.composer",
+            "campaign.activity.action.post",
+            "campaign.activity.action.withdraw",
+            "campaign.activity.action.redact",
+        ),
+        prerequisites=(
+            "You can read the Campaign and its discussion.",
+            "Posting requires the separate campaign discussion-post permission.",
+        ),
+        steps=(
+            "Open Collaboration in the selected Campaign workspace.",
+            "Optionally select a stable version or enter the stable ID of another supported Campaign evidence reference.",
+            "Mention only collaborators who already have access, then post the bounded comment.",
+            "Withdraw your own mistaken entry or ask an authorized moderator to redact content that must no longer be displayed.",
+            "Use Tenant audit for system events and durable action evidence; do not treat discussion as workflow state.",
+        ),
+        outcome="An attributable human discussion entry that does not mutate Campaign versions or impersonate audit evidence.",
+        verification="Reload Collaboration, follow the typed reference, and confirm any withdrawal or redaction appears as a tombstone while the referenced version remains unchanged.",
+        required_capabilities=(),
+        related_modules=("notifications", "audit"),
+        limitations=(
+            "Notifications is optional; the discussion remains available when mention delivery is not configured.",
+            "Comments do not approve, validate, build, queue, send, or otherwise transition a Campaign.",
+        ),
+        translations={
+            "de": {
+                "title": "Kampagnenarbeit besprechen, ohne Nachweise zu verändern",
+                "summary": "Den geregelten Diskussionsverlauf für menschliche Abstimmung mit stabilen Kampagnennachweisen verwenden.",
+                "body": "Die Zusammenarbeit wird unabhängig von der Kampagnenbearbeitung berechtigt. Lesende benötigen weiterhin Zugriff auf die übergeordnete Kampagne; Veröffentlichung und Moderation verwenden eigene Berechtigungen. Veröffentlichter Text ist unveränderlich. Verfassende können eigene Einträge zurücknehmen, Moderierende können Einträge schwärzen. Dabei bleiben Person, Zeitstempel, Referenz, Nachweis-Hash, Platzhalter und Auditnachweis erhalten. Kommentare können Kampagnenversionen, Empfänger-Importläufe, Anlagenregeln, Sendeaufträge oder Berichte referenzieren, ohne historische Versionen zu verändern. Erwähnungen erzeugen nur bei verfügbarem Benachrichtigungsmodul eine interne Benachrichtigung und nur für aktive Personen mit bestehendem Kampagnenzugriff. Diskussion ist kein Systemzustand und ersetzt keinen Auditnachweis.",
+            }
+        },
+    ),
     _workflow_topic(
         topic_id="campaigns.workflow.reuse-content-library",
         title="Reuse Campaign content through Templates",

src/govoplan_campaign/backend/dsar_provider.py

@@ -10,6 +10,7 @@ from sqlalchemy.orm import Session
 from govoplan_campaign.backend.db.models import (
     AttachmentInstance,
     Campaign,
+    CampaignCollaborationEntry,
     CampaignIssue,
     CampaignJob,
     CampaignMessageAction,
@@ -646,6 +647,62 @@ class CampaignDsarProvider:
         version_ids: set[str],
     ) -> None:
         if subject_user_id is not None:
+            collaboration_rows = _bounded_rows(
+                db.query(CampaignCollaborationEntry)
+                .filter(
+                    CampaignCollaborationEntry.tenant_id == tenant_id,
+                    or_(
+                        CampaignCollaborationEntry.actor_user_id == subject_user_id,
+                        CampaignCollaborationEntry.withdrawn_by_user_id == subject_user_id,
+                        CampaignCollaborationEntry.redacted_by_user_id == subject_user_id,
+                        cast(CampaignCollaborationEntry.mention_user_ids, Text).contains(
+                            f'"{subject_user_id}"'
+                        ),
+                    ),
+                )
+                .order_by(CampaignCollaborationEntry.id)
+            )
+            for entry in collaboration_rows:
+                match_fields = _matching_fields(
+                    entry,
+                    subject_user_id,
+                    (
+                        "actor_user_id",
+                        "withdrawn_by_user_id",
+                        "redacted_by_user_id",
+                    ),
+                )
+                if subject_user_id in (entry.mention_user_ids or []):
+                    match_fields.append("mention_user_ids")
+                authored = entry.actor_user_id == subject_user_id
+                append(  # type: ignore[operator]
+                    _record(
+                        "campaign_collaboration_entry",
+                        entry.id,
+                        "campaign_collaboration_evidence",
+                        "Campaign collaboration entry",
+                        {
+                            "match_fields": match_fields,
+                            "campaign_id": entry.campaign_id,
+                            "campaign_version_id": entry.campaign_version_id,
+                            "visibility": entry.visibility,
+                            "reference_kind": entry.reference_kind,
+                            "reference_id": entry.reference_id,
+                            "posted_text": entry.content if authored else None,
+                            "content_disclosed": authored and entry.content is not None,
+                            "content_sha256": entry.content_sha256,
+                            "withdrawn_at": _iso(entry.withdrawn_at),
+                            "redacted_at": _iso(entry.redacted_at),
+                        },
+                        observed_at=entry.updated_at,
+                        immutable=True,
+                        retention_reason=(
+                            "Append-only discussion identity, reference, hash, tombstone, and moderation evidence are retained with Campaign access and Audit evidence."
+                        ),
+                        source_path=f"/campaigns/{entry.campaign_id}/activity",
+                    )
+                )
+
             shares = _bounded_rows(
                 db.query(CampaignShare)
                 .filter(

src/govoplan_campaign/backend/manifest.py

@@ -50,6 +50,7 @@ from govoplan_core.core.templates import (
     CAPABILITY_TEMPLATE_RENDERER,
 )
 from govoplan_core.core.operations import OperationalCheckProviderRegistration
+from govoplan_core.core.notifications import CAPABILITY_NOTIFICATIONS_DISPATCH
 from govoplan_core.core.provider_governance import declared_module_architecture
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
@@ -102,6 +103,24 @@ PERMISSIONS = (
         "Open campaign metadata, versions and permitted message summaries.",
         "Campaigns",
     ),
+    _permission(
+        "campaigns:discussion:read",
+        "View campaign discussions",
+        "Read human collaboration entries attached to campaigns the user can already access.",
+        "Campaign collaboration",
+    ),
+    _permission(
+        "campaigns:discussion:post",
+        "Post campaign discussions",
+        "Post and withdraw the user's own append-only campaign collaboration entries.",
+        "Campaign collaboration",
+    ),
+    _permission(
+        "campaigns:discussion:moderate",
+        "Moderate campaign discussions",
+        "Read moderator-only entries and redact campaign collaboration content while retaining tombstones.",
+        "Campaign collaboration",
+    ),
     _permission(
         "campaigns:campaign:create",
         "Create campaigns",
@@ -302,6 +321,9 @@ ROLE_TEMPLATES = (
             "campaigns:campaign:validate",
             "campaigns:campaign:build",
             "campaigns:ownership:accept_group",
+            "campaigns:discussion:read",
+            "campaigns:discussion:post",
+            "campaigns:discussion:moderate",
             "campaigns:recipient:read",
             "campaigns:recipient:write",
             "campaigns:recipient:import",
@@ -316,6 +338,8 @@ ROLE_TEMPLATES = (
             "campaigns:campaign:read",
             "campaigns:campaign:validate",
             "campaigns:campaign:review",
+            "campaigns:discussion:read",
+            "campaigns:discussion:post",
             "campaigns:recipient:read",
             "campaigns:report:read",
         ),
@@ -332,6 +356,8 @@ ROLE_TEMPLATES = (
             "campaigns:campaign:send",
             "campaigns:campaign:retry",
             "campaigns:campaign:reconcile",
+            "campaigns:discussion:read",
+            "campaigns:discussion:post",
             "campaigns:diagnostic:read",
             "campaigns:recipient:read",
             "campaigns:report:read",
@@ -385,7 +411,7 @@ def _campaigns_router(context: ModuleContext):
 manifest = ModuleManifest(
     id="campaigns",
     name="Campaigns",
-    version="0.1.19",
+    version="0.1.20",
     required_capabilities=(
         CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
         CAPABILITY_AUTH_PERMISSION_EVALUATOR,
@@ -393,6 +419,7 @@ manifest = ModuleManifest(
     optional_capabilities=(
         CAPABILITY_ACCESS_REFERENCE_OPTIONS,
         CAPABILITY_APPROVAL_REQUESTS,
+        CAPABILITY_NOTIFICATIONS_DISPATCH,
     ),
     optional_dependencies=(
         "files",
@@ -517,6 +544,12 @@ manifest = ModuleManifest(
             version_max_exclusive="0.2.0",
             optional=True,
         ),
+        ModuleInterfaceRequirement(
+            name=CAPABILITY_NOTIFICATIONS_DISPATCH,
+            version_min="0.1.0",
+            version_max_exclusive="0.2.0",
+            optional=True,
+        ),
         ModuleInterfaceRequirement(
             name="search.source",
             version_min="1.0.0",
@@ -606,11 +639,19 @@ manifest = ModuleManifest(
                     "campaigns.route.operator-redirect",
                     OPERATOR_QUEUE_SURFACE_ID,
                     REPORTS_SURFACE_ID,
+                    "campaigns.page.activity",
                 ),
                 order=40,
             ),
         ),
         view_surfaces=(
+            ViewSurface(
+                id="campaigns.page.activity",
+                module_id="campaigns",
+                kind="page",
+                label="Campaign collaboration",
+                order=45,
+            ),
             ViewSurface(
                 id="campaigns.widget.activity",
                 module_id="campaigns",
@@ -656,6 +697,7 @@ manifest = ModuleManifest(
             campaign_models.CampaignSchedule,
             campaign_models.CampaignScheduleOccurrence,
             campaign_models.CampaignShare,
+            campaign_models.CampaignCollaborationEntry,
             campaign_models.RecipientImportMappingProfile,
             campaign_models.CampaignVersion,
             campaign_models.CampaignJob,
@@ -678,6 +720,7 @@ manifest = ModuleManifest(
             campaign_models.CampaignSchedule,
             campaign_models.CampaignScheduleOccurrence,
             campaign_models.CampaignShare,
+            campaign_models.CampaignCollaborationEntry,
             campaign_models.RecipientImportMappingProfile,
             campaign_models.CampaignVersion,
             campaign_models.CampaignJob,
@@ -737,14 +780,71 @@ manifest = ModuleManifest(
                 "help_contexts": ["campaigns.quick_access.campaigns"],
             },
         ),
+        DocumentationTopic(
+            id="campaigns.admin.collaboration-governance",
+            title="Govern Campaign collaboration permissions and retention",
+            summary="Configure discussion access separately from Campaign editing and retain auditable moderation tombstones.",
+            body=(
+                "Campaign collaboration uses separate read, post, and moderate permissions in addition to parent Campaign read access. "
+                "The built-in manager role can moderate, while reviewer and sender roles can read and post without receiving Campaign edit permission. "
+                "A read share is sufficient as the parent resource grant; comments never upgrade it to write access. Moderator-only visibility is filtered server-side. "
+                "Posted content has no edit API. Author withdrawal and moderator redaction remove displayed content while retaining the stable entry, SHA-256 evidence, actor snapshot, timestamp, typed reference, tombstone, and bounded Audit event. "
+                "Mention targets are restricted to active users who already have Campaign ownership or share access. If the optional Notifications dispatch capability is available, Campaign emits content-free inbox notifications; provider failure does not make Notifications a required Campaign dependency. "
+                "Operators must apply institutional retention and privacy policy to collaboration rows and Audit evidence together and must not represent comments as approvals, workflow transitions, or system events."
+            ),
+            layer="configured",
+            documentation_types=("admin",),
+            audience=("campaign_manager", "module_admin", "privacy_officer", "records_manager"),
+            order=41,
+            conditions=(
+                DocumentationCondition(
+                    required_modules=("campaigns",),
+                    any_scopes=(
+                        "campaigns:discussion:moderate",
+                        "access:roles:manage",
+                    ),
+                ),
+            ),
+            links=(
+                DocumentationLink(
+                    label="Campaign collaboration",
+                    href="/campaigns",
+                    kind="runtime",
+                ),
+                DocumentationLink(
+                    label="Campaign handbook",
+                    href="govoplan-campaign/docs/CAMPAIGN_HANDBOOK.md",
+                    kind="repository",
+                ),
+            ),
+            related_modules=("access", "audit", "notifications"),
+            metadata={
+                "kind": "configuration",
+                "route": "/campaigns/{campaign_id}/activity",
+                "screen": "Campaign collaboration",
+                "help_contexts": [
+                    "campaign.activity",
+                    "campaign.activity.composer",
+                    "campaign.activity.action.post",
+                    "campaign.activity.action.withdraw",
+                    "campaign.activity.action.redact",
+                ],
+                "permission_scopes": [
+                    "campaigns:discussion:read",
+                    "campaigns:discussion:post",
+                    "campaigns:discussion:moderate",
+                ],
+                "verification": "Test a read-only collaborator, a poster without Campaign edit, and a moderator; confirm moderator visibility, mention access filtering, tombstones, and Audit records.",
+            },
+        ),
         DocumentationTopic(
             id="campaigns.privacy.data-subject-requests",
             title="Review Campaign data in a data-subject request",
-            summary="Collect recipient, version, delivery, report, and artifact metadata without rewriting immutable evidence.",
+            summary="Collect recipient, collaboration, version, delivery, report, and artifact metadata without rewriting immutable evidence.",
             body=(
                 "Campaign's DSAR provider searches the effective tenant by normalized recipient email, direct membership references, and namespaced Campaign job, entry, version, or Campaign references. "
-                "It isolates matching inline-recipient fields and job metadata, and reports built versions, delivery attempts, Postbox and print outcomes, message-action corrections, recipient-specific report projections, generated-message digests, and attachment metadata. It does not export EML bytes, object or local paths, provider target snapshots, worker claims, idempotency material, secrets, credentials, or unrelated recipient addresses. "
-                "Built, locked, published, terminal, delivered, or corrected records remain retained with a reason and continue through Campaign's configured retention/redaction process. Draft recipient content and user-owned attachment content require coordinated manual review because the same data may occur in version JSON, jobs, and generated artifacts. The provider can idempotently delete a personal recipient-import mapping profile and revoke an active Campaign share aimed at the subject. It never rewrites delivered evidence or deletes generated artifacts directly. Campaign reports are derived projections rather than a separate personal-data store."
+                "It isolates matching inline-recipient fields and job metadata, and reports built versions, delivery attempts, Postbox and print outcomes, message-action corrections, recipient-specific report projections, generated-message digests, attachment metadata, and collaboration entries authored, mentioned, or moderated by the subject. Authored collaboration text is included; text authored by somebody else is not copied merely because the subject was mentioned. It does not export EML bytes, object or local paths, provider target snapshots, worker claims, idempotency material, secrets, credentials, or unrelated recipient addresses. "
+                "Built, locked, published, terminal, delivered, corrected, withdrawn, or redacted records remain retained with a reason and continue through Campaign's configured retention/redaction process. Collaboration tombstones, hashes, and Audit evidence remain immutable. Draft recipient content and user-owned attachment content require coordinated manual review because the same data may occur in version JSON, jobs, and generated artifacts. The provider can idempotently delete a personal recipient-import mapping profile and revoke an active Campaign share aimed at the subject. It never rewrites delivered evidence or deletes generated artifacts directly. Campaign reports are derived projections rather than a separate personal-data store."
             ),
             layer="configured",
             documentation_types=("admin",),

src/govoplan_campaign/backend/migrations/versions/c7d8e9f0a1b2_v0120_campaign_collaboration.py

@@ -0,0 +1,89 @@
+"""add governed campaign collaboration entries
+
+revision = "c7d8e9f0a1b2"
+down_revision = "b6c7d8e9f0a1"
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = "c7d8e9f0a1b2"
+down_revision = "b6c7d8e9f0a1"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if inspector.has_table("campaign_collaboration_entries"):
+        return
+    op.create_table(
+        "campaign_collaboration_entries",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("campaign_id", sa.String(length=36), nullable=False),
+        sa.Column("campaign_version_id", sa.String(length=36), nullable=True),
+        sa.Column("reference_kind", sa.String(length=40), nullable=True),
+        sa.Column("reference_id", sa.String(length=500), nullable=True),
+        sa.Column("reference_label", sa.String(length=255), nullable=True),
+        sa.Column("actor_user_id", sa.String(length=36), nullable=True),
+        sa.Column("actor_label_snapshot", sa.String(length=255), nullable=False),
+        sa.Column(
+            "visibility",
+            sa.String(length=30),
+            nullable=False,
+            server_default="collaborators",
+        ),
+        sa.Column("content", sa.Text(), nullable=True),
+        sa.Column("content_sha256", sa.String(length=64), nullable=False),
+        sa.Column("mention_user_ids", sa.JSON(), nullable=False, server_default="[]"),
+        sa.Column("withdrawn_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("withdrawn_by_user_id", sa.String(length=36), nullable=True),
+        sa.Column("redacted_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("redacted_by_user_id", sa.String(length=36), nullable=True),
+        sa.Column("tombstone_reason", sa.String(length=500), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["actor_user_id"], ["access_users.id"], ondelete="SET NULL"),
+        sa.ForeignKeyConstraint(["campaign_id"], ["campaigns.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(
+            ["campaign_version_id"],
+            ["campaign_versions.id"],
+            ondelete="RESTRICT",
+        ),
+        sa.ForeignKeyConstraint(
+            ["redacted_by_user_id"],
+            ["access_users.id"],
+            ondelete="SET NULL",
+        ),
+        sa.ForeignKeyConstraint(
+            ["withdrawn_by_user_id"],
+            ["access_users.id"],
+            ondelete="SET NULL",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    for name, columns in (
+        ("ix_campaign_collaboration_entries_tenant_id", ["tenant_id"]),
+        ("ix_campaign_collaboration_entries_campaign_id", ["campaign_id"]),
+        ("ix_campaign_collaboration_entries_campaign_version_id", ["campaign_version_id"]),
+        ("ix_campaign_collaboration_entries_reference_kind", ["reference_kind"]),
+        ("ix_campaign_collaboration_entries_actor_user_id", ["actor_user_id"]),
+        ("ix_campaign_collaboration_entries_visibility", ["visibility"]),
+        ("ix_campaign_collaboration_entries_withdrawn_at", ["withdrawn_at"]),
+        ("ix_campaign_collaboration_entries_redacted_at", ["redacted_at"]),
+        (
+            "ix_campaign_collaboration_entries_thread",
+            ["tenant_id", "campaign_id", "created_at", "id"],
+        ),
+    ):
+        op.create_index(name, "campaign_collaboration_entries", columns, unique=False)
+
+
+def downgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if inspector.has_table("campaign_collaboration_entries"):
+        op.drop_table("campaign_collaboration_entries")

src/govoplan_campaign/backend/router.py

@@ -4,6 +4,7 @@ from fastapi import APIRouter
 
 from govoplan_campaign.backend.routes.attachments import router as attachments_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
 from govoplan_campaign.backend.routes.jobs import router as jobs_router
 from govoplan_campaign.backend.routes.operations import router as operations_router
@@ -17,6 +18,7 @@ router = APIRouter()
 for workflow_router in (
     operations_router,
     campaigns_router,
+    collaboration_router,
     versions_router,
     jobs_router,
     reports_router,

src/govoplan_campaign/backend/routes/collaboration.py

@@ -0,0 +1,571 @@
+from __future__ import annotations
+
+import base64
+import binascii
+import hashlib
+import json
+from datetime import UTC, datetime
+from typing import Literal
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy import and_, or_
+from sqlalchemy.orm import Session
+
+from govoplan_campaign.backend.db.models import (
+    Campaign,
+    CampaignCollaborationEntry,
+    CampaignJob,
+    CampaignShare,
+    CampaignVersion,
+)
+from govoplan_campaign.backend.path_security import _attachment_rules
+from govoplan_campaign.backend.route_support import (
+    _access_directory,
+    _get_campaign_for_principal,
+    _require_permission,
+)
+from govoplan_campaign.backend.runtime import get_registry
+from govoplan_campaign.backend.schemas import (
+    CampaignCollaborationCreateRequest,
+    CampaignCollaborationEntryResponse,
+    CampaignCollaborationListResponse,
+    CampaignCollaborationModerationRequest,
+    CampaignCollaborationReferenceInput,
+    CampaignCollaborationReferenceResponse,
+)
+from govoplan_core.api.v1.schemas import (
+    ReferenceOptionListResponse,
+    ReferenceOptionResponse,
+)
+from govoplan_core.audit.logging import audit_from_principal
+from govoplan_core.auth import ApiPrincipal, has_scope, require_scope
+from govoplan_core.core.notifications import (
+    NotificationDispatchRequest,
+    notification_dispatch_provider,
+)
+from govoplan_core.core.references import (
+    access_scope_reference_page,
+    access_scope_reference_provider_available,
+)
+from govoplan_core.db.session import get_session
+from govoplan_core.security.time import utc_now
+
+
+router = APIRouter(prefix="/campaigns", tags=["campaign collaboration"])
+
+
+@router.get(
+    "/{campaign_id}/collaboration/mention-options",
+    response_model=ReferenceOptionListResponse,
+)
+def search_campaign_collaboration_mentions(
+    campaign_id: str,
+    q: str = "",
+    selected: list[str] = Query(default=[]),
+    limit: int = Query(default=50, ge=1, le=100),
+    cursor: str | None = None,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("campaigns:discussion:post")),
+) -> ReferenceOptionListResponse:
+    campaign = _get_campaign_for_principal(session, campaign_id, principal)
+    _require_permission(principal, "campaigns:campaign:read")
+    try:
+        page = access_scope_reference_page(
+            get_registry(),
+            principal,
+            scope_type="user",
+            reference_kind="membership",
+            query=q,
+            selected_values=selected,
+            limit=limit,
+            cursor=cursor,
+            administrative=True,
+            session=session,
+        )
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=str(exc),
+        ) from exc
+    allowed = [
+        option
+        for option in page.options
+        if _mentioned_user_has_campaign_access(
+            session,
+            campaign=campaign,
+            user_id=option.value,
+        )
+    ]
+    return ReferenceOptionListResponse(
+        options=[ReferenceOptionResponse(**option.to_dict()) for option in allowed],
+        provider_available=access_scope_reference_provider_available(get_registry()),
+        next_cursor=page.next_cursor,
+        has_more=page.has_more,
+    )
+
+
+@router.get(
+    "/{campaign_id}/collaboration",
+    response_model=CampaignCollaborationListResponse,
+)
+def list_campaign_collaboration(
+    campaign_id: str,
+    limit: int = Query(default=25, ge=1, le=50),
+    cursor: str | None = None,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("campaigns:discussion:read")),
+) -> CampaignCollaborationListResponse:
+    campaign = _get_campaign_for_principal(session, campaign_id, principal)
+    _require_permission(principal, "campaigns:campaign:read")
+    query = session.query(CampaignCollaborationEntry).filter(
+        CampaignCollaborationEntry.tenant_id == principal.tenant_id,
+        CampaignCollaborationEntry.campaign_id == campaign.id,
+    )
+    if not has_scope(principal, "campaigns:discussion:moderate"):
+        query = query.filter(CampaignCollaborationEntry.visibility == "collaborators")
+    if cursor:
+        created_at, entry_id = _decode_cursor(cursor)
+        query = query.filter(
+            or_(
+                CampaignCollaborationEntry.created_at < created_at,
+                and_(
+                    CampaignCollaborationEntry.created_at == created_at,
+                    CampaignCollaborationEntry.id < entry_id,
+                ),
+            )
+        )
+    rows = (
+        query.order_by(
+            CampaignCollaborationEntry.created_at.desc(),
+            CampaignCollaborationEntry.id.desc(),
+        )
+        .limit(limit + 1)
+        .all()
+    )
+    has_more = len(rows) > limit
+    items = rows[:limit]
+    return CampaignCollaborationListResponse(
+        items=[_entry_response(item) for item in items],
+        next_cursor=_encode_cursor(items[-1]) if has_more and items else None,
+        has_more=has_more,
+    )
+
+
+@router.post(
+    "/{campaign_id}/collaboration",
+    response_model=CampaignCollaborationEntryResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def create_campaign_collaboration_entry(
+    campaign_id: str,
+    payload: CampaignCollaborationCreateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("campaigns:discussion:post")),
+) -> CampaignCollaborationEntryResponse:
+    campaign = _get_campaign_for_principal(session, campaign_id, principal)
+    _require_permission(principal, "campaigns:campaign:read")
+    if payload.visibility == "moderators":
+        _require_permission(principal, "campaigns:discussion:moderate")
+    reference = _validated_reference(
+        session,
+        campaign=campaign,
+        reference=payload.reference,
+    )
+    mentions = _validated_mentions(
+        session,
+        campaign=campaign,
+        user_ids=payload.mention_user_ids,
+        actor_user_id=principal.user.id,
+    )
+    actor_label = (
+        getattr(principal.user, "display_name", None)
+        or getattr(principal.user, "email", None)
+        or principal.user.id
+    )
+    entry = CampaignCollaborationEntry(
+        tenant_id=principal.tenant_id,
+        campaign_id=campaign.id,
+        campaign_version_id=reference[0] if reference else None,
+        reference_kind=reference[1] if reference else None,
+        reference_id=reference[2] if reference else None,
+        reference_label=reference[3] if reference else None,
+        actor_user_id=principal.user.id,
+        actor_label_snapshot=str(actor_label)[:255],
+        visibility=payload.visibility,
+        content=payload.content,
+        content_sha256=hashlib.sha256(payload.content.encode("utf-8")).hexdigest(),
+        mention_user_ids=mentions,
+    )
+    session.add(entry)
+    session.flush()
+    _enqueue_mention_notifications(
+        session,
+        campaign=campaign,
+        entry=entry,
+        mention_user_ids=mentions,
+    )
+    audit_from_principal(
+        session,
+        principal,
+        action="campaign.collaboration.posted",
+        object_type="campaign_collaboration_entry",
+        object_id=entry.id,
+        details={
+            "campaign_id": campaign.id,
+            "campaign_version_id": entry.campaign_version_id,
+            "visibility": entry.visibility,
+            "reference_kind": entry.reference_kind,
+            "reference_id": entry.reference_id,
+            "mention_count": len(mentions),
+            "content_sha256": entry.content_sha256,
+            "content_disclosed": False,
+        },
+        commit=True,
+    )
+    session.refresh(entry)
+    return _entry_response(entry)
+
+
+@router.post(
+    "/{campaign_id}/collaboration/{entry_id}/withdraw",
+    response_model=CampaignCollaborationEntryResponse,
+)
+def withdraw_campaign_collaboration_entry(
+    campaign_id: str,
+    entry_id: str,
+    payload: CampaignCollaborationModerationRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("campaigns:discussion:post")),
+) -> CampaignCollaborationEntryResponse:
+    _get_campaign_for_principal(session, campaign_id, principal)
+    _require_permission(principal, "campaigns:campaign:read")
+    entry = _entry_for_campaign(session, campaign_id=campaign_id, entry_id=entry_id, principal=principal)
+    if entry.actor_user_id != principal.user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Only the author can withdraw this collaboration entry.",
+        )
+    if entry.redacted_at is not None or entry.withdrawn_at is not None:
+        return _entry_response(entry)
+    entry.content = None
+    entry.withdrawn_at = utc_now()
+    entry.withdrawn_by_user_id = principal.user.id
+    entry.tombstone_reason = payload.reason
+    audit_from_principal(
+        session,
+        principal,
+        action="campaign.collaboration.withdrawn",
+        object_type="campaign_collaboration_entry",
+        object_id=entry.id,
+        details={
+            "campaign_id": campaign_id,
+            "content_sha256": entry.content_sha256,
+            "reason_recorded": bool(payload.reason),
+            "content_disclosed": False,
+        },
+        commit=True,
+    )
+    session.refresh(entry)
+    return _entry_response(entry)
+
+
+@router.post(
+    "/{campaign_id}/collaboration/{entry_id}/redact",
+    response_model=CampaignCollaborationEntryResponse,
+)
+def redact_campaign_collaboration_entry(
+    campaign_id: str,
+    entry_id: str,
+    payload: CampaignCollaborationModerationRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("campaigns:discussion:moderate")),
+) -> CampaignCollaborationEntryResponse:
+    _get_campaign_for_principal(session, campaign_id, principal)
+    _require_permission(principal, "campaigns:campaign:read")
+    entry = _entry_for_campaign(session, campaign_id=campaign_id, entry_id=entry_id, principal=principal)
+    if entry.redacted_at is not None:
+        return _entry_response(entry)
+    entry.content = None
+    entry.redacted_at = utc_now()
+    entry.redacted_by_user_id = principal.user.id
+    entry.tombstone_reason = payload.reason
+    audit_from_principal(
+        session,
+        principal,
+        action="campaign.collaboration.redacted",
+        object_type="campaign_collaboration_entry",
+        object_id=entry.id,
+        details={
+            "campaign_id": campaign_id,
+            "content_sha256": entry.content_sha256,
+            "reason_recorded": bool(payload.reason),
+            "previously_withdrawn": entry.withdrawn_at is not None,
+            "content_disclosed": False,
+        },
+        commit=True,
+    )
+    session.refresh(entry)
+    return _entry_response(entry)
+
+
+def _entry_for_campaign(
+    session: Session,
+    *,
+    campaign_id: str,
+    entry_id: str,
+    principal: ApiPrincipal,
+) -> CampaignCollaborationEntry:
+    entry = session.get(CampaignCollaborationEntry, entry_id)
+    if (
+        entry is None
+        or entry.tenant_id != principal.tenant_id
+        or entry.campaign_id != campaign_id
+    ):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Collaboration entry not found")
+    return entry
+
+
+def _validated_reference(
+    session: Session,
+    *,
+    campaign: Campaign,
+    reference: CampaignCollaborationReferenceInput | None,
+) -> tuple[str | None, str, str, str] | None:
+    if reference is None:
+        return None
+    kind = reference.kind
+    reference_id = reference.id
+    version: CampaignVersion | None = None
+    default_label = kind.replace("_", " ").title()
+    if kind == "campaign_version":
+        version = session.get(CampaignVersion, reference_id)
+        if version is not None:
+            default_label = f"Version {version.version_number}"
+    elif kind == "delivery_job":
+        job = session.get(CampaignJob, reference_id)
+        if job is None or job.campaign_id != campaign.id or job.tenant_id != campaign.tenant_id:
+            raise _invalid_reference()
+        version = session.get(CampaignVersion, job.campaign_version_id)
+        default_label = f"Delivery job {job.id[:8]}"
+    elif kind in {"recipient_import_batch", "attachment_rule"}:
+        version_id, separator, child_id = reference_id.partition(":")
+        if not separator or not version_id or not child_id:
+            raise _invalid_reference()
+        version = session.get(CampaignVersion, version_id)
+        if version is not None and kind == "recipient_import_batch":
+            raw = version.raw_json if isinstance(version.raw_json, dict) else {}
+            entries = raw.get("entries") if isinstance(raw.get("entries"), dict) else {}
+            imports = entries.get("imports") if isinstance(entries, dict) else []
+            if not any(isinstance(item, dict) and str(item.get("id") or "") == child_id for item in imports or []):
+                raise _invalid_reference()
+            default_label = "Recipient import batch"
+        elif version is not None:
+            raw = version.raw_json if isinstance(version.raw_json, dict) else {}
+            if child_id not in {path for path, _rule in _attachment_rules(raw)}:
+                raise _invalid_reference()
+            default_label = "Attachment rule"
+    else:
+        referenced_campaign_id, separator, remainder = reference_id.partition(":")
+        version_id, separator_two, report_kind = remainder.partition(":")
+        if (
+            not separator
+            or not separator_two
+            or referenced_campaign_id != campaign.id
+            or not version_id
+            or not report_kind
+        ):
+            raise _invalid_reference()
+        version = session.get(CampaignVersion, version_id)
+        default_label = report_kind.replace("_", " ").title()
+    if version is None or version.campaign_id != campaign.id:
+        raise _invalid_reference()
+    return version.id, kind, reference_id, default_label[:255]
+
+
+def _invalid_reference() -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+        detail="The collaboration reference is not stable evidence owned by this campaign.",
+    )
+
+
+def _validated_mentions(
+    session: Session,
+    *,
+    campaign: Campaign,
+    user_ids: list[str],
+    actor_user_id: str,
+) -> list[str]:
+    mentions = [user_id for user_id in user_ids if user_id != actor_user_id]
+    invalid = [
+        user_id
+        for user_id in mentions
+        if not _mentioned_user_has_campaign_access(
+            session,
+            campaign=campaign,
+            user_id=user_id,
+        )
+    ]
+    if invalid:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail="Mentioned users must be active and already have access to this campaign.",
+        )
+    return mentions
+
+
+def _mentioned_user_has_campaign_access(
+    session: Session,
+    *,
+    campaign: Campaign,
+    user_id: str,
+) -> bool:
+    directory = _access_directory()
+    user = next(
+        (candidate for candidate in directory.users_for_tenant(campaign.tenant_id) if candidate.id == user_id),
+        None,
+    )
+    if user is None or user.status != "active":
+        return False
+    if campaign.owner_user_id == user_id:
+        return True
+    group_ids = {
+        group.id
+        for group in directory.groups_for_user(user_id, tenant_id=campaign.tenant_id)
+    }
+    if campaign.owner_group_id and campaign.owner_group_id in group_ids:
+        return True
+    clauses = [
+        and_(
+            CampaignShare.target_type == "user",
+            CampaignShare.target_id == user_id,
+        )
+    ]
+    if group_ids:
+        clauses.append(
+            and_(
+                CampaignShare.target_type == "group",
+                CampaignShare.target_id.in_(sorted(group_ids)),
+            )
+        )
+    return (
+        session.query(CampaignShare.id)
+        .filter(
+            CampaignShare.tenant_id == campaign.tenant_id,
+            CampaignShare.campaign_id == campaign.id,
+            CampaignShare.revoked_at.is_(None),
+            or_(*clauses),
+        )
+        .first()
+        is not None
+    )
+
+
+def _enqueue_mention_notifications(
+    session: Session,
+    *,
+    campaign: Campaign,
+    entry: CampaignCollaborationEntry,
+    mention_user_ids: list[str],
+) -> None:
+    provider = notification_dispatch_provider(get_registry())
+    if provider is None or not mention_user_ids:
+        return
+    try:
+        with session.begin_nested():
+            for user_id in mention_user_ids:
+                provider.enqueue_notification(
+                    session,
+                    NotificationDispatchRequest(
+                        tenant_id=campaign.tenant_id,
+                        source_module="campaigns",
+                        source_resource_type="campaign_collaboration_entry",
+                        source_resource_id=entry.id,
+                        event_kind="campaign.collaboration.mentioned",
+                        channel="inbox",
+                        recipient_type="user",
+                        recipient_id=user_id,
+                        subject=f"Mentioned in campaign: {campaign.name}",
+                        body_text=(
+                            f"{entry.actor_label_snapshot} mentioned you in the campaign collaboration thread."
+                        ),
+                        action_url=f"/campaigns/{campaign.id}/activity",
+                        payload={
+                            "campaign_id": campaign.id,
+                            "entry_id": entry.id,
+                            "content_disclosed": False,
+                        },
+                    ),
+                    enqueue_delivery=False,
+                )
+    except Exception:
+        # Collaboration remains available when the optional Notifications
+        # provider is absent or temporarily unhealthy.
+        return
+
+
+def _entry_response(entry: CampaignCollaborationEntry) -> CampaignCollaborationEntryResponse:
+    tombstone: Literal["withdrawn", "redacted"] | None = None
+    if entry.redacted_at is not None:
+        tombstone = "redacted"
+    elif entry.withdrawn_at is not None:
+        tombstone = "withdrawn"
+    reference = None
+    if entry.reference_kind and entry.reference_id:
+        reference = CampaignCollaborationReferenceResponse(
+            kind=entry.reference_kind,  # type: ignore[arg-type]
+            id=entry.reference_id,
+            label=entry.reference_label,
+        )
+    return CampaignCollaborationEntryResponse(
+        id=entry.id,
+        campaign_id=entry.campaign_id,
+        actor_user_id=entry.actor_user_id,
+        actor_label=entry.actor_label_snapshot,
+        visibility=entry.visibility,  # type: ignore[arg-type]
+        content=entry.content if tombstone is None else None,
+        content_sha256=entry.content_sha256,
+        mention_user_ids=list(entry.mention_user_ids or []),
+        reference=reference,
+        tombstone=tombstone,
+        tombstone_reason=entry.tombstone_reason,
+        withdrawn_at=entry.withdrawn_at,
+        redacted_at=entry.redacted_at,
+        created_at=entry.created_at,
+    )
+
+
+def _encode_cursor(entry: CampaignCollaborationEntry) -> str:
+    created_at = entry.created_at
+    if created_at.tzinfo is None:
+        # SQLite returns timezone-aware columns as naive UTC values.
+        created_at = created_at.replace(tzinfo=UTC)
+    payload = json.dumps(
+        {"created_at": created_at.astimezone(UTC).isoformat(), "id": entry.id},
+        separators=(",", ":"),
+        sort_keys=True,
+    ).encode("utf-8")
+    return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=")
+
+
+def _decode_cursor(value: str) -> tuple[datetime, str]:
+    try:
+        padded = value + "=" * (-len(value) % 4)
+        payload = json.loads(base64.urlsafe_b64decode(padded).decode("utf-8"))
+        created_at = datetime.fromisoformat(str(payload["created_at"]))
+        entry_id = str(payload["id"])
+        if created_at.tzinfo is None or not entry_id or len(entry_id) > 36:
+            raise ValueError
+        return created_at, entry_id
+    except (
+        KeyError,
+        TypeError,
+        ValueError,
+        UnicodeDecodeError,
+        json.JSONDecodeError,
+        binascii.Error,
+    ) as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail="Invalid collaboration cursor.",
+        ) from exc

src/govoplan_campaign/backend/schemas.py

@@ -42,6 +42,97 @@ class CampaignUpdateRequest(BaseModel):
     description: str | None = None
 
 
+CampaignCollaborationReferenceKind = Literal[
+    "campaign_version",
+    "recipient_import_batch",
+    "attachment_rule",
+    "delivery_job",
+    "report",
+]
+
+
+class CampaignCollaborationReferenceInput(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    kind: CampaignCollaborationReferenceKind
+    id: str = Field(min_length=1, max_length=500)
+
+    @field_validator("id")
+    @classmethod
+    def strip_reference_text(cls, value: str) -> str:
+        clean = value.strip()
+        if not clean:
+            raise ValueError("A collaboration reference ID cannot be empty.")
+        return clean
+
+
+class CampaignCollaborationCreateRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    content: str = Field(min_length=1, max_length=8_000)
+    visibility: Literal["collaborators", "moderators"] = "collaborators"
+    reference: CampaignCollaborationReferenceInput | None = None
+    mention_user_ids: list[str] = Field(default_factory=list, max_length=20)
+
+    @field_validator("content")
+    @classmethod
+    def strip_content(cls, value: str) -> str:
+        if not value.strip():
+            raise ValueError("A collaboration entry cannot be empty.")
+        return value.strip()
+
+    @field_validator("mention_user_ids")
+    @classmethod
+    def normalize_mentions(cls, value: list[str]) -> list[str]:
+        normalized = list(dict.fromkeys(item.strip() for item in value if item.strip()))
+        if len(normalized) > 20:
+            raise ValueError("A collaboration entry can mention at most 20 users.")
+        if any(len(item) > 64 for item in normalized):
+            raise ValueError("A collaboration mention contains an invalid user ID.")
+        return normalized
+
+
+class CampaignCollaborationModerationRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    reason: str | None = Field(default=None, max_length=500)
+
+    @field_validator("reason")
+    @classmethod
+    def strip_reason(cls, value: str | None) -> str | None:
+        clean = value.strip() if value is not None else None
+        return clean or None
+
+
+class CampaignCollaborationReferenceResponse(BaseModel):
+    kind: CampaignCollaborationReferenceKind
+    id: str
+    label: str | None = None
+
+
+class CampaignCollaborationEntryResponse(BaseModel):
+    id: str
+    campaign_id: str
+    actor_user_id: str | None = None
+    actor_label: str
+    visibility: Literal["collaborators", "moderators"]
+    content: str | None = None
+    content_sha256: str
+    mention_user_ids: list[str] = Field(default_factory=list)
+    reference: CampaignCollaborationReferenceResponse | None = None
+    tombstone: Literal["withdrawn", "redacted"] | None = None
+    tombstone_reason: str | None = None
+    withdrawn_at: datetime | None = None
+    redacted_at: datetime | None = None
+    created_at: datetime
+
+
+class CampaignCollaborationListResponse(BaseModel):
+    items: list[CampaignCollaborationEntryResponse]
+    next_cursor: str | None = None
+    has_more: bool = False
+
+
 class CampaignLifecycleMutationRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")