feat: add governed payment request slice

src/govoplan_payments/__init__.py

@@ -0,0 +1,3 @@
+"""GovOPlaN Payments module."""
+
+__version__ = "0.1.19"

src/govoplan_payments/backend/__init__.py

@@ -0,0 +1 @@
+"""Payments backend."""

src/govoplan_payments/backend/db/__init__.py

@@ -0,0 +1,7 @@
+from govoplan_payments.backend.db.models import (
+    PaymentEvent,
+    PaymentObligation,
+    PaymentReconciliation,
+)
+
+__all__ = ["PaymentEvent", "PaymentObligation", "PaymentReconciliation"]

src/govoplan_payments/backend/db/models.py

@@ -0,0 +1,166 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+import uuid
+
+from sqlalchemy import BigInteger, DateTime, ForeignKey, Index, JSON, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from govoplan_core.db.base import Base, TimestampMixin
+
+
+def new_uuid() -> str:
+    return str(uuid.uuid4())
+
+
+class PaymentObligation(Base, TimestampMixin):
+    __tablename__ = "payment_obligations"
+    __table_args__ = (
+        UniqueConstraint("tenant_id", "payment_id", name="uq_payment_obligation"),
+        UniqueConstraint(
+            "tenant_id",
+            "source_module",
+            "idempotency_key",
+            name="uq_payment_request_idempotency",
+        ),
+        UniqueConstraint(
+            "tenant_id", "payment_reference", name="uq_payment_reference"
+        ),
+        Index(
+            "ix_payment_obligation_source",
+            "tenant_id",
+            "source_module",
+            "source_resource_type",
+            "source_resource_id",
+        ),
+        Index(
+            "ix_payment_obligation_state",
+            "tenant_id",
+            "status",
+            "requested_at",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    payment_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    payment_reference: Mapped[str] = mapped_column(
+        String(32), nullable=False, index=True
+    )
+    source_module: Mapped[str] = mapped_column(String(120), nullable=False, index=True)
+    source_resource_type: Mapped[str] = mapped_column(
+        String(120), nullable=False, index=True
+    )
+    source_resource_id: Mapped[str] = mapped_column(
+        String(255), nullable=False, index=True
+    )
+    amount_minor: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    currency: Mapped[str] = mapped_column(String(3), nullable=False, index=True)
+    subject: Mapped[str] = mapped_column(Text, nullable=False)
+    status: Mapped[str] = mapped_column(String(30), nullable=False, index=True)
+    idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False)
+    request_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
+    requested_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, index=True
+    )
+    requested_by_ref: Mapped[str] = mapped_column(
+        String(255), nullable=False, index=True
+    )
+    due_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True, index=True
+    )
+    settled_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True, index=True
+    )
+    context_refs: Mapped[dict[str, str]] = mapped_column(
+        JSON, default=dict, nullable=False
+    )
+    details: Mapped[dict[str, Any]] = mapped_column(
+        "metadata", JSON, default=dict, nullable=False
+    )
+
+
+class PaymentReconciliation(Base, TimestampMixin):
+    __tablename__ = "payment_reconciliations"
+    __table_args__ = (
+        UniqueConstraint(
+            "tenant_id", "reconciliation_id", name="uq_payment_reconciliation"
+        ),
+        UniqueConstraint(
+            "tenant_id",
+            "payment_row_id",
+            "idempotency_key",
+            name="uq_payment_reconciliation_idempotency",
+        ),
+        Index(
+            "ix_payment_reconciliation_payment",
+            "tenant_id",
+            "payment_row_id",
+            "recorded_at",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    reconciliation_id: Mapped[str] = mapped_column(
+        String(36), nullable=False, index=True
+    )
+    payment_row_id: Mapped[str] = mapped_column(
+        ForeignKey("payment_obligations.id", ondelete="RESTRICT"),
+        nullable=False,
+        index=True,
+    )
+    mode: Mapped[str] = mapped_column(String(30), nullable=False, index=True)
+    amount_minor: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    currency: Mapped[str] = mapped_column(String(3), nullable=False)
+    transaction_reference: Mapped[str] = mapped_column(
+        String(255), nullable=False, index=True
+    )
+    evidence_ref: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
+    idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False)
+    request_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
+    received_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, index=True
+    )
+    recorded_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, index=True
+    )
+    recorded_by_ref: Mapped[str] = mapped_column(
+        String(255), nullable=False, index=True
+    )
+    details: Mapped[dict[str, Any]] = mapped_column(
+        "metadata", JSON, default=dict, nullable=False
+    )
+
+
+class PaymentEvent(Base, TimestampMixin):
+    __tablename__ = "payment_events"
+    __table_args__ = (
+        UniqueConstraint("tenant_id", "event_id", name="uq_payment_event"),
+        Index(
+            "ix_payment_event_stream",
+            "tenant_id",
+            "payment_row_id",
+            "occurred_at",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    event_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    payment_row_id: Mapped[str] = mapped_column(
+        ForeignKey("payment_obligations.id", ondelete="RESTRICT"),
+        nullable=False,
+        index=True,
+    )
+    event_type: Mapped[str] = mapped_column(String(120), nullable=False, index=True)
+    status: Mapped[str] = mapped_column(String(30), nullable=False, index=True)
+    occurred_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, index=True
+    )
+    actor_ref: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
+    payload: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
+
+
+__all__ = ["PaymentEvent", "PaymentObligation", "PaymentReconciliation"]

src/govoplan_payments/backend/manifest.py

@@ -0,0 +1,236 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from govoplan_core.core.module_guards import (
+    drop_table_retirement_provider,
+    persistent_table_uninstall_guard,
+)
+from govoplan_core.core.modules import (
+    CapabilityDocumentation,
+    DocumentationLink,
+    DocumentationTopic,
+    MigrationSpec,
+    ModuleContext,
+    ModuleInterfaceProvider,
+    ModuleManifest,
+    PermissionDefinition,
+    RoleTemplate,
+)
+from govoplan_core.core.payments import CAPABILITY_PAYMENT_REQUESTS
+from govoplan_core.core.provider_governance import declared_module_architecture
+from govoplan_core.db.base import Base
+from govoplan_payments.backend.db import models as payment_models
+from govoplan_payments.backend.service import SqlPaymentRequestProvider
+
+
+MODULE_ID = "payments"
+MODULE_NAME = "Payments"
+MODULE_VERSION = "0.1.19"
+READ_SCOPE = "payments:payment:read"
+WRITE_SCOPE = "payments:payment:write"
+RECONCILE_SCOPE = "payments:payment:reconcile"
+ADMIN_SCOPE = "payments:payment:admin"
+
+
+def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
+    module_id, resource, action = scope.split(":", 2)
+    return PermissionDefinition(
+        scope=scope,
+        label=label,
+        description=description,
+        category=MODULE_NAME,
+        level="tenant",
+        module_id=module_id,
+        resource=resource,
+        action=action,
+    )
+
+
+def _router(_context: ModuleContext):
+    from govoplan_payments.backend.router import router
+
+    return router
+
+
+def _payment_requests(_context: ModuleContext) -> SqlPaymentRequestProvider:
+    return SqlPaymentRequestProvider()
+
+
+def _tenant_summary(session: object, tenant_id: str) -> dict[str, int]:
+    if not hasattr(session, "query"):
+        return {"payment_requests": 0, "paid_payments": 0, "reconciliations": 0}
+    obligations = session.query(payment_models.PaymentObligation).filter(
+        payment_models.PaymentObligation.tenant_id == tenant_id
+    )
+    return {
+        "payment_requests": obligations.count(),
+        "paid_payments": obligations.filter(
+            payment_models.PaymentObligation.status == "paid"
+        ).count(),
+        "reconciliations": session.query(payment_models.PaymentReconciliation)
+        .filter(payment_models.PaymentReconciliation.tenant_id == tenant_id)
+        .count(),
+    }
+
+
+manifest = ModuleManifest(
+    id=MODULE_ID,
+    name=MODULE_NAME,
+    version=MODULE_VERSION,
+    optional_dependencies=(
+        "files",
+        "audit",
+        "cases",
+        "workflow_engine",
+        "ledger",
+        "xrechnung",
+    ),
+    permissions=(
+        _permission(
+            READ_SCOPE,
+            "View payment requests",
+            "View tenant payment obligations, state, and reconciliation evidence.",
+        ),
+        _permission(
+            WRITE_SCOPE,
+            "Create payment requests",
+            "Create replay-safe payment obligations for an owning procedure.",
+        ),
+        _permission(
+            RECONCILE_SCOPE,
+            "Reconcile manual payments",
+            "Confirm an exact payment with immutable external evidence.",
+        ),
+        _permission(
+            ADMIN_SCOPE,
+            "Administer Payments",
+            "Administer payment access, retention, recovery, and future providers.",
+        ),
+    ),
+    role_templates=(
+        RoleTemplate(
+            slug="payment_operator",
+            name="Payment operator",
+            description="Create payment obligations and reconcile evidenced receipts.",
+            permissions=(READ_SCOPE, WRITE_SCOPE, RECONCILE_SCOPE),
+        ),
+        RoleTemplate(
+            slug="payment_auditor",
+            name="Payment auditor",
+            description="Inspect payment state and reconciliation evidence.",
+            permissions=(READ_SCOPE,),
+        ),
+    ),
+    provides_interfaces=(
+        ModuleInterfaceProvider(name=CAPABILITY_PAYMENT_REQUESTS, version="1.0.0"),
+    ),
+    capability_factories={CAPABILITY_PAYMENT_REQUESTS: _payment_requests},
+    capability_documentation={
+        CAPABILITY_PAYMENT_REQUESTS: CapabilityDocumentation(
+            label="Payment request and reconciliation",
+            summary="Creates replay-safe obligations and records exact, evidence-bound manual settlement.",
+            contract_version="1.0.0",
+        ),
+    },
+    route_factory=_router,
+    migration_spec=MigrationSpec(
+        module_id=MODULE_ID,
+        metadata=Base.metadata,
+        script_location=str(Path(__file__).with_name("migrations") / "versions"),
+        retirement_supported=True,
+        retirement_provider=drop_table_retirement_provider(
+            payment_models.PaymentEvent,
+            payment_models.PaymentReconciliation,
+            payment_models.PaymentObligation,
+            label=MODULE_NAME,
+        ),
+        retirement_notes=(
+            "Destructive retirement removes payment obligations and reconciliation evidence "
+            "and requires a verified database snapshot plus an accounting/records decision."
+        ),
+    ),
+    uninstall_guard_providers=(
+        persistent_table_uninstall_guard(
+            payment_models.PaymentEvent,
+            payment_models.PaymentReconciliation,
+            payment_models.PaymentObligation,
+            label=MODULE_NAME,
+        ),
+    ),
+    tenant_summary_providers=(_tenant_summary,),
+    documentation=(
+        DocumentationTopic(
+            id="payments.requests-and-reconciliation",
+            title="Payment requests and manual reconciliation",
+            summary="Create an exact obligation and mark it paid only with matching, immutable evidence.",
+            body=(
+                "Payments owns the tenant-bound payment ID, human payment reference, requested amount and currency, lifecycle events, and reconciliation evidence. "
+                "A Case, Workflow, or other procedure calls the payments.requests capability with its own source reference and a replay key; it keeps the returned payment ID instead of writing Payments tables. "
+                "The first supported receipt path is manual reconciliation of a full payment. The operator must record the exact amount and currency, external transaction reference, received time, and a same-tenant EvidenceReference carrying a version or checksum. A mismatch, duplicate settlement under another key, cross-tenant evidence, partial amount, or timezone-free timestamp fails closed. "
+                "Successful requests and reconciliations append payment events and API actions add audit evidence when Audit is installed. There is no silent correction: reversal, refund, partial payment, online checkout, provider callbacks, Ledger posting, and XRechnung remain explicit future flows."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("operator", "module_admin", "auditor", "product_owner"),
+            links=(
+                DocumentationLink(
+                    label="Payments boundary and recovery",
+                    href="govoplan-payments/docs/PAYMENTS_DOMAIN.md",
+                    kind="repository",
+                ),
+            ),
+            metadata={
+                "help_contexts": [
+                    "payments.request",
+                    "payments.reconciliation.manual",
+                    "payments.state.requested",
+                    "payments.state.paid",
+                ],
+                "privacy_notes": [
+                    "Procedure context uses stable references; applicant names, bank account details, and submitted form values are not required.",
+                    "The immutable evidence remains owned by its provider; Payments stores only the typed EvidenceReference.",
+                ],
+                "consequence_classes": {
+                    "request_payment": "Creates a durable amount/currency obligation and a stable applicant payment reference.",
+                    "reconcile_manual": "Marks the exact obligation paid and appends evidence; a future governed adjustment is required to reverse it.",
+                },
+            },
+        ),
+    ),
+    architecture=declared_module_architecture(
+        layer="domain_capability",
+        kind="domain",
+        maturity="vertical_slice",
+        documentation_ref="docs/PAYMENTS_DOMAIN.md",
+        test_ref="tests/test_payments.py",
+        known_limits=(
+            "Only full manual payment reconciliation is implemented; partial payments, refunds, reversals, and corrections need explicit governed flows.",
+            "No online payment provider, callback, ledger posting, XRechnung, applicant payment page, or dedicated operator WebUI is included yet.",
+        ),
+        supported_authority_modes=("native_authoritative",),
+        owned_concepts=(
+            "payment obligation",
+            "payment reference",
+            "payment reconciliation",
+            "payment lifecycle evidence",
+        ),
+        non_owned_concepts=(
+            "case",
+            "workflow",
+            "invoice",
+            "accounting entry",
+            "evidence binary",
+            "external payment execution",
+        ),
+        reference_packages=("product.service-to-decision",),
+        migration_docs=("docs/PAYMENTS_DOMAIN.md",),
+        recovery_docs=("docs/PAYMENTS_DOMAIN.md",),
+        security_docs=("docs/PAYMENTS_DOMAIN.md",),
+        operations_docs=("docs/PAYMENTS_DOMAIN.md",),
+    ),
+)
+
+
+def get_manifest() -> ModuleManifest:
+    return manifest

src/govoplan_payments/backend/migrations/__init__.py

@@ -0,0 +1 @@
+"""Payments migrations."""

src/govoplan_payments/backend/migrations/versions/e7b9c1d3f5a7_payments_baseline.py

@@ -0,0 +1,194 @@
+"""Add replay-safe payment obligations and manual reconciliation evidence.
+
+Revision ID: e7b9c1d3f5a7
+Revises: None
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "e7b9c1d3f5a7"
+down_revision = None
+branch_labels = None
+depends_on = "4f2a9c8e7b6d"
+
+
+def upgrade() -> None:
+    op.create_table(
+        "payment_obligations",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("payment_id", sa.String(length=36), nullable=False),
+        sa.Column("payment_reference", sa.String(length=32), nullable=False),
+        sa.Column("source_module", sa.String(length=120), nullable=False),
+        sa.Column("source_resource_type", sa.String(length=120), nullable=False),
+        sa.Column("source_resource_id", sa.String(length=255), nullable=False),
+        sa.Column("amount_minor", sa.BigInteger(), nullable=False),
+        sa.Column("currency", sa.String(length=3), nullable=False),
+        sa.Column("subject", sa.Text(), nullable=False),
+        sa.Column("status", sa.String(length=30), nullable=False),
+        sa.Column("idempotency_key", sa.String(length=255), nullable=False),
+        sa.Column("request_sha256", sa.String(length=64), nullable=False),
+        sa.Column("requested_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("requested_by_ref", sa.String(length=255), nullable=False),
+        sa.Column("due_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("settled_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("context_refs", sa.JSON(), nullable=False),
+        sa.Column("metadata", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_payment_obligations")),
+        sa.UniqueConstraint("tenant_id", "payment_id", name="uq_payment_obligation"),
+        sa.UniqueConstraint(
+            "tenant_id",
+            "source_module",
+            "idempotency_key",
+            name="uq_payment_request_idempotency",
+        ),
+        sa.UniqueConstraint(
+            "tenant_id", "payment_reference", name="uq_payment_reference"
+        ),
+    )
+    for column in (
+        "tenant_id",
+        "payment_id",
+        "payment_reference",
+        "source_module",
+        "source_resource_type",
+        "source_resource_id",
+        "currency",
+        "status",
+        "requested_at",
+        "requested_by_ref",
+        "due_at",
+        "settled_at",
+    ):
+        op.create_index(
+            op.f(f"ix_payment_obligations_{column}"),
+            "payment_obligations",
+            [column],
+            unique=False,
+        )
+    op.create_index(
+        "ix_payment_obligation_source",
+        "payment_obligations",
+        ["tenant_id", "source_module", "source_resource_type", "source_resource_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_payment_obligation_state",
+        "payment_obligations",
+        ["tenant_id", "status", "requested_at"],
+        unique=False,
+    )
+
+    op.create_table(
+        "payment_reconciliations",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("reconciliation_id", sa.String(length=36), nullable=False),
+        sa.Column("payment_row_id", sa.String(length=36), nullable=False),
+        sa.Column("mode", sa.String(length=30), nullable=False),
+        sa.Column("amount_minor", sa.BigInteger(), nullable=False),
+        sa.Column("currency", sa.String(length=3), nullable=False),
+        sa.Column("transaction_reference", sa.String(length=255), nullable=False),
+        sa.Column("evidence_ref", sa.JSON(), nullable=False),
+        sa.Column("idempotency_key", sa.String(length=255), nullable=False),
+        sa.Column("request_sha256", sa.String(length=64), nullable=False),
+        sa.Column("received_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("recorded_by_ref", sa.String(length=255), nullable=False),
+        sa.Column("metadata", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["payment_row_id"],
+            ["payment_obligations.id"],
+            name=op.f("fk_payment_reconciliations_payment_row_id_payment_obligations"),
+            ondelete="RESTRICT",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_payment_reconciliations")),
+        sa.UniqueConstraint(
+            "tenant_id", "reconciliation_id", name="uq_payment_reconciliation"
+        ),
+        sa.UniqueConstraint(
+            "tenant_id",
+            "payment_row_id",
+            "idempotency_key",
+            name="uq_payment_reconciliation_idempotency",
+        ),
+    )
+    for column in (
+        "tenant_id",
+        "reconciliation_id",
+        "payment_row_id",
+        "mode",
+        "transaction_reference",
+        "received_at",
+        "recorded_at",
+        "recorded_by_ref",
+    ):
+        op.create_index(
+            op.f(f"ix_payment_reconciliations_{column}"),
+            "payment_reconciliations",
+            [column],
+            unique=False,
+        )
+    op.create_index(
+        "ix_payment_reconciliation_payment",
+        "payment_reconciliations",
+        ["tenant_id", "payment_row_id", "recorded_at"],
+        unique=False,
+    )
+
+    op.create_table(
+        "payment_events",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("event_id", sa.String(length=36), nullable=False),
+        sa.Column("payment_row_id", sa.String(length=36), nullable=False),
+        sa.Column("event_type", sa.String(length=120), nullable=False),
+        sa.Column("status", sa.String(length=30), nullable=False),
+        sa.Column("occurred_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("actor_ref", sa.String(length=255), nullable=False),
+        sa.Column("payload", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["payment_row_id"],
+            ["payment_obligations.id"],
+            name=op.f("fk_payment_events_payment_row_id_payment_obligations"),
+            ondelete="RESTRICT",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_payment_events")),
+        sa.UniqueConstraint("tenant_id", "event_id", name="uq_payment_event"),
+    )
+    for column in (
+        "tenant_id",
+        "event_id",
+        "payment_row_id",
+        "event_type",
+        "status",
+        "occurred_at",
+        "actor_ref",
+    ):
+        op.create_index(
+            op.f(f"ix_payment_events_{column}"),
+            "payment_events",
+            [column],
+            unique=False,
+        )
+    op.create_index(
+        "ix_payment_event_stream",
+        "payment_events",
+        ["tenant_id", "payment_row_id", "occurred_at"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("payment_events")
+    op.drop_table("payment_reconciliations")
+    op.drop_table("payment_obligations")

src/govoplan_payments/backend/router.py

@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.orm import Session
+
+from govoplan_core.audit.logging import audit_event
+from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
+from govoplan_core.core.institutional import EvidenceReference, InstitutionalContextError
+from govoplan_core.core.payments import (
+    ManualPaymentReconciliationCommand,
+    PaymentRequestCommand,
+)
+from govoplan_core.db.session import get_session
+from govoplan_payments.backend.manifest import (
+    READ_SCOPE,
+    RECONCILE_SCOPE,
+    WRITE_SCOPE,
+)
+from govoplan_payments.backend.schemas import (
+    ManualPaymentReconciliationCreate,
+    PaymentListResponse,
+    PaymentRequestCreate,
+)
+from govoplan_payments.backend.service import (
+    PaymentConflict,
+    PaymentError,
+    SqlPaymentRequestProvider,
+)
+
+
+router = APIRouter(prefix="/payments", tags=["payments"])
+provider = SqlPaymentRequestProvider()
+
+
+@router.get("/requests", response_model=PaymentListResponse)
+def api_list_payment_requests(
+    payment_status: str | None = Query(default=None, alias="status"),
+    source_resource_id: str | None = Query(default=None, max_length=255),
+    limit: int = Query(default=100, ge=1, le=200),
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> PaymentListResponse:
+    _require(principal, READ_SCOPE)
+    try:
+        items = provider.list_payments(
+            session,
+            tenant_id=principal.tenant_id,
+            status=payment_status,
+            source_resource_id=source_resource_id,
+            limit=limit,
+        )
+    except PaymentError as exc:
+        raise _error(exc) from exc
+    return PaymentListResponse(payments=[dict(item) for item in items])
+
+
+@router.get("/requests/{payment_id}", response_model=dict[str, Any])
+def api_get_payment_request(
+    payment_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> dict[str, object]:
+    _require(principal, READ_SCOPE)
+    item = provider.get_payment(
+        session,
+        tenant_id=principal.tenant_id,
+        payment_id=payment_id,
+    )
+    if item is None:
+        raise HTTPException(status_code=404, detail="Payment request not found")
+    return dict(item)
+
+
+@router.post(
+    "/requests",
+    response_model=dict[str, Any],
+    status_code=status.HTTP_201_CREATED,
+)
+def api_create_payment_request(
+    payload: PaymentRequestCreate,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> dict[str, object]:
+    _require(principal, WRITE_SCOPE)
+    try:
+        item = provider.request_payment(
+            session,
+            PaymentRequestCommand(
+                tenant_id=principal.tenant_id,
+                source_module=payload.source_module,
+                source_resource_type=payload.source_resource_type,
+                source_resource_id=payload.source_resource_id,
+                amount_minor=payload.amount_minor,
+                currency=payload.currency,
+                subject=payload.subject,
+                idempotency_key=payload.idempotency_key,
+                requested_at=datetime.now(UTC),
+                requested_by_ref=_actor_ref(principal),
+                due_at=payload.due_at,
+                context_refs=payload.context_refs,
+                metadata=payload.metadata,
+            ),
+        )
+        _audit(
+            session,
+            principal,
+            action="payments.requested",
+            payment=item,
+        )
+        session.commit()
+    except (PaymentError, InstitutionalContextError) as exc:
+        session.rollback()
+        raise _error(exc) from exc
+    return dict(item)
+
+
+@router.post(
+    "/requests/{payment_id}/manual-reconciliations",
+    response_model=dict[str, Any],
+)
+def api_reconcile_manual_payment(
+    payment_id: str,
+    payload: ManualPaymentReconciliationCreate,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> dict[str, object]:
+    _require(principal, RECONCILE_SCOPE)
+    try:
+        item = provider.reconcile_manual_payment(
+            session,
+            ManualPaymentReconciliationCommand(
+                tenant_id=principal.tenant_id,
+                payment_id=payment_id,
+                amount_minor=payload.amount_minor,
+                currency=payload.currency,
+                transaction_reference=payload.transaction_reference,
+                evidence_ref=EvidenceReference.from_mapping(payload.evidence_ref),
+                idempotency_key=payload.idempotency_key,
+                received_at=payload.received_at,
+                recorded_at=datetime.now(UTC),
+                recorded_by_ref=_actor_ref(principal),
+                metadata=payload.metadata,
+            ),
+        )
+        _audit(
+            session,
+            principal,
+            action="payments.reconciled.manual",
+            payment=item,
+        )
+        session.commit()
+    except (PaymentError, InstitutionalContextError) as exc:
+        session.rollback()
+        raise _error(exc) from exc
+    return dict(item)
+
+
+def _require(principal: ApiPrincipal, scope: str) -> None:
+    if not has_scope(principal, scope):
+        raise HTTPException(status_code=403, detail=f"Missing scope: {scope}")
+
+
+def _actor_ref(principal: ApiPrincipal) -> str:
+    if principal.api_key_id:
+        return f"api_key:{principal.api_key_id}"
+    account_id = str(getattr(principal, "account_id", "") or "").strip()
+    if account_id:
+        return f"account:{account_id}"
+    user_id = str(getattr(getattr(principal, "user", None), "id", "") or "").strip()
+    if user_id:
+        return f"user:{user_id}"
+    raise PaymentError("Payment action requires an acting identity.")
+
+
+def _audit(
+    session: Session,
+    principal: ApiPrincipal,
+    *,
+    action: str,
+    payment: dict[str, object] | Any,
+) -> None:
+    item = dict(payment)
+    source = item.get("source") if isinstance(item.get("source"), dict) else {}
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(getattr(principal, "user", None), "id", None),
+        api_key_id=principal.api_key_id,
+        action=action,
+        object_type="payment",
+        object_id=str(item.get("payment_id") or ""),
+        details={
+            "payment_reference": item.get("payment_reference"),
+            "status": item.get("status"),
+            "amount_minor": item.get("amount_minor"),
+            "currency": item.get("currency"),
+            "source_module": source.get("module"),
+            "source_resource_type": source.get("resource_type"),
+            "source_resource_id": source.get("resource_id"),
+            "replayed": item.get("replayed"),
+        },
+    )
+
+
+def _error(exc: Exception) -> HTTPException:
+    return HTTPException(
+        status_code=409 if isinstance(exc, PaymentConflict) else 400,
+        detail=str(exc),
+    )
+
+
+__all__ = ["router"]

src/govoplan_payments/backend/schemas.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class PaymentRequestCreate(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    source_module: str = Field(min_length=1, max_length=120)
+    source_resource_type: str = Field(min_length=1, max_length=120)
+    source_resource_id: str = Field(min_length=1, max_length=255)
+    amount_minor: int = Field(ge=1, le=9_000_000_000_000)
+    currency: str = Field(min_length=3, max_length=3)
+    subject: str = Field(min_length=1, max_length=1000)
+    idempotency_key: str = Field(min_length=1, max_length=255)
+    due_at: datetime | None = None
+    context_refs: dict[str, str] = Field(default_factory=dict)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class ManualPaymentReconciliationCreate(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    amount_minor: int = Field(ge=1, le=9_000_000_000_000)
+    currency: str = Field(min_length=3, max_length=3)
+    transaction_reference: str = Field(min_length=1, max_length=255)
+    evidence_ref: dict[str, Any]
+    idempotency_key: str = Field(min_length=1, max_length=255)
+    received_at: datetime
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class PaymentListResponse(BaseModel):
+    payments: list[dict[str, Any]]
+
+
+__all__ = [
+    "ManualPaymentReconciliationCreate",
+    "PaymentListResponse",
+    "PaymentRequestCreate",
+]

src/govoplan_payments/backend/service.py

@@ -0,0 +1,434 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from datetime import UTC, datetime
+import hashlib
+import json
+import re
+import uuid
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.payments import (
+    ManualPaymentReconciliationCommand,
+    PaymentRequestCommand,
+)
+from govoplan_payments.backend.db.models import (
+    PaymentEvent,
+    PaymentObligation,
+    PaymentReconciliation,
+)
+
+
+PAYMENT_STATES = frozenset({"requested", "paid"})
+_CURRENCY_RE = re.compile(r"^[A-Z]{3}$")
+_MODULE_RE = re.compile(r"^[a-z][a-z0-9_]*$")
+
+
+class PaymentError(ValueError):
+    pass
+
+
+class PaymentConflict(PaymentError):
+    pass
+
+
+class SqlPaymentRequestProvider:
+    def request_payment(
+        self,
+        session: Session,
+        command: PaymentRequestCommand,
+    ) -> Mapping[str, object]:
+        normalized = _normalized_request(command)
+        digest = _digest(normalized)
+        tenant_id = str(normalized["tenant_id"])
+        source_module = str(normalized["source_module"])
+        idempotency_key = str(normalized["idempotency_key"])
+        existing = (
+            session.query(PaymentObligation)
+            .filter(
+                PaymentObligation.tenant_id == tenant_id,
+                PaymentObligation.source_module == source_module,
+                PaymentObligation.idempotency_key == idempotency_key,
+            )
+            .one_or_none()
+        )
+        if existing is not None:
+            if existing.request_sha256 != digest:
+                raise PaymentConflict(
+                    "Payment request idempotency conflict: the key was already used for a different request."
+                )
+            return payment_payload(session, existing, replayed=True)
+
+        payment_id = str(uuid.uuid4())
+        item = PaymentObligation(
+            tenant_id=tenant_id,
+            payment_id=payment_id,
+            payment_reference=f"PAY-{payment_id.replace('-', '')[:12].upper()}",
+            source_module=source_module,
+            source_resource_type=str(normalized["source_resource_type"]),
+            source_resource_id=str(normalized["source_resource_id"]),
+            amount_minor=int(normalized["amount_minor"]),
+            currency=str(normalized["currency"]),
+            subject=str(normalized["subject"]),
+            status="requested",
+            idempotency_key=idempotency_key,
+            request_sha256=digest,
+            requested_at=command.requested_at,
+            requested_by_ref=str(normalized["requested_by_ref"]),
+            due_at=command.due_at,
+            context_refs=dict(normalized["context_refs"]),
+            details=dict(normalized["metadata"]),
+        )
+        session.add(item)
+        session.flush()
+        session.add(
+            PaymentEvent(
+                tenant_id=item.tenant_id,
+                event_id=str(uuid.uuid4()),
+                payment_row_id=item.id,
+                event_type="payments.requested",
+                status=item.status,
+                occurred_at=command.requested_at,
+                actor_ref=str(normalized["requested_by_ref"]),
+                payload={
+                    "source_module": item.source_module,
+                    "source_resource_type": item.source_resource_type,
+                    "source_resource_id": item.source_resource_id,
+                    "amount_minor": item.amount_minor,
+                    "currency": item.currency,
+                },
+            )
+        )
+        session.flush()
+        return payment_payload(session, item)
+
+    def get_payment(
+        self,
+        session: Session,
+        *,
+        tenant_id: str,
+        payment_id: str,
+    ) -> Mapping[str, object] | None:
+        item = (
+            session.query(PaymentObligation)
+            .filter(
+                PaymentObligation.tenant_id == _text(tenant_id, "Tenant", 36),
+                PaymentObligation.payment_id
+                == _text(payment_id, "Payment ID", 36),
+            )
+            .one_or_none()
+        )
+        return payment_payload(session, item) if item is not None else None
+
+    def list_payments(
+        self,
+        session: Session,
+        *,
+        tenant_id: str,
+        status: str | None = None,
+        source_resource_id: str | None = None,
+        limit: int = 100,
+    ) -> tuple[Mapping[str, object], ...]:
+        query = session.query(PaymentObligation).filter(
+            PaymentObligation.tenant_id == _text(tenant_id, "Tenant", 36)
+        )
+        if status is not None:
+            if status not in PAYMENT_STATES:
+                raise PaymentError(f"Unsupported payment status: {status!r}.")
+            query = query.filter(PaymentObligation.status == status)
+        if source_resource_id:
+            query = query.filter(
+                PaymentObligation.source_resource_id == source_resource_id
+            )
+        return tuple(
+            payment_payload(session, item)
+            for item in query.order_by(PaymentObligation.requested_at.desc()).limit(
+                max(1, min(int(limit), 200))
+            )
+        )
+
+    def reconcile_manual_payment(
+        self,
+        session: Session,
+        command: ManualPaymentReconciliationCommand,
+    ) -> Mapping[str, object]:
+        normalized = _normalized_reconciliation(command)
+        digest = _digest(normalized)
+        tenant_id = str(normalized["tenant_id"])
+        payment_id = str(normalized["payment_id"])
+        idempotency_key = str(normalized["idempotency_key"])
+        item = (
+            session.query(PaymentObligation)
+            .filter(
+                PaymentObligation.tenant_id == tenant_id,
+                PaymentObligation.payment_id == payment_id,
+            )
+            .with_for_update()
+            .one_or_none()
+        )
+        if item is None:
+            raise PaymentError("Payment request is unavailable.")
+        existing = (
+            session.query(PaymentReconciliation)
+            .filter(
+                PaymentReconciliation.tenant_id == tenant_id,
+                PaymentReconciliation.payment_row_id == item.id,
+                PaymentReconciliation.idempotency_key == idempotency_key,
+            )
+            .one_or_none()
+        )
+        if existing is not None:
+            if existing.request_sha256 != digest:
+                raise PaymentConflict(
+                    "Payment reconciliation idempotency conflict: the key was already used for different evidence."
+                )
+            return payment_payload(session, item, replayed=True)
+        if item.status == "paid":
+            raise PaymentConflict(
+                "Payment is already reconciled; correct or reverse it through a future governed adjustment flow."
+            )
+        if command.amount_minor != item.amount_minor or command.currency.upper() != item.currency:
+            raise PaymentConflict(
+                "Manual reconciliation must match the requested amount and currency exactly."
+            )
+
+        reconciliation = PaymentReconciliation(
+            tenant_id=tenant_id,
+            reconciliation_id=str(uuid.uuid4()),
+            payment_row_id=item.id,
+            mode="manual",
+            amount_minor=int(normalized["amount_minor"]),
+            currency=str(normalized["currency"]),
+            transaction_reference=str(normalized["transaction_reference"]),
+            evidence_ref=command.evidence_ref.to_dict(),
+            idempotency_key=idempotency_key,
+            request_sha256=digest,
+            received_at=command.received_at,
+            recorded_at=command.recorded_at,
+            recorded_by_ref=str(normalized["recorded_by_ref"]),
+            details=dict(normalized["metadata"]),
+        )
+        session.add(reconciliation)
+        item.status = "paid"
+        item.settled_at = command.received_at
+        session.add(item)
+        session.flush()
+        session.add(
+            PaymentEvent(
+                tenant_id=item.tenant_id,
+                event_id=str(uuid.uuid4()),
+                payment_row_id=item.id,
+                event_type="payments.reconciled.manual",
+                status=item.status,
+                occurred_at=command.recorded_at,
+                actor_ref=command.recorded_by_ref,
+                payload={
+                    "reconciliation_id": reconciliation.reconciliation_id,
+                    "received_at": command.received_at.isoformat(),
+                    "evidence_owner_module": command.evidence_ref.owner_module,
+                    "evidence_id": command.evidence_ref.evidence_id,
+                },
+            )
+        )
+        session.flush()
+        return payment_payload(session, item)
+
+
+def payment_payload(
+    session: Session,
+    item: PaymentObligation,
+    *,
+    replayed: bool = False,
+) -> dict[str, object]:
+    reconciliation = (
+        session.query(PaymentReconciliation)
+        .filter(PaymentReconciliation.payment_row_id == item.id)
+        .order_by(PaymentReconciliation.recorded_at.desc())
+        .first()
+    )
+    events = (
+        session.query(PaymentEvent)
+        .filter(PaymentEvent.payment_row_id == item.id)
+        .order_by(PaymentEvent.occurred_at.asc())
+        .all()
+    )
+    return {
+        "payment_id": item.payment_id,
+        "tenant_id": item.tenant_id,
+        "payment_reference": item.payment_reference,
+        "source": {
+            "module": item.source_module,
+            "resource_type": item.source_resource_type,
+            "resource_id": item.source_resource_id,
+        },
+        "amount_minor": item.amount_minor,
+        "currency": item.currency,
+        "subject": item.subject,
+        "status": item.status,
+        "requested_at": _aware(item.requested_at).isoformat(),
+        "requested_by_ref": item.requested_by_ref,
+        "due_at": _iso(item.due_at),
+        "settled_at": _iso(item.settled_at),
+        "context_refs": dict(item.context_refs or {}),
+        "metadata": dict(item.details or {}),
+        "reconciliation": (
+            {
+                "reconciliation_id": reconciliation.reconciliation_id,
+                "mode": reconciliation.mode,
+                "amount_minor": reconciliation.amount_minor,
+                "currency": reconciliation.currency,
+                "transaction_reference": reconciliation.transaction_reference,
+                "evidence_ref": dict(reconciliation.evidence_ref),
+                "received_at": _aware(reconciliation.received_at).isoformat(),
+                "recorded_at": _aware(reconciliation.recorded_at).isoformat(),
+                "recorded_by_ref": reconciliation.recorded_by_ref,
+            }
+            if reconciliation is not None
+            else None
+        ),
+        "events": [
+            {
+                "event_id": event.event_id,
+                "event_type": event.event_type,
+                "status": event.status,
+                "occurred_at": _aware(event.occurred_at).isoformat(),
+                "actor_ref": event.actor_ref,
+                "payload": dict(event.payload or {}),
+            }
+            for event in events
+        ],
+        "replayed": replayed,
+    }
+
+
+def _normalized_request(command: PaymentRequestCommand) -> dict[str, object]:
+    tenant_id = _text(command.tenant_id, "Tenant", 36)
+    source_module = _text(command.source_module, "Source module", 120)
+    if not _MODULE_RE.fullmatch(source_module):
+        raise PaymentError("Payment source module must be a valid module ID.")
+    source_resource_type = _text(
+        command.source_resource_type, "Source resource type", 120
+    )
+    source_resource_id = _text(command.source_resource_id, "Source resource ID", 255)
+    _amount(command.amount_minor)
+    _currency(command.currency)
+    subject = _text(command.subject, "Payment subject", 1000)
+    idempotency_key = _text(command.idempotency_key, "Idempotency key", 255)
+    requested_by_ref = _text(command.requested_by_ref, "Request actor", 255)
+    _aware_required(command.requested_at, "Requested time")
+    if command.due_at is not None:
+        _aware_required(command.due_at, "Due time")
+        if command.due_at < command.requested_at:
+            raise PaymentError("Payment due time cannot precede the request.")
+    context_refs = _context_refs(command.context_refs)
+    return {
+        "tenant_id": tenant_id,
+        "source_module": source_module,
+        "source_resource_type": source_resource_type,
+        "source_resource_id": source_resource_id,
+        "amount_minor": command.amount_minor,
+        "currency": command.currency.upper(),
+        "subject": subject,
+        "idempotency_key": idempotency_key,
+        "requested_at": command.requested_at.isoformat(),
+        "requested_by_ref": requested_by_ref,
+        "due_at": command.due_at.isoformat() if command.due_at else None,
+        "context_refs": context_refs,
+        "metadata": dict(command.metadata),
+    }
+
+
+def _normalized_reconciliation(
+    command: ManualPaymentReconciliationCommand,
+) -> dict[str, object]:
+    tenant_id = _text(command.tenant_id, "Tenant", 36)
+    payment_id = _text(command.payment_id, "Payment ID", 36)
+    _amount(command.amount_minor)
+    _currency(command.currency)
+    transaction_reference = _text(
+        command.transaction_reference, "Transaction reference", 255
+    )
+    idempotency_key = _text(command.idempotency_key, "Idempotency key", 255)
+    recorded_by_ref = _text(command.recorded_by_ref, "Recording actor", 255)
+    _aware_required(command.received_at, "Payment received time")
+    _aware_required(command.recorded_at, "Reconciliation recorded time")
+    evidence = command.evidence_ref
+    if evidence.tenant_id != tenant_id:
+        raise PaymentError("Payment evidence belongs to another tenant.")
+    if not evidence.version and not evidence.checksum:
+        raise PaymentError(
+            "Manual reconciliation requires versioned or checksum-bound evidence."
+        )
+    return {
+        "tenant_id": tenant_id,
+        "payment_id": payment_id,
+        "amount_minor": command.amount_minor,
+        "currency": command.currency.upper(),
+        "transaction_reference": transaction_reference,
+        "idempotency_key": idempotency_key,
+        "evidence_ref": evidence.to_dict(),
+        "received_at": command.received_at.isoformat(),
+        "recorded_at": command.recorded_at.isoformat(),
+        "recorded_by_ref": recorded_by_ref,
+        "metadata": dict(command.metadata),
+    }
+
+
+def _digest(payload: Mapping[str, object]) -> str:
+    serialized = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
+    return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
+
+
+def _context_refs(value: Mapping[str, str]) -> dict[str, str]:
+    if len(value) > 20:
+        raise PaymentError("Payment context accepts at most 20 references.")
+    return {
+        _text(key, "Context reference key", 120): _text(
+            reference, "Context reference", 255
+        )
+        for key, reference in value.items()
+    }
+
+
+def _text(value: object, label: str, max_length: int) -> str:
+    candidate = str(value or "").strip()
+    if not candidate or len(candidate) > max_length:
+        raise PaymentError(f"{label} must contain 1 to {max_length} characters.")
+    return candidate
+
+
+def _amount(value: int) -> None:
+    if (
+        isinstance(value, bool)
+        or not isinstance(value, int)
+        or not 1 <= value <= 9_000_000_000_000
+    ):
+        raise PaymentError("Payment amount must be a positive integer in minor units.")
+
+
+def _currency(value: str) -> None:
+    if not _CURRENCY_RE.fullmatch(str(value or "").strip().upper()):
+        raise PaymentError("Payment currency must be a three-letter ISO code.")
+
+
+def _aware_required(value: datetime, label: str) -> None:
+    if value.tzinfo is None:
+        raise PaymentError(f"{label} must include a timezone.")
+
+
+def _aware(value: datetime) -> datetime:
+    return value if value.tzinfo is not None else value.replace(tzinfo=UTC)
+
+
+def _iso(value: datetime | None) -> str | None:
+    return _aware(value).isoformat() if value is not None else None
+
+
+__all__ = [
+    "PAYMENT_STATES",
+    "PaymentConflict",
+    "PaymentError",
+    "SqlPaymentRequestProvider",
+    "payment_payload",
+]

src/govoplan_payments/py.typed

@@ -0,0 +1 @@
+