feat(inspections): INSP-001..004 Phase 1 generisches Pruefart-System additiv (Dual-Write, FAILED-Ampel)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV

arbeitskacheln/09_inspections.md

@@ -22,9 +22,35 @@ Details zu INSP-001 … INSP-006 (vollständig).
 - **Audit:** Änderung geloggt
 - **Akzeptanzkriterien:** mehrere Prüfarten anlegen, an AssetType koppelbar.
 - **Tests:** CRUD-Test.
-- **DoD:** mind. 3 unterschiedliche Prüfarten (UVV, Medizinprodukt, Elektro) anlegbar.
-  **Lücke:** MABEA hat nur zwei hartcodierte Prüftypen (generische Geräteprüfung +
-  Fahrzeug-HU/UVV als Sonderfall), keine generische Prüfart-Stammdatentabelle.
+- **Umsetzungsentscheidung 2026-09-10 (nach Vor-/Nachteils-Abwägung mit dem
+  Nutzer): additiver Phasenplan statt Big-Bang-Umbau.** Ein vollständiger
+  Ersatz der bestehenden `GeraetInstanz.status`/`pruefdatum`/`naechste_pruefung`-
+  und `Fahrzeugdetails.naechste_hu`/`naechste_uvv`-Felder hätte Dashboard-
+  Prüftermine, den Fristen-Dienst (FOUND-008) und ASSET-010 (Custody nutzt
+  `naechste_pruefung`) mitbetroffen — deutlich höheres Regressionsrisiko als
+  nötig für den geäußerten Bedarf. Stattdessen:
+  - **Phase 1 (umgesetzt 2026-09-10):** `pruefart`/`objekttyp_pruefart`/
+    `pruefung`-Tabellen additiv angelegt, bestehende Felder bleiben
+    unverändert bestehen und werden weiterhin von Dashboard/Fristen-Dienst/
+    ASSET-010 gelesen. `PATCH /geraete/{id}` schreibt bei gesetztem
+    `pruefdatum` zusätzlich (Dual-Write) einen `Pruefung`-Eintrag.
+  - **Phase 2 (nicht in dieser Session, eigener künftiger Schritt):**
+    Lesepfade (Dashboard-Prüftermine, Fristen-Dienst-Ampel, ASSET-010)
+    schrittweise auf die neue `pruefung`-Tabelle umstellen. Alte Felder
+    (`GeraetInstanz.pruefdatum`/`naechste_pruefung`,
+    `Fahrzeugdetails.naechste_hu`/`naechste_uvv`) erst entfernen, wenn ALLE
+    Lesepfade umgestellt sind UND ein Übergangszeitraum ohne Probleme lief —
+    explizit NICHT vorher anfassen.
+- **DoD:** ✅ Phase 1 umgesetzt (2026-09-10). `app/models/pruefung.py`
+  (`Pruefart`, `ObjekttypPruefart`, `Pruefung`), Migration 0045 mit Seed
+  (Gerätesicherheitsprüfung/HU/UVV) und Backfill aus bestehenden
+  `GeraetInstanz.pruefdatum`-Werten (Fahrzeug-HU/UVV NICHT zurückbefüllt —
+  `Fahrzeugdetails` speichert nur das nächste Fälligkeitsdatum, keine reale
+  Quelle für ein historisches `geprueft_am`). `GET/POST /pruefarten`,
+  `app/services/pruefung.py`. **Ursprüngliches DoD-Kriterium** (≥3 Prüfarten
+  anlegbar, an Objekttyp koppelbar) erfüllt über Seed-Daten +
+  `objekttyp_pruefart`-Tabelle (Koppel-UI folgt bei Bedarf in Phase 2, aktuell
+  nur über die Tabelle direkt befüllbar, kein Admin-UI-Screen dafür).
 
 ## INSP-002 — Prüfintervalle
 
@@ -45,9 +71,13 @@ Details zu INSP-001 … INSP-006 (vollständig).
 - **Audit:** Änderung geloggt
 - **Akzeptanzkriterien:** Intervall setzen, nächster Termin korrekt berechnet.
 - **Tests:** Berechnungstest (inkl. Monatsende-Sonderfall, z.B. 31.1. + 1 Monat).
-- **DoD:** deckt sich mit MABEA (`pruefintervall_monate` an Objektposition,
-  `_plus_monate()`-Hilfsfunktion bereits vorhanden) — bisher nur implizit „die eine
-  Geräteprüfung", nicht an eine generische Prüfart gekoppelt.
+- **DoD:** ✅ Phase 1 umgesetzt (2026-09-10) — `objekttyp_pruefart`
+  (objekttyp_id, pruefart_id, intervall_monate) als generische Stammdaten-
+  Kopplung angelegt (Tabelle existiert, noch ohne eigenes Admin-UI, siehe
+  INSP-001-DoD). Die tatsächliche Fälligkeits-BERECHNUNG bleibt in Phase 1
+  weiterhin `pruefintervall_monate` an `Objektposition` +
+  `_plus_monate()` (unverändert, kein Regressionsrisiko) — Umstellung der
+  Berechnung auf `objekttyp_pruefart` ist Teil von Phase 2.
 
 ## INSP-003 — Prüfdurchführung/-ergebnis
 
@@ -69,8 +99,19 @@ Details zu INSP-001 … INSP-006 (vollständig).
 - **Akzeptanzkriterien:** Prüfung erfassen aktualisiert nächsten Fälligkeitstermin
   korrekt.
 - **Tests:** Erfassungstest.
-- **DoD:** deckt sich mit MABEA (`GeraetInstanz.pruefdatum`/status, Fahrzeug-HU/UVV-Felder)
-  — aber ohne strukturiertes Ergebnis-/Prüfer-Feld, nur Statuswechsel.
+- **DoD:** ✅ Phase 1 umgesetzt (2026-09-10). `pruefung`-Tabelle (geprueft_am,
+  pruefer_id, ergebnis: bestanden/nicht_bestanden/mit_maengeln, bemerkung) —
+  ECHTE Historie, ein Eintrag pro Durchführung statt nur "letzter Status".
+  Dual-Write-Hook in `app/services/geraet_instanz.py:aendere_status()`: jedes
+  `PATCH /geraete/{id}` mit gesetztem `pruefdatum` legt zusätzlich einen
+  `Pruefung`-Eintrag an (Ergebnis-Mapping: `einsatzbereit`→`bestanden`,
+  `defekt`→`nicht_bestanden`, `in_reparatur`→`mit_maengeln`), OHNE das
+  bestehende Verhalten (Historie-Log, Fehlbestand-Erzeugung,
+  `naechste_pruefung`-Berechnung) zu verändern. Fahrzeug-HU/UVV hat noch
+  keinen eigenen Erfassen-Endpunkt (Admin setzt `Fahrzeugdetails.naechste_hu`/
+  `naechste_uvv` bisher direkt) — Dual-Write dafür ist Teil von Phase 2, sobald
+  dafür ein echter Erfassen-Workflow existiert. `GET /geraete/{id}/pruefungen`
+  liefert die volle Historie.
 
 ## INSP-004 — Status-Ampel (VALID/DUE_SOON/OVERDUE/FAILED)
 
@@ -90,10 +131,16 @@ Details zu INSP-001 … INSP-006 (vollständig).
 - **Akzeptanzkriterien:** alle vier Zustände korrekt unterscheidbar (inkl. FAILED bei
   „nicht bestanden" unabhängig vom Datum).
 - **Tests:** Statustest je Zustand.
-- **DoD:** deckt sich mit MABEA-Berechnungsmuster (`_ablauf_status`:
-  gueltig/bald_ablaufend/abgelaufen). **FAILED als eigener Zustand fehlt in MABEA:** kein
-  strukturiertes Prüfergebnis vorhanden, daher kein „nicht bestanden" unabhängig vom
-  Datum abbildbar.
+- **DoD:** ✅ Phase 1 umgesetzt (2026-09-10), **FAILED-Lücke geschlossen.**
+  `app/services/pruefung.py:dringlichkeit_fuer_geraet()` liefert fünf
+  Zustände statt der bisherigen drei (`_ablauf_status`:
+  gueltig/bald_ablaufend/abgelaufen): `fehlgeschlagen` (jüngste `Pruefung` hat
+  `ergebnis=nicht_bestanden` — **unabhängig vom Datum**, das ist exakt die
+  ursprünglich fehlende FAILED-Fähigkeit), sonst `ok`/`bald`/`kritisch`/
+  `ueberfaellig` aus `app/services/fristen.py:ampel_aus_tagen()` (FOUND-008,
+  wiederverwendet statt dupliziert) bezogen auf `GeraetInstanz.naechste_pruefung`.
+  `GET /geraete/{id}/pruefungs-ampel`. Nur für Geräte (Phase 1), Fahrzeug-
+  HU/UVV-Ampel bleibt vorerst beim bestehenden Dashboard-Mechanismus.
 
 ## INSP-005 — Prüfdokument-Anbindung
 
@@ -147,3 +194,14 @@ Feld erfasst (INSP-003 unvollständig), und FAILED als eigener, datumsunabhängi
 fehlt (INSP-004 unvollständig). Die Berechnungsmechanik (Intervall→Fälligkeit→Ampel) ist
 aber ausgereift und direkt wiederverwendbar, sobald die generische Prüfart-Struktur
 nachgezogen wird.
+
+**Update 2026-09-10 (Phase 1 additiv umgesetzt):** INSP-001/002/003/004 sind jetzt
+additiv abgedeckt (`pruefart`/`objekttyp_pruefart`/`pruefung`-Tabellen, Dual-Write aus
+`PATCH /geraete/{id}`, FAILED-Ampel-Zustand). Bestehende Lesepfade (Dashboard,
+Fristen-Dienst, ASSET-010) sind bewusst NICHT umgestellt — das ist Phase 2, ein
+eigener künftiger Schritt, kein Teil dieser Umsetzung. INSP-005 (Prüfdokument-
+Anbindung) und INSP-006 (Prüftermin-Übersicht) bleiben unverändert auf ihrem
+bisherigen Stand (siehe deren eigene DoD-Einträge oben). **Kein Frontend in Phase 1**
+(reines Backend, additiv, kein UI-Bedarf ohne Lesepfad-Umstellung) — Prüfarten-
+Verwaltungs-UI, Prüfungshistorie-Anzeige und FAILED-Ampel-Darstellung im Portal
+folgen mit Phase 2.

backend/alembic/versions/0045_pruefung.py

@@ -0,0 +1,114 @@
+"""INSP-001/002/003 Phase 1: generische Pruefart/Pruefung-Tabellen (additiv,
+Dual-Write neben bestehenden GeraetInstanz-Feldern - kein Ersatz, siehe
+09_inspections.md Phasenplan).
+
+Backfill: für jede GeraetInstanz mit gesetztem `pruefdatum` wird ein
+`pruefung`-Eintrag mit Prüfart "Gerätesicherheitsprüfung" angelegt
+(`pruefer_id` bleibt NULL, da der ursprüngliche Prüfer nicht gespeichert war).
+Fahrzeug-HU/UVV wird NICHT zurückbefüllt - `Fahrzeugdetails` speichert nur das
+nächste Fälligkeitsdatum, nicht das tatsächliche letzte Prüfdatum, es gäbe
+also keine reale Quelle für `geprueft_am`.
+
+Revision ID: 0045_pruefung
+Revises: 0044_material_gtin
+Create Date: 2026-09-10
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0045_pruefung"
+down_revision: Union[str, None] = "0044_material_gtin"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+_SEED_PRUEFARTEN = [
+    ("Gerätesicherheitsprüfung", "Generische Geräteprüfung (bisher einzige Prüfart in MABEA)", None),
+    ("HU", "Hauptuntersuchung (Fahrzeug)", "§29 StVZO"),
+    ("UVV", "Prüfung nach Unfallverhütungsvorschrift (Fahrzeug)", "DGUV Vorschrift 70"),
+]
+
+
+def upgrade() -> None:
+    op.execute("CREATE TYPE pruefung_ergebnis AS ENUM ('bestanden', 'nicht_bestanden', 'mit_maengeln')")
+    op.execute(
+        """
+        CREATE TABLE pruefart (
+            id SERIAL PRIMARY KEY,
+            name TEXT NOT NULL UNIQUE,
+            beschreibung TEXT,
+            gesetzliche_grundlage TEXT
+        )
+        """
+    )
+    op.execute(
+        """
+        CREATE TABLE objekttyp_pruefart (
+            objekttyp_id INTEGER NOT NULL REFERENCES objekttyp(id),
+            pruefart_id INTEGER NOT NULL REFERENCES pruefart(id),
+            intervall_monate INTEGER,
+            PRIMARY KEY (objekttyp_id, pruefart_id)
+        )
+        """
+    )
+    op.execute(
+        """
+        CREATE TABLE pruefung (
+            id UUID PRIMARY KEY,
+            geraet_instanz_id UUID REFERENCES geraet_instanz(id),
+            objekt_id INTEGER REFERENCES objekt(id),
+            pruefart_id INTEGER NOT NULL REFERENCES pruefart(id),
+            geprueft_am DATE NOT NULL,
+            pruefer_id INTEGER REFERENCES benutzer(id),
+            ergebnis pruefung_ergebnis NOT NULL,
+            bemerkung TEXT,
+            erstellt_am TIMESTAMPTZ NOT NULL
+        )
+        """
+    )
+    op.execute("CREATE INDEX ix_pruefung_geraet_instanz ON pruefung (geraet_instanz_id)")
+    op.execute("CREATE INDEX ix_pruefung_objekt ON pruefung (objekt_id)")
+
+    bind = op.get_bind()
+    for name, beschreibung, grundlage in _SEED_PRUEFARTEN:
+        bind.execute(
+            sa.text(
+                "INSERT INTO pruefart (name, beschreibung, gesetzliche_grundlage) "
+                "VALUES (:name, :beschreibung, :grundlage)"
+            ),
+            {"name": name, "beschreibung": beschreibung, "grundlage": grundlage},
+        )
+
+    geraetesicherheitspruefung_id = bind.execute(
+        sa.text("SELECT id FROM pruefart WHERE name = 'Gerätesicherheitsprüfung'")
+    ).scalar_one()
+
+    geraete_mit_pruefdatum = bind.execute(
+        sa.text("SELECT id, pruefdatum, status FROM geraet_instanz WHERE pruefdatum IS NOT NULL")
+    ).mappings().all()
+    for zeile in geraete_mit_pruefdatum:
+        ergebnis = "bestanden" if zeile["status"] == "einsatzbereit" else (
+            "nicht_bestanden" if zeile["status"] == "defekt" else "mit_maengeln"
+        )
+        bind.execute(
+            sa.text(
+                "INSERT INTO pruefung (id, geraet_instanz_id, pruefart_id, geprueft_am, ergebnis, bemerkung, erstellt_am) "
+                "VALUES (gen_random_uuid(), :geraet_instanz_id, :pruefart_id, :geprueft_am, :ergebnis, "
+                "'Backfill aus bestehendem Datenbestand (Migration 0045)', now())"
+            ),
+            {
+                "geraet_instanz_id": zeile["id"],
+                "pruefart_id": geraetesicherheitspruefung_id,
+                "geprueft_am": zeile["pruefdatum"],
+                "ergebnis": ergebnis,
+            },
+        )
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE pruefung")
+    op.execute("DROP TABLE objekttyp_pruefart")
+    op.execute("DROP TABLE pruefart")
+    op.execute("DROP TYPE IF EXISTS pruefung_ergebnis")

backend/app/api/v1/api.py

@@ -23,6 +23,7 @@ from app.api.v1.endpoints import (
     objekte,
     permission,
     personal,
+    pruefung,
     readiness_regel,
     readiness_snapshot,
     stammdaten,
@@ -53,6 +54,7 @@ api_router.include_router(fehlbestaende.router, tags=["fehlbestaende"])
 api_router.include_router(eskalation.router, tags=["eskalation"])
 api_router.include_router(historie.router, tags=["historie"])
 api_router.include_router(geraet_instanz.router, tags=["geraet_instanz"])
+api_router.include_router(pruefung.router, tags=["pruefung"])
 api_router.include_router(custody.router, tags=["custody"])
 api_router.include_router(dashboard.router, tags=["dashboard"])
 api_router.include_router(personal.router, tags=["personal"])

backend/app/api/v1/endpoints/pruefung.py

@@ -0,0 +1,60 @@
+import uuid
+
+from fastapi import APIRouter, Depends, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import require_roles
+from app.db.session import get_db
+from app.models.auth import RolleTyp
+from app.models.pruefung import Pruefart, Pruefung
+from app.schemas.pruefung import PruefartCreate, PruefartRead, PruefungRead, PruefungsAmpelRead
+from app.services.pruefung import (
+    dringlichkeit_fuer_geraet,
+    erstelle_pruefart,
+    liste_pruefarten,
+    liste_pruefungen_fuer_geraet,
+)
+
+router = APIRouter()
+
+_lesend = require_roles(
+    RolleTyp.mitarbeiter,
+    RolleTyp.materialverantwortlicher,
+    RolleTyp.leitungsverantwortlicher,
+    RolleTyp.administration,
+)
+_administration = require_roles(RolleTyp.administration)
+
+
+@router.get("/pruefarten", response_model=list[PruefartRead])
+async def liste_pruefarten_endpunkt(
+    db: AsyncSession = Depends(get_db), _=Depends(_lesend)
+) -> list[Pruefart]:
+    return await liste_pruefarten(db)
+
+
+@router.post("/pruefarten", response_model=PruefartRead, status_code=status.HTTP_201_CREATED)
+async def erstelle_pruefart_endpunkt(
+    payload: PruefartCreate, db: AsyncSession = Depends(get_db), _=Depends(_administration)
+) -> Pruefart:
+    return await erstelle_pruefart(
+        db,
+        name=payload.name,
+        beschreibung=payload.beschreibung,
+        gesetzliche_grundlage=payload.gesetzliche_grundlage,
+    )
+
+
+@router.get("/geraete/{geraet_instanz_id}/pruefungen", response_model=list[PruefungRead])
+async def liste_pruefungen_endpunkt(
+    geraet_instanz_id: uuid.UUID, db: AsyncSession = Depends(get_db), _=Depends(_lesend)
+) -> list[Pruefung]:
+    return await liste_pruefungen_fuer_geraet(db, geraet_instanz_id)
+
+
+@router.get("/geraete/{geraet_instanz_id}/pruefungs-ampel", response_model=PruefungsAmpelRead)
+async def pruefungs_ampel_endpunkt(
+    geraet_instanz_id: uuid.UUID, db: AsyncSession = Depends(get_db), _=Depends(_lesend)
+) -> PruefungsAmpelRead:
+    dringlichkeit = await dringlichkeit_fuer_geraet(db, geraet_instanz_id)
+    return PruefungsAmpelRead(dringlichkeit=dringlichkeit)

backend/app/models/__init__.py

@@ -18,6 +18,7 @@ from app.models.notiz import AkteNotiz
 from app.models.objekt import Objekt, ObjektStatus
 from app.models.objektposition import Objektposition, ObjektpositionStatus
 from app.models.permission import Berechtigung, BenutzerRolleZuordnung, Rolle, RolleBerechtigung
+from app.models.pruefung import ObjekttypPruefart, Pruefart, Pruefung, PruefungErgebnis
 from app.models.readiness_regel import ReadinessRegel
 from app.models.readiness_snapshot import ReadinessSnapshot, ReadinessZustand
 from app.models.personal import (
@@ -87,6 +88,10 @@ __all__ = [
     "BenutzerRolleZuordnung",
     "Rolle",
     "RolleBerechtigung",
+    "ObjekttypPruefart",
+    "Pruefart",
+    "Pruefung",
+    "PruefungErgebnis",
     "ReadinessRegel",
     "ReadinessSnapshot",
     "ReadinessZustand",

backend/app/models/pruefung.py

@@ -0,0 +1,74 @@
+import enum
+import uuid
+from datetime import date, datetime
+
+from sqlalchemy import Date, ForeignKey, Integer, String, Text
+from sqlalchemy.dialects.postgresql import ENUM as PgEnum, TIMESTAMP, UUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class PruefungErgebnis(str, enum.Enum):
+    bestanden = "bestanden"
+    nicht_bestanden = "nicht_bestanden"
+    mit_maengeln = "mit_maengeln"
+
+
+pruefung_ergebnis_pg = PgEnum(PruefungErgebnis, name="pruefung_ergebnis", create_type=False)
+
+# INSP-001: Konstante statt "magic string" an mehreren Call-Sites - Backfill-
+# Migration UND Dual-Write-Hook (app/services/geraet_instanz.py) referenzieren
+# denselben Namen.
+GERAETESICHERHEITSPRUEFUNG = "Gerätesicherheitsprüfung"
+
+
+class Pruefart(Base):
+    """INSP-001: generische Prüfart-Stammdaten (UVV/HU/Medizinprodukt/Elektro/
+    Sichtprüfung/...) - Phase 1 (additiv, 2026-09-10): ergänzt die bestehenden
+    GeraetInstanz-/Fahrzeugdetails-Felder als strukturierte Historie, ersetzt
+    sie noch nicht (siehe Kachel-Dokumentation zum Phasenplan)."""
+
+    __tablename__ = "pruefart"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String, unique=True, nullable=False)
+    beschreibung: Mapped[str | None] = mapped_column(Text)
+    gesetzliche_grundlage: Mapped[str | None] = mapped_column(String)
+
+
+class ObjekttypPruefart(Base):
+    """INSP-002: welche Prüfart für welchen Objekttyp vorgesehen ist +
+    Standard-Intervall - analog zum bestehenden Wartungsplan-Muster
+    (Objekttyp-Ebene statt Einzelinstanz)."""
+
+    __tablename__ = "objekttyp_pruefart"
+
+    objekttyp_id: Mapped[int] = mapped_column(ForeignKey("objekttyp.id"), primary_key=True)
+    pruefart_id: Mapped[int] = mapped_column(ForeignKey("pruefart.id"), primary_key=True)
+    intervall_monate: Mapped[int | None] = mapped_column(Integer)
+
+
+class Pruefung(Base):
+    """INSP-003: EIN Eintrag je tatsächlich durchgeführter Prüfung (echte
+    Historie, nicht nur "letzter Status") - genau eins von
+    `geraet_instanz_id`/`objekt_id` ist gesetzt, je nachdem ob es eine
+    Geräte- oder eine Fahrzeug-Prüfung (HU/UVV) ist (Anwendungslogik erzwingt
+    das, kein DB-CHECK-Constraint - zwei nullable FKs sind hier einfacher als
+    eine polymorphe Zwischentabelle für nur zwei Zieltypen)."""
+
+    __tablename__ = "pruefung"
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    geraet_instanz_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("geraet_instanz.id")
+    )
+    objekt_id: Mapped[int | None] = mapped_column(ForeignKey("objekt.id"))
+    pruefart_id: Mapped[int] = mapped_column(ForeignKey("pruefart.id"), nullable=False)
+    geprueft_am: Mapped[date] = mapped_column(Date, nullable=False)
+    # Nullable: Backfill aus bestehendem Datenbestand kennt den ursprünglichen
+    # Prüfer nicht (nur das Prüfdatum war bisher gespeichert).
+    pruefer_id: Mapped[int | None] = mapped_column(ForeignKey("benutzer.id"))
+    ergebnis: Mapped[PruefungErgebnis] = mapped_column(pruefung_ergebnis_pg, nullable=False)
+    bemerkung: Mapped[str | None] = mapped_column(Text)
+    erstellt_am: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False)

backend/app/schemas/pruefung.py

@@ -0,0 +1,37 @@
+import uuid
+from datetime import date, datetime
+
+from pydantic import BaseModel, ConfigDict
+
+from app.models.pruefung import PruefungErgebnis
+
+
+class PruefartCreate(BaseModel):
+    name: str
+    beschreibung: str | None = None
+    gesetzliche_grundlage: str | None = None
+
+
+class PruefartRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+    name: str
+    beschreibung: str | None
+    gesetzliche_grundlage: str | None
+
+
+class PruefungRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    geraet_instanz_id: uuid.UUID | None
+    objekt_id: int | None
+    pruefart_id: int
+    geprueft_am: date
+    pruefer_id: int | None
+    ergebnis: PruefungErgebnis
+    bemerkung: str | None
+    erstellt_am: datetime
+
+
+class PruefungsAmpelRead(BaseModel):
+    dringlichkeit: str

backend/app/services/geraet_instanz.py

@@ -9,10 +9,18 @@ from app.models.fehlbestand import Fehlbestand, FehlbestandStatus
 from app.models.geraet_instanz import GeraetInstanz, GeraetStatus
 from app.models.objekt import Objekt
 from app.models.objektposition import Objektposition
+from app.models.pruefung import GERAETESICHERHEITSPRUEFUNG, Pruefart, PruefungErgebnis
 from app.services import benachrichtigung as benachrichtigung_service
 from app.services import historie as historie_service
+from app.services import pruefung as pruefung_service
 from app.services.datumshilfen import plus_monate as _plus_monate
 
+_STATUS_ZU_ERGEBNIS = {
+    GeraetStatus.einsatzbereit: PruefungErgebnis.bestanden,
+    GeraetStatus.defekt: PruefungErgebnis.nicht_bestanden,
+    GeraetStatus.in_reparatur: PruefungErgebnis.mit_maengeln,
+}
+
 
 async def liste_fuer_position(db: AsyncSession, objektposition_id) -> list[GeraetInstanz]:
     result = await db.execute(
@@ -77,6 +85,19 @@ async def aendere_status(
         if objektposition is not None and objektposition.pruefintervall_monate:
             instanz.naechste_pruefung = _plus_monate(pruefdatum, objektposition.pruefintervall_monate)
 
+        pruefart_id = (
+            await db.execute(select(Pruefart.id).where(Pruefart.name == GERAETESICHERHEITSPRUEFUNG))
+        ).scalar_one()
+        await pruefung_service.erstelle_pruefung_fuer_geraet(
+            db,
+            geraet_instanz_id=instanz.id,
+            pruefart_id=pruefart_id,
+            geprueft_am=pruefdatum,
+            pruefer_id=benutzer_id,
+            ergebnis=_STATUS_ZU_ERGEBNIS[status],
+            bemerkung=bemerkung,
+        )
+
     await db.flush()
 
     await historie_service.log(

backend/app/services/pruefung.py

@@ -0,0 +1,82 @@
+import uuid
+from datetime import date, datetime, timezone
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.geraet_instanz import GeraetInstanz
+from app.models.pruefung import Pruefart, Pruefung, PruefungErgebnis
+from app.services.fristen import ampel_aus_tagen
+
+_WARNZEITRAUM_TAGE = 30
+
+
+async def erstelle_pruefart(
+    db: AsyncSession, *, name: str, beschreibung: str | None, gesetzliche_grundlage: str | None
+) -> Pruefart:
+    pruefart = Pruefart(name=name, beschreibung=beschreibung, gesetzliche_grundlage=gesetzliche_grundlage)
+    db.add(pruefart)
+    await db.flush()
+    return pruefart
+
+
+async def liste_pruefarten(db: AsyncSession) -> list[Pruefart]:
+    result = await db.execute(select(Pruefart))
+    return list(result.scalars().all())
+
+
+async def erstelle_pruefung_fuer_geraet(
+    db: AsyncSession,
+    *,
+    geraet_instanz_id: uuid.UUID,
+    pruefart_id: int,
+    geprueft_am: date,
+    pruefer_id: int | None,
+    ergebnis: PruefungErgebnis,
+    bemerkung: str | None,
+) -> Pruefung:
+    pruefung = Pruefung(
+        geraet_instanz_id=geraet_instanz_id,
+        pruefart_id=pruefart_id,
+        geprueft_am=geprueft_am,
+        pruefer_id=pruefer_id,
+        ergebnis=ergebnis,
+        bemerkung=bemerkung,
+        erstellt_am=datetime.now(timezone.utc),
+    )
+    db.add(pruefung)
+    return pruefung
+
+
+async def liste_pruefungen_fuer_geraet(db: AsyncSession, geraet_instanz_id: uuid.UUID) -> list[Pruefung]:
+    result = await db.execute(
+        select(Pruefung)
+        .where(Pruefung.geraet_instanz_id == geraet_instanz_id)
+        .order_by(Pruefung.geprueft_am.desc())
+    )
+    return list(result.scalars().all())
+
+
+async def dringlichkeit_fuer_geraet(db: AsyncSession, geraet_instanz_id: uuid.UUID) -> str:
+    """INSP-004: 'fehlgeschlagen' ist ein neuer Ampel-Zustand, den es im
+    bisherigen GeraetInstanz-Statussystem nicht gab - eine nicht bestandene
+    Prüfung überstrahlt jede datumsbasierte Einstufung, unabhängig davon wie
+    weit `naechste_pruefung` noch in der Zukunft liegt."""
+    result = await db.execute(
+        select(Pruefung)
+        .where(Pruefung.geraet_instanz_id == geraet_instanz_id)
+        .order_by(Pruefung.geprueft_am.desc())
+        .limit(1)
+    )
+    juengste = result.scalars().first()
+    if juengste is None:
+        return "unbekannt"
+    if juengste.ergebnis == PruefungErgebnis.nicht_bestanden:
+        return "fehlgeschlagen"
+
+    instanz = await db.get(GeraetInstanz, geraet_instanz_id)
+    if instanz is None or instanz.naechste_pruefung is None:
+        return "unbekannt"
+
+    verbleibende_tage = (instanz.naechste_pruefung - date.today()).days
+    return ampel_aus_tagen(verbleibende_tage, _WARNZEITRAUM_TAGE).value

backend/tests/test_pruefung.py

@@ -0,0 +1,102 @@
+import pytest
+from sqlalchemy import select
+
+from app.models.pruefung import Pruefung, PruefungErgebnis
+from tests.conftest import auth_header, login
+
+
+async def _erstelle_geraet(client, objekt_mit_position, token, seriennummer: str) -> str:
+    objekt, material = objekt_mit_position
+    positionen = await client.get(f"/api/v1/objekte/{objekt.id}/positionen", headers=auth_header(token))
+    position_id = positionen.json()[0]["id"]
+    erstellt = await client.post(
+        f"/api/v1/objektpositionen/{position_id}/geraete",
+        json={"seriennummer": seriennummer},
+        headers=auth_header(token),
+    )
+    return erstellt.json()["id"]
+
+
+@pytest.mark.asyncio
+async def test_pruefung_dual_write_bestanden(
+    client, db_session, objekt_mit_position, materialverantwortlicher_user
+):
+    token = await login(client, "materialverantwortlicher1")
+    geraet_id = await _erstelle_geraet(client, objekt_mit_position, token, "SN-INSP-001")
+
+    antwort = await client.patch(
+        f"/api/v1/geraete/{geraet_id}",
+        json={"status": "einsatzbereit", "pruefdatum": "2026-01-15"},
+        headers=auth_header(token),
+    )
+    assert antwort.status_code == 200
+
+    pruefungen = (
+        await db_session.execute(select(Pruefung).where(Pruefung.geraet_instanz_id == geraet_id))
+    ).scalars().all()
+    assert len(pruefungen) == 1
+    assert pruefungen[0].ergebnis == PruefungErgebnis.bestanden
+
+
+@pytest.mark.asyncio
+async def test_pruefung_dual_write_nicht_bestanden_und_ampel(
+    client, db_session, objekt_mit_position, materialverantwortlicher_user
+):
+    token = await login(client, "materialverantwortlicher1")
+    geraet_id = await _erstelle_geraet(client, objekt_mit_position, token, "SN-INSP-002")
+
+    antwort = await client.patch(
+        f"/api/v1/geraete/{geraet_id}",
+        json={"status": "defekt", "pruefdatum": "2026-01-15"},
+        headers=auth_header(token),
+    )
+    assert antwort.status_code == 200
+
+    pruefungen = (
+        await db_session.execute(select(Pruefung).where(Pruefung.geraet_instanz_id == geraet_id))
+    ).scalars().all()
+    assert len(pruefungen) == 1
+    assert pruefungen[0].ergebnis == PruefungErgebnis.nicht_bestanden
+
+    ampel = await client.get(
+        f"/api/v1/geraete/{geraet_id}/pruefungs-ampel", headers=auth_header(token)
+    )
+    assert ampel.status_code == 200
+    assert ampel.json()["dringlichkeit"] == "fehlgeschlagen"
+
+
+@pytest.mark.asyncio
+async def test_pruefungen_liste_sortiert_absteigend(
+    client, db_session, objekt_mit_position, materialverantwortlicher_user
+):
+    token = await login(client, "materialverantwortlicher1")
+    geraet_id = await _erstelle_geraet(client, objekt_mit_position, token, "SN-INSP-003")
+
+    await client.patch(
+        f"/api/v1/geraete/{geraet_id}",
+        json={"status": "einsatzbereit", "pruefdatum": "2026-01-01"},
+        headers=auth_header(token),
+    )
+    await client.patch(
+        f"/api/v1/geraete/{geraet_id}",
+        json={"status": "einsatzbereit", "pruefdatum": "2026-02-01"},
+        headers=auth_header(token),
+    )
+
+    liste = await client.get(f"/api/v1/geraete/{geraet_id}/pruefungen", headers=auth_header(token))
+    assert liste.status_code == 200
+    daten = liste.json()
+    assert len(daten) == 2
+    assert daten[0]["geprueft_am"] == "2026-02-01"
+    assert daten[1]["geprueft_am"] == "2026-01-01"
+
+
+@pytest.mark.asyncio
+async def test_pruefart_erstellen_nur_administration(client, mitarbeiter_user):
+    token = await login(client, "mitarbeiter1")
+    antwort = await client.post(
+        "/api/v1/pruefarten",
+        json={"name": "Neue Pruefart"},
+        headers=auth_header(token),
+    )
+    assert antwort.status_code == 403