feat: add resumable external work hand-offs

tests/test_triggers.py

@@ -12,6 +12,10 @@ from govoplan_core.core.access import (
     PrincipalRef,
 )
 from govoplan_core.core.automation import AutomationPrincipalResolution
+from govoplan_core.core.campaigns import (
+    CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION,
+    CampaignWorkHandoffInspection,
+)
 from govoplan_core.core.events import EventTenantRef, PlatformEvent
 from govoplan_core.core.recovery import RecoveryCheckpoint, RecoveryOperation
 from govoplan_core.core.runtime_coordination import (
@@ -39,6 +43,7 @@ from govoplan_workflow_engine.backend.schemas import (
     WorkflowNode,
 )
 from govoplan_workflow_engine.backend.service import (
+    WorkflowConflictError,
     activate_definition,
     create_definition,
 )
@@ -82,17 +87,58 @@ class AutomationProvider:
         )
 
 
+class CampaignHandoffProvider:
+    def __init__(self) -> None:
+        self.allowed = True
+        self.status = "completed"
+        self.revision = 2
+        self.inspections: list[tuple[str, int | None]] = []
+
+    def prepare_handoff(self, _session, _principal, *, request):
+        raise AssertionError("The external wait must not create Campaign work.")
+
+    def inspect_handoff(
+        self,
+        _session,
+        _principal,
+        *,
+        tenant_id,
+        assignment_id,
+        expected_revision=None,
+    ):
+        assert tenant_id == "tenant-1"
+        self.inspections.append((assignment_id, expected_revision))
+        return CampaignWorkHandoffInspection(
+            allowed=self.allowed,
+            status=self.status,
+            assignment_revision=self.revision,
+            action_url="/campaigns/campaign-1/work?assignment=assignment-1",
+            assignment_ref=f"campaign-work-assignment:assignment-1:r{self.revision}",
+            reason=None if self.allowed else "Campaign access was revoked.",
+            provenance={"access_rechecked": True},
+        )
+
+
 class Registry:
     def __init__(self) -> None:
         self.provider = AutomationProvider()
+        self.campaign = CampaignHandoffProvider()
 
     def has_capability(self, name: str) -> bool:
-        return name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER
+        return (
+            name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER
+            or (
+                name == CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION
+                and self.campaign is not None
+            )
+        )
 
     def capability(self, name: str):
         if not self.has_capability(name):
             raise KeyError(name)
-        return self.provider
+        if name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER:
+            return self.provider
+        return self.campaign
 
 
 def graph(start_type: str, *, wait: WorkflowNode | None = None) -> WorkflowGraph:
@@ -123,6 +169,83 @@ def graph(start_type: str, *, wait: WorkflowNode | None = None) -> WorkflowGraph
     return WorkflowGraph(nodes=nodes, edges=edges)
 
 
+def external_handoff_graph(*, timeout_after: str = "1h") -> WorkflowGraph:
+    return WorkflowGraph(
+        nodes=[
+            WorkflowNode(id="start", type="workflow.start.manual"),
+            WorkflowNode(
+                id="campaign_work",
+                type="workflow.external_handoff",
+                label="Complete Campaign review",
+                config={
+                    "provider_capability": CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION,
+                    "event_type": "campaign.work.changed",
+                    "event_filter": {
+                        "payload": {"assignment_id": "$input.assignment_id"}
+                    },
+                    "outcome_path": "payload.outcome",
+                    "terminal_outcomes": {
+                        "completed": "completed",
+                        "rejected": "rejected",
+                        "cancelled": "cancelled",
+                    },
+                    "observed_outcomes": ["assigned", "accepted", "reassigned"],
+                    "external_id": "$input.assignment_id",
+                    "expected_revision": "$input.assignment_revision",
+                    "action_url": "$input.action_url",
+                    "immutable_ref": "$input.assignment_ref",
+                    "optional_capabilities": "$input.optional_capabilities",
+                    "timeout_after": timeout_after,
+                    "view_surface_ids": ["campaigns.page.work"],
+                },
+            ),
+            WorkflowNode(id="completed", type="workflow.end.completed"),
+            WorkflowNode(
+                id="rejected",
+                type="workflow.end.cancelled",
+                config={"reason": "Campaign work rejected"},
+            ),
+            WorkflowNode(
+                id="cancelled",
+                type="workflow.end.cancelled",
+                config={"reason": "Campaign work cancelled"},
+            ),
+            WorkflowNode(
+                id="timed_out",
+                type="workflow.end.cancelled",
+                config={"reason": "Campaign work timed out"},
+            ),
+        ],
+        edges=[
+            WorkflowEdge(id="start-work", source="start", target="campaign_work"),
+            WorkflowEdge(
+                id="work-completed",
+                source="campaign_work",
+                source_port="completed",
+                target="completed",
+            ),
+            WorkflowEdge(
+                id="work-rejected",
+                source="campaign_work",
+                source_port="rejected",
+                target="rejected",
+            ),
+            WorkflowEdge(
+                id="work-cancelled",
+                source="campaign_work",
+                source_port="cancelled",
+                target="cancelled",
+            ),
+            WorkflowEdge(
+                id="work-timeout",
+                source="campaign_work",
+                source_port="timed_out",
+                target="timed_out",
+            ),
+        ],
+    )
+
+
 class WorkflowTriggerTests(unittest.TestCase):
     def setUp(self) -> None:
         self.engine = create_engine("sqlite:///:memory:")
@@ -175,6 +298,18 @@ class WorkflowTriggerTests(unittest.TestCase):
         )
         return definition
 
+    def _external_input(self) -> dict[str, object]:
+        return {
+            "assignment_id": "assignment-1",
+            "assignment_revision": 1,
+            "action_url": "/campaigns/campaign-1/work?assignment=assignment-1",
+            "assignment_ref": "campaign-work-assignment:assignment-1:r1",
+            "optional_capabilities": {
+                "tasks": False,
+                "notifications": False,
+            },
+        }
+
     def test_schedule_registration_dispatch_and_replay_are_durable(self) -> None:
         definition = self._definition(
             graph("workflow.start.schedule"),
@@ -280,6 +415,184 @@ class WorkflowTriggerTests(unittest.TestCase):
         self.assertEqual(1, result["waits_timed_out"])
         self.assertEqual("completed", instance.status)
 
+    def test_external_handoff_observes_duplicate_safe_events_and_resumes(self) -> None:
+        definition = self._definition(
+            external_handoff_graph(),
+            automation=False,
+        )
+        instance, _replayed = start_instance(
+            self.session,
+            tenant_id="tenant-1",
+            definition_id=definition.id,
+            actor_id="account-1",
+            principal=principal(),
+            registry=self.registry,
+            payload=WorkflowInstanceStartRequest(
+                idempotency_key="campaign-handoff-1",
+                input=self._external_input(),
+            ),
+        )
+        step = self.session.get(WorkflowInstanceStep, instance.current_step_id)
+        state = self.session.scalar(select(WorkflowWaitState))
+        assert step is not None and state is not None
+        self.assertEqual("external_handoff", state.mode)
+        self.assertEqual("assigned", step.handoff["state"])
+        self.assertEqual(
+            ["notifications", "tasks"],
+            step.handoff["unavailable_optional_capabilities"],
+        )
+        dispatcher = SqlWorkflowTriggerDispatcher(registry=self.registry)
+        accepted = PlatformEvent(
+            type="campaign.work.changed",
+            module_id="campaigns",
+            event_id="campaign-event-accepted",
+            tenant=EventTenantRef(id="tenant-1"),
+            payload={
+                "assignment_id": "assignment-1",
+                "assignment_revision": 2,
+                "outcome": "accepted",
+            },
+        )
+
+        observed = dispatcher.ingest_event(self.session, event=accepted)
+        duplicate = dispatcher.ingest_event(self.session, event=accepted)
+
+        self.assertEqual(0, observed["waits_triggered"])
+        self.assertEqual(0, duplicate["waits_triggered"])
+        self.assertEqual("accepted", step.handoff["state"])
+        self.assertEqual(
+            1,
+            self.session.query(WorkflowInstanceEvent)
+            .filter(
+                WorkflowInstanceEvent.kind
+                == "workflow.external_handoff.observed"
+            )
+            .count(),
+        )
+
+        completed = dispatcher.ingest_event(
+            self.session,
+            event=PlatformEvent(
+                type="campaign.work.changed",
+                module_id="campaigns",
+                event_id="campaign-event-completed",
+                tenant=EventTenantRef(id="tenant-1"),
+                payload={
+                    "assignment_id": "assignment-1",
+                    "assignment_revision": 2,
+                    "outcome": "completed",
+                },
+            ),
+        )
+        result = dispatcher.dispatch_due(self.session)
+
+        self.assertEqual(1, completed["waits_triggered"])
+        self.assertEqual(1, result["waits_resumed"])
+        self.assertEqual("completed", instance.status)
+        self.assertEqual(
+            [("assignment-1", 2)],
+            self.registry.campaign.inspections,
+        )
+        self.assertEqual(
+            "campaign-work-assignment:assignment-1:r2",
+            step.external_ref,
+        )
+
+    def test_external_handoff_revoked_access_blocks_until_rechecked(self) -> None:
+        definition = self._definition(
+            external_handoff_graph(),
+            automation=False,
+        )
+        instance, _replayed = start_instance(
+            self.session,
+            tenant_id="tenant-1",
+            definition_id=definition.id,
+            actor_id="account-1",
+            principal=principal(),
+            registry=self.registry,
+            payload=WorkflowInstanceStartRequest(
+                idempotency_key="campaign-handoff-revoked",
+                input=self._external_input(),
+            ),
+        )
+        dispatcher = SqlWorkflowTriggerDispatcher(registry=self.registry)
+        dispatcher.ingest_event(
+            self.session,
+            event=PlatformEvent(
+                type="campaign.work.changed",
+                module_id="campaigns",
+                tenant=EventTenantRef(id="tenant-1"),
+                payload={
+                    "assignment_id": "assignment-1",
+                    "assignment_revision": 2,
+                    "outcome": "completed",
+                },
+            ),
+        )
+        self.registry.campaign.allowed = False
+
+        blocked = dispatcher.dispatch_due(self.session)
+        step = self.session.get(WorkflowInstanceStep, instance.current_step_id)
+        assert step is not None
+
+        self.assertEqual(1, blocked["waits_skipped"])
+        self.assertEqual("waiting", instance.status)
+        self.assertEqual("blocked", step.handoff["state"])
+        self.assertIn("revoked", str(step.handoff["message"]))
+
+        self.registry.campaign.allowed = True
+        resumed = dispatcher.dispatch_due(self.session)
+
+        self.assertEqual(1, resumed["waits_resumed"])
+        self.assertEqual("completed", instance.status)
+
+    def test_external_handoff_timeout_and_optional_provider_absence(self) -> None:
+        definition = self._definition(
+            external_handoff_graph(timeout_after="1s"),
+            automation=False,
+        )
+        instance, _replayed = start_instance(
+            self.session,
+            tenant_id="tenant-1",
+            definition_id=definition.id,
+            actor_id="account-1",
+            principal=principal(),
+            registry=self.registry,
+            payload=WorkflowInstanceStartRequest(
+                idempotency_key="campaign-handoff-timeout",
+                input=self._external_input(),
+            ),
+        )
+
+        result = SqlWorkflowTriggerDispatcher(
+            registry=self.registry
+        ).dispatch_due(
+            self.session,
+            now=datetime.now(tz=UTC) + timedelta(seconds=2),
+        )
+
+        self.assertEqual(1, result["waits_timed_out"])
+        self.assertEqual("cancelled", instance.status)
+
+        unavailable = self._definition(
+            external_handoff_graph(),
+            automation=False,
+        )
+        self.registry.campaign = None  # type: ignore[assignment]
+        with self.assertRaisesRegex(WorkflowConflictError, "is not available"):
+            start_instance(
+                self.session,
+                tenant_id="tenant-1",
+                definition_id=unavailable.id,
+                actor_id="account-1",
+                principal=principal(),
+                registry=self.registry,
+                payload=WorkflowInstanceStartRequest(
+                    idempotency_key="campaign-handoff-unavailable",
+                    input=self._external_input(),
+                ),
+            )
+
     def test_parent_workflow_outcome_starts_pinned_child(self) -> None:
         parent = self._definition(
             graph("workflow.start.manual"),