Sprint 5: Mindermengen-Genehmigung + automatischer Ablauf, Historie für Kernkette

- MindermengenGenehmigung-Modell/Service (Prompt 04): reine fachliche Bewertung,
  ändert NIEMALS Ist-Menge/Fehlmenge/Fehlbestand-Status (Leitplanke, U2)
- Automatischer Ablauf bei Start der nächsten Kontrolle desselben Objekts (Karte 08,
  U6) - kein Zeit-Trigger, der Kontrollstart selbst ist der Ablauf-Zeitpunkt
- Genehmigung endet gemeinsam mit dem Fehlbestand, wenn dieser durch Nachfüllung
  erledigt wird (U7), unabhängig davon ob vorher eine neue Kontrolle stattfand
- Historie-Service (app/services/historie.py) + Verdrahtung in die komplette
  Kernkette: kontrolle_gestartet/-abgeschlossen/-abgebrochen/-uebernommen,
  istmenge_erfasst, fehlbestand_entstanden/-erledigt, nachfuellung_erfasst,
  mindermenge_genehmigt/-abgelaufen/-beendet_durch_erledigung (U14)
  Hinweis: Stammdaten-/Vorlagen-/Benutzerverwaltung noch nicht retrofittet -
  Sprint 5 deckt bewusst die im Testkonzept referenzierte Kernkette ab, keine
  flächendeckende Audit-Abdeckung aller CRUD-Endpunkte.
- Tests: U2, U6, U7, vollständige Historie-Kette nach Prompt-13.3-Beispiel (U14)

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

DEVLOG.md

@@ -1241,3 +1241,66 @@ Keine Commits in dieser Session.
 - backend/tests/test_kontrolle.py            | 238 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 
 ---
+## 2026-09-03 23:41 – 23:46 (5m)
+**Beschreibung:** Claude Code Session
+**Projekt:** asb-material
+
+### Commits
+- 18fba28 Sprint 4: Fehlbestand-Statusmaschine komplett, Nachfüllung sofort/später/teilweise
+
+### Geänderte Dateien
+- DEVLOG.md                                     |  42 ++++++++++++++++++++++
+- backend/app/api/v1/api.py                     |  13 ++++++-
+- backend/app/api/v1/endpoints/fehlbestaende.py |  86 +++++++++++++++++++++++++++++++++++++++++++++
+- backend/app/api/v1/endpoints/kontrollen.py    |  13 +++++--
+- backend/app/models/__init__.py                |   2 ++
+- backend/app/models/nachfuellung.py            |  26 ++++++++++++++
+- backend/app/schemas/fehlbestand.py            |  46 ++++++++++++++++++++++++
+- backend/app/schemas/kontrolle.py              |   1 +
+- backend/app/services/fehlbestand.py           |  69 ++++++++++++++++++++++++++++++++++++
+- backend/app/services/kontrolle.py             |  42 ++++++++++++----------
+- backend/tests/test_fehlbestand.py             | 138 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+---
+## 2026-09-03 23:46 – 23:46 (0m)
+**Beschreibung:** Claude Code Session
+**Projekt:** asb-material
+
+### Commits
+Keine Commits in dieser Session.
+
+### Geänderte Dateien
+- DEVLOG.md                                     |  42 ++++++++++++++++++++++
+- backend/app/api/v1/api.py                     |  13 ++++++-
+- backend/app/api/v1/endpoints/fehlbestaende.py |  86 +++++++++++++++++++++++++++++++++++++++++++++
+- backend/app/api/v1/endpoints/kontrollen.py    |  13 +++++--
+- backend/app/models/__init__.py                |   2 ++
+- backend/app/models/nachfuellung.py            |  26 ++++++++++++++
+- backend/app/schemas/fehlbestand.py            |  46 ++++++++++++++++++++++++
+- backend/app/schemas/kontrolle.py              |   1 +
+- backend/app/services/fehlbestand.py           |  69 ++++++++++++++++++++++++++++++++++++
+- backend/app/services/kontrolle.py             |  42 ++++++++++++----------
+- backend/tests/test_fehlbestand.py             | 138 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+---
+## 2026-09-03 23:47 – 23:47 (0m)
+**Beschreibung:** Claude Code Session
+**Projekt:** asb-material
+
+### Commits
+Keine Commits in dieser Session.
+
+### Geänderte Dateien
+- DEVLOG.md                                     |  42 ++++++++++++++++++++++
+- backend/app/api/v1/api.py                     |  13 ++++++-
+- backend/app/api/v1/endpoints/fehlbestaende.py |  86 +++++++++++++++++++++++++++++++++++++++++++++
+- backend/app/api/v1/endpoints/kontrollen.py    |  13 +++++--
+- backend/app/models/__init__.py                |   2 ++
+- backend/app/models/nachfuellung.py            |  26 ++++++++++++++
+- backend/app/schemas/fehlbestand.py            |  46 ++++++++++++++++++++++++
+- backend/app/schemas/kontrolle.py              |   1 +
+- backend/app/services/fehlbestand.py           |  69 ++++++++++++++++++++++++++++++++++++
+- backend/app/services/kontrolle.py             |  42 ++++++++++++----------
+- backend/tests/test_fehlbestand.py             | 138 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+---

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

@@ -6,7 +6,7 @@ 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.auth import KnotenTyp, RolleTyp, Systemknoten
 from app.models.fehlbestand import Fehlbestand
 from app.schemas.fehlbestand import (
     FehlbestandRead,
@@ -14,7 +14,11 @@ from app.schemas.fehlbestand import (
     NachfuellungRead,
     NachfuellungResponse,
 )
+from app.schemas.mindermenge import MindermengeGenehmigenRequest, MindermengeRead
 from app.services.fehlbestand import FehlbestandBereitsErledigtError, nachfuellen
+from app.services.mindermenge import BereitsGenehmigtError
+from app.services.mindermenge import FehlbestandBereitsErledigtError as MindermengeAufErledigtemError
+from app.services.mindermenge import genehmigen
 
 router = APIRouter()
 
@@ -29,6 +33,11 @@ _verantwortliche = require_roles(
 )
 
 
+async def _hauptserver_id(db: AsyncSession) -> int:
+    result = await db.execute(select(Systemknoten.id).where(Systemknoten.typ == KnotenTyp.haupt))
+    return result.scalar_one()
+
+
 @router.get("/fehlbestaende", response_model=list[FehlbestandRead])
 async def liste_fehlbestaende(
     status_filter: str | None = None,
@@ -70,9 +79,14 @@ async def erfasse_nachfuellung(
     if fehlbestand is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Fehlbestand nicht gefunden")
 
+    zustaendiger_server_id = await _hauptserver_id(db)
     try:
         nachfuellung, ueberbestand = await nachfuellen(
-            db, fehlbestand=fehlbestand, menge=payload.menge, benutzer_id=current_user.id
+            db,
+            fehlbestand=fehlbestand,
+            menge=payload.menge,
+            benutzer_id=current_user.id,
+            zustaendiger_server_id=zustaendiger_server_id,
         )
     except FehlbestandBereitsErledigtError as exc:
         raise HTTPException(
@@ -84,3 +98,41 @@ async def erfasse_nachfuellung(
         fehlbestand=FehlbestandRead.model_validate(fehlbestand),
         ueberbestand=ueberbestand,
     )
+
+
+@router.post(
+    "/fehlbestaende/{fehlbestand_id}/mindermenge",
+    response_model=MindermengeRead,
+    status_code=status.HTTP_201_CREATED,
+)
+async def genehmige_mindermenge(
+    fehlbestand_id: uuid.UUID,
+    payload: MindermengeGenehmigenRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user=Depends(_verantwortliche),
+) -> MindermengeRead:
+    """Prompt 04: NIEMALS durch Mitarbeiter (Rollen-Dependency erzwingt das bereits),
+    ändert NIEMALS Ist-Menge/Fehlmenge/Fehlbestand-Status (U2)."""
+    fehlbestand = await db.get(Fehlbestand, fehlbestand_id)
+    if fehlbestand is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Fehlbestand nicht gefunden")
+
+    zustaendiger_server_id = await _hauptserver_id(db)
+    try:
+        genehmigung = await genehmigen(
+            db,
+            fehlbestand=fehlbestand,
+            begruendung=payload.begruendung,
+            genehmigt_von=current_user.id,
+            zustaendiger_server_id=zustaendiger_server_id,
+        )
+    except MindermengeAufErledigtemError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Fehlbestand ist bereits erledigt"
+        ) from exc
+    except BereitsGenehmigtError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Fehlbestand hat bereits eine aktive Genehmigung"
+        ) from exc
+
+    return MindermengeRead.model_validate(genehmigung)

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

@@ -154,8 +154,9 @@ async def abschliesse(
 ) -> Kontrolle:
     kontrolle = await _hole_kontrolle_oder_404(db, kontrolle_id)
     _pruefe_besitz(kontrolle, current_user)
+    zustaendiger_server_id = await _hauptserver_id(db)
     try:
-        await abschliesse_kontrolle(db, kontrolle=kontrolle)
+        await abschliesse_kontrolle(db, kontrolle=kontrolle, zustaendiger_server_id=zustaendiger_server_id)
     except UnvollstaendigError as exc:
         raise HTTPException(
             status_code=status.HTTP_409_CONFLICT,
@@ -176,5 +177,8 @@ async def abbreche(
 ) -> Kontrolle:
     kontrolle = await _hole_kontrolle_oder_404(db, kontrolle_id)
     _pruefe_besitz(kontrolle, current_user)
-    await abbreche_kontrolle(db, kontrolle=kontrolle, grund=payload.grund)
+    zustaendiger_server_id = await _hauptserver_id(db)
+    await abbreche_kontrolle(
+        db, kontrolle=kontrolle, grund=payload.grund, zustaendiger_server_id=zustaendiger_server_id
+    )
     return kontrolle

backend/app/models/__init__.py

@@ -2,6 +2,7 @@ from app.models.auth import Benutzer, BenutzerRolle, RolleTyp, Systemknoten, Kno
 from app.models.fehlbestand import Fehlbestand, FehlbestandStatus
 from app.models.historie import Historie
 from app.models.kontrolle import Kontrolle, KontrollStatus, Kontrollposition
+from app.models.mindermenge import MindermengeStatus, MindermengenGenehmigung
 from app.models.nachfuellung import Nachfuellung
 from app.models.objekt import Objekt, ObjektStatus
 from app.models.objektposition import Objektposition, ObjektpositionStatus
@@ -21,6 +22,8 @@ __all__ = [
     "Kontrolle",
     "KontrollStatus",
     "Kontrollposition",
+    "MindermengeStatus",
+    "MindermengenGenehmigung",
     "Nachfuellung",
     "Objekt",
     "ObjektStatus",

backend/app/models/mindermenge.py

@@ -0,0 +1,45 @@
+import enum
+import uuid
+from datetime import datetime
+
+from sqlalchemy import ForeignKey, String
+from sqlalchemy.dialects.postgresql import ENUM as PgEnum, TIMESTAMP, UUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class MindermengeStatus(str, enum.Enum):
+    aktiv = "aktiv"
+    abgelaufen = "abgelaufen"
+    beendet_durch_erledigung = "beendet_durch_erledigung"
+
+
+mindermenge_status_pg = PgEnum(MindermengeStatus, name="mindermenge_status", create_type=False)
+
+
+class MindermengenGenehmigung(Base):
+    """Prompt 04: reine fachliche Bewertung, ändert NIEMALS Ist-Menge/Fehlmenge/
+    Fehlbestand-Status (Leitplanke, U2). Gilt automatisch nur bis zur nächsten
+    Kontrolle desselben Objekts (Karte 08, U6) oder bis der Fehlbestand durch
+    Nachfüllung erledigt wird (U7) - beides ohne separaten Zeit-Trigger."""
+
+    __tablename__ = "mindermengen_genehmigung"
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    fehlbestand_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("fehlbestand.id"), nullable=False
+    )
+    genehmigt_von: Mapped[int] = mapped_column(ForeignKey("benutzer.id"), nullable=False)
+    begruendung: Mapped[str] = mapped_column(String, nullable=False)
+    genehmigt_am: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False)
+    ausloesende_kontrolle_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("kontrolle.id"), nullable=False
+    )
+    status: Mapped[MindermengeStatus] = mapped_column(
+        mindermenge_status_pg, nullable=False, default=MindermengeStatus.aktiv
+    )
+    beendet_am: Mapped[datetime | None] = mapped_column(TIMESTAMP(timezone=True))
+    beendende_kontrolle_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("kontrolle.id")
+    )

backend/app/schemas/mindermenge.py

@@ -0,0 +1,23 @@
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+from app.models.mindermenge import MindermengeStatus
+
+
+class MindermengeGenehmigenRequest(BaseModel):
+    begruendung: str
+
+
+class MindermengeRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    fehlbestand_id: uuid.UUID
+    genehmigt_von: int
+    begruendung: str
+    genehmigt_am: datetime
+    ausloesende_kontrolle_id: uuid.UUID
+    status: MindermengeStatus
+    beendet_am: datetime | None
+    beendende_kontrolle_id: uuid.UUID | None

backend/app/services/fehlbestand.py

@@ -5,8 +5,10 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.fehlbestand import Fehlbestand, FehlbestandStatus
+from app.models.mindermenge import MindermengeStatus, MindermengenGenehmigung
 from app.models.nachfuellung import Nachfuellung
 from app.models.objektposition import Objektposition
+from app.services import historie as historie_service
 
 
 class FehlbestandBereitsErledigtError(Exception):
@@ -14,7 +16,12 @@ class FehlbestandBereitsErledigtError(Exception):
 
 
 async def nachfuellen(
-    db: AsyncSession, *, fehlbestand: Fehlbestand, menge: Decimal, benutzer_id: int
+    db: AsyncSession,
+    *,
+    fehlbestand: Fehlbestand,
+    menge: Decimal,
+    benutzer_id: int,
+    zustaendiger_server_id: int,
 ) -> tuple[Nachfuellung, Decimal]:
     """Vier-Kernbegriffe (Prompt 02.9): Nachfüllung ist die einzige Aktion, die
     Objektposition.istmenge tatsächlich ändert. Deckt sowohl Sofort-Nachfüllung
@@ -50,7 +57,8 @@ async def nachfuellen(
 
     fehlbestand.istmenge = neue_istmenge
     fehlbestand.fehlmenge = neue_fehlmenge
-    if neue_fehlmenge <= 0:
+    wird_erledigt = neue_fehlmenge <= 0
+    if wird_erledigt:
         fehlbestand.status = FehlbestandStatus.erledigt
         fehlbestand.erledigt_am = datetime.now(timezone.utc)
     else:
@@ -66,4 +74,47 @@ async def nachfuellen(
     )
     db.add(nachfuellung)
     await db.flush()
+
+    await historie_service.log(
+        db,
+        zustaendiger_server_id=zustaendiger_server_id,
+        benutzer_id=benutzer_id,
+        ereignistyp="nachfuellung_erfasst",
+        entitaet_typ="fehlbestand",
+        entitaet_id=fehlbestand.id,
+        neuer_wert={"menge": str(menge), "neue_istmenge": str(neue_istmenge), "ueberbestand": str(ueberbestand)},
+    )
+
+    if wird_erledigt:
+        await historie_service.log(
+            db,
+            zustaendiger_server_id=zustaendiger_server_id,
+            benutzer_id=benutzer_id,
+            ereignistyp="fehlbestand_erledigt",
+            entitaet_typ="fehlbestand",
+            entitaet_id=fehlbestand.id,
+            neuer_wert={"status": "erledigt"},
+        )
+        # U7 (Prompt 04.4): wird die Abweichung behoben, erledigt sich eine noch
+        # aktive Mindermengen-Genehmigung gemeinsam mit dem Fehlbestand.
+        result = await db.execute(
+            select(MindermengenGenehmigung).where(
+                MindermengenGenehmigung.fehlbestand_id == fehlbestand.id,
+                MindermengenGenehmigung.status == MindermengeStatus.aktiv,
+            )
+        )
+        genehmigung = result.scalar_one_or_none()
+        if genehmigung is not None:
+            genehmigung.status = MindermengeStatus.beendet_durch_erledigung
+            genehmigung.beendet_am = datetime.now(timezone.utc)
+            await historie_service.log(
+                db,
+                zustaendiger_server_id=zustaendiger_server_id,
+                benutzer_id=benutzer_id,
+                ereignistyp="mindermenge_beendet_durch_erledigung",
+                entitaet_typ="mindermengen_genehmigung",
+                entitaet_id=genehmigung.id,
+                neuer_wert={"status": "beendet_durch_erledigung"},
+            )
+    await db.flush()
     return nachfuellung, ueberbestand

backend/app/services/historie.py

@@ -0,0 +1,37 @@
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.historie import Historie
+
+
+async def log(
+    db: AsyncSession,
+    *,
+    zustaendiger_server_id: int,
+    benutzer_id: int | None,
+    ereignistyp: str,
+    entitaet_typ: str,
+    entitaet_id: Any,
+    alter_wert: dict | None = None,
+    neuer_wert: dict | None = None,
+    begruendung: str | None = None,
+) -> Historie:
+    """Prompt 13: append-only Audit-Log. Deckt Sprint 5 die Kernkette ab (Kontrolle,
+    Fehlbestand, Nachfüllung, Mindermenge) - Stammdaten-/Vorlagen-/Benutzerverwaltung
+    folgen bei Bedarf mit demselben Muster, hier noch nicht flächendeckend verdrahtet."""
+    eintrag = Historie(
+        erzeugt_von_server_id=zustaendiger_server_id,
+        zeitpunkt=datetime.now(timezone.utc),
+        benutzer_id=benutzer_id,
+        ereignistyp=ereignistyp,
+        entitaet_typ=entitaet_typ,
+        entitaet_id=str(entitaet_id),
+        alter_wert=alter_wert,
+        neuer_wert=neuer_wert,
+        begruendung=begruendung,
+    )
+    db.add(eintrag)
+    await db.flush()
+    return eintrag

backend/app/services/kontrolle.py

@@ -6,10 +6,11 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.fehlbestand import Fehlbestand, FehlbestandStatus
-from app.models.historie import Historie
 from app.models.kontrolle import Kontrolle, KontrollStatus, Kontrollposition
+from app.models.mindermenge import MindermengeStatus, MindermengenGenehmigung
 from app.models.objekt import Objekt
 from app.models.objektposition import Objektposition, ObjektpositionStatus
+from app.services import historie as historie_service
 from app.services.objekte import sollmenge_effektiv, sollmenge_map
 
 
@@ -59,19 +60,16 @@ async def starte_kontrolle(
         # Fehlbestände dürfen nicht stehen bleiben (Prompt 16.6: kein Fehlbestand
         # durch eine abgebrochene Kontrolle).
         await _verwirf_kontrolldaten(db, laufende.id)
-        db.add(
-            Historie(
-                erzeugt_von_server_id=zustaendiger_server_id,
-                zeitpunkt=datetime.now(timezone.utc),
+        await historie_service.log(
+            db,
+            zustaendiger_server_id=zustaendiger_server_id,
             benutzer_id=benutzer_id,
             ereignistyp="kontrolle_uebernommen",
             entitaet_typ="kontrolle",
-                entitaet_id=str(laufende.id),
+            entitaet_id=laufende.id,
             alter_wert={"benutzer_id": laufende.benutzer_id},
             neuer_wert={"benutzer_id": benutzer_id},
         )
-        )
-        await db.flush()
 
     neue_kontrolle = Kontrolle(
         erzeugt_von_server_id=zustaendiger_server_id,
@@ -82,9 +80,51 @@ async def starte_kontrolle(
     )
     db.add(neue_kontrolle)
     await db.flush()
+
+    await _lasse_mindermengen_ablaufen(
+        db, objekt_id=objekt_id, neue_kontrolle=neue_kontrolle, zustaendiger_server_id=zustaendiger_server_id
+    )
+
+    await historie_service.log(
+        db,
+        zustaendiger_server_id=zustaendiger_server_id,
+        benutzer_id=benutzer_id,
+        ereignistyp="kontrolle_gestartet",
+        entitaet_typ="kontrolle",
+        entitaet_id=neue_kontrolle.id,
+        neuer_wert={"objekt_id": objekt_id, "status": "in_bearbeitung"},
+    )
     return neue_kontrolle
 
 
+async def _lasse_mindermengen_ablaufen(
+    db: AsyncSession, *, objekt_id: int, neue_kontrolle: Kontrolle, zustaendiger_server_id: int
+) -> None:
+    """Karte 08 / Prompt 04.4 / Sprintplan U6: Genehmigung gilt automatisch nur bis
+    zur nächsten Kontrolle desselben Objekts - kein Zeit-Trigger nötig, der Start
+    der nächsten Kontrolle IST der Ablauf-Zeitpunkt."""
+    result = await db.execute(
+        select(MindermengenGenehmigung)
+        .join(Fehlbestand, MindermengenGenehmigung.fehlbestand_id == Fehlbestand.id)
+        .where(Fehlbestand.objekt_id == objekt_id, MindermengenGenehmigung.status == MindermengeStatus.aktiv)
+    )
+    for genehmigung in result.scalars().all():
+        genehmigung.status = MindermengeStatus.abgelaufen
+        genehmigung.beendet_am = datetime.now(timezone.utc)
+        genehmigung.beendende_kontrolle_id = neue_kontrolle.id
+        await historie_service.log(
+            db,
+            zustaendiger_server_id=zustaendiger_server_id,
+            benutzer_id=None,
+            ereignistyp="mindermenge_abgelaufen",
+            entitaet_typ="mindermengen_genehmigung",
+            entitaet_id=genehmigung.id,
+            alter_wert={"status": "aktiv"},
+            neuer_wert={"status": "abgelaufen", "beendende_kontrolle_id": str(neue_kontrolle.id)},
+        )
+    await db.flush()
+
+
 async def _verwirf_kontrolldaten(db: AsyncSession, kontrolle_id) -> None:
     fehlbestaende = await db.execute(select(Fehlbestand).where(Fehlbestand.kontrolle_id == kontrolle_id))
     for fehlbestand in fehlbestaende.scalars().all():
@@ -141,6 +181,16 @@ async def erfasse_position(
         kontrollposition.abweichung = abweichung
     await db.flush()
 
+    await historie_service.log(
+        db,
+        zustaendiger_server_id=zustaendiger_server_id,
+        benutzer_id=kontrolle.benutzer_id,
+        ereignistyp="istmenge_erfasst",
+        entitaet_typ="kontrollposition",
+        entitaet_id=kontrollposition.id,
+        neuer_wert={"material_id": material_id, "sollmenge": str(sollmenge), "istmenge": str(istmenge)},
+    )
+
     fehlbestand_id = await _synchronisiere_fehlbestand(
         db,
         kontrolle=kontrolle,
@@ -191,6 +241,22 @@ async def _synchronisiere_fehlbestand(
                 status=FehlbestandStatus.offen,
             )
             db.add(bestehender)
+            await db.flush()
+            await historie_service.log(
+                db,
+                zustaendiger_server_id=zustaendiger_server_id,
+                benutzer_id=kontrolle.benutzer_id,
+                ereignistyp="fehlbestand_entstanden",
+                entitaet_typ="fehlbestand",
+                entitaet_id=bestehender.id,
+                neuer_wert={
+                    "objekt_id": objekt.id,
+                    "material_id": material_id,
+                    "sollmenge": str(sollmenge),
+                    "istmenge": str(istmenge),
+                    "fehlmenge": str(unterdeckung),
+                },
+            )
         else:
             # Korrektur vor Abschluss (z. B. Tippfehler nochmal erfasst) - solange die
             # Kontrolle noch läuft, ist der von IHR erzeugte Fehlbestand noch nicht
@@ -224,16 +290,27 @@ async def fehlende_material_ids(db: AsyncSession, kontrolle: Kontrolle) -> list[
     return sorted(erwartete - erfasste)
 
 
-async def abschliesse_kontrolle(db: AsyncSession, *, kontrolle: Kontrolle) -> None:
+async def abschliesse_kontrolle(db: AsyncSession, *, kontrolle: Kontrolle, zustaendiger_server_id: int) -> None:
     fehlende = await fehlende_material_ids(db, kontrolle)
     if fehlende:
         raise UnvollstaendigError(fehlende)
     kontrolle.status = KontrollStatus.abgeschlossen
     kontrolle.beendet_am = datetime.now(timezone.utc)
     await db.flush()
+    await historie_service.log(
+        db,
+        zustaendiger_server_id=zustaendiger_server_id,
+        benutzer_id=kontrolle.benutzer_id,
+        ereignistyp="kontrolle_abgeschlossen",
+        entitaet_typ="kontrolle",
+        entitaet_id=kontrolle.id,
+        neuer_wert={"status": "abgeschlossen"},
+    )
 
 
-async def abbreche_kontrolle(db: AsyncSession, *, kontrolle: Kontrolle, grund: str | None) -> None:
+async def abbreche_kontrolle(
+    db: AsyncSession, *, kontrolle: Kontrolle, grund: str | None, zustaendiger_server_id: int
+) -> None:
     """Prompt 16.6: keine Ist-Übernahme (Kontrolle rührt Objektposition ohnehin nie
     an), kein Fehlbestand aus dieser Kontrolle bleibt bestehen. Kontrolle selbst
     bleibt als Datensatz erhalten (Status "abgebrochen"), nicht gelöscht."""
@@ -242,3 +319,12 @@ async def abbreche_kontrolle(db: AsyncSession, *, kontrolle: Kontrolle, grund: s
     kontrolle.abbruch_grund = grund
     kontrolle.beendet_am = datetime.now(timezone.utc)
     await db.flush()
+    await historie_service.log(
+        db,
+        zustaendiger_server_id=zustaendiger_server_id,
+        benutzer_id=kontrolle.benutzer_id,
+        ereignistyp="kontrolle_abgebrochen",
+        entitaet_typ="kontrolle",
+        entitaet_id=kontrolle.id,
+        neuer_wert={"status": "abgebrochen", "grund": grund},
+    )

backend/app/services/mindermenge.py

@@ -0,0 +1,67 @@
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.fehlbestand import Fehlbestand, FehlbestandStatus
+from app.models.mindermenge import MindermengeStatus, MindermengenGenehmigung
+from app.services import historie as historie_service
+
+
+class FehlbestandBereitsErledigtError(Exception):
+    """Ein bereits erledigter Fehlbestand braucht keine Mindermengen-Bewertung mehr."""
+
+
+class BereitsGenehmigtError(Exception):
+    """Für diesen Fehlbestand existiert bereits eine aktive Genehmigung."""
+
+
+async def genehmigen(
+    db: AsyncSession,
+    *,
+    fehlbestand: Fehlbestand,
+    begruendung: str,
+    genehmigt_von: int,
+    zustaendiger_server_id: int,
+) -> MindermengenGenehmigung:
+    """Prompt 04: reine fachliche Bewertung - ändert NIEMALS Ist-Menge, Fehlmenge
+    oder den Fehlbestand-Status selbst (Leitplanke, U2). Verknüpft mit der Kontrolle,
+    die den Fehlbestand ausgelöst hat ("aktuelle Kontrolle des Objekts", Prompt 04.3.4)."""
+    if fehlbestand.status == FehlbestandStatus.erledigt:
+        raise FehlbestandBereitsErledigtError()
+    if fehlbestand.kontrolle_id is None:
+        # Sollte praktisch nicht vorkommen (jeder Fehlbestand entsteht aus einer
+        # Kontrolle, Sprint 3), aber das Schema erlaubt kontrolle_id=NULL.
+        raise ValueError("Fehlbestand ohne auslösende Kontrolle kann nicht genehmigt werden")
+
+    result = await db.execute(
+        select(MindermengenGenehmigung).where(
+            MindermengenGenehmigung.fehlbestand_id == fehlbestand.id,
+            MindermengenGenehmigung.status == MindermengeStatus.aktiv,
+        )
+    )
+    if result.scalar_one_or_none() is not None:
+        raise BereitsGenehmigtError()
+
+    genehmigung = MindermengenGenehmigung(
+        fehlbestand_id=fehlbestand.id,
+        genehmigt_von=genehmigt_von,
+        begruendung=begruendung,
+        genehmigt_am=datetime.now(timezone.utc),
+        ausloesende_kontrolle_id=fehlbestand.kontrolle_id,
+        status=MindermengeStatus.aktiv,
+    )
+    db.add(genehmigung)
+    await db.flush()
+
+    await historie_service.log(
+        db,
+        zustaendiger_server_id=zustaendiger_server_id,
+        benutzer_id=genehmigt_von,
+        ereignistyp="mindermenge_genehmigt",
+        entitaet_typ="mindermengen_genehmigung",
+        entitaet_id=genehmigung.id,
+        neuer_wert={"fehlbestand_id": str(fehlbestand.id), "status": "aktiv"},
+        begruendung=begruendung,
+    )
+    return genehmigung

backend/tests/conftest.py

@@ -91,6 +91,21 @@ async def mitarbeiter2_user(db_session):
     return benutzer
 
 
+@pytest_asyncio.fixture
+async def materialverantwortlicher_user(db_session):
+    benutzer = Benutzer(
+        name="Test Materialverantwortlicher",
+        login="materialverantwortlicher1",
+        passwort_hash=hash_password("test-passwort-123"),
+        aktiv=True,
+    )
+    db_session.add(benutzer)
+    await db_session.flush()
+    db_session.add(BenutzerRolle(benutzer_id=benutzer.id, rolle=RolleTyp.materialverantwortlicher))
+    await db_session.flush()
+    return benutzer
+
+
 @pytest_asyncio.fixture
 async def admin_user(db_session):
     benutzer = Benutzer(

backend/tests/test_historie.py

@@ -0,0 +1,62 @@
+import pytest
+from sqlalchemy import select
+
+from app.models.historie import Historie
+from tests.conftest import auth_header, login
+
+
+@pytest.mark.asyncio
+async def test_historie_kette_soll10_ist8_genehmigt_dann_nachgefuellt(
+    client, db_session, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    """U14/Prompt 13.3 Beispiel-Kette: Soll 10, Ist 8 -> Fehlbestand entstanden ->
+    keine Nachfüllung möglich -> Mindermenge genehmigt -> später 2 nachgefüllt ->
+    Ist wieder 10 -> Fehlbestand erledigt, Genehmigung endet gemeinsam. Jeder
+    Schritt muss als chronologisch korrekter Historieneintrag auffindbar sein."""
+    objekt, material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+
+    start = await client.post(
+        f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(mitarbeiter_token)
+    )
+    kontrolle_id = start.json()["id"]
+
+    put_response = await client.put(
+        f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
+        json={"istmenge": "8"},
+        headers=auth_header(mitarbeiter_token),
+    )
+    fehlbestand_id = put_response.json()["fehlbestand_id"]
+    assert fehlbestand_id is not None
+
+    await client.post(f"/api/v1/kontrollen/{kontrolle_id}/abschliessen", headers=auth_header(mitarbeiter_token))
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
+        json={"begruendung": "Lieferengpass, Ersatz bestellt"},
+        headers=auth_header(verantwortlicher_token),
+    )
+
+    await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/nachfuellungen",
+        json={"menge": "2"},
+        headers=auth_header(mitarbeiter_token),
+    )
+
+    result = await db_session.execute(select(Historie).order_by(Historie.zeitpunkt))
+    ereignisse = [h.ereignistyp for h in result.scalars().all()]
+
+    erwartete_reihenfolge = [
+        "kontrolle_gestartet",
+        "istmenge_erfasst",
+        "fehlbestand_entstanden",
+        "kontrolle_abgeschlossen",
+        "mindermenge_genehmigt",
+        "nachfuellung_erfasst",
+        "fehlbestand_erledigt",
+        "mindermenge_beendet_durch_erledigung",
+    ]
+    # Reine Teilfolge-Prüfung (keine anderen Events dazwischen erwartet, aber falls
+    # doch, ist die Reihenfolge der genannten Ereignisse das eigentlich Wichtige).
+    assert ereignisse == erwartete_reihenfolge

backend/tests/test_mindermenge.py

@@ -0,0 +1,187 @@
+import pytest
+from sqlalchemy import select
+
+from app.models.fehlbestand import Fehlbestand
+from app.models.mindermenge import MindermengenGenehmigung
+from app.models.objektposition import Objektposition
+from tests.conftest import auth_header, login
+
+
+async def _fehlbestand_ueber_kontrolle(client, token, objekt, material, istmenge: str, abschliessen=False):
+    start = await client.post(
+        f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(token)
+    )
+    kontrolle_id = start.json()["id"]
+    put_response = await client.put(
+        f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
+        json={"istmenge": istmenge},
+        headers=auth_header(token),
+    )
+    fehlbestand_id = put_response.json()["fehlbestand_id"]
+    if abschliessen:
+        await client.post(f"/api/v1/kontrollen/{kontrolle_id}/abschliessen", headers=auth_header(token))
+    return kontrolle_id, fehlbestand_id
+
+
+@pytest.mark.asyncio
+async def test_genehmigung_aendert_nichts_am_fehlbestand(
+    client, db_session, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    """U2 - Leitplanke: Genehmigung ändert NIEMALS Ist-Menge/Fehlmenge/Status."""
+    objekt, material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+    _, fehlbestand_id = await _fehlbestand_ueber_kontrolle(
+        client, mitarbeiter_token, objekt, material, "8", abschliessen=True
+    )
+
+    vor = await db_session.get(Fehlbestand, fehlbestand_id)
+    vor_status, vor_istmenge, vor_fehlmenge = vor.status, vor.istmenge, vor.fehlmenge
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    response = await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
+        json={"begruendung": "Lieferengpass, Ersatz bestellt"},
+        headers=auth_header(verantwortlicher_token),
+    )
+    assert response.status_code == 201
+    assert response.json()["status"] == "aktiv"
+
+    await db_session.refresh(vor)
+    assert vor.status == vor_status
+    assert vor.istmenge == vor_istmenge
+    assert vor.fehlmenge == vor_fehlmenge
+
+    objektposition = (
+        await db_session.execute(
+            select(Objektposition).where(
+                Objektposition.objekt_id == objekt.id, Objektposition.material_id == material.id
+            )
+        )
+    ).scalar_one()
+    assert float(objektposition.istmenge) == 0  # unverändert (Kontrolle/Genehmigung ändern nie Ist-Menge)
+
+
+@pytest.mark.asyncio
+async def test_mitarbeiter_darf_nicht_genehmigen(
+    client, objekt_mit_position, mitarbeiter_user
+):
+    objekt, material = objekt_mit_position
+    token = await login(client, "mitarbeiter1")
+    _, fehlbestand_id = await _fehlbestand_ueber_kontrolle(client, token, objekt, material, "8", abschliessen=True)
+
+    response = await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
+        json={"begruendung": "Versuch"},
+        headers=auth_header(token),
+    )
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_genehmigung_laeuft_bei_naechster_kontrolle_ab(
+    client, db_session, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    """U6 / Karte 08: Genehmigung gilt automatisch nur bis zur nächsten Kontrolle."""
+    objekt, material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+    _, fehlbestand_id = await _fehlbestand_ueber_kontrolle(
+        client, mitarbeiter_token, objekt, material, "8", abschliessen=True
+    )
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    genehmigung_response = await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
+        json={"begruendung": "Lieferengpass"},
+        headers=auth_header(verantwortlicher_token),
+    )
+    genehmigung_id = genehmigung_response.json()["id"]
+
+    naechste_kontrolle = await client.post(
+        f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(mitarbeiter_token)
+    )
+    assert naechste_kontrolle.status_code == 201
+    neue_kontrolle_id = naechste_kontrolle.json()["id"]
+
+    genehmigung = await db_session.get(MindermengenGenehmigung, genehmigung_id)
+    assert genehmigung.status.value == "abgelaufen"
+    assert str(genehmigung.beendende_kontrolle_id) == neue_kontrolle_id
+
+
+@pytest.mark.asyncio
+async def test_genehmigung_endet_gemeinsam_mit_erledigung(
+    client, db_session, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    """U7: wird die Abweichung behoben, erledigt sich Fehlbestand UND Genehmigung gemeinsam."""
+    objekt, material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+    _, fehlbestand_id = await _fehlbestand_ueber_kontrolle(
+        client, mitarbeiter_token, objekt, material, "8", abschliessen=True
+    )
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    genehmigung_response = await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
+        json={"begruendung": "Lieferengpass"},
+        headers=auth_header(verantwortlicher_token),
+    )
+    genehmigung_id = genehmigung_response.json()["id"]
+
+    nachfuellung = await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/nachfuellungen",
+        json={"menge": "2"},
+        headers=auth_header(mitarbeiter_token),
+    )
+    assert nachfuellung.json()["fehlbestand"]["status"] == "erledigt"
+
+    genehmigung = await db_session.get(MindermengenGenehmigung, genehmigung_id)
+    assert genehmigung.status.value == "beendet_durch_erledigung"
+
+
+@pytest.mark.asyncio
+async def test_doppelte_genehmigung_ist_409(
+    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    objekt, material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+    _, fehlbestand_id = await _fehlbestand_ueber_kontrolle(
+        client, mitarbeiter_token, objekt, material, "8", abschliessen=True
+    )
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    erste = await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
+        json={"begruendung": "Lieferengpass"},
+        headers=auth_header(verantwortlicher_token),
+    )
+    assert erste.status_code == 201
+
+    zweite = await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
+        json={"begruendung": "Nochmal"},
+        headers=auth_header(verantwortlicher_token),
+    )
+    assert zweite.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_genehmigung_auf_erledigten_fehlbestand_ist_409(
+    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    objekt, material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+    _, fehlbestand_id = await _fehlbestand_ueber_kontrolle(
+        client, mitarbeiter_token, objekt, material, "8", abschliessen=True
+    )
+    await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/nachfuellungen",
+        json={"menge": "2"},
+        headers=auth_header(mitarbeiter_token),
+    )
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    response = await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
+        json={"begruendung": "Zu spät"},
+        headers=auth_header(verantwortlicher_token),
+    )
+    assert response.status_code == 409