feat: extend resident permit service journey

tests/fixtures/resident_parking_permit_journey.json

@@ -21,10 +21,17 @@
     "version": "3",
     "fields": {
       "applicant_name": "Ada Lovelace",
+      "applicant_email": "ada.lovelace@example.test",
       "residence_address": "Musterstrasse 17, 10115 Berlin",
       "licence_plate": "B-AL 1843"
     }
   },
+  "status_access": {
+    "mode": "email_link",
+    "email_field_key": "applicant_email",
+    "token_ttl_seconds": 1800,
+    "request_limit_per_hour": 3
+  },
   "assisted_intake": {
     "channel": "counter",
     "affected_party_ref": "party:resident-ada-lovelace",
@@ -59,6 +66,14 @@
     "delivery_channel": "postbox",
     "remedy": "review:administrative-court"
   },
+  "payment": {
+    "mode": "manual",
+    "amount_minor": 3000,
+    "currency": "EUR",
+    "subject": "Resident parking permit fee",
+    "due_days": 14,
+    "evidence_owner": "files"
+  },
   "records": {
     "file_plan_key": "traffic.resident-parking-permits",
     "retention_policy_ref": "records:resident-parking-permit"
@@ -67,14 +82,17 @@
     "automated": [
       "The published service and exact form revision drive digital intake.",
       "An authenticated assisted session uses the same exact form and validation rules while retaining purpose, authority, channel, party, accessibility, source, correction, and read-back provenance.",
+      "The configured applicant email issues a short-lived, hash-only status link through Notifications and exposes only the bounded status timeline.",
       "An idempotent replay returns the same persisted submission.",
       "The human review handoff survives a database-session restart and remains visible in Tasks until completion.",
       "The formal decision retains party, mandate, legal-basis, evidence, delivery, review, and exact revision references.",
+      "The Case-bound payment handoff creates a replay-safe obligation and accepts a full manual receipt only with exact amount, currency, transaction reference, and immutable evidence.",
       "Forms Runtime, Cases, and Decisions can expose exact snapshots for explicit eAkte filing."
     ],
     "manual_or_target": [
       "Complete the digital journey with keyboard and screen reader at desktop and mobile widths.",
       "Complete the assisted operator journey with keyboard and screen reader at desktop and mobile widths.",
+      "Open, resend, expire, and revoke the applicant status link with keyboard and screen reader at desktop and mobile widths.",
       "Verify the configured Postbox or external delivery provider, including unknown outcome and reconciliation.",
       "Restore the pinned composition and reconstruct the exact form, case, decision, delivery evidence, and eAkte chronology.",
       "Transfer through a named archive profile and retain independently signed target evidence."

tests/test_institutional_service_journey.py

@@ -5,6 +5,7 @@ from datetime import UTC, datetime, timedelta
 import json
 from pathlib import Path
 from types import SimpleNamespace
+from urllib.parse import parse_qs, urlparse
 import unittest
 
 from sqlalchemy import create_engine
@@ -15,6 +16,7 @@ from govoplan_core.core.access import PrincipalRef
 from govoplan_core.core.institutional import (
     CAPABILITY_FORM_DEFINITIONS,
     CAPABILITY_SERVICE_DEFINITIONS,
+    EvidenceReference,
     FormDefinition,
     FormFieldDefinition,
     InstitutionalReference,
@@ -23,6 +25,11 @@ from govoplan_core.core.institutional import (
     TemporalRevision,
     service_launch_capability,
 )
+from govoplan_core.core.notifications import CAPABILITY_NOTIFICATIONS_DISPATCH
+from govoplan_core.core.payments import (
+    ManualPaymentReconciliationCommand,
+    PaymentRequestCommand,
+)
 from govoplan_core.core.runtime_coordination import (
     DistributedLease,
     RuntimeIdentity,
@@ -50,6 +57,9 @@ from govoplan_forms_runtime.backend.db.models import (
     FormInstanceRevision,
     FormIntakeProfile,
     FormIntakeSession,
+    FormStatusAccessGrant,
+    FormStatusAccessPolicy,
+    FormStatusAccessToken,
 )
 from govoplan_forms_runtime.backend.intake import FormIntakeService
 from govoplan_forms_runtime.backend.service import (
@@ -57,7 +67,14 @@ from govoplan_forms_runtime.backend.service import (
     FormRuntimeService,
     FormsServiceLauncher,
 )
+from govoplan_forms_runtime.backend.status_access import FormStatusAccessService
 from govoplan_portal.backend.service_directory import PortalServiceDirectory
+from govoplan_payments.backend.db.models import (
+    PaymentEvent,
+    PaymentObligation,
+    PaymentReconciliation,
+)
+from govoplan_payments.backend.service import SqlPaymentRequestProvider
 from govoplan_tasks.backend.aggregation import aggregate_work_items
 from govoplan_workflow_engine.backend.db.models import (
     WorkflowDefinition,
@@ -173,9 +190,29 @@ class _FormRegistry(_Registry):
         self.capabilities[service_launch_capability("form")] = FormsServiceLauncher(
             self
         )
+        self.notifications = _NotificationProvider()
+        self.capabilities[CAPABILITY_NOTIFICATIONS_DISPATCH] = self.notifications
 
     def has(self, module_id: str) -> bool:
-        return module_id in {"portal", "forms", "forms_runtime"}
+        return module_id in {"portal", "forms", "forms_runtime", "notifications"}
+
+
+class _NotificationProvider:
+    def __init__(self) -> None:
+        self.requests: list[object] = []
+
+    def tenant_id_for_notification(self, session, *, notification_id):
+        return "tenant-1"
+
+    def enqueue_notification(self, session, request, *, enqueue_delivery=True):
+        self.requests.append(request)
+        return {"id": f"notification-{len(self.requests)}"}
+
+    def deliver_notification(self, session, *, notification_id):
+        return {"id": notification_id}
+
+    def deliver_pending(self, session, *, tenant_id=None, limit=50):
+        return {"delivered": 0}
 
 
 class _WorkflowTaskRegistry:
@@ -248,8 +285,10 @@ class InstitutionalServiceJourneyTests(unittest.TestCase):
     def test_reference_fixture_names_remaining_manual_target_evidence(self) -> None:
         self.assertEqual("Anwohnerparkausweis", JOURNEY["title_de"])
         self.assertEqual("de-DE", JOURNEY["locale"])
-        self.assertEqual(6, len(JOURNEY["acceptance"]["automated"]))
-        self.assertEqual(5, len(JOURNEY["acceptance"]["manual_or_target"]))
+        self.assertEqual("email_link", JOURNEY["status_access"]["mode"])
+        self.assertEqual("manual", JOURNEY["payment"]["mode"])
+        self.assertEqual(8, len(JOURNEY["acceptance"]["automated"]))
+        self.assertEqual(6, len(JOURNEY["acceptance"]["manual_or_target"]))
 
     def test_portal_launches_exact_form_revision_and_persists_submission(self) -> None:
         engine = create_engine("sqlite+pysqlite:///:memory:")
@@ -290,6 +329,13 @@ class InstitutionalServiceJourneyTests(unittest.TestCase):
                             required=True,
                             constraints={"min_length": 2},
                         ),
+                        FormFieldDefinition(
+                            key="applicant_email",
+                            label="Applicant email",
+                            value_type="email",
+                            required=True,
+                            constraints={"min_length": 5},
+                        ),
                         FormFieldDefinition(
                             key="residence_address",
                             label="Primary residence address",
@@ -395,6 +441,9 @@ class InstitutionalServiceJourneyTests(unittest.TestCase):
             FormIntakeProfile.__table__,
             FormIntakeSession.__table__,
             FormAssistedConfirmation.__table__,
+            FormStatusAccessPolicy.__table__,
+            FormStatusAccessGrant.__table__,
+            FormStatusAccessToken.__table__,
         ):
             table.create(engine)
         sessions = sessionmaker(bind=engine)
@@ -424,6 +473,9 @@ class InstitutionalServiceJourneyTests(unittest.TestCase):
                             FormFieldDefinition(
                                 key=key,
                                 label=key.replace("_", " ").title(),
+                                value_type=(
+                                    "email" if key == "applicant_email" else "text"
+                                ),
                                 required=True,
                                 constraints={"min_length": 2},
                             )
@@ -435,6 +487,20 @@ class InstitutionalServiceJourneyTests(unittest.TestCase):
                     ),
                 )
                 registry = _FormRegistry(_service())
+                status_access = JOURNEY["status_access"]
+                FormStatusAccessService(registry).upsert_policy(
+                    session,
+                    principal,
+                    definition_ref=form.reference,
+                    mode=status_access["mode"],
+                    enabled=True,
+                    email_field_key=status_access["email_field_key"],
+                    token_ttl_seconds=status_access["token_ttl_seconds"],
+                    request_limit_per_hour=status_access[
+                        "request_limit_per_hour"
+                    ],
+                    recorded_at=NOW,
+                )
                 intake = FormIntakeService(registry)
                 profile = intake.create_profile(
                     session,
@@ -536,6 +602,50 @@ class InstitutionalServiceJourneyTests(unittest.TestCase):
                 self.assertEqual("submitted", submitted.status)
                 self.assertEqual(current.revision, confirmation.instance_revision)
                 self.assertEqual(assisted["affected_party_ref"], confirmation.confirmed_by_ref)
+                status_service = FormStatusAccessService(registry)
+                access = status_service.access_summary_for_instance(
+                    resumed,
+                    tenant_id="tenant-1",
+                    instance_id=instance_id,
+                )
+                self.assertIsNotNone(access)
+                tracking_id = str(access["tracking_id"])
+                challenge = status_service.public_access_challenge(
+                    resumed,
+                    tracking_id=tracking_id,
+                )
+                self.assertEqual("email_link", challenge["mode"])
+                self.assertFalse(
+                    status_service.request_email_link(
+                        resumed,
+                        tracking_id=tracking_id,
+                        email="wrong@example.test",
+                        requested_at=NOW + timedelta(minutes=5),
+                    )
+                )
+                self.assertTrue(
+                    status_service.request_email_link(
+                        resumed,
+                        tracking_id=tracking_id,
+                        email=JOURNEY["form"]["fields"]["applicant_email"],
+                        requested_at=NOW + timedelta(minutes=6),
+                    )
+                )
+                notification = registry.notifications.requests[-1]
+                query = parse_qs(urlparse(notification.action_url).query)
+                projection = status_service.get_public_projection(
+                    resumed,
+                    tracking_id=tracking_id,
+                    token=query["token"][0],
+                    observed_at=NOW + timedelta(minutes=7),
+                )
+                self.assertEqual("submitted", projection["status"])
+                self.assertEqual(JOURNEY["title"], projection["title"])
+                self.assertEqual(
+                    ["submitted"],
+                    [item["status"] for item in projection["timeline"]],
+                )
+                self.assertNotIn("values", projection)
         finally:
             engine.dispose()
 
@@ -682,6 +792,76 @@ class InstitutionalServiceJourneyTests(unittest.TestCase):
             bind_process_runtime_identity(None)
             engine.dispose()
 
+    def test_case_bound_payment_handoff_is_replay_safe_and_evidence_bound(
+        self,
+    ) -> None:
+        engine = create_engine("sqlite+pysqlite:///:memory:")
+        for table in (
+            PaymentObligation.__table__,
+            PaymentReconciliation.__table__,
+            PaymentEvent.__table__,
+        ):
+            table.create(engine)
+        session = Session(engine)
+        provider = SqlPaymentRequestProvider()
+        payment = JOURNEY["payment"]
+        try:
+            command = PaymentRequestCommand(
+                tenant_id="tenant-1",
+                source_module="cases",
+                source_resource_type="case",
+                source_resource_id="case-1",
+                amount_minor=payment["amount_minor"],
+                currency=payment["currency"],
+                subject=payment["subject"],
+                idempotency_key="resident-permit-case-1-fee",
+                requested_at=NOW + timedelta(days=1),
+                requested_by_ref="workflow:resident-parking-permit-review",
+                due_at=NOW + timedelta(days=1 + payment["due_days"]),
+                context_refs={
+                    "case": "case-1",
+                    "workflow": "workflow:resident-parking-permit-review",
+                },
+            )
+            requested = provider.request_payment(session, command)
+            replay = provider.request_payment(session, command)
+            self.assertEqual(requested["payment_id"], replay["payment_id"])
+            self.assertTrue(replay["replayed"])
+            self.assertEqual("case-1", requested["source"]["resource_id"])
+
+            paid = provider.reconcile_manual_payment(
+                session,
+                ManualPaymentReconciliationCommand(
+                    tenant_id="tenant-1",
+                    payment_id=str(requested["payment_id"]),
+                    amount_minor=payment["amount_minor"],
+                    currency=payment["currency"],
+                    transaction_reference="BANK-RPP-2026-0001",
+                    evidence_ref=EvidenceReference(
+                        kind="document",
+                        owner_module=payment["evidence_owner"],
+                        evidence_id="file-payment-rpp-1",
+                        tenant_id="tenant-1",
+                        version="1",
+                        checksum="b" * 64,
+                    ),
+                    idempotency_key="resident-permit-bank-receipt-1",
+                    received_at=NOW + timedelta(days=2),
+                    recorded_at=NOW + timedelta(days=2, minutes=5),
+                    recorded_by_ref="account:payment-officer-1",
+                ),
+            )
+            session.commit()
+            self.assertEqual("paid", paid["status"])
+            self.assertEqual(
+                "BANK-RPP-2026-0001",
+                paid["reconciliation"]["transaction_reference"],
+            )
+            self.assertEqual(2, len(paid["events"]))
+        finally:
+            session.close()
+            engine.dispose()
+
 
 if __name__ == "__main__":
     unittest.main()