2 Commits
Author SHA1 Message Date
zemion 51dc33d8ae docs(erp): complete German reference contract
Module Package Release / publish-packages (push) Successful in 10s
2026-08-23 19:31:23 +02:00
zemion a12e096366 feat(erp): plan payable exports and reconcile bookings
Module Package Release / publish-packages (push) Successful in 13s
2026-08-23 17:53:23 +02:00
11 changed files with 1298 additions and 0 deletions
+270
View File
@@ -0,0 +1,270 @@
name: Module Package Release
on:
push:
tags:
- "v*"
workflow_dispatch:
inputs:
release_tag:
description: Existing protected version tag to publish
required: true
type: string
jobs:
publish-packages:
runs-on: ubuntu-latest
env:
GITEA_REPOSITORY: ${{ gitea.repository }}
steps:
- uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5
with:
fetch-depth: 0
- uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065
with:
python-version: "3.12"
- uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020
with:
node-version: "22"
- name: Select and validate protected release tag
shell: bash
env:
REQUESTED_TAG: ${{ inputs.release_tag }}
TRIGGER_TAG: ${{ gitea.ref_name }}
run: |
set -euo pipefail
tag="${REQUESTED_TAG:-$TRIGGER_TAG}"
case "$tag" in
v[0-9]*.[0-9]*.[0-9]*) ;;
*) echo "Release tag must start with a SemVer-shaped vX.Y.Z value" >&2; exit 1 ;;
esac
git fetch --force origin "refs/tags/$tag:refs/tags/$tag" refs/heads/main:refs/remotes/origin/main
tag_commit="$(git rev-list -n 1 "$tag")"
git merge-base --is-ancestor "$tag_commit" refs/remotes/origin/main || {
echo "Release tag is not contained in main" >&2
exit 1
}
git checkout --detach "$tag"
printf 'RELEASE_TAG=%s\n' "$tag" >> "$GITEA_ENV"
printf 'SOURCE_DATE_EPOCH=%s\n' "$(git show -s --format=%ct HEAD)" >> "$GITEA_ENV"
- name: Validate package versions
run: |
python - <<'PY'
import json
from pathlib import Path
import os
import re
import tomllib
tag = os.environ["RELEASE_TAG"]
expected = tag.removeprefix("v")
project = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))["project"]
if project.get("version") != expected:
raise SystemExit(f"pyproject version {project.get('version')!r} does not match {tag}")
if re.fullmatch(r"govoplan-[a-z0-9-]+", str(project.get("name", ""))) is None:
raise SystemExit("Python distribution name must use the govoplan-* namespace")
webui = Path("webui/package.json")
if webui.is_file():
package = json.loads(webui.read_text(encoding="utf-8"))
if package.get("version") != expected:
raise SystemExit(f"WebUI version {package.get('version')!r} does not match {tag}")
if re.fullmatch(r"@govoplan/[a-z0-9-]+-webui", str(package.get("name", ""))) is None:
raise SystemExit("WebUI package name must use the @govoplan/*-webui namespace")
release = Path("webui/package.release.json")
if release.is_file():
release_package = json.loads(release.read_text(encoding="utf-8"))
if (
release_package.get("name") != package.get("name")
or release_package.get("version") != expected
):
raise SystemExit("WebUI release package identity does not match package.json and the release tag")
PY
- name: Build immutable package artifacts
shell: bash
run: |
set -euo pipefail
python -m pip install --disable-pip-version-check build==1.5.0 twine==7.0.0
rm -rf dist .package-webui
python -m build --wheel --outdir dist
python -m twine check dist/*.whl
if [[ -f webui/package.json ]]; then
mkdir .package-webui
cp -a webui/. .package-webui/
rm -rf .package-webui/node_modules .package-webui/dist
if [[ -f .package-webui/package.release.json ]]; then
cp .package-webui/package.release.json .package-webui/package.json
fi
node <<'NODE'
const fs = require("node:fs");
const path = ".package-webui/package.json";
const packageJson = JSON.parse(fs.readFileSync(path, "utf8"));
const groups = ["dependencies", "optionalDependencies", "peerDependencies"];
for (const group of groups) {
for (const [name, specifier] of Object.entries(packageJson[group] || {})) {
if (!name.startsWith("@govoplan/")) continue;
if (typeof specifier !== "string") {
throw new Error(`${group}.${name} must use a string version`);
}
const packageSlug = name.slice("@govoplan/".length);
if (!packageSlug.endsWith("-webui")) {
throw new Error(`${group}.${name} is outside the WebUI package namespace`);
}
const repository = `govoplan-${packageSlug.slice(0, -"-webui".length)}`;
const escapedRepository = repository.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
const gitTag = specifier.match(
new RegExp(
`^git\\+(?:ssh://git@|https://)git\\.add-ideas\\.de/(?:GovOPlaN|add-ideas)/${escapedRepository}\\.git#v([0-9]+\\.[0-9]+\\.[0-9]+)$`,
),
);
if (gitTag) {
packageJson[group][name] = gitTag[1];
continue;
}
if (specifier.startsWith("file:") || specifier.startsWith("git+")) {
throw new Error(
`${group}.${name} must resolve to an exact registry version for publication`,
);
}
}
}
delete packageJson.private;
fs.writeFileSync(path, `${JSON.stringify(packageJson, null, 2)}\n`);
NODE
npm pkg delete private --prefix .package-webui
(cd .package-webui && npm pack --ignore-scripts --pack-destination ../dist)
fi
python - <<'PY'
import hashlib
import json
from pathlib import Path
import os
import subprocess
artifacts = []
for path in sorted(Path("dist").iterdir()):
if path.suffix not in {".whl", ".tgz"}:
continue
digest = hashlib.sha256(path.read_bytes()).hexdigest()
artifacts.append({"filename": path.name, "sha256": digest, "size": path.stat().st_size})
payload = {
"schema_version": "1",
"repository": os.environ["GITEA_REPOSITORY"],
"tag": os.environ["RELEASE_TAG"],
"commit": subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip(),
"artifacts": artifacts,
}
Path("dist/package-artifacts.json").write_text(
json.dumps(payload, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
PY
- name: Retain package hash evidence
uses: actions/upload-artifact@a8a3f3ad30e3422c9c7b888a15615d19a852ae32
with:
name: module-packages-${{ gitea.ref_name }}
path: dist/package-artifacts.json
- name: Check immutable registry state
shell: bash
env:
PACKAGE_TOKEN: ${{ secrets.GOVOPLAN_PACKAGE_TOKEN }}
run: |
set -euo pipefail
test -n "$PACKAGE_TOKEN"
python - <<'PY'
import hashlib
import json
import os
from pathlib import Path
import tomllib
from urllib.error import HTTPError
from urllib.parse import quote
from urllib.request import Request, urlopen
api_root = "https://git.add-ideas.de/api/v1/packages/GovOPlaN"
token = os.environ["PACKAGE_TOKEN"]
def should_publish(kind, name, version, path):
package_url = "/".join(
(api_root, kind, quote(name, safe=""), quote(version, safe=""), "files")
)
request = Request(
package_url,
headers={"Accept": "application/json", "Authorization": f"token {token}"},
)
try:
with urlopen(request, timeout=30) as response:
files = json.load(response)
except HTTPError as exc:
if exc.code == 404:
print(f"{kind} package {name}=={version} is not published yet")
return True
raise
if not isinstance(files, list) or len(files) != 1:
raise SystemExit(
f"immutable {kind} package {name}=={version} has an unexpected file set"
)
expected_sha256 = hashlib.sha256(path.read_bytes()).hexdigest()
if files[0].get("sha256") != expected_sha256:
raise SystemExit(
f"immutable {kind} package {name}=={version} already exists with a different SHA-256"
)
print(f"verified existing {kind} package {name}=={version} ({expected_sha256})")
return False
project = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))["project"]
wheels = tuple(Path("dist").glob("*.whl"))
if len(wheels) != 1:
raise SystemExit("release build must contain exactly one wheel")
publish_pypi = should_publish(
"pypi", str(project["name"]), str(project["version"]), wheels[0]
)
tarballs = tuple(Path("dist").glob("*.tgz"))
if len(tarballs) > 1:
raise SystemExit("release build must contain at most one npm package")
publish_npm = False
if tarballs:
webui = json.loads(
Path(".package-webui/package.json").read_text(encoding="utf-8")
)
publish_npm = should_publish(
"npm", str(webui["name"]), str(webui["version"]), tarballs[0]
)
with Path(os.environ["GITEA_ENV"]).open("a", encoding="utf-8") as env_file:
env_file.write(f"PUBLISH_PYPI={int(publish_pypi)}\n")
env_file.write(f"PUBLISH_NPM={int(publish_npm)}\n")
PY
- name: Publish wheel and WebUI package
shell: bash
env:
PACKAGE_USERNAME: ${{ secrets.GOVOPLAN_PACKAGE_USERNAME }}
PACKAGE_TOKEN: ${{ secrets.GOVOPLAN_PACKAGE_TOKEN }}
run: |
set -euo pipefail
test -n "$PACKAGE_USERNAME"
test -n "$PACKAGE_TOKEN"
if [[ "$PUBLISH_PYPI" == 1 ]]; then
TWINE_USERNAME="$PACKAGE_USERNAME" TWINE_PASSWORD="$PACKAGE_TOKEN" \
python -m twine upload --non-interactive \
--repository-url https://git.add-ideas.de/api/packages/GovOPlaN/pypi \
dist/*.whl
else
echo "Exact wheel is already present; skipping immutable retry."
fi
shopt -s nullglob
webui_packages=(dist/*.tgz)
if (( ${#webui_packages[@]} )) && [[ "$PUBLISH_NPM" == 1 ]]; then
npmrc="$(mktemp)"
trap 'rm -f "$npmrc"' EXIT
chmod 600 "$npmrc"
printf '%s\n' \
'@govoplan:registry=https://git.add-ideas.de/api/packages/GovOPlaN/npm/' \
"//git.add-ideas.de/api/packages/GovOPlaN/npm/:_authToken=$PACKAGE_TOKEN" \
> "$npmrc"
NPM_CONFIG_USERCONFIG="$npmrc" npm publish "./${webui_packages[0]}" \
--ignore-scripts --access public \
--registry https://git.add-ideas.de/api/packages/GovOPlaN/npm/
elif (( ${#webui_packages[@]} )); then
echo "Exact WebUI package is already present; skipping immutable retry."
fi
+7
View File
@@ -3,3 +3,10 @@
<!-- govoplan-repository-type:start -->
**Repository type:** module (domain).
<!-- govoplan-repository-type:end -->
Provider-neutral contracts for digest-bound payable export planning and
external booking reconciliation. This release intentionally performs no ERP
effect until a named product, schema, transport, and target recovery test have
been approved.
See [the payable export and reconciliation contract](docs/PAYABLE_EXPORT_AND_RECONCILIATION.md).
+54
View File
@@ -0,0 +1,54 @@
# Payable export and booking reconciliation
ERP is an integration boundary, not GovOPlaN's accounting system. Procurement,
Payments, or Ledger remains authoritative for an approved payable. A configured
external ERP remains authoritative for its booking identifier and booking
status. The connector joins those facts only through explicit correlation and
evidence.
## Export plan
`PayableExportInput` contains an immutable payable revision, stable invoice and
creditor references, an invoice-document SHA-256, currency, and an amount in
integer minor units. Floating-point amounts are rejected. Optional budget and
cost-centre references and up to 100 unique evidence references are bound to the
same projection.
`ErpPayableProfile` names the target provider and company code, exact payable
schema and version, mapping revision, non-secret connection reference, and a
complete raw-to-normalized booking-status mapping. No provider or schema is
selected by the module.
The plan serializes this input as canonical JSON and binds profile, input,
payload, payable revision, and idempotency identity with SHA-256. It is always
marked `dispatch_allowed=False`. A later product adapter must prove its API,
credentials, idempotency, outcome-unknown lookup, and recovery behavior before
it can execute the plan.
## Reconciliation
An observation must identify the tenant, payable, provider, exact export-plan
digest, external booking and revision, observation time, amount, currency, and
evidence digest. The decision is deterministic:
- `received` and `validated` wait for a terminal outcome;
- `booked` records the correlated booking;
- `rejected` and `cancelled` record a rejection;
- `reversed` records a separate reversal;
- unknown statuses and binding mismatches are quarantined.
Missing responses and timeouts never mean that the export failed. Operators
must query the target by the stable plan correlation before retrying.
## Datenschutz und Betrieb
ERP speichert in dieser Ausbaustufe weder Rechnungen noch Verbindlichkeiten,
Pläne, Anbieterantworten oder Zugangsdaten. Das fachlich verantwortliche Modul
bleibt für Aufbewahrung, Auskunft, Berichtigung und Löschung zuständig. Profile
enthalten nur eine Verbindungsreferenz; Geheimnisse dürfen weder im Plan noch im
Nachweis erscheinen.
Vor einer echten Anbindung sind Produkt und Version, Schema und Transport,
Buchungskreis, Anmeldedaten, Statusabbildung, Korrelationssuche, Stornoverhalten
und eine Testumgebung festzulegen und gegen Erfolg, Ablehnung, Zeitüberschreitung,
Doppelzustellung und Wiederherstellung zu prüfen.
+31
View File
@@ -0,0 +1,31 @@
[build-system]
requires = ["setuptools>=69", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "govoplan-erp"
version = "0.1.20"
description = "GovOPlaN governed ERP payable export and reconciliation module."
readme = "README.md"
requires-python = ">=3.12"
authors = [{ name = "GovOPlaN" }]
dependencies = [
"govoplan-core>=0.1.37",
"govoplan-access>=0.1.18",
]
[tool.setuptools.packages.find]
where = ["src"]
[tool.setuptools.package-data]
govoplan_erp = ["py.typed"]
[project.entry-points."govoplan.modules"]
erp = "govoplan_erp.backend.manifest:get_manifest"
[tool.pytest.ini_options]
pythonpath = ["src"]
[tool.ruff]
line-length = 100
target-version = "py312"
+3
View File
@@ -0,0 +1,3 @@
"""GovOPlaN ERP integration module."""
__all__: list[str] = []
+1
View File
@@ -0,0 +1 @@
"""Governed ERP integration contracts."""
+328
View File
@@ -0,0 +1,328 @@
from __future__ import annotations
from datetime import UTC, datetime
from govoplan_core.core.access import (
CAPABILITY_AUTH_PERMISSION_EVALUATOR,
CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
)
from govoplan_core.core.modules import (
DocumentationCondition,
DocumentationLink,
DocumentationTopic,
ModuleManifest,
PermissionDefinition,
RoleTemplate,
)
from govoplan_core.core.provider_governance import (
ExternalProviderDeclaration,
ExternalProviderRuntimeState,
ExternalProviderStateContext,
ExternalProviderStateProviderRegistration,
ProviderBehaviorDeclaration,
ProviderObjectDeclaration,
declared_module_architecture,
)
MODULE_ID = "erp"
MODULE_VERSION = "0.1.20"
READ_SCOPE = "erp:payables:read"
PLAN_SCOPE = "erp:payables:plan"
RECONCILE_SCOPE = "erp:bookings:reconcile"
ADMIN_SCOPE = "erp:integration:admin"
ERP_PROVIDER_ID = "erp.payables_target"
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="ERP",
level="tenant",
module_id=module_id,
resource=resource,
action=action,
)
ERP_PROVIDER = ExternalProviderDeclaration(
id=ERP_PROVIDER_ID,
module_id=MODULE_ID,
label="Configured ERP payable and booking target",
maturity="read",
operations=("read", "preview", "dry_run"),
objects=(
ProviderObjectDeclaration(
object_type="payable_export",
field_groups=("identity", "amount", "coding", "evidence"),
authority_modes=("native_authoritative",),
default_authority_mode="native_authoritative",
),
ProviderObjectDeclaration(
object_type="booking_observation",
field_groups=("correlation", "status", "amount", "evidence"),
authority_modes=("external_authoritative",),
default_authority_mode="external_authoritative",
),
),
behavior=ProviderBehaviorDeclaration(
revision_tokens=(
"Payable revisions and mapping revisions bind export plans; external revisions "
"bind booking observations."
),
concurrency=(
"Only the exact payable revision and reviewed profile digest may be reconciled."
),
freshness=(
"Every booking observation carries an aware timestamp and external revision."
),
health=(
"Profile mapping, export planning, target transport, correlation lookup, and "
"reconciliation are reported separately."
),
max_read_items=1000,
idempotency=(
"The exact payable projection digest is the default idempotency identity."
),
retry=(
"No export retry is allowed until the target is queried by the exact plan correlation."
),
timeout_seconds=30,
conflicts=(
"Tenant, payable, provider, profile, plan, currency, and amount mismatches are quarantined."
),
outcome_unknown=(
"A missing response is unknown, never evidence that a payable was or was not booked."
),
outcome_unknown_supported=True,
evidence=(
"Payable, profile, payload, plan, external revision, and observation digests form evidence."
),
correction=(
"Correct the source payable or mapping and issue a new revision-bound export plan."
),
rollback="An external booking is not assumed to be transactionally reversible.",
compensation=(
"Reversals are separate correlated observations and never overwrite the original booking."
),
reconciliation=(
"Look up the external booking by stable plan correlation and compare amount, currency, "
"profile, revision, and evidence before retry."
),
outage=(
"Module-owned payable state remains authoritative while export and reconciliation wait."
),
classifications=("confidential", "restricted"),
purposes=("payable export", "booking reconciliation"),
retention=(
"Owning finance modules retain payables; ERP and Audit retain only governed integration evidence."
),
secret_handling=(
"Profiles contain only a connection reference; credentials never enter plans or evidence."
),
),
documentation_topic_ids=("erp.payable-reconciliation",),
)
def _provider_states(
context: ExternalProviderStateContext,
) -> tuple[ExternalProviderRuntimeState, ...]:
del context
return (
ExternalProviderRuntimeState(
provider_id=ERP_PROVIDER_ID,
observed_at=datetime.now(UTC),
configured=False,
active=False,
health="inactive",
freshness="not_applicable",
conflict="not_applicable",
recovery="unsupported",
detail="No target-tested ERP product binding is configured; dispatch is disabled.",
),
)
manifest = ModuleManifest(
id=MODULE_ID,
name="ERP",
version=MODULE_VERSION,
dependencies=("access",),
optional_dependencies=("procurement", "payments", "ledger", "files", "audit", "policy"),
required_capabilities=(
CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
CAPABILITY_AUTH_PERMISSION_EVALUATOR,
),
permissions=(
_permission(READ_SCOPE, "View ERP integration", "Read non-secret profiles, plans, and reconciliation evidence."),
_permission(PLAN_SCOPE, "Plan payable export", "Create a digest-bound, effect-free payable export plan."),
_permission(RECONCILE_SCOPE, "Reconcile ERP booking", "Interpret a correlated external booking observation."),
_permission(ADMIN_SCOPE, "Administer ERP integration", "Configure and verify ERP mappings and recovery behavior."),
),
role_templates=(
RoleTemplate(
slug="erp_integration_operator",
name="ERP integration operator",
description="Plan payable exports and reconcile correlated booking observations.",
permissions=(READ_SCOPE, PLAN_SCOPE, RECONCILE_SCOPE),
),
RoleTemplate(
slug="erp_integration_administrator",
name="ERP integration administrator",
description="Configure and verify governed ERP target bindings.",
permissions=(READ_SCOPE, PLAN_SCOPE, RECONCILE_SCOPE, ADMIN_SCOPE),
),
),
external_providers=(ERP_PROVIDER,),
external_provider_state_providers=(
ExternalProviderStateProviderRegistration(
module_id=MODULE_ID,
provider_id=ERP_PROVIDER_ID,
provider=_provider_states,
),
),
documentation=(
DocumentationTopic(
id="erp.boundary",
title="ERP integration boundary",
summary="Exchange governed finance projections without making GovOPlaN a replacement ERP or moving payable authority into the connector.",
body=(
"ERP owns product profiles, mapping and transport plans, correlation, and external booking observations. Procurement, Payments, and Ledger continue to own approvals, payables, payments, and accounting projections. An export plan is not evidence of external booking, and an unconfigured provider declaration is not a production integration."
),
layer="available",
documentation_types=("admin", "user"),
audience=("user", "operator", "module_admin", "auditor"),
related_modules=("procurement", "payments", "ledger", "audit"),
links=(
DocumentationLink(
label="ERP integration boundary",
href="docs/PAYABLE_EXPORT_AND_RECONCILIATION.md",
kind="repository",
),
),
translations={
"de": {
"title": "Integrationsgrenze des ERP-Moduls",
"summary": "Gesteuerte Finanzprojektionen austauschen, ohne GovOPlaN zum Ersatz-ERP zu machen oder die Verantwortung für Verbindlichkeiten in den Konnektor zu verlagern.",
"body": "ERP verantwortet Produktprofile, Zuordnung und Transportpläne, Korrelation sowie externe Buchungsbeobachtungen. Procurement, Payments und Ledger bleiben für Freigaben, Verbindlichkeiten, Zahlungen und Buchhaltungsprojektionen zuständig. Ein Übergabeplan ist kein Buchungsnachweis; eine unkonfigurierte Anbieterdeklaration ist keine produktive Integration.",
}
},
metadata={"kind": "reference"},
order=90,
),
DocumentationTopic(
id="erp.payable-reconciliation",
title="Plan payable exports and reconcile ERP bookings",
summary=(
"Bind an exact payable revision to an effect-free export plan and quarantine "
"uncorrelated or unmapped external booking observations."
),
body=(
"Procurement, Payments, or Ledger owns the payable and supplies integer minor-unit "
"amounts, stable references, an invoice digest, and a revision. ERP combines this "
"projection with an administrator-reviewed product profile, schema, company code, "
"mapping revision, connection reference, and status mapping. The resulting canonical "
"payload and plan are digest-bound but cannot dispatch in this release. Reconciliation "
"accepts only an observation tied to the exact plan, provider, tenant, payable, amount, "
"currency, and evidence. Unknown statuses or mismatches are quarantined. Bookings, "
"rejections, and reversals remain separate evidence-bearing outcomes."
),
layer="configured",
documentation_types=("admin", "user"),
audience=("user", "operator", "module_admin", "auditor"),
related_modules=("procurement", "payments", "ledger", "files", "audit"),
conditions=(
DocumentationCondition(
any_scopes=(READ_SCOPE, PLAN_SCOPE, RECONCILE_SCOPE, ADMIN_SCOPE),
),
),
links=(
DocumentationLink(
label="ERP payable and booking contract",
href="docs/PAYABLE_EXPORT_AND_RECONCILIATION.md",
kind="repository",
),
),
translations={
"de": {
"title": "Kreditorische Übergaben planen und ERP-Buchungen abgleichen",
"summary": "Eine exakte Verbindlichkeitsrevision an einen wirkungsfreien Übergabeplan binden und nicht korrelierte oder unbekannte ERP-Buchungsstände sperren.",
"body": "Procurement, Payments oder Ledger verantwortet die Verbindlichkeit und liefert ganzzahlige Nebenwährungseinheiten, stabile Referenzen, die Rechnungsprüfsumme und eine Revision. ERP verbindet diese Projektion mit einem administrativ geprüften Produktprofil, Schema, Buchungskreis, Mapping-Revision, Verbindungsreferenz und einer Statuszuordnung. Nutzdaten und Plan sind kanonisch prüfsummengebunden, können in dieser Version aber nicht versendet werden. Der Abgleich akzeptiert nur Beobachtungen, die exakt zu Plan, Anbieter, Mandant, Verbindlichkeit, Betrag, Währung und Nachweis passen. Unbekannte Zustände und Abweichungen werden isoliert. Buchung, Ablehnung und Storno bleiben getrennte nachweisgebundene Ergebnisse.",
}
},
metadata={
"kind": "workflow",
"prerequisites": [
"The owning finance module supplies an approved immutable payable revision.",
"An administrator has reviewed the target schema and raw-status mapping.",
"The connection reference resolves through a deployment-owned credential boundary.",
],
"steps": [
"Build and review the exact canonical export payload and plan digests.",
"Dispatch only through a separately target-tested adapter.",
"Read the booking by stable plan correlation after success, timeout, or retry.",
"Record, wait, or quarantine the deterministic reconciliation decision.",
],
"limitations": [
"No ERP product, schema, endpoint, or transport is selected by this release.",
"Plans cannot dispatch and observations are not persisted by this module.",
],
"consequences": [
"Changing the payable or mapping produces a new plan and idempotency identity.",
"A timeout never implies success or failure.",
"Reversals remain linked outcomes rather than destructive status replacement.",
],
},
structured_translation_version="1",
structured_translations={
"de": {
"prerequisites": [
"Das fachlich verantwortliche Finanzmodul liefert eine freigegebene unveränderliche Verbindlichkeitsrevision.",
"Eine administrierende Person hat Zielschema und Rohstatuszuordnung geprüft.",
"Die Verbindungsreferenz wird über eine installationsverantwortete Berechtigungsgrenze aufgelöst.",
],
"steps": [
"Exakte kanonische Übergabenutzdaten und Planprüfsummen erzeugen und prüfen.",
"Nur über einen getrennt am Ziel geprüften Adapter versenden.",
"Die Buchung nach Erfolg, Zeitüberschreitung oder Wiederholung über die stabile Plankorrelation lesen.",
"Die deterministische Abgleichentscheidung erfassen, abwarten oder isolieren.",
],
"limitations": [
"Diese Version wählt weder ERP-Produkt, Schema, Endpunkt noch Transport aus.",
"Pläne können nicht versenden und Beobachtungen werden von diesem Modul nicht gespeichert.",
],
"consequences": [
"Eine Änderung an Verbindlichkeit oder Zuordnung erzeugt einen neuen Plan und eine neue Idempotenzidentität.",
"Eine Zeitüberschreitung bedeutet weder Erfolg noch Fehlschlag.",
"Stornos bleiben verknüpfte Ergebnisse, statt den ursprünglichen Status destruktiv zu ersetzen.",
],
}
},
order=100,
),
),
architecture=declared_module_architecture(
layer="data_reporting_integration",
kind="integration",
maturity="vertical_slice",
documentation_ref="docs/PAYABLE_EXPORT_AND_RECONCILIATION.md",
test_ref="tests/test_payables.py",
known_limits=(
"A named ERP product, schema, transport, credentials, and target test are required before dispatch.",
),
supported_authority_modes=("native_authoritative", "external_authoritative"),
owned_concepts=("ERP payable profile", "payable export plan", "booking observation mapping"),
non_owned_concepts=("invoice", "payable approval", "ledger entry", "payment"),
recovery_docs=("docs/PAYABLE_EXPORT_AND_RECONCILIATION.md",),
security_docs=("docs/PAYABLE_EXPORT_AND_RECONCILIATION.md",),
operations_docs=("docs/PAYABLE_EXPORT_AND_RECONCILIATION.md",),
),
)
def get_manifest() -> ModuleManifest:
return manifest
+451
View File
@@ -0,0 +1,451 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import date, datetime
import hashlib
import json
import re
from typing import Literal
BookingState = Literal[
"received",
"validated",
"booked",
"rejected",
"reversed",
"cancelled",
]
ReconciliationOutcome = Literal[
"pending",
"booked",
"rejected",
"reversed",
"conflict",
]
ReconciliationAction = Literal[
"wait",
"record_booking",
"record_rejection",
"record_reversal",
"quarantine",
]
_SOURCE_MODULE = re.compile(r"^[a-z][a-z0-9_-]{0,63}$")
_CURRENCY = re.compile(r"^[A-Z]{3}$")
_MAX_AMOUNT_MINOR = 10**15
class ErpPayableError(RuntimeError):
"""Stable ERP planning or reconciliation failure without invoice contents."""
@dataclass(frozen=True, slots=True)
class PayableExportInput:
"""An exact module-owned payable projection prepared for external export."""
tenant_id: str
source_module: str
payable_id: str
payable_revision: int
invoice_reference: str
creditor_reference: str
currency: str
gross_amount_minor: int
due_date: date | None
cost_center_reference: str | None
budget_reference: str | None
invoice_document_sha256: str
evidence_references: tuple[str, ...] = ()
def __post_init__(self) -> None:
for name in (
"tenant_id",
"payable_id",
"invoice_reference",
"creditor_reference",
):
object.__setattr__(self, name, _text(getattr(self, name), name))
source_module = _text(self.source_module, "source_module", maximum=64)
if _SOURCE_MODULE.fullmatch(source_module) is None:
raise ValueError("ERP source_module must be a stable lowercase module identifier.")
object.__setattr__(self, "source_module", source_module)
if not isinstance(self.payable_revision, int) or isinstance(self.payable_revision, bool):
raise ValueError("ERP payable_revision must be an integer.")
if self.payable_revision < 1:
raise ValueError("ERP payable_revision must be positive.")
currency = str(self.currency or "").strip().upper()
if _CURRENCY.fullmatch(currency) is None:
raise ValueError("ERP currency must be a three-letter ISO-style code.")
object.__setattr__(self, "currency", currency)
if not isinstance(self.gross_amount_minor, int) or isinstance(
self.gross_amount_minor, bool
):
raise ValueError("ERP amounts must use integer minor units, never floating point.")
if not 0 < self.gross_amount_minor <= _MAX_AMOUNT_MINOR:
raise ValueError("ERP gross_amount_minor is outside the supported positive range.")
for name in ("cost_center_reference", "budget_reference"):
value = getattr(self, name)
if value is not None:
object.__setattr__(self, name, _text(value, name))
object.__setattr__(
self,
"invoice_document_sha256",
_sha256(self.invoice_document_sha256, "invoice_document_sha256"),
)
if len(self.evidence_references) > 100:
raise ValueError("ERP payable evidence is limited to 100 references.")
evidence = tuple(_text(value, "evidence_reference") for value in self.evidence_references)
if len(evidence) != len(set(evidence)):
raise ValueError("ERP payable evidence references must be unique.")
object.__setattr__(self, "evidence_references", evidence)
@property
def input_sha256(self) -> str:
return _digest(self.to_payload())
def to_payload(self) -> dict[str, object]:
return {
"tenant_id": self.tenant_id,
"source_module": self.source_module,
"payable_id": self.payable_id,
"payable_revision": self.payable_revision,
"invoice_reference": self.invoice_reference,
"creditor_reference": self.creditor_reference,
"currency": self.currency,
"gross_amount_minor": self.gross_amount_minor,
"due_date": self.due_date.isoformat() if self.due_date else None,
"cost_center_reference": self.cost_center_reference,
"budget_reference": self.budget_reference,
"invoice_document_sha256": self.invoice_document_sha256,
"evidence_references": list(self.evidence_references),
}
@dataclass(frozen=True, slots=True)
class ErpPayableProfile:
"""Product-neutral, non-secret mapping for one external finance target."""
profile_id: str
provider_id: str
company_code: str
payable_schema: str
payable_schema_version: str
mapping_revision: str
connection_ref: str
booking_status_mapping: tuple[tuple[str, BookingState], ...]
def __post_init__(self) -> None:
for name in (
"profile_id",
"provider_id",
"company_code",
"payable_schema",
"payable_schema_version",
"mapping_revision",
"connection_ref",
):
object.__setattr__(self, name, _text(getattr(self, name), name))
if not self.booking_status_mapping:
raise ValueError("ERP profile requires an explicit booking-status mapping.")
if len(self.booking_status_mapping) > 100:
raise ValueError("ERP profile supports at most 100 booking-status mappings.")
normalized: list[tuple[str, BookingState]] = []
seen: set[str] = set()
allowed = {"received", "validated", "booked", "rejected", "reversed", "cancelled"}
for raw_status, state in self.booking_status_mapping:
raw = _text(raw_status, "raw_booking_status", maximum=100)
key = raw.casefold()
if key in seen:
raise ValueError("ERP raw booking statuses must be unique ignoring case.")
if state not in allowed:
raise ValueError("ERP normalized booking state is unsupported.")
seen.add(key)
normalized.append((raw, state))
object.__setattr__(self, "booking_status_mapping", tuple(normalized))
@property
def profile_sha256(self) -> str:
return _digest(
{
"profile_id": self.profile_id,
"provider_id": self.provider_id,
"company_code": self.company_code,
"payable_schema": self.payable_schema,
"payable_schema_version": self.payable_schema_version,
"mapping_revision": self.mapping_revision,
"connection_ref": self.connection_ref,
"booking_status_mapping": [list(item) for item in self.booking_status_mapping],
}
)
def normalized_status(self, raw_status: str) -> BookingState | None:
candidate = str(raw_status or "").strip().casefold()
return next(
(state for raw, state in self.booking_status_mapping if raw.casefold() == candidate),
None,
)
@dataclass(frozen=True, slots=True)
class PayableExportPlan:
tenant_id: str
payable_id: str
payable_revision: int
currency: str
gross_amount_minor: int
provider_id: str
profile_id: str
profile_sha256: str
input_sha256: str
idempotency_key: str
payload_json: bytes
payload_sha256: str
plan_sha256: str
dispatch_allowed: bool = False
@dataclass(frozen=True, slots=True)
class BookingObservation:
tenant_id: str
payable_id: str
provider_id: str
external_booking_id: str
raw_status: str
external_revision: str
observed_at: datetime
exported_plan_sha256: str
evidence_sha256: str
currency: str
gross_amount_minor: int
def __post_init__(self) -> None:
for name in (
"tenant_id",
"payable_id",
"provider_id",
"external_booking_id",
"raw_status",
"external_revision",
):
object.__setattr__(self, name, _text(getattr(self, name), name))
if self.observed_at.tzinfo is None or self.observed_at.utcoffset() is None:
raise ValueError("ERP booking observations require a timezone-aware timestamp.")
object.__setattr__(
self,
"exported_plan_sha256",
_sha256(self.exported_plan_sha256, "exported_plan_sha256"),
)
object.__setattr__(
self,
"evidence_sha256",
_sha256(self.evidence_sha256, "evidence_sha256"),
)
currency = str(self.currency or "").strip().upper()
if _CURRENCY.fullmatch(currency) is None:
raise ValueError("ERP observation currency must be a three-letter code.")
object.__setattr__(self, "currency", currency)
if not isinstance(self.gross_amount_minor, int) or isinstance(
self.gross_amount_minor, bool
):
raise ValueError("ERP observation amounts must use integer minor units.")
@dataclass(frozen=True, slots=True)
class BookingReconciliationDecision:
outcome: ReconciliationOutcome
action: ReconciliationAction
observed_state: BookingState | None
reason: str
external_booking_id: str
evidence_sha256: str
decision_sha256: str
def build_payable_export_plan(
profile: ErpPayableProfile,
payable: PayableExportInput,
*,
idempotency_key: str | None = None,
) -> PayableExportPlan:
payload = {
"schema": profile.payable_schema,
"schema_version": profile.payable_schema_version,
"company_code": profile.company_code,
"mapping_revision": profile.mapping_revision,
"payable": payable.to_payload(),
}
payload_json = _canonical_json(payload)
payload_sha256 = hashlib.sha256(payload_json).hexdigest()
key = (
_text(idempotency_key, "idempotency_key")
if idempotency_key is not None
else f"erp-payable:{payable.input_sha256}"
)
plan_payload = {
"tenant_id": payable.tenant_id,
"payable_id": payable.payable_id,
"payable_revision": payable.payable_revision,
"currency": payable.currency,
"gross_amount_minor": payable.gross_amount_minor,
"provider_id": profile.provider_id,
"profile_id": profile.profile_id,
"profile_sha256": profile.profile_sha256,
"input_sha256": payable.input_sha256,
"idempotency_key": key,
"payload_sha256": payload_sha256,
}
return PayableExportPlan(
tenant_id=payable.tenant_id,
payable_id=payable.payable_id,
payable_revision=payable.payable_revision,
currency=payable.currency,
gross_amount_minor=payable.gross_amount_minor,
provider_id=profile.provider_id,
profile_id=profile.profile_id,
profile_sha256=profile.profile_sha256,
input_sha256=payable.input_sha256,
idempotency_key=key,
payload_json=payload_json,
payload_sha256=payload_sha256,
plan_sha256=_digest(plan_payload),
)
def reconcile_booking(
profile: ErpPayableProfile,
plan: PayableExportPlan,
observation: BookingObservation,
) -> BookingReconciliationDecision:
conflict = _binding_conflict(profile, plan, observation)
if conflict is not None:
return _decision(observation, "conflict", "quarantine", None, conflict)
state = profile.normalized_status(observation.raw_status)
if state is None:
return _decision(
observation,
"conflict",
"quarantine",
None,
"The external booking status is not present in the reviewed mapping.",
)
if state in {"received", "validated"}:
return _decision(
observation,
"pending",
"wait",
state,
"The external payable is acknowledged but has no terminal booking outcome.",
)
if state == "booked":
return _decision(
observation,
"booked",
"record_booking",
state,
"The exact exported payable has a mapped external booking outcome.",
)
if state == "reversed":
return _decision(
observation,
"reversed",
"record_reversal",
state,
"The external system reports a reversal of the correlated booking.",
)
return _decision(
observation,
"rejected",
"record_rejection",
state,
"The external system rejected or cancelled the correlated payable.",
)
def _binding_conflict(
profile: ErpPayableProfile,
plan: PayableExportPlan,
observation: BookingObservation,
) -> str | None:
checks = (
(plan.profile_sha256 == profile.profile_sha256, "The reviewed ERP profile changed."),
(plan.profile_id == profile.profile_id, "The ERP profile identity does not match."),
(plan.provider_id == profile.provider_id, "The ERP provider identity does not match."),
(observation.tenant_id == plan.tenant_id, "The observation belongs to another tenant."),
(observation.payable_id == plan.payable_id, "The observation belongs to another payable."),
(observation.provider_id == plan.provider_id, "The observation came from another provider."),
(observation.currency == plan.currency, "The observation currency does not match the export."),
(
observation.gross_amount_minor == plan.gross_amount_minor,
"The observation amount does not match the export.",
),
(
observation.exported_plan_sha256 == plan.plan_sha256,
"The observation is not bound to the exact export plan.",
),
)
return next((reason for valid, reason in checks if not valid), None)
def _decision(
observation: BookingObservation,
outcome: ReconciliationOutcome,
action: ReconciliationAction,
state: BookingState | None,
reason: str,
) -> BookingReconciliationDecision:
payload = {
"outcome": outcome,
"action": action,
"observed_state": state,
"reason": reason,
"external_booking_id": observation.external_booking_id,
"external_revision": observation.external_revision,
"observed_at": observation.observed_at.isoformat(),
"evidence_sha256": observation.evidence_sha256,
}
return BookingReconciliationDecision(
outcome=outcome,
action=action,
observed_state=state,
reason=reason,
external_booking_id=observation.external_booking_id,
evidence_sha256=observation.evidence_sha256,
decision_sha256=_digest(payload),
)
def _text(value: object, label: str, *, maximum: int = 255) -> str:
normalized = str(value or "").strip()
if not normalized or len(normalized) > maximum or any(ord(char) < 32 for char in normalized):
raise ValueError(
f"ERP {label.replace('_', ' ')} is required, bounded, and must not contain controls."
)
return normalized
def _sha256(value: object, label: str) -> str:
normalized = str(value or "").strip().lower().removeprefix("sha256:")
if len(normalized) != 64 or any(char not in "0123456789abcdef" for char in normalized):
raise ValueError(f"ERP {label.replace('_', ' ')} must be a SHA-256 digest.")
return normalized
def _canonical_json(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
def _digest(value: object) -> str:
return hashlib.sha256(_canonical_json(value)).hexdigest()
__all__ = [
"BookingObservation",
"BookingReconciliationDecision",
"ErpPayableError",
"ErpPayableProfile",
"PayableExportInput",
"PayableExportPlan",
"build_payable_export_plan",
"reconcile_booking",
]
+1
View File
@@ -0,0 +1 @@
+16
View File
@@ -0,0 +1,16 @@
from __future__ import annotations
from govoplan_core.core.modules import documentation_structured_translation_issues
from govoplan_erp.backend.manifest import manifest
def test_erp_german_reference_contract_is_complete() -> None:
topics = {topic.id: topic for topic in manifest.documentation}
assert topics["erp.boundary"].metadata["kind"] == "reference"
workflow = topics["erp.payable-reconciliation"]
assert workflow.metadata["kind"] == "workflow"
assert workflow.structured_translation_version == "1"
assert "de" in workflow.structured_translations
assert documentation_structured_translation_issues(workflow) == ()
for topic in topics.values():
assert all(topic.translations["de"].get(key) for key in ("title", "summary", "body"))
+136
View File
@@ -0,0 +1,136 @@
from __future__ import annotations
from datetime import UTC, date, datetime
import json
import pytest
from govoplan_erp.backend.manifest import get_manifest
from govoplan_erp.backend.payables import (
BookingObservation,
ErpPayableProfile,
PayableExportInput,
build_payable_export_plan,
reconcile_booking,
)
def _profile() -> ErpPayableProfile:
return ErpPayableProfile(
profile_id="reviewed-target-v1",
provider_id="erp-target",
company_code="1000",
payable_schema="govoplan-neutral-payable",
payable_schema_version="1",
mapping_revision="mapping-3",
connection_ref="core-credential:erp-target",
booking_status_mapping=(
("RECEIVED", "received"),
("POSTED", "booked"),
("REJECTED", "rejected"),
("REVERSED", "reversed"),
),
)
def _payable() -> PayableExportInput:
return PayableExportInput(
tenant_id="tenant-a",
source_module="payments",
payable_id="payable-7",
payable_revision=3,
invoice_reference="invoice-2026-7",
creditor_reference="creditor-4",
currency="eur",
gross_amount_minor=123_45,
due_date=date(2026, 9, 15),
cost_center_reference="cost-12",
budget_reference="budget-2026",
invoice_document_sha256="a" * 64,
evidence_references=("xrechnung:handoff-8",),
)
def _observation(plan, *, status: str = "POSTED", plan_sha256: str | None = None):
return BookingObservation(
tenant_id=plan.tenant_id,
payable_id=plan.payable_id,
provider_id=plan.provider_id,
external_booking_id="booking-99",
raw_status=status,
external_revision="external-4",
observed_at=datetime(2026, 8, 23, 12, tzinfo=UTC),
exported_plan_sha256=plan_sha256 or plan.plan_sha256,
evidence_sha256="b" * 64,
currency="EUR",
gross_amount_minor=123_45,
)
def test_plan_is_canonical_digest_bound_and_effect_free() -> None:
first = build_payable_export_plan(_profile(), _payable())
second = build_payable_export_plan(_profile(), _payable())
assert first == second
assert first.dispatch_allowed is False
assert first.idempotency_key.startswith("erp-payable:")
payload = json.loads(first.payload_json)
assert payload["payable"]["gross_amount_minor"] == 12345
assert payload["mapping_revision"] == "mapping-3"
def test_money_rejects_floating_point_and_invalid_currency() -> None:
values = _payable().to_payload()
with pytest.raises(ValueError, match="integer minor units"):
PayableExportInput(
**{
**values,
"due_date": date(2026, 9, 15),
"gross_amount_minor": 123.45,
"evidence_references": (),
}
)
@pytest.mark.parametrize(
("raw_status", "outcome", "action"),
[
("RECEIVED", "pending", "wait"),
("POSTED", "booked", "record_booking"),
("REJECTED", "rejected", "record_rejection"),
("REVERSED", "reversed", "record_reversal"),
],
)
def test_reconciliation_maps_reviewed_statuses(raw_status, outcome, action) -> None:
profile = _profile()
plan = build_payable_export_plan(profile, _payable())
decision = reconcile_booking(profile, plan, _observation(plan, status=raw_status))
assert decision.outcome == outcome
assert decision.action == action
assert len(decision.decision_sha256) == 64
def test_unknown_or_wrong_plan_observation_is_quarantined() -> None:
profile = _profile()
plan = build_payable_export_plan(profile, _payable())
unknown = reconcile_booking(profile, plan, _observation(plan, status="NEW-VENDOR-STATE"))
wrong_plan = reconcile_booking(
profile,
plan,
_observation(plan, plan_sha256="c" * 64),
)
assert (unknown.outcome, unknown.action) == ("conflict", "quarantine")
assert (wrong_plan.outcome, wrong_plan.action) == ("conflict", "quarantine")
def test_manifest_documents_unconfigured_provider_and_german_workflow() -> None:
manifest = get_manifest()
assert manifest.version == "0.1.20"
assert manifest.external_providers[0].id == "erp.payables_target"
assert manifest.architecture is not None
assert manifest.architecture.target_tested_providers == ()
assert manifest.documentation[0].translations["de"]["title"]