Implement correlated campaign invitations

docs/CALENDAR_INTEGRATION_CONCEPT.md

@@ -219,6 +219,18 @@ Tasks/workflow remain owners of assignment, status, SLA logic, and completion se
 
 Mail remains owner of SMTP/IMAP profiles and mailbox transport. Notifications remains owner of delivery channels and delivery policy.
 
+The versioned `calendar.invitations` capability is the concrete Campaign/Mail
+boundary. Campaign renders and freezes one `METHOD:REQUEST` attachment per
+recipient, then upserts the correlated VEVENT only after delivery acceptance.
+Calendar owns attendee `PARTSTAT`, response timestamps, bounded evidence,
+CalDAV outbox state, and batched correlation/summary queries. Mail can forward
+`METHOD:REPLY` parts discovered by an authorized, checkpointed IMAP
+delivery-status source. Replaying the same mailbox evidence is idempotent.
+Campaign reports read current Calendar state and retain only the invitation
+request and mirror result in their own delivery provenance. Recurring Campaign
+invitation series remain a separate workflow rather than being inferred from
+unrelated recipient rows.
+
 ### Documents And DMS
 
 `govoplan-dms` can link documents to events for:

src/govoplan_calendar/backend/capabilities.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import hashlib
+from collections import Counter
 from collections.abc import Mapping, Sequence
 from datetime import datetime
+from types import SimpleNamespace
 
 from sqlalchemy.orm import Session
 
@@ -14,13 +16,18 @@ from govoplan_core.core.calendar import (
     CalendarExternalProfileRef,
     CalendarExternalProfileRequest,
     CalendarInvitationAttendeeRequest,
+    CalendarInvitationCalendarRef,
     CalendarInvitationProvider,
     CalendarInvitationRef,
     CalendarInvitationRequest,
     CalendarSchedulingProvider,
 )
 from govoplan_core.security.time import utc_now
-from govoplan_calendar.backend.db.models import CalendarEvent
+from govoplan_calendar.backend.db.models import (
+    CalendarEvent,
+    CalendarSyncSource,
+)
+from govoplan_calendar.backend.ical import ICalendarError, event_to_ics, parse_vevents
 from govoplan_calendar.backend.schemas import (
     CalendarEventCreateRequest,
     CalendarEventUpdateRequest,
@@ -30,6 +37,7 @@ from govoplan_calendar.backend.service import (
     CalendarError,
     create_sync_source,
     create_event,
+    list_calendars as list_calendar_collections,
     list_freebusy,
     update_event,
 )
@@ -159,13 +167,89 @@ def _invitation_ref(event: CalendarEvent) -> CalendarInvitationRef:
         attendees=tuple(dict(item) for item in event.attendees or []),
         external_state=state,
         outbox_operation_id=operation_id,
+        reply_ingress="icalendar-reply",
         degraded_reasons=(
             "Recurring campaign invitations require a separate series workflow.",
-            "Mail-delivered iCalendar replies require a Mail adapter to call record_response.",
         ),
     )
 
 
+def _invitation_render_event(request: CalendarInvitationRequest) -> SimpleNamespace:
+    return SimpleNamespace(
+        uid=_invitation_uid(request),
+        recurrence_id=None,
+        sequence=0,
+        summary=request.summary,
+        description=request.description,
+        location=request.location,
+        status="CONFIRMED",
+        transparency="OPAQUE",
+        classification=request.classification,
+        start_at=request.start_at,
+        end_at=request.end_at,
+        duration_seconds=None,
+        all_day=False,
+        timezone=request.timezone,
+        organizer=dict(request.organizer) if request.organizer else None,
+        attendees=[_attendee_record(item) for item in request.attendees],
+        categories=list(request.categories),
+        rrule=None,
+        rdate=[],
+        exdate=[],
+        reminders=[],
+        attachments=[],
+        related_to=[],
+        icalendar={"method": "REQUEST"},
+    )
+
+
+def _calendar_source(
+    session: Session,
+    *,
+    tenant_id: str,
+    calendar_id: str,
+) -> CalendarSyncSource | None:
+    return (
+        session.query(CalendarSyncSource)
+        .filter(
+            CalendarSyncSource.tenant_id == tenant_id,
+            CalendarSyncSource.calendar_id == calendar_id,
+            CalendarSyncSource.deleted_at.is_(None),
+        )
+        .order_by(CalendarSyncSource.created_at.desc())
+        .first()
+    )
+
+
+def _invitation_summary(events: Sequence[CalendarEvent]) -> dict[str, object]:
+    attendee_statuses: Counter[str] = Counter()
+    external_states: Counter[str] = Counter()
+    degraded_reasons: set[str] = set()
+    replied = 0
+    for event in events:
+        ref = _invitation_ref(event)
+        external_states[ref.external_state] += 1
+        degraded_reasons.update(ref.degraded_reasons)
+        for attendee in ref.attendees:
+            status = _attendee_status(attendee)
+            attendee_statuses[status] += 1
+            if status != "NEEDS-ACTION":
+                replied += 1
+    attendee_total = sum(attendee_statuses.values())
+    return {
+        "available": True,
+        "invitation_count": len(events),
+        "attendee_count": attendee_total,
+        "response_count": replied,
+        "pending_response_count": attendee_statuses.get("NEEDS-ACTION", 0),
+        "by_participation_status": dict(attendee_statuses),
+        "by_external_state": dict(external_states),
+        "reply_ingress": "icalendar-reply",
+        "recurrence_supported": False,
+        "degraded_reasons": sorted(degraded_reasons),
+    }
+
+
 class SqlCalendarSchedulingProvider(CalendarSchedulingProvider):
     def list_freebusy(
         self,
@@ -342,6 +426,54 @@ class SqlCalendarExternalProfileProvider(CalendarExternalProfileProvider):
 
 
 class SqlCalendarInvitationProvider(CalendarInvitationProvider):
+    def list_calendars(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        user_id: str | None = None,
+        group_ids: Sequence[str] = (),
+        can_admin: bool = False,
+    ) -> tuple[CalendarInvitationCalendarRef, ...]:
+        db = self._session(session)
+        calendars = list_calendar_collections(
+            db,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            group_ids=group_ids,
+            can_admin=can_admin,
+        )
+        result: list[CalendarInvitationCalendarRef] = []
+        for calendar in calendars:
+            source = _calendar_source(
+                db,
+                tenant_id=tenant_id,
+                calendar_id=calendar.id,
+            )
+            writable = source is None or (
+                source.source_kind == "caldav"
+                and source.sync_enabled
+                and source.sync_direction == "two_way"
+            )
+            result.append(
+                CalendarInvitationCalendarRef(
+                    id=calendar.id,
+                    name=calendar.name,
+                    color=calendar.color,
+                    timezone=calendar.timezone,
+                    source_kind=source.source_kind if source is not None else "local",
+                    writable=writable,
+                )
+            )
+        return tuple(result)
+
+    def render_invitation(self, request: CalendarInvitationRequest) -> str:
+        self._validate_request(request)
+        try:
+            return event_to_ics(_invitation_render_event(request))
+        except (ICalendarError, TypeError, ValueError) as exc:
+            raise CalendarCapabilityError(str(exc)) from exc
+
     def upsert_invitation(
         self,
         session: object,
@@ -444,6 +576,64 @@ class SqlCalendarInvitationProvider(CalendarInvitationProvider):
         )
         return _invitation_ref(event) if event is not None else None
 
+    def get_invitations(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        correlation_ids: Sequence[str],
+    ) -> dict[str, CalendarInvitationRef]:
+        normalized = tuple(
+            dict.fromkeys(
+                str(value).strip()
+                for value in correlation_ids
+                if str(value).strip()
+            )
+        )
+        if not normalized:
+            return {}
+        if len(normalized) > 500:
+            raise CalendarCapabilityError(
+                "At most 500 invitation correlation IDs may be queried at once."
+            )
+        events = (
+            self._session(session)
+            .query(CalendarEvent)
+            .filter(
+                CalendarEvent.tenant_id == tenant_id,
+                CalendarEvent.correlation_key.in_(normalized),
+                CalendarEvent.deleted_at.is_(None),
+            )
+            .all()
+        )
+        return {
+            event.correlation_key: _invitation_ref(event)
+            for event in events
+            if event.correlation_key
+        }
+
+    def summarize_invitations(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        source_module: str,
+        source_resource_type: str,
+        source_resource_id: str | None,
+    ) -> Mapping[str, object]:
+        query = self._session(session).query(CalendarEvent).filter(
+            CalendarEvent.tenant_id == tenant_id,
+            CalendarEvent.producer_module == source_module,
+            CalendarEvent.producer_resource_type == source_resource_type,
+            CalendarEvent.deleted_at.is_(None),
+        )
+        query = query.filter(
+            CalendarEvent.producer_resource_id == source_resource_id
+            if source_resource_id is not None
+            else CalendarEvent.producer_resource_id.is_(None)
+        )
+        return _invitation_summary(query.all())
+
     def record_response(
         self,
         session: object,
@@ -482,21 +672,35 @@ class SqlCalendarInvitationProvider(CalendarInvitationProvider):
         target = attendee_address.strip().casefold()
         attendees: list[dict[str, object]] = []
         matched = False
+        changed = False
         response_time = responded_at or utc_now()
         for raw in event.attendees or []:
             item = dict(raw)
             if _attendee_address(item) == target:
+                current_evidence = item.get("response_evidence")
+                same_evidence = (
+                    bool(evidence)
+                    and isinstance(current_evidence, Mapping)
+                    and dict(current_evidence) == dict(evidence)
+                )
+                if _attendee_status(item) == status and same_evidence:
+                    matched = True
+                    attendees.append(item)
+                    continue
                 params = dict(item.get("params") or {})
                 params["PARTSTAT"] = [status]
                 item["params"] = params
                 item["response_at"] = response_time.isoformat()
                 item["response_evidence"] = dict(evidence or {})
                 matched = True
+                changed = True
             attendees.append(item)
         if not matched:
             raise CalendarCapabilityError(
                 "The reply address is not an attendee of this invitation."
             )
+        if not changed:
+            return _invitation_ref(event)
         metadata = dict(event.metadata_ or {})
         invitation_metadata = dict(metadata.get("calendar_invitation") or {})
         invitation_metadata["last_response_at"] = response_time.isoformat()
@@ -516,6 +720,78 @@ class SqlCalendarInvitationProvider(CalendarInvitationProvider):
             raise CalendarCapabilityError(str(exc)) from exc
         return _invitation_ref(event)
 
+    def record_icalendar_reply(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        icalendar: str,
+        received_at: datetime | None = None,
+        evidence: Mapping[str, object] | None = None,
+    ) -> tuple[CalendarInvitationRef, ...]:
+        try:
+            replies = parse_vevents(icalendar)
+        except ICalendarError as exc:
+            raise CalendarCapabilityError(str(exc)) from exc
+        recorded: list[CalendarInvitationRef] = []
+        for reply in replies:
+            metadata = reply.get("icalendar")
+            method = metadata.get("method") if isinstance(metadata, Mapping) else None
+            if str(method or "").upper() != "REPLY":
+                continue
+            uid = str(reply.get("uid") or "").strip()
+            if not uid:
+                continue
+            event = (
+                self._session(session)
+                .query(CalendarEvent)
+                .filter(
+                    CalendarEvent.tenant_id == tenant_id,
+                    CalendarEvent.uid == uid,
+                    CalendarEvent.correlation_key.is_not(None),
+                    CalendarEvent.producer_module == "campaigns",
+                    CalendarEvent.deleted_at.is_(None),
+                )
+                .order_by(CalendarEvent.created_at.asc())
+                .first()
+            )
+            if event is None:
+                continue
+            for attendee in reply.get("attendees") or []:
+                if not isinstance(attendee, Mapping):
+                    continue
+                status = _attendee_status(attendee)
+                address = _attendee_address(attendee)
+                if (
+                    not address
+                    or status not in _PARTICIPATION_STATUSES - {"NEEDS-ACTION"}
+                ):
+                    continue
+                response_evidence = dict(evidence or {})
+                response_evidence.update(
+                    {
+                        "method": "REPLY",
+                        "uid": uid,
+                        "sequence": int(reply.get("sequence") or 0),
+                    }
+                )
+                try:
+                    recorded.append(
+                        self.record_response(
+                            session,
+                            tenant_id=tenant_id,
+                            attendee_address=address,
+                            participation_status=status,
+                            uid=uid,
+                            responded_at=received_at,
+                            evidence=response_evidence,
+                        )
+                    )
+                except CalendarCapabilityError as exc:
+                    if "not an attendee" not in str(exc):
+                        raise
+        return tuple(recorded)
+
     @staticmethod
     def _session(session: object) -> Session:
         if not isinstance(session, Session):

src/govoplan_calendar/backend/manifest.py

@@ -17,6 +17,7 @@ from govoplan_core.core.calendar import (
 )
 from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard
 from govoplan_core.core.modules import (
+    DocumentationCondition,
     DocumentationTopic,
     FrontendModule,
     FrontendRoute,
@@ -497,7 +498,7 @@ manifest = ModuleManifest(
     provides_interfaces=(
         ModuleInterfaceProvider(name="calendar.outbox", version="0.1.8"),
         ModuleInterfaceProvider(name="calendar.scheduling", version="0.1.8"),
-        ModuleInterfaceProvider(name="calendar.invitations", version="0.1.0"),
+        ModuleInterfaceProvider(name="calendar.invitations", version="0.2.0"),
         ModuleInterfaceProvider(name="calendar.external_profiles", version="0.1.0"),
     ),
     permissions=PERMISSIONS,
@@ -554,6 +555,27 @@ manifest = ModuleManifest(
             related_modules=("connectors", "audit", "ops"),
             metadata={"kind": "reference"},
         ),
+        DocumentationTopic(
+            id="calendar.campaign-invitations-and-replies",
+            title="Track Campaign invitations and attendee replies",
+            summary="Mirror accepted Campaign invitation deliveries as correlated VEVENTs and keep attendee participation state authoritative in Calendar.",
+            body="Campaign can render a METHOD:REQUEST attachment before delivery and create or update the correlated Calendar event only after delivery acceptance. Calendar stores correlation, attendee PARTSTAT, response time, bounded evidence, synchronization state, and any degraded behavior. When Mail watches an authorized IMAP folder, METHOD:REPLY parts are forwarded idempotently to Calendar. Campaign reports query the current state in batches instead of copying it into Campaign records. Recurring Campaign invitation series remain a separate workflow.",
+            documentation_types=("admin", "user"),
+            audience=("calendar_manager", "campaign_manager", "operator"),
+            conditions=(
+                DocumentationCondition(
+                    required_modules=("calendar", "campaigns"),
+                    required_capabilities=(CAPABILITY_CALENDAR_INVITATIONS,),
+                    any_scopes=(
+                        "calendar:calendar:read",
+                        "calendar:calendar:write",
+                        "calendar:calendar:admin",
+                    ),
+                ),
+            ),
+            related_modules=("campaigns", "mail", "audit"),
+            metadata={"kind": "workflow"},
+        ),
         DocumentationTopic(
             id="calendar.outbound-change-recovery",
             title="Recover synchronized calendar writes",

tests/test_capabilities.py

@@ -12,6 +12,7 @@ from govoplan_calendar.backend.capabilities import (
     SqlCalendarInvitationProvider,
     SqlCalendarSchedulingProvider,
 )
+from govoplan_calendar.backend.db.models import CalendarEvent
 from govoplan_calendar.backend.schemas import CalendarCollectionCreateRequest
 from govoplan_calendar.backend.service import create_calendar
 from govoplan_core.core.calendar import (
@@ -131,6 +132,10 @@ class CalendarInvitationCapabilityTests(unittest.TestCase):
             summary=summary,
             start_at=datetime(2026, 8, 5, 9, tzinfo=timezone.utc),
             end_at=datetime(2026, 8, 5, 10, tzinfo=timezone.utc),
+            organizer={
+                "value": "mailto:organizer@example.test",
+                "params": {"CN": ["Organizer"]},
+            },
             attendees=(
                 CalendarInvitationAttendeeRequest(
                     address="ada@example.test",
@@ -173,6 +178,92 @@ class CalendarInvitationCapabilityTests(unittest.TestCase):
         self.assertFalse(updated.recurrence_supported)
         self.assertTrue(updated.degraded_reasons)
 
+    def test_invitation_render_batch_lookup_and_summary(self) -> None:
+        request = self.request()
+        payload = self.provider.render_invitation(request)
+        created = self.provider.upsert_invitation(
+            self.session,
+            tenant_id="tenant-1",
+            user_id=None,
+            request=request,
+        )
+
+        self.assertIn("METHOD:REQUEST", payload)
+        self.assertIn(f"UID:{created.uid}", payload)
+        self.assertEqual(
+            created,
+            self.provider.get_invitations(
+                self.session,
+                tenant_id="tenant-1",
+                correlation_ids=(created.correlation_id,),
+            )[created.correlation_id],
+        )
+        summary = self.provider.summarize_invitations(
+            self.session,
+            tenant_id="tenant-1",
+            source_module="campaigns",
+            source_resource_type="campaign_recipient",
+            source_resource_id="recipient-1",
+        )
+        self.assertEqual(1, summary["invitation_count"])
+        self.assertEqual(1, summary["pending_response_count"])
+        calendars = self.provider.list_calendars(
+            self.session,
+            tenant_id="tenant-1",
+        )
+        self.assertEqual((self.calendar.id,), tuple(item.id for item in calendars))
+        self.assertTrue(calendars[0].writable)
+
+    def test_icalendar_reply_updates_attendee_status(self) -> None:
+        created = self.provider.upsert_invitation(
+            self.session,
+            tenant_id="tenant-1",
+            user_id=None,
+            request=self.request(),
+        )
+        reply = "\r\n".join(
+            (
+                "BEGIN:VCALENDAR",
+                "VERSION:2.0",
+                "METHOD:REPLY",
+                "BEGIN:VEVENT",
+                f"UID:{created.uid}",
+                "DTSTART:20260805T090000Z",
+                "ATTENDEE;CN=Ada;PARTSTAT=TENTATIVE:mailto:ada@example.test",
+                "END:VEVENT",
+                "END:VCALENDAR",
+                "",
+            )
+        )
+
+        recorded = self.provider.record_icalendar_reply(
+            self.session,
+            tenant_id="tenant-1",
+            icalendar=reply,
+            evidence={"profile_id": "mail-profile-1", "uid": "42"},
+        )
+        sequence_after_first_reply = self.session.get(
+            CalendarEvent,
+            created.event_id,
+        ).sequence
+        replayed = self.provider.record_icalendar_reply(
+            self.session,
+            tenant_id="tenant-1",
+            icalendar=reply,
+            evidence={"profile_id": "mail-profile-1", "uid": "42"},
+        )
+
+        self.assertEqual(1, len(recorded))
+        self.assertEqual(1, len(replayed))
+        self.assertEqual(
+            "TENTATIVE",
+            recorded[0].attendees[0]["params"]["PARTSTAT"][0],
+        )
+        self.assertEqual(
+            sequence_after_first_reply,
+            self.session.get(CalendarEvent, created.event_id).sequence,
+        )
+
     def test_reply_must_match_an_existing_attendee(self) -> None:
         created = self.provider.upsert_invitation(
             self.session,