Resolve campaign worker jobs to tenants

src/govoplan_campaign/backend/capabilities.py

@@ -690,6 +690,10 @@ def policy_context_capability(context: object) -> CampaignPolicyContextService:
 
 
 class CampaignDeliveryTaskService(CampaignDeliveryTaskProvider):
+    def tenant_id_for_job(self, session: object, *, job_id: str) -> str | None:
+        job = session.get(CampaignJob, job_id)  # type: ignore[attr-defined]
+        return job.tenant_id if job is not None else None
+
     def send_campaign_job(self, session: object, *, job_id: str, enqueue_imap_task: bool = True) -> Mapping[str, object]:
         from govoplan_campaign.backend.sending.jobs import send_campaign_job
 

src/govoplan_campaign/backend/manifest.py

@@ -796,7 +796,7 @@ manifest = ModuleManifest(
             id="campaigns.mail-profile-operations",
             title="Operate profile-backed campaign delivery",
             summary="Workers re-authorize and resolve Mail profiles at execution time while Campaign retains only opaque Mail-owned revisions and outcomes.",
-            body="A legacy snapshot, unauthorized or inactive profile, profile-reference mismatch, or changed SMTP/IMAP transport revision stops delivery. Preserve the record, migrate or correct the profile selection, revalidate, rebuild, and only then queue again. Password-only rotation remains possible without copying secrets into Campaign. Uncertain SMTP and IMAP effects remain blocked until an evidence-backed operator reconciliation.",
+            body="A legacy snapshot, unauthorized or inactive profile, profile-reference mismatch, or changed SMTP/IMAP transport revision stops delivery. Preserve the record, migrate or correct the profile selection, revalidate, rebuild, and only then queue again. Password-only rotation remains possible without copying secrets into Campaign. Uncertain SMTP and IMAP effects remain blocked until an evidence-backed operator reconciliation. If Campaign becomes unavailable to the tenant after a job was accepted, the worker leaves the job untouched and reports an operator action instead of sending or dropping it.",
             layer="configured",
             documentation_types=("admin",),
             audience=("campaign_sender", "campaign_operator", "mail_admin"),

tests/test_sending_jobs.py

@@ -60,6 +60,19 @@ class CampaignQueueSelectionTests(unittest.TestCase):
         self.assertIsNotNone(capability)
         configure.assert_called_once_with(registry=registry, settings=settings)
 
+    def test_delivery_task_capability_resolves_job_tenant_before_effect(self):
+        capability = delivery_tasks_capability(
+            SimpleNamespace(registry=object(), settings=object())
+        )
+        session = SimpleNamespace(
+            get=lambda _model, _job_id: SimpleNamespace(tenant_id="tenant-1")
+        )
+
+        self.assertEqual(
+            "tenant-1",
+            capability.tenant_id_for_job(session, job_id="job-1"),
+        )
+
     def test_selects_queueable_jobs_without_reclassifying_retry_states(self):
         skipped_send = _job("1", send_status=JobSendStatus.FAILED_TEMPORARY.value)
         skipped_queue = _job("2", queue_status=JobQueueStatus.PAUSED.value)