Section I backend — audit trail + topology source

docs/WALKTHROUGH.md

@@ -879,3 +879,43 @@ Die rekonstruierte Erzählung entlang der Zeit um einen **Anker-Alarm** — der
 **Warum existiert es / wo sitzt es?**
 …
 ```
+
+## I — Audit-Trail & Topologie-Quelle (Sektion I, Backend · Teil 1)
+
+> **Die Plattform sieht sich selbst — ehrlich.** Sektion I macht zwei Dinge sichtbar: WER/welches System hat WANN welche Erkenntnis abgerufen oder welche HITL-Entscheidung getroffen (Audit-Trail, zugleich AI-Act-/Art.-50-Nachweis-Beleg), und mit welchen Quellen/Konsumenten FOREMAN verbunden ist (Topologie). Das Zielbild der Designstudie §4I ist ein Multi-System-Mesh und als **[VISION]** markiert; dieser Teil baut die *ehrlich abgeleitete* Teilmenge — kein erfundener Knoten, kein erfundener Akteur. Voller Vertrag: GROUND_TRUTH §22.
+
+### Unveränderliches `audit_logs` (Migration 0010)
+
+**Was tut es?**
+Erweitert das nackte `audit_logs`-Skelett (id/user_id/action/target/created_at) additiv um den echten Trail: `actor` (HMAC-Token, nie Klartext), `actor_role`, `action_type` (CHECK), `target_kind`/`target_id`/`machine_id`, `origin` (CHECK), `detail` (JSONB), `occurred_at`. Ein PL/pgSQL-Trigger `trg_audit_logs_append_only` weist `UPDATE`/`DELETE` ab.
+
+**Warum existiert es / wo sitzt es?**
+Ein Audit-Trail, der sich ändern lässt, ist kein Beleg. Die Unveränderlichkeit ist deshalb **DB-seitig** erzwungen (Defense-in-Depth, Vorbild die `failure_*`-CheckConstraints), nicht nur app-seitig. Bewusst kein `TRUNCATE`-Trigger — TRUNCATE feuert keine Row-Trigger, und Test-/Reset-Pfade müssen die Tabelle leeren können. Der Legacy-`user_id`-FK bleibt erhalten, wird aber nicht befüllt: der namentliche Nachweis lebt im QM-System (System of Record), FOREMAN führt nur das Token (§8).
+
+### Writer (`audit/writer.py`)
+
+**Was tut es?**
+Ein reiner Zeilen-Bauer (`build_audit_log`: `AuditEntry` → ORM-Zeile, spiegelt `action_type` in die Legacy-`action`-NOT-NULL-Spalte) plus zwei Schreibwege. `record(session, entry)` schreibt IN die übergebene Session (atomar). `emit_mcp_retrieval(...)` schreibt best-effort auf EIGENER Session + Commit und schluckt jeden Fehler (loggt nur).
+
+**Warum existiert es / wo sitzt es?**
+Zwei Quellen, zwei Transaktions-Bedürfnisse: die HITL-Quittierung MUSS atomar mit dem Geschäfts-Write sein (eine Quittierung ohne ihren Beleg darf es nicht geben) → in-session. Der MCP-Abruf darf den read-only-Tool-Pfad NIE brechen → eigener Sink, best-effort. Beide bauen dieselbe Zeile.
+
+### Zwei reale Schreibpfade (HITL + MCP)
+
+**Was tut es?**
+HITL: die Drift-Quittier-Route (`reasoners/drift/router.py`) schreibt nach dem `flush` einen `hitl_acknowledge`-Eintrag in dieselbe Transaktion. MCP: der Tool-Wrapper `_measured` (`mcp/tools.py`) emittiert im `finally` einen `mcp_retrieval`-Eintrag mit Ziel-Kontext aus dem Abruf.
+
+**Warum existiert es / wo sitzt es?**
+Es gibt heute genau zwei reale HITL-/Abruf-Spuren — keine erfundenen. Die Quittier-Route liegt am Drift-Reasoner (nicht in `alarms.py`, das keine Ack-Route hat). Beim MCP ist `_measured` der ÄUSSERE Context-Manager, `_read_session` der innere: beim Block-Exit schließt die read-only-Session zuerst, dann läuft das `finally` — der Audit-Sink öffnet eine eigene Session, die Read-Invariante (I, §17.1) bleibt unangetastet. Der Akteur: `mcp/auth.py` kennt nur EINEN geteilten Token, keine Per-Client-Identität → der `actor` ist ein pseudonymisiertes Single-Consumer-Label, ehrlich genau eine Grenze; per-Client-Attribution ist [VISION].
+
+### Read-API + Topologie (`audit/service.py`, `topology/service.py`, Router)
+
+**Was tut es?**
+`GET /api/v1/audit` (gefiltert, paginiert, jüngste zuerst, nur Manager/Admin). `GET /api/v1/topology` leitet die Knoten ehrlich aus realen Quellen ab: Eingänge aus `data_points.source` + jüngster `readings`-Aktivität (simulation als intern), Gedächtnis-Substrat per Live-Smoke-Probe (`?probe=false` abschaltbar), F7-MCP-Grenze gespeist aus den `mcp_retrieval`-Audit-Zeilen. Status nur wo messbar (sonst `unbekannt`, nie grün); benannte Drittsysteme bleiben in einer separaten `[VISION]`-Liste.
+
+**Warum existiert es / wo sitzt es?**
+Die Linie quer durch FOREMAN: kein Fake. Eine Quelle ohne jüngste Daten wird `unbekannt`/`inaktiv`, nie verbunden gefärbt. ERP/Energiemanagement/externe Simulationssoftware existieren nicht als Integration → sie stehen ehrlich als Vision, nicht als grüner Knoten. Rollen-Split (Studie-Matrix): Audit nur Manager/Admin; Topologie Manager voll, Schichtleiter nur Verbindungsstatus (sein Datenqualitäts-Thema, kein Audit), Werker/Techniker kein Zugang. Schöne Kopplung: Teil A (Audit) speist Teil B (Topologie-MCP-Aktivität).
+
+### Gates (lokal grün)
+
+mypy strict 0, ruff clean + Format clean. Migration 0010 up/down getestet, Trigger blockt UPDATE/DELETE nachgewiesen (eigene ephemere DB je Lauf, eindeutiger Name). 607 Backend-Tests grün (ohne F-PRED, lokal Windows nativ) + 30 neue Sektion-I-Tests; Coverage ≥ 80 % auf `audit/`/`topology/` + den neuen Routern. MCP-Read-Only-Invariante nachgewiesen (Tool-Pfad mutiert keine Domänendaten; Audit-Sink committet separat). Hidden-Term-Scan über die neuen Außen-Strings sauber.

migrations/versions/0010_audit_trail_topology.py

@@ -0,0 +1,113 @@
+"""audit trail + topology source (Sektion I)
+
+Erweitert das nackte `audit_logs`-Skelett zum echten, UNVERÄNDERLICHEN Audit-Trail
+(zugleich AI-Act-/Art.-50-Nachweis-Beleg, §10.5): additive Spalten (actor/actor_role/
+action_type/target_kind/target_id/machine_id/origin/detail/occurred_at), CHECK-Constraints
+auf den geschlossenen Vokabularen (Defense-in-Depth analog failure_*), Lese-Indizes und
+ein Append-Only-Trigger, der UPDATE/DELETE DB-seitig abweist. Bestehende Spalten bleiben
+unangetastet; der Legacy-`user_id`-FK wird vom Schreibpfad nicht mehr befüllt (§8).
+
+Revision ID: 0010
+Revises: 0009
+Create Date: 2026-06-22
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision: str = "0010"
+down_revision: str | None = "0009"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+# Append-only erzwingen: jeder UPDATE/DELETE auf audit_logs wird abgewiesen. Bewusst
+# KEIN BEFORE-TRUNCATE-Trigger — TRUNCATE feuert keine Row-Trigger, und die Test-/
+# Admin-Reset-Pfade (TRUNCATE … CASCADE) müssen die Tabelle leeren können.
+# Je ein einzelnes Statement pro op.execute (asyncpg-Prepared-Statement erlaubt kein
+# Mehrfach-Kommando, vgl. 0002).
+_CREATE_FUNCTION = """
+CREATE OR REPLACE FUNCTION foreman_block_audit_logs_mutation()
+RETURNS trigger
+LANGUAGE plpgsql AS $$
+BEGIN
+    RAISE EXCEPTION
+        'audit_logs ist append-only: % ist nicht erlaubt (Sektion I, Migration 0010).',
+        TG_OP;
+END;
+$$;
+"""
+
+_CREATE_TRIGGER = """
+CREATE TRIGGER trg_audit_logs_append_only
+    BEFORE UPDATE OR DELETE ON audit_logs
+    FOR EACH ROW
+    EXECUTE FUNCTION foreman_block_audit_logs_mutation();
+"""
+
+
+def upgrade() -> None:
+    # --- Additive Spalten (bestehende id/user_id/action/target/created_at unberührt) ---
+    op.add_column("audit_logs", sa.Column("actor", sa.String(128), nullable=True))
+    op.add_column("audit_logs", sa.Column("actor_role", sa.String(32), nullable=True))
+    op.add_column("audit_logs", sa.Column("action_type", sa.String(64), nullable=True))
+    op.add_column("audit_logs", sa.Column("target_kind", sa.String(32), nullable=True))
+    op.add_column("audit_logs", sa.Column("target_id", sa.BigInteger(), nullable=True))
+    op.add_column("audit_logs", sa.Column("machine_id", sa.BigInteger(), nullable=True))
+    op.add_column("audit_logs", sa.Column("origin", sa.String(16), nullable=True))
+    op.add_column("audit_logs", sa.Column("detail", postgresql.JSONB(), nullable=True))
+    op.add_column(
+        "audit_logs",
+        sa.Column(
+            "occurred_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=True,
+        ),
+    )
+
+    # --- CHECK-Constraints (geschlossene, erweiterbare Vokabulare; NULL für Altzeilen) ---
+    op.create_check_constraint(
+        "ck_audit_logs_action_type",
+        "audit_logs",
+        "action_type IS NULL OR action_type IN ('hitl_acknowledge', 'mcp_retrieval')",
+    )
+    op.create_check_constraint(
+        "ck_audit_logs_origin",
+        "audit_logs",
+        "origin IS NULL OR origin IN ('dashboard', 'mcp', 'system')",
+    )
+
+    # --- Lese-Indizes (jüngste zuerst + häufige Filter) ---
+    op.create_index("ix_audit_logs_occurred", "audit_logs", ["occurred_at"])
+    op.create_index("ix_audit_logs_action_occurred", "audit_logs", ["action_type", "occurred_at"])
+    op.create_index("ix_audit_logs_machine", "audit_logs", ["machine_id"])
+    op.create_index("ix_audit_logs_target", "audit_logs", ["target_kind", "target_id"])
+
+    # --- Unveränderlichkeit DB-seitig (Defense-in-Depth) ---
+    op.execute(_CREATE_FUNCTION)
+    op.execute(_CREATE_TRIGGER)
+
+
+def downgrade() -> None:
+    op.execute("DROP TRIGGER IF EXISTS trg_audit_logs_append_only ON audit_logs;")
+    op.execute("DROP FUNCTION IF EXISTS foreman_block_audit_logs_mutation();")
+    op.drop_index("ix_audit_logs_target", table_name="audit_logs")
+    op.drop_index("ix_audit_logs_machine", table_name="audit_logs")
+    op.drop_index("ix_audit_logs_action_occurred", table_name="audit_logs")
+    op.drop_index("ix_audit_logs_occurred", table_name="audit_logs")
+    op.drop_constraint("ck_audit_logs_origin", "audit_logs", type_="check")
+    op.drop_constraint("ck_audit_logs_action_type", "audit_logs", type_="check")
+    op.drop_column("audit_logs", "occurred_at")
+    op.drop_column("audit_logs", "detail")
+    op.drop_column("audit_logs", "origin")
+    op.drop_column("audit_logs", "machine_id")
+    op.drop_column("audit_logs", "target_id")
+    op.drop_column("audit_logs", "target_kind")
+    op.drop_column("audit_logs", "action_type")
+    op.drop_column("audit_logs", "actor_role")
+    op.drop_column("audit_logs", "actor")

src/foreman/api/routers/audit.py

@@ -0,0 +1,60 @@
+# ============================================================
+#  FOREMAN — api/routers/audit.py (Sektion I)
+#  Zweck: Read-API des Audit-Trails — GET /api/v1/audit, gefiltert + paginiert,
+#         jüngste zuerst. Nur Manager (Studie-Rollenmatrix); Schichtleiter/
+#         Techniker/Werker erhalten 403. `actor` bleibt pseudonym.
+#  Architektur-Einordnung: HTTP-Schicht (Schicht 2).
+#  Konvention (§6): Type Hints überall, deutsche Kommentare, englische Bezeichner.
+# ============================================================
+from __future__ import annotations
+
+from collections.abc import Sequence
+from datetime import datetime
+from typing import Annotated
+
+from fastapi import APIRouter, HTTPException, Query, status
+
+from foreman.api.deps import CurrentUser, SessionDep
+from foreman.audit.schemas import AuditEntryRead
+from foreman.audit.service import list_audit
+from foreman.db.models import AuditLog
+from foreman.realtime.authz import ROLE_MANAGER
+
+router = APIRouter(prefix="/audit", tags=["audit"])
+
+# Audit-Einsicht: Manager (es gibt keine separate „admin"-Rolle → Manager).
+# Schichtleiter/Techniker/Werker bewusst ausgeschlossen (Plattform-/Audit-Kontext).
+_AUDIT_ROLES = frozenset({ROLE_MANAGER})
+
+
+@router.get("", response_model=list[AuditEntryRead])
+async def list_audit_entries(
+    session: SessionDep,
+    user: CurrentUser,
+    action_type: str | None = Query(default=None),
+    target_kind: str | None = Query(default=None),
+    target_id: int | None = Query(default=None),
+    actor: str | None = Query(default=None),
+    machine_id: int | None = Query(default=None),
+    since: Annotated[datetime | None, Query()] = None,
+    until: Annotated[datetime | None, Query()] = None,
+    limit: int = Query(default=100, ge=1, le=1000),
+    offset: int = Query(default=0, ge=0),
+) -> Sequence[AuditLog]:
+    """Audit-Trail (jüngste zuerst), gefiltert. Nur Manager — sonst 403."""
+    if user.role not in _AUDIT_ROLES:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, detail="Kein Zugriff auf den Audit-Trail"
+        )
+    return await list_audit(
+        session,
+        action_type=action_type,
+        target_kind=target_kind,
+        target_id=target_id,
+        actor=actor,
+        machine_id=machine_id,
+        since=since,
+        until=until,
+        limit=limit,
+        offset=offset,
+    )

src/foreman/api/routers/topology.py

@@ -0,0 +1,51 @@
+# ============================================================
+#  FOREMAN — api/routers/topology.py (Sektion I)
+#  Zweck: Read-API der Systemtopologie — GET /api/v1/topology. Manager voll
+#         (inkl. Audit-abgeleiteter MCP-Aktivität); Schichtleiter nur Verbindungs-
+#         status (kein Audit-Bezug); Werker/Techniker 403.
+#  Architektur-Einordnung: HTTP-Schicht (Schicht 2).
+#  Konvention (§6): Type Hints überall, deutsche Kommentare, englische Bezeichner.
+# ============================================================
+from __future__ import annotations
+
+from datetime import timedelta
+
+from fastapi import APIRouter, HTTPException, Query, status
+
+from foreman.api.deps import CurrentUser, SessionDep, SubstrateClientDep
+from foreman.mcp.auth import get_mcp_settings
+from foreman.realtime.authz import ROLE_MANAGER, ROLE_SHIFT_LEAD
+from foreman.topology.schemas import TopologyView
+from foreman.topology.service import build_topology
+
+router = APIRouter(prefix="/topology", tags=["topology"])
+
+# Volle Sicht (inkl. Audit-abgeleiteter MCP-Aktivität): nur Manager. (Die Studie §4I
+# nennt „Manager/Admin"; FOREMAN kennt keine separate admin-Rolle → durchgesetzt für manager.)
+_FULL_ROLES = frozenset({ROLE_MANAGER})
+# Nur Verbindungsstatus (kein Audit): zusätzlich Schichtleiter.
+_STATUS_ROLES = frozenset({ROLE_MANAGER, ROLE_SHIFT_LEAD})
+
+
+@router.get("", response_model=TopologyView)
+async def get_topology(
+    session: SessionDep,
+    user: CurrentUser,
+    substrate_client: SubstrateClientDep,
+    probe: bool = Query(default=True, description="Substrat live proben (schreibt Smoke-Marker)."),
+    fresh_within_minutes: int = Query(default=60, ge=1, le=10080),
+) -> TopologyView:
+    """Systemtopologie (ehrlich abgeleitet). Manager voll · Schichtleiter nur Status · sonst 403."""
+    if user.role not in _STATUS_ROLES:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, detail="Kein Zugriff auf die Systemtopologie"
+        )
+    return await build_topology(
+        session,
+        substrate_client=substrate_client,
+        mcp_token_configured=get_mcp_settings().token is not None,
+        fresh_window=timedelta(minutes=fresh_within_minutes),
+        probe_substrate=probe,
+        # Schichtleiter: kein Audit-Bezug → MCP-Knoten ohne Audit-Details, nur Status.
+        include_audit=user.role in _FULL_ROLES,
+    )

src/foreman/audit/__init__.py

@@ -0,0 +1,7 @@
+# ============================================================
+#  FOREMAN — audit/ (Sektion I)
+#  Zweck: Audit-Trail-Subsystem — Writer (in-session + best-effort out-of-band),
+#         Read-Service und Schemas. Der Trail ist zugleich der AI-Act-/Art.-50-
+#         Transparenz-/Nachweis-Beleg (§10.5) und unveränderlich (DB-Trigger, 0010).
+#  Architektur-Einordnung: Plattform-/Audit-Schicht (Schicht 2).
+# ============================================================

src/foreman/audit/schemas.py

@@ -0,0 +1,56 @@
+# ============================================================
+#  FOREMAN — audit/schemas.py (Sektion I)
+#  Zweck: Pydantic-Verträge des Audit-Trails — `AuditEntry` als interner
+#         Schreib-Eingang (Writer) und `AuditEntryRead` als pseudonyme
+#         Lese-Ausgabe der Read-API. `actor` ist stets ein HMAC-Token (§8).
+#  Architektur-Einordnung: Audit-Schicht (Schicht 2).
+#  Konvention (§6): Type Hints überall, deutsche Kommentare, englische Bezeichner.
+# ============================================================
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict
+
+
+class AuditEntry(BaseModel):
+    """Interner, unveränderlicher Schreib-Eingang für eine Audit-Zeile.
+
+    Wird vom Writer aus den realen Quellen (HITL-Quittierung, MCP-Abruf) gebaut.
+    `actor` ist bereits pseudonymisiert (HMAC-Token), nie Klartext.
+    """
+
+    model_config = ConfigDict(frozen=True)
+
+    action_type: str
+    origin: str
+    actor: str | None = None
+    actor_role: str | None = None
+    target_kind: str | None = None
+    target_id: int | None = None
+    machine_id: int | None = None
+    detail: dict[str, Any] | None = None
+    occurred_at: datetime | None = None
+
+
+class AuditEntryRead(BaseModel):
+    """Pseudonyme Lese-Sicht einer Audit-Zeile (Read-API, nur Manager).
+
+    Bewusst OHNE die Legacy-Spalten `action`/`target` und OHNE `user_id` — der
+    typisierte, pseudonyme Trail ist die maßgebliche Außensicht.
+    """
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    occurred_at: datetime | None
+    created_at: datetime
+    action_type: str | None
+    actor: str | None  # HMAC-Token, nie Klartext
+    actor_role: str | None
+    origin: str | None
+    target_kind: str | None
+    target_id: int | None
+    machine_id: int | None
+    detail: dict[str, Any] | None

src/foreman/audit/service.py

@@ -0,0 +1,50 @@
+# ============================================================
+#  FOREMAN — audit/service.py (Sektion I)
+#  Zweck: Lese-Kern des Audit-Trails — eine gefilterte, paginierte Abfrage,
+#         jüngste zuerst. Reine Query-Schicht (keine Rollen-Logik; die sitzt im
+#         Router). `actor` bleibt pseudonym (HMAC-Token), nie aufgelöst.
+#  Architektur-Einordnung: Audit-Schicht (Schicht 2).
+#  Konvention (§6): Type Hints überall, deutsche Kommentare, englische Bezeichner.
+# ============================================================
+from __future__ import annotations
+
+from collections.abc import Sequence
+from datetime import datetime
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from foreman.db.models import AuditLog
+
+
+async def list_audit(
+    session: AsyncSession,
+    *,
+    action_type: str | None = None,
+    target_kind: str | None = None,
+    target_id: int | None = None,
+    actor: str | None = None,
+    machine_id: int | None = None,
+    since: datetime | None = None,
+    until: datetime | None = None,
+    limit: int = 100,
+    offset: int = 0,
+) -> Sequence[AuditLog]:
+    """Listet Audit-Zeilen (jüngste zuerst), optional gefiltert. Pseudonym."""
+    stmt = select(AuditLog).order_by(AuditLog.occurred_at.desc(), AuditLog.id.desc())
+    if action_type is not None:
+        stmt = stmt.where(AuditLog.action_type == action_type)
+    if target_kind is not None:
+        stmt = stmt.where(AuditLog.target_kind == target_kind)
+    if target_id is not None:
+        stmt = stmt.where(AuditLog.target_id == target_id)
+    if actor is not None:
+        stmt = stmt.where(AuditLog.actor == actor)
+    if machine_id is not None:
+        stmt = stmt.where(AuditLog.machine_id == machine_id)
+    if since is not None:
+        stmt = stmt.where(AuditLog.occurred_at >= since)
+    if until is not None:
+        stmt = stmt.where(AuditLog.occurred_at <= until)
+    result = await session.scalars(stmt.limit(limit).offset(offset))
+    return result.all()