fix(xrechnung): bound validator output and report reads

Enforce capture limits while draining both subprocess pipes, kill and reap interrupted validators, and bound report allocation before parsing. Preserve result precedence and add synthetic-process regressions with EN/DE documentation.

Refs #2

.gitea/workflows/module-package-release.yml

@@ -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

README.md

@@ -1,5 +1,9 @@
 # govoplan-xrechnung
 
+Optional connector for fail-closed inbound XRechnung validation and digest-bound handoff. A deployment must explicitly select and pin the exact XRechnung and KoSIT validator configuration; no standard version is activated by default.
+
+See [the inbound validation profile](docs/INBOUND_VALIDATION.md).
+
 <!-- govoplan-repository-type:start -->
 **Repository type:** connector (standard).
 <!-- govoplan-repository-type:end -->

docs/INBOUND_VALIDATION.md

@@ -0,0 +1,85 @@
+# Inbound XRechnung validation
+
+This module validates an inbound XML invoice through a locally installed, exact KoSIT validator profile. It does not choose the active standard release, download validation artifacts at runtime, approve a payable, or book an invoice.
+
+## Pinned profile
+
+An administrator records the XRechnung version, validator version, configuration release, absolute Java and JAR paths, JAR SHA-256, configuration root and complete deterministic tree SHA-256, scenario file, and expected minimum validation-step count. Every run rechecks the executable and both artifact digests. Symbolic links and files outside the configuration root are rejected.
+
+## Deployment-owned profile allow-list
+
+`KoSITProfileRegistry` allows a deployment to approve several exact profiles
+without GovOPlaN choosing a standard version. Each approval binds the verified
+artifact digest, approving actor, decision reference, receive-time window, and
+the state `approved`, `suspended`, or `retired`. A deployment may configure one
+explicit default, or require the invoking workflow to select a profile every
+time. A default is never inferred.
+
+Selection uses the invoice receive time, which permits an administrator to
+model an overlap or transition window. Suspended, retired, unknown, expired,
+not-yet-valid, or subsequently modified profiles fail closed. Artifact digests
+are rechecked on selection and again during validation.
+
+Administrierende können damit mehrere konkrete Profile zeitlich begrenzt
+freigeben, ohne dass GovOPlaN eine XRechnungs-Version vorgibt. Übergangsfristen
+und ein Standardprofil sind ausdrücklich konfigurierte Entscheidungen. Ohne
+Standard muss jeder aufrufende Prozess ein freigegebenes Profil benennen.
+
+The configuration tree must be installed through deployment governance. A newer published release never becomes active automatically. Updating any artifact creates a different profile digest and requires regression evidence with accepted, rejected, malformed, and deliberately broken technical fixtures.
+
+## Fail-closed validation
+
+Inbound XML is size-bounded and parsed with external entities and DTD processing disabled before Java runs. KoSIT runs without a shell, with an absolute executable, fixed argument vector, bounded time, bounded captured output, and an isolated result directory.
+
+The validator's stdout and stderr share a fixed 2 MiB capture limit, including
+the single separator byte used when inspecting their combined output. Both
+streams are drained while the validator runs. If the next byte exceeds that
+limit, GovOPlaN kills and reaps the direct validator process and returns
+`failed` with the reason `KoSIT runner output exceeded the safety limit.`
+Conformance and assessment remain `unknown`, and no handoff is allowed even
+if a valid-looking report was already written. The existing timeout (120 seconds
+by default, configurable from 1 to 600 seconds) also kills and reaps the
+validator; cancellation propagates after the same cleanup. Output is not
+included in these technical-failure reasons. Operators should check the
+approved engine and configuration using synthetic inputs before retrying an
+output-limit failure. The pipe runner uses POSIX process facilities; deployment
+resource limits and descendant-process isolation remain outside this capture
+bound.
+
+Report reads are bounded to the existing 16 MiB limit plus one probe byte.
+Reports exactly at the limit remain eligible for normal interpretation;
+oversized reports cannot supply trusted evidence and yield `incomplete` with
+unknown conformance unless an earlier runner failure already determines the
+outcome. An output-limit failure skips reading the report entirely. This read
+bound does not limit how much report data the validator can write to disk.
+
+Standardausgabe und Standardfehlerausgabe des Validators teilen eine feste
+Erfassungsgrenze von 2 MiB einschließlich eines Trennbytes. Beide Ausgaben
+werden während des Laufs gelesen. Bei Überschreitung beendet GovOPlaN den
+direkten Validatorprozess und wartet dessen Abschluss ab. Das Ergebnis lautet
+`failed` mit unbekannter Konformität und Bewertung (`unknown`); auch ein
+scheinbar gültiger Bericht erlaubt keine Übergabe. Das bestehende Zeitlimit
+(standardmäßig 120 Sekunden, konfigurierbar von 1 bis 600 Sekunden) beendet den
+Prozess ebenfalls; ein Abbruch wird nach der Prozessbereinigung weitergegeben.
+Diese technischen Fehlergründe enthalten keine Runner-Ausgabe. Betreibende
+sollten das freigegebene Profil vor einem erneuten Versuch mit synthetischen
+Eingaben prüfen. Der Pipe-Runner verwendet POSIX-Prozessfunktionen;
+Ressourcengrenzen der Installation und die Isolation von Kindprozessen bleiben
+außerhalb dieser Erfassungsgrenze.
+
+Berichte werden höchstens bis zur bestehenden Grenze von 16 MiB zuzüglich
+eines Prüfbytes gelesen. Berichte genau an der Grenze werden normal
+ausgewertet; übergroße Berichte liefern keinen vertrauenswürdigen Nachweis und
+ergeben `incomplete` mit unbekannter Konformität, sofern nicht bereits ein
+vorrangiger Runner-Fehler das Ergebnis bestimmt. Bei Überschreitung der
+Runner-Ausgabegrenze wird der Bericht gar nicht gelesen. Die Lesegrenze
+begrenzt nicht die Berichtsmenge, die der Validator auf Datenträger schreiben
+kann.
+
+GovOPlaN does not trust the report alone. A technically complete result requires a zero process status, no technical error marker in runner output, a bounded well-formed VARL report, a matched scenario, at least the configured number of complete validation steps, and exactly one assessment. Formal validity and the accept/reject recommendation are retained separately because warnings can make them differ. This compensates for the known risk that a partial report can look valid after a transformation failure.
+
+Semantic invalidity is different from technical failure. A complete reject report is `invalid`; a timeout, crash, partial report, or inconsistent assessment is `unknown`. Neither result may be handed off as valid.
+
+## Governed handoff
+
+Only a technically complete, formally `valid`, and explicitly accepted result creates a handoff. The handoff binds tenant, source reference, invoice SHA-256, VARL report SHA-256, profile ID and profile SHA-256, and validation time. Procurement or Payments owns the later payable workflow; Files owns retained invoice bytes; Records may file an exact revision. Revalidation is mandatory after invoice or profile changes.

pyproject.toml

@@ -0,0 +1,32 @@
+[build-system]
+requires = ["setuptools>=69", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "govoplan-xrechnung"
+version = "0.1.21"
+description = "GovOPlaN inbound XRechnung validation connector."
+readme = "README.md"
+requires-python = ">=3.12"
+authors = [{ name = "GovOPlaN" }]
+dependencies = [
+  "defusedxml>=0.7.1",
+  "govoplan-core>=0.1.37",
+  "govoplan-access>=0.1.18",
+]
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.setuptools.package-data]
+govoplan_xrechnung = ["py.typed"]
+
+[project.entry-points."govoplan.modules"]
+xrechnung = "govoplan_xrechnung.backend.manifest:get_manifest"
+
+[tool.pytest.ini_options]
+pythonpath = ["src"]
+
+[tool.ruff]
+line-length = 100
+target-version = "py312"

src/govoplan_xrechnung/__init__.py

@@ -0,0 +1,3 @@
+"""GovOPlaN XRechnung connector."""
+
+__version__ = "0.1.21"

src/govoplan_xrechnung/backend/__init__.py

@@ -0,0 +1,2 @@
+"""Inbound XRechnung validation boundary."""
+

src/govoplan_xrechnung/backend/manifest.py

@@ -0,0 +1,262 @@
+from __future__ import annotations
+
+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 declared_module_architecture
+
+
+MODULE_ID = "xrechnung"
+MODULE_VERSION = "0.1.21"
+READ_SCOPE = "xrechnung:validation:read"
+EXECUTE_SCOPE = "xrechnung:validation:execute"
+HANDOFF_SCOPE = "xrechnung:handoff:create"
+
+
+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="XRechnung",
+        level="tenant",
+        module_id=module_id,
+        resource=resource,
+        action=action,
+    )
+
+
+manifest = ModuleManifest(
+    id=MODULE_ID,
+    name="XRechnung",
+    version=MODULE_VERSION,
+    dependencies=("access",),
+    optional_dependencies=("files", "procurement", "payments", "records", "audit"),
+    required_capabilities=(
+        CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
+        CAPABILITY_AUTH_PERMISSION_EVALUATOR,
+    ),
+    permissions=(
+        _permission(READ_SCOPE, "View XRechnung validation", "Read validation profiles, outcomes, and non-secret diagnostics."),
+        _permission(EXECUTE_SCOPE, "Validate inbound XRechnung", "Run a pinned local KoSIT profile against an inbound invoice."),
+        _permission(HANDOFF_SCOPE, "Handoff validated XRechnung", "Create a digest-bound handoff only from technically complete valid evidence."),
+    ),
+    role_templates=(
+        RoleTemplate(
+            slug="xrechnung_processor",
+            name="XRechnung processor",
+            description="Validate inbound invoices and create governed valid-invoice handoffs.",
+            permissions=(READ_SCOPE, EXECUTE_SCOPE, HANDOFF_SCOPE),
+        ),
+        RoleTemplate(
+            slug="xrechnung_reviewer",
+            name="XRechnung reviewer",
+            description="Inspect validation profiles, results, and diagnostics without creating effects.",
+            permissions=(READ_SCOPE,),
+        ),
+    ),
+    documentation=(
+        DocumentationTopic(
+            id="xrechnung.inbound-validation",
+            title="Validate and hand off an inbound XRechnung",
+            summary="Verify safe invoice XML with an exact locally pinned KoSIT engine and rule tree, then create a digest-bound handoff only from complete valid evidence.",
+            body=(
+                "An administrator must approve an exact XRechnung version, KoSIT validator release, configuration release, configuration-tree digest, engine digest, scenario file, and minimum validation-step count. A deployment-owned allow-list records approval identity, decision reference, receive-time window, suspension or retirement state, and an optional explicit default; GovOPlaN selects no version itself. The connector verifies those artifacts before each run and never downloads rules while processing an invoice. It rejects dangerous XML before invoking Java. Runner exit, bounded output, technical error markers, VARL identity, matched scenario, configured validation-step count, validity flag, and exactly one accept or reject assessment are checked independently. Technical failure or partial reports yield unknown conformance and can never be handed off. A valid handoff binds tenant, source reference, invoice digest, report digest, and profile digest for an owning Procurement or Payments workflow."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "operator", "module_admin", "auditor"),
+            related_modules=("files", "procurement", "payments", "records", "audit"),
+            conditions=(
+                DocumentationCondition(
+                    any_scopes=(READ_SCOPE, EXECUTE_SCOPE, HANDOFF_SCOPE),
+                ),
+            ),
+            links=(
+                DocumentationLink(
+                    label="Inbound validation profile",
+                    href="docs/INBOUND_VALIDATION.md",
+                    kind="repository",
+                ),
+            ),
+            translations={
+                "de": {
+                    "title": "Eingehende XRechnung prüfen und übergeben",
+                    "summary": "Sicheres Rechnungs-XML mit exakt lokal festgelegter KoSIT-Engine und Regelstruktur prüfen und nur aus vollständigem gültigem Nachweis eine prüfsummengebundene Übergabe erzeugen.",
+                    "body": "Administrierende müssen eine exakte XRechnungs-Version, KoSIT-Validator-Version, Konfigurationsversion, Prüfsumme der Konfigurationsstruktur, Engine-Prüfsumme, Szenariodatei und Mindestzahl der Prüfschritte freigeben. Eine installationsbezogene Positivliste erfasst Freigabestelle, Entscheidungsreferenz, Eingangszeitfenster, Sperrung oder Ausmusterung und einen optional ausdrücklich gesetzten Standard; GovOPlaN wählt selbst keine Version. Der Konnektor prüft die Artefakte vor jedem Lauf und lädt während der Rechnungsverarbeitung keine Regeln herunter. Gefährliches XML wird vor dem Java-Aufruf abgewiesen. Prozessende, begrenzte Ausgabe, technische Fehlermarker, VARL-Identität, erkanntes Szenario, konfigurierte Prüfschrittzahl, Gültigkeitskennzeichen und genau eine Annahme- oder Ablehnungsempfehlung werden unabhängig geprüft. Technische Fehler oder Teilberichte ergeben unbekannte Konformität und dürfen niemals übergeben werden. Eine gültige Übergabe bindet Mandant, Quellreferenz, Rechnungsprüfsumme, Berichtsprüfsumme und Profilprüfsumme für einen fachlich verantwortlichen Procurement- oder Payments-Ablauf.",
+                }
+            },
+            metadata={
+                "kind": "workflow",
+                "prerequisites": [
+                    "An exact KoSIT engine and self-contained XRechnung configuration have been installed locally.",
+                    "The administrator has approved all artifact digests and the expected validation-step count.",
+                    "The profile is approved for the invoice receive time in the deployment allow-list.",
+                    "The owning invoice workflow supplies a stable tenant and source reference.",
+                ],
+                "steps": [
+                    "Verify the executable, engine digest, complete configuration-tree digest, and scenario location.",
+                    "Select an explicit allow-listed profile or the deployment's explicitly configured default.",
+                    "Reject unsafe or oversized XML, then run the pinned profile with bounded time and output.",
+                    "Check technical completeness separately from the report's semantic validity.",
+                    "Create a handoff only when the report is complete, formally valid, and accepts the invoice.",
+                ],
+                "limitations": [
+                    "No XRechnung/configuration version is activated by default in this release.",
+                    "The connector creates a handoff contract but does not own payable approval or booking.",
+                    "A syntactically valid report is not trusted when runner output signals a technical failure.",
+                    "Runner stdout and stderr share a fixed 2 MiB capture limit, including one separator byte; the limit is enforced while the validator runs.",
+                    "Reports are limited to 16 MiB and read with at most one extra byte to detect overflow. Pipe capture requires POSIX; deployment resource limits and descendant-process isolation remain separate.",
+                ],
+                "consequences": [
+                    "Changing any pinned artifact changes the profile digest and requires a new validation.",
+                    "Invalid invoices retain diagnostics but cannot enter the valid-invoice handoff.",
+                    "Technical failure produces unknown conformance, never a semantic rejection or acceptance.",
+                    "Exceeding the output limit stops and reaps the validator and reports technical failure with unknown conformance; operators should check the approved profile using synthetic inputs before retrying.",
+                ],
+                "verification": "Confirm artifact and profile digests, technical outcome, validation-step count, formal validity, VARL assessment, invoice digest, and report digest before following the handoff reference.",
+            },
+            structured_translation_version="1",
+            structured_translations={
+                "de": {
+                    "prerequisites": [
+                        "Eine exakte KoSIT-Engine und eine eigenständige XRechnung-Konfiguration sind lokal installiert.",
+                        "Die Administration hat alle Artefaktprüfsummen und die erwartete Zahl der Prüfschritte freigegeben.",
+                        "Das Profil ist in der installationsbezogenen Positivliste für den Eingangszeitpunkt der Rechnung freigegeben.",
+                        "Der fachlich verantwortliche Rechnungsablauf liefert einen stabilen Mandanten- und Quellverweis.",
+                    ],
+                    "steps": [
+                        "Ausführbare Datei, Engine-Prüfsumme, vollständige Prüfsumme der Konfigurationsstruktur und Szenariopfad prüfen.",
+                        "Ein ausdrückliches Positivlistenprofil oder den ausdrücklich konfigurierten Installationsstandard auswählen.",
+                        "Unsicheres oder übergroßes XML abweisen und anschließend das festgelegte Profil mit begrenzter Zeit und Ausgabe ausführen.",
+                        "Technische Vollständigkeit getrennt von der semantischen Gültigkeit des Berichts prüfen.",
+                        "Nur bei vollständigem, formal gültigem und die Rechnung annehmendem Bericht eine Übergabe erzeugen.",
+                    ],
+                    "limitations": [
+                        "Diese Version aktiviert standardmäßig keine XRechnung- oder Konfigurationsversion.",
+                        "Der Konnektor erzeugt einen Übergabevertrag, führt aber weder Freigabe der Verbindlichkeit noch Buchung.",
+                        "Ein syntaktisch gültiger Bericht ist nicht vertrauenswürdig, wenn die Runner-Ausgabe einen technischen Fehler meldet.",
+                        "Standardausgabe und Standardfehlerausgabe teilen eine feste Erfassungsgrenze von 2 MiB einschließlich eines Trennbytes; die Grenze wird während des Validatorlaufs durchgesetzt.",
+                        "Berichte sind auf 16 MiB begrenzt und werden mit höchstens einem zusätzlichen Byte zur Erkennung einer Überschreitung gelesen. Die Pipe-Erfassung erfordert POSIX; Ressourcengrenzen der Installation und die Isolation nachgelagerter Prozesse bleiben getrennt.",
+                    ],
+                    "consequences": [
+                        "Jede Änderung eines festgelegten Artefakts ändert die Profilprüfsumme und erfordert eine neue Validierung.",
+                        "Ungültige Rechnungen bewahren Diagnosen, dürfen aber nicht in die gültige Rechnungsübergabe gelangen.",
+                        "Technischer Fehler ergibt unbekannte Konformität, niemals semantische Ablehnung oder Annahme.",
+                        "Bei Überschreitung der Ausgabegrenze wird der Validator beendet und sein Prozessabschluss abgewartet; das Ergebnis ist ein technischer Fehler mit unbekannter Konformität. Betreibende sollten das freigegebene Profil vor einem erneuten Versuch mit synthetischen Eingaben prüfen.",
+                    ],
+                    "verification": "Vor dem Folgen des Übergabeverweises Artefakt- und Profilprüfsummen, technisches Ergebnis, Prüfschrittzahl, formale Gültigkeit, VARL-Bewertung, Rechnungsprüfsumme und Berichtsprüfsumme bestätigen.",
+                }
+            },
+            order=100,
+        ),
+        DocumentationTopic(
+            id="xrechnung.reference.validation-profile-and-handoff",
+            title="XRechnung validation-profile and handoff consequences",
+            summary=(
+                "Understand receive-time profile selection, immutable artifact approval, "
+                "technical uncertainty, and the evidence boundary of a valid-invoice handoff."
+            ),
+            body=(
+                "A validation profile names exact engine, configuration, scenario, and artifact "
+                "digests. Its approval has an actor, decision reference, receive-time window, and "
+                "approved, suspended, or retired state. Selection uses the invoice receive time and "
+                "never infers a product version. Unknown, not-yet-valid, expired, suspended, retired, "
+                "or digest-mismatched profiles fail closed. Validation evidence distinguishes runner "
+                "completion, report completeness, formal validity, and semantic assessment. Stdout and "
+                "stderr share a fixed 2 MiB capture limit, including one separator byte, enforced during "
+                "execution. Overflow stops and reaps the validator; the result is technical failure "
+                "with unknown conformance and no handoff, even if a report looks valid. Reports are "
+                "read with a 16 MiB limit plus one probe byte; oversized reports cannot supply trusted "
+                "validation evidence. A handoff "
+                "binds the exact invoice, report, and profile digests but does not approve a payable, "
+                "book an invoice, retain the source file, or replace Records custody."
+            ),
+            layer="available",
+            documentation_types=("admin", "user"),
+            audience=("user", "operator", "module_admin", "auditor"),
+            related_modules=("files", "procurement", "payments", "records", "audit"),
+            links=(
+                DocumentationLink(
+                    label="Inbound validation profile",
+                    href="docs/INBOUND_VALIDATION.md",
+                    kind="repository",
+                ),
+            ),
+            translations={
+                "de": {
+                    "title": "Folgen von XRechnung-Prüfprofil und Übergabe",
+                    "summary": (
+                        "Profilauswahl nach Eingangszeit, unveränderliche Artefaktfreigabe, technische Ungewissheit und die Nachweisgrenze einer gültigen Rechnungsübergabe verstehen."
+                    ),
+                    "body": (
+                        "Ein Prüfprofil bezeichnet exakte Prüfsummen von Engine, Konfiguration, Szenario und Artefakten. Seine "
+                        "Freigabe enthält Akteur, Entscheidungsverweis, Eingangszeitfenster und den Zustand freigegeben, gesperrt "
+                        "oder ausgemustert. Die Auswahl verwendet den Eingangszeitpunkt der Rechnung und leitet niemals eine "
+                        "Produktversion ab. Unbekannte, noch nicht gültige, abgelaufene, gesperrte, ausgemusterte oder bei der "
+                        "Prüfsumme abweichende Profile werden geschlossen abgewiesen. Der Validierungsnachweis unterscheidet "
+                        "Runner-Abschluss, Berichtsvollständigkeit, formale Gültigkeit und semantische Bewertung. Standardausgabe "
+                        "und Standardfehlerausgabe teilen eine feste Erfassungsgrenze von 2 MiB einschließlich eines Trennbytes, "
+                        "die während der Ausführung durchgesetzt wird. Bei Überschreitung wird der Validator beendet und sein "
+                        "Prozessabschluss abgewartet; das Ergebnis ist ein technischer Fehler mit unbekannter Konformität ohne "
+                        "Übergabe, selbst wenn ein Bericht gültig erscheint. Berichte werden mit einer Grenze von 16 MiB "
+                        "zuzüglich eines Prüfbytes gelesen; übergroße Berichte liefern keinen vertrauenswürdigen "
+                        "Validierungsnachweis. Eine Übergabe "
+                        "bindet exakte Rechnungs-, Berichts- und Profilprüfsummen, genehmigt aber keine Verbindlichkeit, bucht keine "
+                        "Rechnung, bewahrt keine Quelldatei und ersetzt nicht die Verwahrung durch Records."
+                    ),
+                }
+            },
+            metadata={
+                "kind": "reference",
+                "consequence_classes": {
+                    "approve_profile": "Makes one exact verified artifact set eligible only in its receive-time window.",
+                    "suspend_or_retire_profile": "Blocks selection without rewriting prior validation evidence.",
+                    "fail_closed": "Treats unknown profile or technical outcome as untrusted rather than valid or invalid.",
+                    "exceed_output_limit": "Stops and reaps the validator with a technical-failure reason and blocks handoff regardless of report content.",
+                    "create_handoff": "Binds valid evidence for an owner workflow without approving, booking, or storing the invoice.",
+                },
+            },
+            structured_translation_version="1",
+            structured_translations={
+                "de": {
+                    "consequence_classes": {
+                        "approve_profile": "Macht genau einen geprüften Artefaktsatz ausschließlich in seinem Eingangszeitfenster auswählbar.",
+                        "suspend_or_retire_profile": "Sperrt die Auswahl, ohne frühere Validierungsnachweise umzuschreiben.",
+                        "fail_closed": "Behandelt unbekanntes Profil oder technisches Ergebnis als nicht vertrauenswürdig statt als gültig oder ungültig.",
+                        "exceed_output_limit": "Beendet den Validator, wartet seinen Prozessabschluss ab, meldet einen technischen Fehler und sperrt die Übergabe unabhängig vom Berichtsinhalt.",
+                        "create_handoff": "Bindet gültige Nachweise für einen Eigentümerablauf, ohne die Rechnung zu genehmigen, zu buchen oder zu speichern.",
+                    }
+                }
+            },
+            order=110,
+        ),
+    ),
+    architecture=declared_module_architecture(
+        layer="data_reporting_integration",
+        kind="integration",
+        maturity="vertical_slice",
+        documentation_ref="docs/INBOUND_VALIDATION.md",
+        test_ref="tests/test_validation.py",
+        known_limits=("A deployment-approved XRechnung and KoSIT configuration profile is required; none is activated by default.",),
+        owned_concepts=("XRechnung validation profile", "validation result", "validated invoice handoff"),
+        non_owned_concepts=("invoice payable", "procurement approval", "booking status", "invoice file storage"),
+        recovery_docs=("docs/INBOUND_VALIDATION.md",),
+        security_docs=("docs/INBOUND_VALIDATION.md",),
+        operations_docs=("docs/INBOUND_VALIDATION.md",),
+    ),
+)
+
+
+def get_manifest() -> ModuleManifest:
+    return manifest

src/govoplan_xrechnung/backend/profiles.py

@@ -0,0 +1,142 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Literal
+
+from govoplan_xrechnung.backend.validation import (
+    KoSITValidationProfile,
+    XRechnungValidationError,
+    verify_profile,
+)
+
+
+ProfileApprovalStatus = Literal["approved", "suspended", "retired"]
+
+
+@dataclass(frozen=True, slots=True)
+class KoSITProfileApproval:
+    """A reviewed deployment decision bound to exact installed artifacts."""
+
+    profile: KoSITValidationProfile
+    profile_sha256: str
+    status: ProfileApprovalStatus
+    approved_at: datetime
+    approved_by: str
+    decision_reference: str
+    accept_from: datetime
+    accept_until: datetime | None = None
+
+    def __post_init__(self) -> None:
+        if self.status not in {"approved", "suspended", "retired"}:
+            raise ValueError("XRechnung profile approval status is unsupported.")
+        for name in ("approved_by", "decision_reference"):
+            value = str(getattr(self, name) or "").strip()
+            if not value or len(value) > 255 or any(ord(char) < 32 for char in value):
+                raise ValueError(
+                    f"XRechnung {name.replace('_', ' ')} is required and bounded."
+                )
+            object.__setattr__(self, name, value)
+        for name in ("approved_at", "accept_from"):
+            _aware(getattr(self, name), name)
+        if self.accept_until is not None:
+            _aware(self.accept_until, "accept_until")
+            if self.accept_until <= self.accept_from:
+                raise ValueError("XRechnung accept_until must be later than accept_from.")
+        normalized_digest = str(self.profile_sha256 or "").strip().lower()
+        if len(normalized_digest) != 64 or any(
+            character not in "0123456789abcdef" for character in normalized_digest
+        ):
+            raise ValueError("XRechnung approved profile digest must be SHA-256.")
+        object.__setattr__(self, "profile_sha256", normalized_digest)
+
+    @classmethod
+    def approve(
+        cls,
+        profile: KoSITValidationProfile,
+        *,
+        approved_at: datetime,
+        approved_by: str,
+        decision_reference: str,
+        accept_from: datetime,
+        accept_until: datetime | None = None,
+    ) -> KoSITProfileApproval:
+        return cls(
+            profile=profile,
+            profile_sha256=verify_profile(profile),
+            status="approved",
+            approved_at=approved_at,
+            approved_by=approved_by,
+            decision_reference=decision_reference,
+            accept_from=accept_from,
+            accept_until=accept_until,
+        )
+
+    def accepts(self, received_at: datetime) -> bool:
+        _aware(received_at, "invoice received_at")
+        return (
+            self.status == "approved"
+            and received_at >= self.accept_from
+            and (self.accept_until is None or received_at < self.accept_until)
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class KoSITProfileRegistry:
+    """Deployment-owned allow-list with an optional, explicit default profile."""
+
+    approvals: tuple[KoSITProfileApproval, ...]
+    default_profile_id: str | None = None
+
+    def __post_init__(self) -> None:
+        if not self.approvals:
+            raise ValueError("XRechnung profile registry requires at least one reviewed entry.")
+        profile_ids = [item.profile.profile_id for item in self.approvals]
+        if len(profile_ids) != len(set(profile_ids)):
+            raise ValueError("XRechnung profile registry contains duplicate profile ids.")
+        profile_digests = [item.profile_sha256 for item in self.approvals]
+        if len(profile_digests) != len(set(profile_digests)):
+            raise ValueError("XRechnung profile registry contains duplicate artifact profiles.")
+        if self.default_profile_id is not None:
+            normalized = str(self.default_profile_id or "").strip()
+            if normalized not in profile_ids:
+                raise ValueError("XRechnung default profile is not present in the allow-list.")
+            object.__setattr__(self, "default_profile_id", normalized)
+
+    def select(
+        self,
+        *,
+        received_at: datetime,
+        profile_id: str | None = None,
+    ) -> KoSITValidationProfile:
+        selected_id = str(profile_id or self.default_profile_id or "").strip()
+        if not selected_id:
+            raise XRechnungValidationError(
+                "No XRechnung profile was selected and the deployment has no default."
+            )
+        approval = next(
+            (item for item in self.approvals if item.profile.profile_id == selected_id),
+            None,
+        )
+        if approval is None:
+            raise XRechnungValidationError(
+                "The selected XRechnung profile is not in the deployment allow-list."
+            )
+        if not approval.accepts(received_at):
+            raise XRechnungValidationError(
+                "The selected XRechnung profile is not approved for the invoice receive time."
+            )
+        observed_digest = verify_profile(approval.profile)
+        if observed_digest != approval.profile_sha256:
+            raise XRechnungValidationError(
+                "The installed XRechnung profile no longer matches its approval evidence."
+            )
+        return approval.profile
+
+
+def _aware(value: datetime, label: str) -> None:
+    if value.tzinfo is None or value.utcoffset() is None:
+        raise ValueError(f"XRechnung {label.replace('_', ' ')} must be timezone-aware.")
+
+
+__all__ = ["KoSITProfileApproval", "KoSITProfileRegistry", "ProfileApprovalStatus"]

src/govoplan_xrechnung/backend/validation.py

@@ -0,0 +1,493 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import selectors
+import subprocess
+import tempfile
+import time
+from typing import Literal
+
+from defusedxml import ElementTree
+
+
+VARL_NAMESPACE = "http://www.xoev.de/de/validator/varl/1"
+MAX_INVOICE_BYTES = 20 * 1024 * 1024
+MAX_RUNNER_OUTPUT_BYTES = 2 * 1024 * 1024
+MAX_REPORT_BYTES = 16 * 1024 * 1024
+_TECHNICAL_OUTPUT = re.compile(
+    rb"(?im)(?:^|\n)\s*(?:\[[^\]]*\]\s*)?(?:ERROR|FATAL)\b|\b(?:Exception|StackOverflowError|OutOfMemoryError)\b"
+)
+
+TechnicalOutcome = Literal["complete", "failed", "incomplete"]
+ConformanceOutcome = Literal["valid", "invalid", "unknown"]
+AssessmentOutcome = Literal["accept", "reject", "unknown"]
+
+
+class XRechnungValidationError(RuntimeError):
+    """Stable validation or configuration error without invoice contents."""
+
+
+class _RunnerOutputLimitExceeded(Exception):
+    """The shared stdout/stderr budget was exhausted during execution."""
+
+
+@dataclass(frozen=True, slots=True)
+class KoSITValidationProfile:
+    """Exact locally installed validator and rule-set selection."""
+
+    profile_id: str
+    xrechnung_version: str
+    validator_version: str
+    configuration_version: str
+    java_executable: Path
+    validator_jar: Path
+    validator_jar_sha256: str
+    configuration_root: Path
+    configuration_tree_sha256: str
+    scenarios_file: Path
+    minimum_validation_steps: int
+
+    def __post_init__(self) -> None:
+        for name in (
+            "profile_id",
+            "xrechnung_version",
+            "validator_version",
+            "configuration_version",
+        ):
+            value = str(getattr(self, name) or "").strip()
+            if not value or len(value) > 255:
+                raise ValueError(f"XRechnung {name.replace('_', ' ')} is required and limited to 255 characters.")
+            object.__setattr__(self, name, value)
+        for name in ("validator_jar_sha256", "configuration_tree_sha256"):
+            object.__setattr__(self, name, _sha256_text(getattr(self, name), name))
+        for name in ("java_executable", "validator_jar", "configuration_root", "scenarios_file"):
+            path = Path(getattr(self, name))
+            if not path.is_absolute():
+                raise ValueError(f"XRechnung {name.replace('_', ' ')} must be an absolute path.")
+            object.__setattr__(self, name, path)
+        if self.minimum_validation_steps < 1:
+            raise ValueError("XRechnung minimum_validation_steps must be positive.")
+
+
+@dataclass(frozen=True, slots=True)
+class InboundInvoice:
+    tenant_id: str
+    source_reference: str
+    document: bytes
+    received_at: datetime
+
+    def __post_init__(self) -> None:
+        if not self.tenant_id.strip() or not self.source_reference.strip():
+            raise ValueError("Inbound invoices require tenant and source references.")
+        if not self.document or len(self.document) > MAX_INVOICE_BYTES:
+            raise ValueError(f"Inbound invoice XML must contain 1 to {MAX_INVOICE_BYTES} bytes.")
+
+    @property
+    def document_sha256(self) -> str:
+        return hashlib.sha256(self.document).hexdigest()
+
+
+@dataclass(frozen=True, slots=True)
+class ValidationDiagnostic:
+    level: str
+    code: str | None
+    text: str
+
+
+@dataclass(frozen=True, slots=True)
+class XRechnungValidationResult:
+    profile_id: str
+    profile_sha256: str
+    document_sha256: str
+    technical_outcome: TechnicalOutcome
+    conformance: ConformanceOutcome
+    assessment: AssessmentOutcome
+    observed_at: datetime
+    report_sha256: str | None
+    validation_step_count: int
+    diagnostics: tuple[ValidationDiagnostic, ...]
+    technical_reason: str | None = None
+
+    @property
+    def handoff_allowed(self) -> bool:
+        return (
+            self.technical_outcome == "complete"
+            and self.conformance == "valid"
+            and self.assessment == "accept"
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ValidatedInvoiceHandoff:
+    tenant_id: str
+    source_reference: str
+    document_sha256: str
+    report_sha256: str
+    validation_profile_id: str
+    validation_profile_sha256: str
+    validated_at: datetime
+    handoff_sha256: str
+
+
+class KoSITValidator:
+    def __init__(self, profile: KoSITValidationProfile, *, timeout_seconds: int = 120) -> None:
+        if not 1 <= timeout_seconds <= 600:
+            raise ValueError("KoSIT timeout must be between 1 and 600 seconds.")
+        self.profile = profile
+        self.timeout_seconds = timeout_seconds
+
+    def validate(self, invoice: InboundInvoice) -> XRechnungValidationResult:
+        profile_sha256 = verify_profile(self.profile)
+        _verify_safe_xml(invoice.document)
+        with tempfile.TemporaryDirectory(prefix="govoplan-xrechnung-") as temporary:
+            workdir = Path(temporary)
+            input_file = workdir / "invoice.xml"
+            input_file.write_bytes(invoice.document)
+            command = [
+                str(self.profile.java_executable),
+                "-jar",
+                str(self.profile.validator_jar),
+                "-s",
+                str(self.profile.scenarios_file),
+                "-r",
+                str(workdir),
+                "-h",
+                str(input_file),
+            ]
+            env = {"LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "TZ": "UTC"}
+            if os.environ.get("JAVA_HOME"):
+                env["JAVA_HOME"] = os.environ["JAVA_HOME"]
+            try:
+                completed = _run_with_bounded_output(
+                    command,
+                    cwd=self.profile.configuration_root,
+                    env=env,
+                    timeout=self.timeout_seconds,
+                )
+            except subprocess.TimeoutExpired:
+                return _technical_result(
+                    self.profile,
+                    profile_sha256,
+                    invoice.document_sha256,
+                    "failed",
+                    "KoSIT validation timed out; no handoff is allowed.",
+                )
+            except _RunnerOutputLimitExceeded:
+                return _technical_result(
+                    self.profile,
+                    profile_sha256,
+                    invoice.document_sha256,
+                    "failed",
+                    "KoSIT runner output exceeded the safety limit.",
+                )
+            output = completed.stdout + b"\n" + completed.stderr
+            report_path = workdir / "invoice-report.xml"
+            report = None
+            if report_path.is_file():
+                with report_path.open("rb") as report_file:
+                    report = report_file.read(MAX_REPORT_BYTES + 1)
+            return interpret_kosit_result(
+                profile=self.profile,
+                profile_sha256=profile_sha256,
+                document_sha256=invoice.document_sha256,
+                return_code=completed.returncode,
+                runner_output=output,
+                report=report,
+            )
+
+
+def _run_with_bounded_output(
+    command: list[str], *, cwd: Path, env: dict[str, str], timeout: int
+) -> subprocess.CompletedProcess[bytes]:
+    """Drain both pipes within one budget, killing and reaping on any interruption."""
+    with subprocess.Popen(  # noqa: S603 - absolute executable, pinned artifacts, fixed argv; no shell.
+        command,
+        cwd=cwd,
+        env=env,
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        bufsize=0,
+    ) as process:
+        try:
+            deadline = time.monotonic() + timeout
+            stdout, stderr = bytearray(), bytearray()
+            # Preserve the existing combined-output cap, including its separator.
+            captured_bytes = 1
+            with selectors.DefaultSelector() as selector:
+                selector.register(process.stdout, selectors.EVENT_READ, stdout)
+                selector.register(process.stderr, selectors.EVENT_READ, stderr)
+                while selector.get_map():
+                    remaining = deadline - time.monotonic()
+                    if remaining <= 0:
+                        raise subprocess.TimeoutExpired(command, timeout)
+                    for key, _events in selector.select(remaining):
+                        # Read at most one byte beyond the budget to detect overflow.
+                        chunk = os.read(
+                            key.fd, min(64 * 1024, MAX_RUNNER_OUTPUT_BYTES - captured_bytes + 1)
+                        )
+                        if not chunk:
+                            selector.unregister(key.fileobj)
+                            continue
+                        captured_bytes += len(chunk)
+                        if captured_bytes > MAX_RUNNER_OUTPUT_BYTES:
+                            raise _RunnerOutputLimitExceeded
+                        key.data.extend(chunk)
+            return_code = process.wait(timeout=max(0, deadline - time.monotonic()))
+        except BaseException:
+            # Do not communicate() here: draining after termination would be unbounded.
+            process.kill()
+            process.wait()
+            raise
+    return subprocess.CompletedProcess(command, return_code, bytes(stdout), bytes(stderr))
+
+
+def verify_profile(profile: KoSITValidationProfile) -> str:
+    if not profile.java_executable.is_file() or not os.access(profile.java_executable, os.X_OK):
+        raise XRechnungValidationError("The pinned Java executable is unavailable or not executable.")
+    if not profile.validator_jar.is_file():
+        raise XRechnungValidationError("The pinned KoSIT validator JAR is unavailable.")
+    if _file_sha256(profile.validator_jar) != profile.validator_jar_sha256:
+        raise XRechnungValidationError("The KoSIT validator JAR digest does not match the approved profile.")
+    root = profile.configuration_root.resolve()
+    if not root.is_dir():
+        raise XRechnungValidationError("The pinned KoSIT configuration root is unavailable.")
+    scenarios = profile.scenarios_file.resolve()
+    if not scenarios.is_file() or not scenarios.is_relative_to(root):
+        raise XRechnungValidationError("The scenarios file must exist inside the pinned configuration root.")
+    observed_tree = configuration_tree_sha256(root)
+    if observed_tree != profile.configuration_tree_sha256:
+        raise XRechnungValidationError("The KoSIT configuration tree digest does not match the approved profile.")
+    descriptor = {
+        "profile_id": profile.profile_id,
+        "xrechnung_version": profile.xrechnung_version,
+        "validator_version": profile.validator_version,
+        "configuration_version": profile.configuration_version,
+        "validator_jar_sha256": profile.validator_jar_sha256,
+        "configuration_tree_sha256": profile.configuration_tree_sha256,
+        "scenarios_file": str(scenarios.relative_to(root)),
+        "minimum_validation_steps": profile.minimum_validation_steps,
+    }
+    return hashlib.sha256(_canonical_json(descriptor)).hexdigest()
+
+
+def configuration_tree_sha256(root: Path) -> str:
+    resolved = root.resolve()
+    if not resolved.is_dir():
+        raise XRechnungValidationError("KoSIT configuration tree is unavailable.")
+    digest = hashlib.sha256()
+    files = sorted(resolved.rglob("*"), key=lambda item: item.relative_to(resolved).as_posix())
+    for path in files:
+        if path.is_symlink():
+            raise XRechnungValidationError("KoSIT configuration trees must not contain symbolic links.")
+        if path.is_dir():
+            continue
+        if not path.is_file():
+            raise XRechnungValidationError("KoSIT configuration trees may contain only regular files and directories.")
+        relative = path.relative_to(resolved).as_posix().encode("utf-8")
+        digest.update(len(relative).to_bytes(4, "big"))
+        digest.update(relative)
+        digest.update(bytes.fromhex(_file_sha256(path)))
+    return digest.hexdigest()
+
+
+def interpret_kosit_result(
+    *,
+    profile: KoSITValidationProfile,
+    profile_sha256: str,
+    document_sha256: str,
+    return_code: int,
+    runner_output: bytes,
+    report: bytes | None,
+) -> XRechnungValidationResult:
+    if return_code != 0:
+        return _technical_result(
+            profile,
+            profile_sha256,
+            document_sha256,
+            "failed",
+            f"KoSIT runner exited with status {return_code}; report semantics are not trusted.",
+        )
+    if _TECHNICAL_OUTPUT.search(runner_output):
+        return _technical_result(
+            profile,
+            profile_sha256,
+            document_sha256,
+            "incomplete",
+            "KoSIT runner reported a technical error; a generated report is not trusted.",
+        )
+    if report is None or not report or len(report) > MAX_REPORT_BYTES:
+        return _technical_result(
+            profile,
+            profile_sha256,
+            document_sha256,
+            "incomplete",
+            "KoSIT did not produce a bounded XML report.",
+        )
+    report_sha256 = hashlib.sha256(report).hexdigest()
+    try:
+        root = ElementTree.fromstring(report)
+    except ElementTree.ParseError:
+        return _technical_result(
+            profile,
+            profile_sha256,
+            document_sha256,
+            "incomplete",
+            "KoSIT produced malformed report XML.",
+            report_sha256=report_sha256,
+        )
+    if root.tag != f"{{{VARL_NAMESPACE}}}report" or root.get("valid") not in {"true", "false"}:
+        return _technical_result(
+            profile,
+            profile_sha256,
+            document_sha256,
+            "incomplete",
+            "KoSIT report is missing the expected VARL report identity or validity flag.",
+            report_sha256=report_sha256,
+        )
+    scenario = root.find(f"{{{VARL_NAMESPACE}}}scenarioMatched")
+    steps = root.findall(f".//{{{VARL_NAMESPACE}}}validationStepResult")
+    assessment = root.find(f"{{{VARL_NAMESPACE}}}assessment")
+    accepts = [] if assessment is None else assessment.findall(f"{{{VARL_NAMESPACE}}}accept")
+    rejects = [] if assessment is None else assessment.findall(f"{{{VARL_NAMESPACE}}}reject")
+    steps_well_formed = all(item.get("valid") in {"true", "false"} for item in steps)
+    if (
+        scenario is None
+        or len(steps) < profile.minimum_validation_steps
+        or not steps_well_formed
+        or (len(accepts), len(rejects)) not in {(1, 0), (0, 1)}
+    ):
+        return _technical_result(
+            profile,
+            profile_sha256,
+            document_sha256,
+            "incomplete",
+            "KoSIT report is partial or lacks the configured validation-step and assessment evidence.",
+            report_sha256=report_sha256,
+            validation_step_count=len(steps),
+        )
+    accepted = len(accepts) == 1
+    valid_flag = root.get("valid") == "true"
+    diagnostics = tuple(
+        ValidationDiagnostic(
+            level=(item.get("level") or "unknown")[:50],
+            code=(item.get("code") or "")[:100] or None,
+            text=" ".join("".join(item.itertext()).split())[:1000],
+        )
+        for item in root.findall(f".//{{{VARL_NAMESPACE}}}message")[:1000]
+    )
+    return XRechnungValidationResult(
+        profile_id=profile.profile_id,
+        profile_sha256=profile_sha256,
+        document_sha256=document_sha256,
+        technical_outcome="complete",
+        conformance="valid" if valid_flag else "invalid",
+        assessment="accept" if accepted else "reject",
+        observed_at=datetime.now(UTC),
+        report_sha256=report_sha256,
+        validation_step_count=len(steps),
+        diagnostics=diagnostics,
+    )
+
+
+def create_validated_handoff(
+    invoice: InboundInvoice,
+    result: XRechnungValidationResult,
+) -> ValidatedInvoiceHandoff:
+    if result.document_sha256 != invoice.document_sha256:
+        raise XRechnungValidationError("Validation result belongs to another invoice document.")
+    if not result.handoff_allowed or not result.report_sha256:
+        raise XRechnungValidationError(
+            "Invoice handoff requires technically complete and semantically valid XRechnung evidence."
+        )
+    payload = {
+        "tenant_id": invoice.tenant_id,
+        "source_reference": invoice.source_reference,
+        "document_sha256": invoice.document_sha256,
+        "report_sha256": result.report_sha256,
+        "validation_profile_id": result.profile_id,
+        "validation_profile_sha256": result.profile_sha256,
+        "validated_at": result.observed_at.isoformat(),
+    }
+    return ValidatedInvoiceHandoff(
+        tenant_id=invoice.tenant_id,
+        source_reference=invoice.source_reference,
+        document_sha256=invoice.document_sha256,
+        report_sha256=result.report_sha256,
+        validation_profile_id=result.profile_id,
+        validation_profile_sha256=result.profile_sha256,
+        validated_at=result.observed_at,
+        handoff_sha256=hashlib.sha256(_canonical_json(payload)).hexdigest(),
+    )
+
+
+def _verify_safe_xml(document: bytes) -> None:
+    try:
+        ElementTree.fromstring(document)
+    except ElementTree.ParseError as exc:
+        raise XRechnungValidationError("Inbound invoice is not well-formed safe XML.") from exc
+
+
+def _technical_result(
+    profile: KoSITValidationProfile,
+    profile_sha256: str,
+    document_sha256: str,
+    outcome: Literal["failed", "incomplete"],
+    reason: str,
+    *,
+    report_sha256: str | None = None,
+    validation_step_count: int = 0,
+) -> XRechnungValidationResult:
+    return XRechnungValidationResult(
+        profile_id=profile.profile_id,
+        profile_sha256=profile_sha256,
+        document_sha256=document_sha256,
+        technical_outcome=outcome,
+        conformance="unknown",
+        assessment="unknown",
+        observed_at=datetime.now(UTC),
+        report_sha256=report_sha256,
+        validation_step_count=validation_step_count,
+        diagnostics=(),
+        technical_reason=reason,
+    )
+
+
+def _file_sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _sha256_text(value: str, label: str) -> str:
+    normalized = str(value or "").strip().lower().removeprefix("sha256:")
+    if len(normalized) != 64 or any(character not in "0123456789abcdef" for character in normalized):
+        raise ValueError(f"XRechnung {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("utf-8")
+
+
+__all__ = [
+    "InboundInvoice",
+    "KoSITValidationProfile",
+    "KoSITValidator",
+    "ValidatedInvoiceHandoff",
+    "ValidationDiagnostic",
+    "XRechnungValidationError",
+    "XRechnungValidationResult",
+    "configuration_tree_sha256",
+    "create_validated_handoff",
+    "interpret_kosit_result",
+    "verify_profile",
+]

src/govoplan_xrechnung/py.typed

@@ -0,0 +1 @@
+

tests/test_documentation.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from govoplan_core.core.modules import documentation_structured_translation_issues
+from govoplan_xrechnung.backend.manifest import manifest
+
+
+def test_xrechnung_german_workflow_and_reference_are_complete() -> None:
+    topics = {topic.id: topic for topic in manifest.documentation}
+    assert len(topics) == 2
+    assert topics["xrechnung.inbound-validation"].metadata["kind"] == "workflow"
+    assert topics["xrechnung.reference.validation-profile-and-handoff"].metadata["kind"] == "reference"
+    for topic in topics.values():
+        assert all(topic.translations["de"].get(key) for key in ("title", "summary", "body"))
+        assert topic.structured_translation_version == "1"
+        assert "de" in topic.structured_translations
+        assert documentation_structured_translation_issues(topic) == ()

tests/test_profiles.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from dataclasses import replace
+from datetime import UTC, datetime
+import hashlib
+from pathlib import Path
+
+import pytest
+
+from govoplan_xrechnung.backend.profiles import KoSITProfileApproval, KoSITProfileRegistry
+from govoplan_xrechnung.backend.validation import (
+    KoSITValidationProfile,
+    XRechnungValidationError,
+    configuration_tree_sha256,
+)
+
+
+def _profile(tmp_path: Path, profile_id: str = "profile-a") -> KoSITValidationProfile:
+    root = tmp_path / profile_id
+    root.mkdir()
+    java = root / "java"
+    java.write_bytes(b"#!/bin/sh\n")
+    java.chmod(0o700)
+    jar = root / "validator.jar"
+    jar.write_bytes(b"validator")
+    config = root / "config"
+    config.mkdir()
+    scenarios = config / "scenarios.xml"
+    scenarios.write_text("<scenarios/>", encoding="utf-8")
+    return KoSITValidationProfile(
+        profile_id=profile_id,
+        xrechnung_version="configured-version",
+        validator_version="configured-validator",
+        configuration_version="configured-rules",
+        java_executable=java.resolve(),
+        validator_jar=jar.resolve(),
+        validator_jar_sha256=hashlib.sha256(jar.read_bytes()).hexdigest(),
+        configuration_root=config.resolve(),
+        configuration_tree_sha256=configuration_tree_sha256(config),
+        scenarios_file=scenarios.resolve(),
+        minimum_validation_steps=1,
+    )
+
+
+def _approval(tmp_path: Path) -> KoSITProfileApproval:
+    return KoSITProfileApproval.approve(
+        _profile(tmp_path),
+        approved_at=datetime(2026, 8, 23, tzinfo=UTC),
+        approved_by="configuration-board",
+        decision_reference="decision:xrechnung-profile-a",
+        accept_from=datetime(2026, 9, 1, tzinfo=UTC),
+        accept_until=datetime(2027, 1, 1, tzinfo=UTC),
+    )
+
+
+def test_registry_requires_explicit_selection_without_default(tmp_path: Path) -> None:
+    registry = KoSITProfileRegistry((_approval(tmp_path),))
+
+    with pytest.raises(XRechnungValidationError, match="no default"):
+        registry.select(received_at=datetime(2026, 10, 1, tzinfo=UTC))
+
+
+def test_explicit_or_configured_default_selects_only_inside_window(tmp_path: Path) -> None:
+    approval = _approval(tmp_path)
+    registry = KoSITProfileRegistry(
+        (approval,),
+        default_profile_id=approval.profile.profile_id,
+    )
+
+    selected = registry.select(received_at=datetime(2026, 10, 1, tzinfo=UTC))
+    assert selected.profile_id == "profile-a"
+
+    with pytest.raises(XRechnungValidationError, match="receive time"):
+        registry.select(received_at=datetime(2027, 1, 1, tzinfo=UTC))
+
+
+@pytest.mark.parametrize("status", ["suspended", "retired"])
+def test_non_approved_profile_cannot_be_selected(tmp_path: Path, status: str) -> None:
+    approval = replace(_approval(tmp_path), status=status)
+    registry = KoSITProfileRegistry((approval,), default_profile_id="profile-a")
+
+    with pytest.raises(XRechnungValidationError, match="not approved"):
+        registry.select(received_at=datetime(2026, 10, 1, tzinfo=UTC))
+
+
+def test_changed_artifact_is_rejected_after_approval(tmp_path: Path) -> None:
+    approval = _approval(tmp_path)
+    registry = KoSITProfileRegistry((approval,), default_profile_id="profile-a")
+    approval.profile.validator_jar.write_bytes(b"changed")
+
+    with pytest.raises(XRechnungValidationError, match="digest"):
+        registry.select(received_at=datetime(2026, 10, 1, tzinfo=UTC))
+
+
+def test_default_must_reference_allow_list(tmp_path: Path) -> None:
+    with pytest.raises(ValueError, match="not present"):
+        KoSITProfileRegistry((_approval(tmp_path),), default_profile_id="missing")

tests/test_validation.py

@@ -0,0 +1,391 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+from datetime import UTC, datetime
+import hashlib
+import os
+from pathlib import Path
+import subprocess
+import sys
+import time
+
+import pytest
+
+from govoplan_xrechnung.backend import validation
+from govoplan_xrechnung.backend.manifest import get_manifest
+from govoplan_xrechnung.backend.validation import (
+    InboundInvoice,
+    KoSITValidationProfile,
+    KoSITValidator,
+    MAX_RUNNER_OUTPUT_BYTES,
+    XRechnungValidationError,
+    configuration_tree_sha256,
+    create_validated_handoff,
+    interpret_kosit_result,
+    verify_profile,
+)
+
+
+def _profile(tmp_path: Path, *, minimum_steps: int = 2) -> KoSITValidationProfile:
+    java = tmp_path / "java"
+    java.write_bytes(b"#!/bin/sh\n")
+    java.chmod(0o700)
+    jar = tmp_path / "validator.jar"
+    jar.write_bytes(b"pinned validator")
+    config = tmp_path / "config"
+    config.mkdir()
+    scenarios = config / "scenarios.xml"
+    scenarios.write_text("<scenarios/>", encoding="utf-8")
+    resources = config / "resources"
+    resources.mkdir()
+    (resources / "rules.xsl").write_text("<stylesheet/>", encoding="utf-8")
+    return KoSITValidationProfile(
+        profile_id="xrechnung-explicit-test",
+        xrechnung_version="explicit-test-only",
+        validator_version="validator-test",
+        configuration_version="configuration-test",
+        java_executable=java.resolve(),
+        validator_jar=jar.resolve(),
+        validator_jar_sha256=hashlib.sha256(jar.read_bytes()).hexdigest(),
+        configuration_root=config.resolve(),
+        configuration_tree_sha256=configuration_tree_sha256(config),
+        scenarios_file=scenarios.resolve(),
+        minimum_validation_steps=minimum_steps,
+    )
+
+
+def _report(*, valid: bool, step_count: int = 2) -> bytes:
+    assessment = "accept" if valid else "reject"
+    flag = "true" if valid else "false"
+    steps = "".join(
+        f'<rep:validationStepResult id="step-{index}" valid="{flag}" />'
+        for index in range(step_count)
+    )
+    return (
+        f'<rep:report xmlns:rep="http://www.xoev.de/de/validator/varl/1" valid="{flag}" varlVersion="1.0.0">'
+        f"<rep:scenarioMatched>{steps}</rep:scenarioMatched>"
+        f"<rep:assessment><rep:{assessment}/></rep:assessment>"
+        "</rep:report>"
+    ).encode()
+
+
+def _invoice() -> InboundInvoice:
+    return InboundInvoice(
+        tenant_id="tenant-a",
+        source_reference="mail:message-1:attachment-1",
+        document=b"<Invoice/>",
+        received_at=datetime(2026, 8, 23, tzinfo=UTC),
+    )
+
+
+def _runner_profile(tmp_path: Path, script: str) -> KoSITValidationProfile:
+    """Install a synthetic executable; never invoke Java or a real invoice validator."""
+    profile = _profile(tmp_path)
+    profile.java_executable.write_text(
+        f"#!{sys.executable}\n"
+        "import os, sys, time\n"
+        "from pathlib import Path\n"
+        f"Path(sys.argv[-1]).with_name('invoice-report.xml').write_bytes({_report(valid=True)!r})\n"
+        + script,
+        encoding="utf-8",
+    )
+    return profile
+
+
+@pytest.fixture
+def runner_processes(monkeypatch: pytest.MonkeyPatch) -> Iterator[list[subprocess.Popen[bytes]]]:
+    processes: list[subprocess.Popen[bytes]] = []
+    real_popen = subprocess.Popen
+
+    def start(*args, **kwargs):
+        process = real_popen(*args, **kwargs)
+        processes.append(process)
+        return process
+
+    monkeypatch.setattr(validation.subprocess, "Popen", start)
+    try:
+        yield processes
+    finally:
+        for process in processes:
+            if process.poll() is None:
+                process.kill()
+                process.wait()
+
+
+def _assert_runner_reaped(processes: list[subprocess.Popen[bytes]]) -> None:
+    assert len(processes) == 1
+    process = processes[0]
+    assert process.returncode is not None
+    assert process.stdout is not None and process.stdout.closed
+    assert process.stderr is not None and process.stderr.closed
+    with pytest.raises(ChildProcessError):
+        os.waitpid(process.pid, os.WNOHANG)
+
+
+@pytest.fixture
+def report_read_sizes(monkeypatch: pytest.MonkeyPatch) -> list[int]:
+    sizes: list[int] = []
+    real_open = Path.open
+
+    class ReportReader:
+        def __init__(self, stream):
+            self.stream = stream
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_args):
+            self.stream.close()
+
+        def read(self, size=-1):
+            sizes.append(size)
+            assert size == validation.MAX_REPORT_BYTES + 1
+            return self.stream.read(size)
+
+    def open_file(path, mode="r", *args, **kwargs):
+        stream = real_open(path, mode, *args, **kwargs)
+        if path.name == "invoice-report.xml" and mode == "rb":
+            return ReportReader(stream)
+        return stream
+
+    monkeypatch.setattr(Path, "open", open_file)
+    return sizes
+
+
+@pytest.mark.parametrize("stream", ["stdout", "stderr", "both"])
+def test_noisy_runner_is_stopped_at_shared_output_limit(
+    tmp_path: Path,
+    runner_processes: list[subprocess.Popen[bytes]],
+    report_read_sizes: list[int],
+    stream: str,
+) -> None:
+    count = MAX_RUNNER_OUTPUT_BYTES
+    writes = {
+        "stdout": f"sys.stdout.buffer.write(b'x' * {count})\nsys.stdout.flush()\n",
+        "stderr": f"sys.stderr.buffer.write(b'x' * {count})\nsys.stderr.flush()\n",
+        "both": (
+            f"sys.stdout.buffer.write(b'x' * {count // 2})\nsys.stdout.flush()\n"
+            f"sys.stderr.buffer.write(b'x' * {count // 2})\nsys.stderr.flush()\n"
+        ),
+    }
+    profile = _runner_profile(tmp_path, writes[stream] + "time.sleep(30)\n")
+
+    result = KoSITValidator(profile, timeout_seconds=2).validate(_invoice())
+
+    assert result.technical_outcome == "failed"
+    assert result.technical_reason == "KoSIT runner output exceeded the safety limit."
+    assert result.conformance == result.assessment == "unknown"
+    assert not result.handoff_allowed
+    assert result.report_sha256 is None
+    assert result.diagnostics == ()
+    assert report_read_sizes == []
+    _assert_runner_reaped(runner_processes)
+
+
+@pytest.mark.parametrize(("extra_bytes", "exit_code"), [(0, 0), (1, 0), (100_000, 0), (100_000, 7)])
+def test_report_reads_are_bounded_and_preserve_exact_limit(
+    tmp_path: Path,
+    runner_processes: list[subprocess.Popen[bytes]],
+    report_read_sizes: list[int],
+    monkeypatch: pytest.MonkeyPatch,
+    extra_bytes: int,
+    exit_code: int,
+) -> None:
+    report_limit = len(_report(valid=True))
+    monkeypatch.setattr(validation, "MAX_REPORT_BYTES", report_limit)
+    profile = _runner_profile(
+        tmp_path,
+        "with Path(sys.argv[-1]).with_name('invoice-report.xml').open('ab') as report:\n"
+        f"    report.write(b' ' * {extra_bytes})\n"
+        f"sys.exit({exit_code})\n",
+    )
+
+    result = KoSITValidator(profile, timeout_seconds=2).validate(_invoice())
+
+    assert report_read_sizes == [report_limit + 1]
+    if extra_bytes:
+        assert result.technical_outcome == ("failed" if exit_code else "incomplete")
+        assert result.technical_reason == (
+            f"KoSIT runner exited with status {exit_code}; report semantics are not trusted."
+            if exit_code else "KoSIT did not produce a bounded XML report."
+        )
+        assert result.conformance == result.assessment == "unknown"
+        assert result.report_sha256 is None
+        assert not result.handoff_allowed
+    else:
+        assert result.technical_outcome == "complete"
+        assert result.report_sha256 == hashlib.sha256(_report(valid=True)).hexdigest()
+        assert result.handoff_allowed
+    _assert_runner_reaped(runner_processes)
+
+
+@pytest.mark.parametrize("excess_bytes", [0, 1])
+def test_completed_runner_output_preserves_exact_combined_limit(
+    tmp_path: Path, runner_processes: list[subprocess.Popen[bytes]], excess_bytes: int
+) -> None:
+    stdout_size = MAX_RUNNER_OUTPUT_BYTES // 2
+    stderr_size = MAX_RUNNER_OUTPUT_BYTES - stdout_size - 1 + excess_bytes
+    profile = _runner_profile(
+        tmp_path,
+        f"sys.stdout.buffer.write(b'x' * {stdout_size})\n"
+        f"sys.stderr.buffer.write(b'x' * {stderr_size})\n",
+    )
+
+    result = KoSITValidator(profile, timeout_seconds=2).validate(_invoice())
+
+    assert result.technical_outcome == ("failed" if excess_bytes else "complete")
+    assert result.handoff_allowed is (excess_bytes == 0)
+    _assert_runner_reaped(runner_processes)
+
+
+@pytest.mark.parametrize("close_pipes", [False, True])
+def test_hanging_runner_times_out_and_is_reaped(
+    tmp_path: Path, runner_processes: list[subprocess.Popen[bytes]], close_pipes: bool
+) -> None:
+    script = "os.close(1)\nos.close(2)\n" if close_pipes else ""
+    profile = _runner_profile(tmp_path, script + "time.sleep(30)\n")
+    started = time.monotonic()
+
+    result = KoSITValidator(profile, timeout_seconds=1).validate(_invoice())
+
+    assert time.monotonic() - started < 5
+    assert result.technical_outcome == "failed"
+    assert result.technical_reason == "KoSIT validation timed out; no handoff is allowed."
+    assert result.conformance == result.assessment == "unknown"
+    assert not result.handoff_allowed
+    _assert_runner_reaped(runner_processes)
+
+
+@pytest.mark.parametrize(
+    ("script", "outcome", "reason"),
+    [
+        ("sys.stderr.write('synthetic failure')\nsys.exit(7)\n", "failed", "status 7"),
+        (
+            "sys.stdout.buffer.write(b'x' * 65532 + b'\\nERROR synthetic failure')\n",
+            "incomplete",
+            "technical error",
+        ),
+        ("sys.stderr.write('ERROR synthetic failure')\n", "incomplete", "technical error"),
+    ],
+)
+def test_runner_failure_and_technical_output_never_trust_valid_report(
+    tmp_path: Path,
+    runner_processes: list[subprocess.Popen[bytes]],
+    script: str,
+    outcome: str,
+    reason: str,
+) -> None:
+    profile = _runner_profile(tmp_path, script)
+
+    result = KoSITValidator(profile, timeout_seconds=2).validate(_invoice())
+
+    assert result.technical_outcome == outcome
+    assert reason in result.technical_reason
+    assert result.conformance == result.assessment == "unknown"
+    assert not result.handoff_allowed
+    _assert_runner_reaped(runner_processes)
+
+
+def test_runner_cancellation_kills_and_reaps_before_propagating(
+    tmp_path: Path,
+    runner_processes: list[subprocess.Popen[bytes]],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    profile = _runner_profile(tmp_path, "time.sleep(30)\n")
+
+    def cancel(_selector, _timeout):
+        raise KeyboardInterrupt
+
+    monkeypatch.setattr(validation.selectors.DefaultSelector, "select", cancel)
+    with pytest.raises(KeyboardInterrupt):
+        KoSITValidator(profile, timeout_seconds=2).validate(_invoice())
+
+    _assert_runner_reaped(runner_processes)
+
+
+def test_profile_verifies_exact_engine_and_complete_configuration_tree(tmp_path: Path) -> None:
+    profile = _profile(tmp_path)
+    first = verify_profile(profile)
+    assert len(first) == 64
+
+    (profile.configuration_root / "resources" / "rules.xsl").write_text("changed", encoding="utf-8")
+    with pytest.raises(XRechnungValidationError, match="tree digest"):
+        verify_profile(profile)
+
+
+def test_complete_valid_report_can_create_digest_bound_handoff(tmp_path: Path) -> None:
+    profile = _profile(tmp_path)
+    invoice = _invoice()
+    result = interpret_kosit_result(
+        profile=profile,
+        profile_sha256=verify_profile(profile),
+        document_sha256=invoice.document_sha256,
+        return_code=0,
+        runner_output=b"INFO validation completed",
+        report=_report(valid=True),
+    )
+
+    handoff = create_validated_handoff(invoice, result)
+
+    assert result.technical_outcome == "complete"
+    assert result.conformance == "valid"
+    assert result.assessment == "accept"
+    assert handoff.document_sha256 == invoice.document_sha256
+    assert len(handoff.handoff_sha256) == 64
+
+
+def test_semantically_invalid_report_is_complete_but_cannot_handoff(tmp_path: Path) -> None:
+    profile = _profile(tmp_path)
+    invoice = _invoice()
+    result = interpret_kosit_result(
+        profile=profile,
+        profile_sha256=verify_profile(profile),
+        document_sha256=invoice.document_sha256,
+        return_code=0,
+        runner_output=b"INFO validation completed",
+        report=_report(valid=False),
+    )
+
+    assert result.technical_outcome == "complete"
+    assert result.conformance == "invalid"
+    assert result.assessment == "reject"
+    with pytest.raises(XRechnungValidationError, match="technically complete"):
+        create_validated_handoff(invoice, result)
+
+
+@pytest.mark.parametrize(
+    ("return_code", "output", "report", "expected"),
+    [
+        (1, b"", _report(valid=True), "failed"),
+        (0, b"ERROR Transformation failed", _report(valid=True), "incomplete"),
+        (0, b"", _report(valid=True, step_count=1), "incomplete"),
+        (0, b"", None, "incomplete"),
+    ],
+)
+def test_technical_failures_never_trust_a_valid_looking_report(
+    tmp_path: Path,
+    return_code: int,
+    output: bytes,
+    report: bytes | None,
+    expected: str,
+) -> None:
+    profile = _profile(tmp_path)
+    result = interpret_kosit_result(
+        profile=profile,
+        profile_sha256=verify_profile(profile),
+        document_sha256="a" * 64,
+        return_code=return_code,
+        runner_output=output,
+        report=report,
+    )
+
+    assert result.technical_outcome == expected
+    assert result.conformance == "unknown"
+    assert result.assessment == "unknown"
+    assert result.handoff_allowed is False
+
+
+def test_manifest_does_not_select_an_active_standard_version() -> None:
+    manifest = get_manifest()
+    assert manifest.version == "0.1.21"
+    assert "none is activated by default" in manifest.architecture.known_limits[0].lower()