feat: extend resident permit service journey

docs/strategy/REFERENCE_JOURNEY_PROGRAM.md

@@ -48,6 +48,23 @@ Decision, Postbox delivery, and Records target. Changing this flagship scenario
 is a product decision; implementations may add further scenarios without
 weakening or silently replacing its acceptance gates.
 
+The reference fixes an email-link applicant-status profile. The exact
+published Form revision names the linked email field and bounded expiry/request
+limits. Submission issues a tracking grant, a matching request delegates mail
+delivery to Notifications using a hash-only short-lived secret, and Portal
+presents only the public lifecycle projection. Forms Runtime's module tests
+also cover authenticated-only and permanent-link variants; the flagship keeps
+email-link mode because it exercises identity minimization, delivery,
+revocation, resend, expiry, and non-enumerating failure behavior in one slice.
+
+The Case-to-payment handoff now has an executable first contract as well. The
+flagship requests a fixed EUR obligation through `payments.requests`, retains
+the Case and Workflow context references, proves exact replay, and reconciles a
+full offline receipt against a Files-owned immutable evidence reference. This
+does not simulate online checkout or accounting: provider callbacks, partial
+payments, corrections, refunds, Ledger posting, and XRechnung remain separate
+governed slices.
+
 The Records vertical now supplies the journey's native file plan, immutable
 record and item revisions, chronology, close/reopen, retention calculation,
 holds, appraisal, independent disposition approval, recovery-ledger evidence,

packages/govoplan-meta/pyproject.toml

@@ -57,6 +57,7 @@ full = [
   "govoplan-mandates==0.1.18",
   "govoplan-notifications==0.1.18",
   "govoplan-parties==0.1.18",
+  "govoplan-payments==0.1.19",
   "govoplan-permits==0.1.18",
   "govoplan-poll==0.1.18",
   "govoplan-portal==0.1.18",

requirements-dev.txt

@@ -30,6 +30,7 @@
 -e ../govoplan-parties
 -e ../govoplan-mandates
 -e ../govoplan-decisions
+-e ../govoplan-payments
 -e ../govoplan-connectors
 -e ../govoplan-datasources
 -e ../govoplan-dataflow

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("email_link", JOURNEY["status_access"]["mode"])
-        self.assertEqual(5, len(JOURNEY["acceptance"]["manual_or_target"]))
+        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()

tools/inventory/endpoint-surface-declarations.json

@@ -1146,6 +1146,38 @@
       "rationale": "This capability-first module intentionally exposes a headless API for other modules and integrations.",
       "repository": "govoplan-parties"
     },
+    {
+      "category": "missing_ui",
+      "method": "GET",
+      "path": "/payments/requests",
+      "rationale": "The first payment vertical slice is capability-first; its guided operator workspace will adopt the centralized layout usage contract.",
+      "repository": "govoplan-payments",
+      "tracking_issue": "https://git.add-ideas.de/GovOPlaN/govoplan-payments/issues/1"
+    },
+    {
+      "category": "missing_ui",
+      "method": "POST",
+      "path": "/payments/requests",
+      "rationale": "The first payment vertical slice is capability-first; its guided operator workspace will adopt the centralized layout usage contract.",
+      "repository": "govoplan-payments",
+      "tracking_issue": "https://git.add-ideas.de/GovOPlaN/govoplan-payments/issues/1"
+    },
+    {
+      "category": "missing_ui",
+      "method": "GET",
+      "path": "/payments/requests/{}",
+      "rationale": "The first payment vertical slice is capability-first; its guided operator workspace will adopt the centralized layout usage contract.",
+      "repository": "govoplan-payments",
+      "tracking_issue": "https://git.add-ideas.de/GovOPlaN/govoplan-payments/issues/1"
+    },
+    {
+      "category": "missing_ui",
+      "method": "POST",
+      "path": "/payments/requests/{}/manual-reconciliations",
+      "rationale": "The first payment vertical slice is capability-first; its guided operator workspace will adopt the centralized layout usage contract.",
+      "repository": "govoplan-payments",
+      "tracking_issue": "https://git.add-ideas.de/GovOPlaN/govoplan-payments/issues/1"
+    },
     {
       "category": "intentionally_headless",
       "method": "DELETE",