feat: add Campaign work orchestration contract

tests/test_access_contracts.py

@@ -71,6 +71,7 @@ from govoplan_core.core.campaigns import (
     CAPABILITY_CAMPAIGNS_MAIL_POLICY_CONTEXT,
     CAPABILITY_CAMPAIGNS_POLICY_CONTEXT,
     CAPABILITY_CAMPAIGNS_RETENTION,
+    CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION,
     CampaignAccessProvider,
     CampaignDeliveryTaskProvider,
     CampaignMailPolicyContext,
@@ -78,6 +79,10 @@ from govoplan_core.core.campaigns import (
     CampaignPolicyContext,
     CampaignPolicyContextProvider,
     CampaignRetentionProvider,
+    CampaignWorkHandoffInspection,
+    CampaignWorkHandoffRef,
+    CampaignWorkHandoffRequest,
+    CampaignWorkOrchestrationProvider,
 )
 from govoplan_core.core.files import CAPABILITY_FILES_ACCESS, FileAccessProvider
 from govoplan_core.core.modules import ModuleContext, ModuleManifest
@@ -464,6 +469,40 @@ class _FakeCampaignRetentionProvider:
         return {"raw_campaign_json": {"eligible": int(dry_run)}}
 
 
+class _FakeCampaignWorkOrchestrationProvider:
+    def prepare_handoff(self, session: object, principal: object, *, request):
+        del session, principal
+        return CampaignWorkHandoffRef(
+            tenant_id=request.tenant_id,
+            campaign_id=request.campaign_id or "campaign-created",
+            campaign_version_id="campaign-version-1",
+            campaign_revision=1,
+            assignment_id="assignment-1",
+            assignment_revision=1,
+            status="open",
+            action_url="/campaigns/campaign-1/work?assignment=assignment-1",
+            campaign_ref="campaign:campaign-1:version:campaign-version-1:r1",
+            assignment_ref="campaign-work-assignment:assignment-1:r1",
+        )
+
+    def inspect_handoff(
+        self,
+        session: object,
+        principal: object,
+        *,
+        tenant_id: str,
+        assignment_id: str,
+        expected_revision: int | None = None,
+    ):
+        del session, principal, tenant_id, assignment_id
+        return CampaignWorkHandoffInspection(
+            allowed=expected_revision in {None, 1},
+            status="open",
+            assignment_revision=1,
+            assignment_ref="campaign-work-assignment:assignment-1:r1",
+        )
+
+
 class _FakeSecretProvider:
     def __init__(self) -> None:
         self._values: dict[str, str] = {}
@@ -528,6 +567,10 @@ class AccessContractTests(unittest.TestCase):
         self.assertEqual("campaigns.mailPolicyContext", CAPABILITY_CAMPAIGNS_MAIL_POLICY_CONTEXT)
         self.assertEqual("campaigns.policyContext", CAPABILITY_CAMPAIGNS_POLICY_CONTEXT)
         self.assertEqual("campaigns.retention", CAPABILITY_CAMPAIGNS_RETENTION)
+        self.assertEqual(
+            "campaigns.workOrchestration",
+            CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION,
+        )
         self.assertEqual("tenancy.tenantResolver", CAPABILITY_TENANCY_TENANT_RESOLVER)
         self.assertEqual("security.secretProvider", CAPABILITY_SECURITY_SECRET_PROVIDER)
         self.assertEqual("audit.sink", CAPABILITY_AUDIT_SINK)
@@ -642,6 +685,10 @@ class AccessContractTests(unittest.TestCase):
         self.assertIsInstance(_FakeCampaignMailPolicyContextProvider(), CampaignMailPolicyContextProvider)
         self.assertIsInstance(_FakeCampaignPolicyContextProvider(), CampaignPolicyContextProvider)
         self.assertIsInstance(_FakeCampaignRetentionProvider(), CampaignRetentionProvider)
+        self.assertIsInstance(
+            _FakeCampaignWorkOrchestrationProvider(),
+            CampaignWorkOrchestrationProvider,
+        )
         self.assertIsInstance(_FakeSecretProvider(), SecretProvider)
         self.assertIsInstance(_FakeAuditSink(), AuditSink)
         self.assertIsInstance(_FakeAuditRecorder(), AuditRecorder)
@@ -676,6 +723,37 @@ class AccessContractTests(unittest.TestCase):
         self.assertEqual({"job_id": "job-1", "status": "appended"}, delivery_provider.append_sent_for_job(object(), job_id="job-1"))
         self.assertEqual({"raw_campaign_json": {"eligible": 1}}, retention_provider.apply_retention(object(), dry_run=True, now=object(), policy_for_campaign_id=lambda campaign_id: object()))
 
+    def test_campaign_work_handoff_contract_requires_one_campaign_source(self) -> None:
+        request = CampaignWorkHandoffRequest(
+            tenant_id="tenant-1",
+            campaign_id="campaign-1",
+            idempotency_key="workflow-step-1",
+            purpose="Review the campaign",
+            assignee_kind="account",
+            assignee_id="account-1",
+        )
+        provider = _FakeCampaignWorkOrchestrationProvider()
+
+        handoff = provider.prepare_handoff(object(), object(), request=request)
+        inspection = provider.inspect_handoff(
+            object(),
+            object(),
+            tenant_id="tenant-1",
+            assignment_id=handoff.assignment_id,
+            expected_revision=handoff.assignment_revision,
+        )
+
+        self.assertEqual("campaign-1", handoff.campaign_id)
+        self.assertTrue(inspection.allowed)
+        with self.assertRaisesRegex(ValueError, "either reference one campaign"):
+            CampaignWorkHandoffRequest(
+                tenant_id="tenant-1",
+                idempotency_key="workflow-step-2",
+                purpose="Review",
+                assignee_kind="account",
+                assignee_id="account-1",
+            )
+
     def test_access_capabilities_register_and_resolve_through_platform_registry(self) -> None:
         directory = _FakeAccessDirectory()
         semantic_directory = _FakeAccessSemanticDirectory()