Karte 12: Eskalationslogik für lange offene Fehlbestände

Zwei Stufen (Erinnerung an Materialverantwortliche, Eskalation an
Leitungsverantwortliche/Administration) mit Tracking-Zeitstempel je Stufe
gegen Mehrfachversand. Zeitschwellen in eskalation_konfiguration (DB,
Singleton), admin-editierbar über GET/PUT /eskalation/konfiguration - bewusst
nicht fix im Code. Prüflauf per POST /eskalation/pruefen angestoßen,
Scheduling (Cron/systemd-Timer) bleibt Betriebsaufgabe.

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

DEVLOG.md

@@ -2655,3 +2655,16 @@ Keine Commits in dieser Session.
 - backend/tests/test_objekte.py           | 49 ++++++++++++++++++++++++++++
 
 ---
+## 2026-09-04 18:01 – 18:01 (0m)
+**Beschreibung:** Claude Code Session
+**Projekt:** asb-material
+
+### Commits
+- 7f06d1c Karte 10: Test-Label-Skript + Druck-/Scan-Testcheckliste
+
+### Geänderte Dateien
+- DEVLOG.md                              | 32 ++++++++++++++++++++++++++++++++
+- arbeitskarten/10_qr_barcode.md         | 15 +++++++++++++++
+- backend/scripts/test_label_erzeugen.py | 34 ++++++++++++++++++++++++++++++++++
+
+---

arbeitskarten/12_eskalation.md

@@ -15,3 +15,8 @@ Zeitschwellen konfigurierbar, nicht fix.
 
 ## Umsetzung
 - Eskalationslogik (Jobs, Schwellenwerte, Empfängerlogik) → Prompt 24 (Roadmap).
+
+## Umsetzungsstand (2026-09-04)
+- `app/services/eskalation.py`: zwei Stufen (Erinnerung an Materialverantwortliche, Eskalation an Leitungsverantwortliche/Administration), Tracking-Zeitstempel je Stufe verhindert Mehrfachversand.
+- Zeitschwellen in `eskalation_konfiguration` (DB, Singleton-Zeile) statt fix im Code - admin-editierbar über `GET/PUT /api/v1/eskalation/konfiguration`.
+- Prüflauf selbst per `POST /api/v1/eskalation/pruefen` (admin-only) angestoßen - **Scheduling (Cron) ist weiterhin Betriebsaufgabe**, kein eigener Scheduler-Prozess im Code (Prompt 19 Deployment-Regel). Auf dem Zielserver z. B. per systemd-Timer/Cron, der diesen Endpunkt periodisch mit Admin-Token aufruft.

backend/DEVLOG.md

@@ -92,3 +92,16 @@ Keine Commits in dieser Session.
 - flutter_app/pubspec.yaml                        |  1 +
 
 ---
+## 2026-09-04 18:01 – 18:04 (2m)
+**Beschreibung:** Claude Code Session
+**Projekt:** asb-material
+
+### Commits
+Keine Commits in dieser Session.
+
+### Geänderte Dateien
+- DEVLOG.md                              | 32 ++++++++++++++++++++++++++++++++
+- arbeitskarten/10_qr_barcode.md         | 15 +++++++++++++++
+- backend/scripts/test_label_erzeugen.py | 34 ++++++++++++++++++++++++++++++++++
+
+---

backend/alembic/versions/0006_add_eskalation_felder.py

@@ -0,0 +1,50 @@
+"""Eskalation offener Fehlbestände (Karte 12): Tracking-Felder + Konfiguration
+
+Revision ID: 0006_add_eskalation_felder
+Revises: 0005_add_objektposition_code
+Create Date: 2026-09-04
+
+Karte 12 / Prompt 24: Zeitstempel je Eskalationsstufe auf fehlbestand (damit
+dieselbe Stufe nicht bei jedem Prüflauf erneut verschickt wird), plus eine
+Singleton-Konfigurationstabelle für die Zeitschwellen - Nutzer-Vorgabe: die
+Schwellen dürfen nicht fix im Code stehen, müssen zur Laufzeit änderbar sein.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0006_add_eskalation_felder"
+down_revision: Union[str, None] = "0005_add_objektposition_code"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        ALTER TABLE fehlbestand
+            ADD COLUMN erinnerung_gesendet_am TIMESTAMPTZ,
+            ADD COLUMN eskalation_gesendet_am TIMESTAMPTZ;
+        """
+    )
+    op.execute(
+        """
+        CREATE TABLE eskalation_konfiguration (
+            id SERIAL PRIMARY KEY,
+            erinnerung_tage INTEGER NOT NULL DEFAULT 3,
+            leitung_tage INTEGER NOT NULL DEFAULT 7
+        );
+        """
+    )
+    op.execute("INSERT INTO eskalation_konfiguration (id, erinnerung_tage, leitung_tage) VALUES (1, 3, 7);")
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE eskalation_konfiguration;")
+    op.execute(
+        """
+        ALTER TABLE fehlbestand
+            DROP COLUMN erinnerung_gesendet_am,
+            DROP COLUMN eskalation_gesendet_am;
+        """
+    )

backend/app/api/v1/api.py

@@ -4,6 +4,7 @@ from app.api.v1.endpoints import (
     auth,
     benutzer,
     dashboard,
+    eskalation,
     fehlbestaende,
     health,
     kontrollen,
@@ -23,4 +24,5 @@ api_router.include_router(vorlagen.router, tags=["vorlagen"])
 api_router.include_router(objekte.router, tags=["objekte"])
 api_router.include_router(kontrollen.router, tags=["kontrollen"])
 api_router.include_router(fehlbestaende.router, tags=["fehlbestaende"])
+api_router.include_router(eskalation.router, tags=["eskalation"])
 api_router.include_router(dashboard.router, tags=["dashboard"])

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

@@ -0,0 +1,47 @@
+from fastapi import APIRouter, Depends
+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.eskalation import EskalationKonfiguration
+from app.schemas.eskalation import (
+    EskalationKonfigurationRead,
+    EskalationKonfigurationUpdate,
+    EskalationPruefungResult,
+)
+from app.services.eskalation import hole_konfiguration, pruefe_offene_fehlbestaende
+
+router = APIRouter()
+
+_admin_only = require_roles(RolleTyp.administration)
+
+
+@router.get("/eskalation/konfiguration", response_model=EskalationKonfigurationRead)
+async def lese_konfiguration(
+    db: AsyncSession = Depends(get_db), _=Depends(_admin_only)
+) -> EskalationKonfiguration:
+    return await hole_konfiguration(db)
+
+
+@router.put("/eskalation/konfiguration", response_model=EskalationKonfigurationRead)
+async def aendere_konfiguration(
+    payload: EskalationKonfigurationUpdate,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(_admin_only),
+) -> EskalationKonfiguration:
+    """Karte 12: Zeitschwellen zur Laufzeit änderbar (Nutzer-Vorgabe: nicht fix
+    im Code)."""
+    konfiguration = await hole_konfiguration(db)
+    konfiguration.erinnerung_tage = payload.erinnerung_tage
+    konfiguration.leitung_tage = payload.leitung_tage
+    await db.flush()
+    return konfiguration
+
+
+@router.post("/eskalation/pruefen", response_model=EskalationPruefungResult)
+async def pruefen(db: AsyncSession = Depends(get_db), _=Depends(_admin_only)) -> dict[str, int]:
+    """Trigger für den Eskalations-Prüflauf (Karte 12). Scheduling selbst ist
+    Betriebsaufgabe (z.B. System-Cron ruft diesen Endpunkt periodisch auf,
+    Prompt 19 Deployment-Regel: kein eigener Scheduler-Prozess im Code)."""
+    return await pruefe_offene_fehlbestaende(db)

backend/app/models/__init__.py

@@ -1,4 +1,5 @@
 from app.models.auth import Benutzer, BenutzerRolle, RolleTyp, Systemknoten, KnotenTyp
+from app.models.eskalation import EskalationKonfiguration
 from app.models.fehlbestand import Fehlbestand, FehlbestandStatus
 from app.models.historie import Historie
 from app.models.kontrolle import Kontrolle, KontrollStatus, Kontrollposition
@@ -16,6 +17,7 @@ __all__ = [
     "RolleTyp",
     "Systemknoten",
     "KnotenTyp",
+    "EskalationKonfiguration",
     "Fehlbestand",
     "FehlbestandStatus",
     "Historie",

backend/app/models/eskalation.py

@@ -0,0 +1,16 @@
+from sqlalchemy import Integer
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class EskalationKonfiguration(Base):
+    """Karte 12: Zeitschwellen für die Fehlbestand-Eskalation müssen zur Laufzeit
+    änderbar sein, nicht fix im Code (Nutzer-Vorgabe) - Singleton-Zeile (id=1),
+    per Admin-Endpunkt lesbar/änderbar statt Settings/Env."""
+
+    __tablename__ = "eskalation_konfiguration"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    erinnerung_tage: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
+    leitung_tage: Mapped[int] = mapped_column(Integer, nullable=False, default=7)

backend/app/models/fehlbestand.py

@@ -45,3 +45,7 @@ class Fehlbestand(Base):
         fehlbestand_status_pg, nullable=False, default=FehlbestandStatus.offen
     )
     erledigt_am: Mapped[datetime | None] = mapped_column(TIMESTAMP(timezone=True))
+    # Karte 12: Eskalationsstufen-Tracking, damit dieselbe Stufe nicht bei jedem
+    # Prüflauf erneut verschickt wird (NULL = Stufe noch nicht ausgelöst).
+    erinnerung_gesendet_am: Mapped[datetime | None] = mapped_column(TIMESTAMP(timezone=True))
+    eskalation_gesendet_am: Mapped[datetime | None] = mapped_column(TIMESTAMP(timezone=True))

backend/app/schemas/eskalation.py

@@ -0,0 +1,17 @@
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class EskalationKonfigurationRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    erinnerung_tage: int
+    leitung_tage: int
+
+
+class EskalationKonfigurationUpdate(BaseModel):
+    erinnerung_tage: int = Field(gt=0)
+    leitung_tage: int = Field(gt=0)
+
+
+class EskalationPruefungResult(BaseModel):
+    erinnerungen: int
+    eskalationen: int

backend/app/services/eskalation.py

@@ -0,0 +1,98 @@
+import logging
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.email import send_email
+from app.models.auth import Benutzer, BenutzerRolle, RolleTyp
+from app.models.eskalation import EskalationKonfiguration
+from app.models.fehlbestand import Fehlbestand, FehlbestandStatus
+from app.models.stammdaten import Material
+
+logger = logging.getLogger(__name__)
+
+
+async def hole_konfiguration(db: AsyncSession) -> EskalationKonfiguration:
+    """Singleton-Zeile (id=1), von Migration 0006 geseedet."""
+    konfiguration = await db.get(EskalationKonfiguration, 1)
+    if konfiguration is None:
+        konfiguration = EskalationKonfiguration(id=1)
+        db.add(konfiguration)
+        await db.flush()
+    return konfiguration
+
+
+async def _empfaenger(db: AsyncSession, rollen: list[RolleTyp]) -> list[str]:
+    result = await db.execute(
+        select(Benutzer.email)
+        .join(BenutzerRolle, BenutzerRolle.benutzer_id == Benutzer.id)
+        .where(BenutzerRolle.rolle.in_(rollen), Benutzer.aktiv.is_(True), Benutzer.email.is_not(None))
+    )
+    return [email for (email,) in result.all() if email]
+
+
+async def _sende(db: AsyncSession, *, fehlbestand: Fehlbestand, rollen: list[RolleTyp], betreff_praefix: str) -> None:
+    empfaenger = await _empfaenger(db, rollen)
+    if not empfaenger:
+        return
+    material = await db.get(Material, fehlbestand.material_id)
+    material_name = material.name if material else str(fehlbestand.material_id)
+    alter_tage = (datetime.now(timezone.utc) - fehlbestand.entstanden_am).days
+
+    subject = f"{betreff_praefix}: {material_name} seit {alter_tage} Tagen offen"
+    body = (
+        f"Material: {material_name}\n"
+        f"Objekt-ID: {fehlbestand.objekt_id}\n"
+        f"Fehlmenge: {fehlbestand.fehlmenge}\n"
+        f"Entstanden am: {fehlbestand.entstanden_am.isoformat()}\n"
+        f"Offen seit: {alter_tage} Tagen\n"
+    )
+    try:
+        await send_email(to=empfaenger, subject=subject, body=body)
+    except Exception:  # noqa: BLE001 - Versandfehler dürfen den Prüflauf nie abbrechen
+        logger.exception("Eskalations-E-Mail für Fehlbestand %s fehlgeschlagen", fehlbestand.id)
+
+
+async def pruefe_offene_fehlbestaende(db: AsyncSession) -> dict[str, int]:
+    """Karte 12: zwei Stufen, Zeitschwellen aus eskalation_konfiguration (DB,
+    zur Laufzeit änderbar, nicht fix im Code). Jede Stufe wird pro Fehlbestand
+    nur einmal ausgelöst (Tracking-Zeitstempel), unabhängig davon, wie oft dieser
+    Prüflauf angestoßen wird - Aufrufer (Cron/manueller Trigger) ist bewusst
+    nicht Teil dieser Funktion, Scheduling ist Betriebsaufgabe (Prompt 19
+    Deployment-Regel)."""
+    konfiguration = await hole_konfiguration(db)
+    jetzt = datetime.now(timezone.utc)
+    erinnerung_schwelle = jetzt - timedelta(days=konfiguration.erinnerung_tage)
+    eskalation_schwelle = jetzt - timedelta(days=konfiguration.leitung_tage)
+
+    result = await db.execute(
+        select(Fehlbestand).where(Fehlbestand.status != FehlbestandStatus.erledigt)
+    )
+    offene = result.scalars().all()
+
+    erinnerungen = 0
+    eskalationen = 0
+    for fehlbestand in offene:
+        if fehlbestand.erinnerung_gesendet_am is None and fehlbestand.entstanden_am <= erinnerung_schwelle:
+            await _sende(
+                db,
+                fehlbestand=fehlbestand,
+                rollen=[RolleTyp.materialverantwortlicher],
+                betreff_praefix="Erinnerung: Fehlbestand offen",
+            )
+            fehlbestand.erinnerung_gesendet_am = jetzt
+            erinnerungen += 1
+
+        if fehlbestand.eskalation_gesendet_am is None and fehlbestand.entstanden_am <= eskalation_schwelle:
+            await _sende(
+                db,
+                fehlbestand=fehlbestand,
+                rollen=[RolleTyp.leitungsverantwortlicher, RolleTyp.administration],
+                betreff_praefix="Eskalation: Fehlbestand lange offen",
+            )
+            fehlbestand.eskalation_gesendet_am = jetzt
+            eskalationen += 1
+
+    await db.flush()
+    return {"erinnerungen": erinnerungen, "eskalationen": eskalationen}

backend/tests/test_eskalation.py

@@ -0,0 +1,73 @@
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from app.models.fehlbestand import Fehlbestand
+from app.services.eskalation import pruefe_offene_fehlbestaende
+from tests.conftest import auth_header, login
+
+
+async def _fehlbestand_erzeugen(client, token, objekt, material, istmenge: str):
+    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),
+    )
+    return put_response.json()["fehlbestand_id"]
+
+
+@pytest.mark.asyncio
+async def test_konfiguration_lesen_und_aendern(client, admin_user):
+    token = await login(client, "admin1")
+    lesen = await client.get("/api/v1/eskalation/konfiguration", headers=auth_header(token))
+    assert lesen.status_code == 200
+    assert lesen.json() == {"erinnerung_tage": 3, "leitung_tage": 7}
+
+    aendern = await client.put(
+        "/api/v1/eskalation/konfiguration",
+        json={"erinnerung_tage": 1, "leitung_tage": 2},
+        headers=auth_header(token),
+    )
+    assert aendern.status_code == 200
+    assert aendern.json() == {"erinnerung_tage": 1, "leitung_tage": 2}
+
+
+@pytest.mark.asyncio
+async def test_mitarbeiter_darf_konfiguration_nicht_lesen(client, mitarbeiter_user):
+    token = await login(client, "mitarbeiter1")
+    response = await client.get("/api/v1/eskalation/konfiguration", headers=auth_header(token))
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_pruefung_setzt_stufen_nur_einmal(
+    client, db_session, objekt_mit_position, mitarbeiter_user, admin_user
+):
+    objekt, material = objekt_mit_position
+    token = await login(client, "mitarbeiter1")
+    fehlbestand_id = await _fehlbestand_erzeugen(client, token, objekt, material, "3")
+
+    fehlbestand = await db_session.get(Fehlbestand, fehlbestand_id)
+    fehlbestand.entstanden_am = datetime.now(timezone.utc) - timedelta(days=10)
+    await db_session.flush()
+
+    admin_token = await login(client, "admin1")
+    await client.put(
+        "/api/v1/eskalation/konfiguration",
+        json={"erinnerung_tage": 3, "leitung_tage": 7},
+        headers=auth_header(admin_token),
+    )
+
+    ergebnis = await pruefe_offene_fehlbestaende(db_session)
+    assert ergebnis == {"erinnerungen": 1, "eskalationen": 1}
+
+    await db_session.refresh(fehlbestand)
+    assert fehlbestand.erinnerung_gesendet_am is not None
+    assert fehlbestand.eskalation_gesendet_am is not None
+
+    wiederholung = await pruefe_offene_fehlbestaende(db_session)
+    assert wiederholung == {"erinnerungen": 0, "eskalationen": 0}