ai/docs/GOV-04-PRUEFPROTOKOLL.md

@@ -0,0 +1,64 @@
+# GOV-04 — Prüfprotokoll: Kostenkontrolle & Quota (Kopplung Core LIC-03)
+
+Datum: 2026-09-27
+Testhost: root@192.168.1.204 (Wegwerf-Rolle/DB `gov04_test`, nach Abschluss entfernt)
+
+## Ausgangslage geprüft
+
+- `ENG-05` (`ai/internal/queue`): Postgres-Warteschlange für Inferenz-Aufrufe — vorausgesetzt, unverändert im Kernverhalten.
+- Core `LIC-03` (`internal/usage`): Nutzungszähler/Quotas je Tenant, HTTP-Kopplung über `internal/resync.Handler.UsageHandler` (Delta-Meldung, "applied genau einmal"). AI ist ein eigenständiges Go-Modul (`ai/go.mod`) und kann `internal/usage` nicht importieren — Muster identisch zu `dms/internal/storage.HTTPUsageReporter` / `mail/internal/storage`: lokaler Zähler/Quota-Spiegel + Delta-Meldung an Core.
+- `GOV-03` (`ai/internal/residency`): Sibling-Muster für tenant-gescopte Governance-Konfiguration in diesem Modul (Store + strukturelle Interfaces statt harter Kopplung).
+
+## Umsetzung
+
+- `ai/internal/quota`: `Store` mit `Check`/`Enforce`/`RecordCall`, Metrik `ai_inference_calls`, Warnschwelle 80 % (analog Core `internal/usage`), `HTTPReporter` meldet jedes Delta an Cores Resync-Endpunkt (Kopplung LIC-03).
+- `ai/migrations/0007_ai_usage_quota.{up,down}.sql`: lokale Tabellen `ai_usage_counters`/`ai_usage_quotas`.
+- `ai/migrations/0008_inference_jobs_tenant.{up,down}.sql`: `tenant_slug`-Spalte auf `inference_jobs` (nullable), damit ENG-05-Jobs einem Tenant zuordenbar sind.
+- `ai/internal/queue`: `QuotaEnforcer`-Interface, `Queue.SetQuotaEnforcer` — `Enqueue` blockt VOR dem Anlegen eines Jobs, wenn die Tenant-Quota erreicht ist (AK2).
+- `ai/internal/queue`: `UsageRecorder`-Interface, `Worker.SetUsageRecorder` — Zählung erst NACH erfolgreichem `Complete`, sodass der Zählerstand tatsächlich ausgeführte Aufrufe abbildet (Prüfung 3), nicht bloße Versuche.
+
+## Akzeptanzkriterien
+
+| # | Kriterium | Ergebnis |
+|---|---|---|
+| 1 | Verbrauch ist je Tenant nachvollziehbar erfasst | Erfüllt — `quota.Store.Check`/`Get`, strikt `tenant_slug`-gescoped, siehe `TestStore_CheckTracksPerTenant`, `TestQuota_TenantIsolation` |
+| 2 | Überschreitung der Quota wird zuverlässig verhindert | Erfüllt — `Queue.Enqueue` blockt vor Jobanlage, siehe `TestStore_EnforceBlocksWhenExceeded`, `TestQuota_BlocksEnqueueAndTracksActualCompletions` |
+| 3 | Näherung an die Grenze wird vorab gemeldet | Erfüllt — `Store.Enforce` löst bei 80 % `WarnFunc` aus, VOR Erreichen der Grenze, siehe `TestStore_EnforceWarnsBeforeExceeding` |
+
+## Prüfungen vor der Abnahme
+
+### 1. Quota-Überschreitung im Test zuverlässig blockiert
+
+Zielwert: bei erreichter Tenant-Quota wird kein weiterer Job angelegt, `Enqueue` liefert `quota.ErrQuotaExceeded`.
+
+Ergebnis: **bestanden**. `TestStore_EnforceBlocksWhenExceeded` (reine Store-Logik) und `TestQuota_BlocksEnqueueAndTracksActualCompletions` (End-to-End über echte Queue + Worker + Postgres): nach 3 tatsächlich abgeschlossenen Aufrufen bei Limit 3 wird der 4. `Enqueue`-Versuch abgelehnt, `Queue.Depth().Pending` bleibt bei 0 — kein Job entsteht.
+
+### 2. Vorabwarnung im Test ausgelöst
+
+Zielwert: bei 80 % Auslastung wird die Warnreaktion ausgelöst, der Aufruf selbst bleibt aber noch erlaubt.
+
+Ergebnis: **bestanden**. `TestStore_EnforceWarnsBeforeExceeding`: bei 8 von 10 konfigurierten Aufrufen (80 %) löst `Enforce` genau eine Warnung aus und lässt den Aufruf weiterhin zu.
+
+### 3. Verbrauchszählung gegen tatsächliche Aufrufe abgeglichen
+
+Zielwert: der Zählerstand entspricht der Anzahl tatsächlich über die Queue ausgeführter (abgeschlossener) Inferenz-Aufrufe, nicht der Anzahl Enqueue-Versuche.
+
+Ergebnis: **bestanden**. `TestQuota_BlocksEnqueueAndTracksActualCompletions`: 3 Jobs werden über `Worker.ProcessOne` real mit einem Mock-Provider ausgeführt (Dequeue → execute → Complete); `quota.Store.Get` liefert danach exakt 3 — die Zählung erfolgt im Worker erst nach erfolgreichem `Complete`, ein abgelehnter/blockierter Enqueue-Versuch erhöht den Zähler nicht. Die Kopplung an Core LIC-03 (Delta-Meldung je erfassten Aufruf) ist zusätzlich durch `TestStore_RecordCallReportsToCore` (Fake-Reporter erfasst genau die gemeldeten Deltas) nachgewiesen.
+
+## Testausführung
+
+```
+TEST_TENANT_DSN=postgres://gov04_test:***@localhost:5432/gov04_test?sslmode=disable \
+  go test ./ai/internal/quota/... ./ai/internal/queue/... -v
+```
+
+Alle 14 Tests (5 `quota`, 9 `queue`, davon 4 neu für GOV-04) **PASS**, inklusive der bereits bestehenden ENG-05/ENG-07-Tests (keine Regression).
+
+`go build ./...` und `go vet ./...` im `ai`-Modul: keine Fehler.
+
+Voller Modul-Testlauf `go test ./...`: alle Pakete bis auf `ai/internal/vectorindex` grün; `TestSearch_LatencyAtScale` in `vectorindex` läuft auf diesem Testhost in einen 10-Minuten-Timeout — vorbestehend, unabhängig von GOV-04 (kein Bezug zu Queue/Quota-Code), nicht Teil dieser Kachel und daher nicht behoben.
+
+## Offene Punkte / Entscheidungen
+
+- Core exponiert (Stand LIC-03) nur einen Write-Endpunkt (`resync.UsageHandler`, Delta anwenden) — keinen synchronen Read/Check-Endpunkt für Module. Enforcement erfolgt daher lokal gegen einen AI-eigenen Zähler-/Quota-Spiegel (identisches Muster zu dms/mail), Core wird informativ nachgeführt. Das ist eine reversible, kleine Design-Entscheidung im Rahmen der bestehenden Konventionen, keine Abweichung von den Akzeptanzkriterien.
+- Setzen der Quota-Obergrenze je Tenant (Admin-Oberfläche) ist Core `LIC-04` und damit ausdrücklich nicht Teil dieser Kachel; `quota.Store.SetQuota` existiert als Schnittstelle dafür.

ai/internal/queue/queue.go

@@ -47,14 +47,24 @@ type Job struct {
 	MaxAttempts int
 	Result      json.RawMessage
 	LastError   *string
+	TenantSlug  string
 }
 
 const DefaultMaxAttempts = 5
 
+// QuotaEnforcer ist die schmale Schnittstelle zu GOV-04 (ai/internal/quota):
+// blockiert Enqueue, wenn die Tenant-Quota bereits erreicht ist. Als
+// Interface hier statt Import, damit queue (ENG-05) nicht von quota
+// abhaengt — quota.Store erfuellt diese Schnittstelle strukturell.
+type QuotaEnforcer interface {
+	Enforce(ctx context.Context, tenantSlug string) error
+}
+
 type Queue struct {
 	pool           *pgxpool.Pool
 	staleLockAfter time.Duration
 	limits         limits.Limits
+	quota          QuotaEnforcer
 }
 
 func NewQueue(pool *pgxpool.Pool, staleLockAfter time.Duration) *Queue {
@@ -65,14 +75,22 @@ func NewQueue(pool *pgxpool.Pool, staleLockAfter time.Duration) *Queue {
 // vor allem fuer Tests mit kleinen, deterministischen Werten.
 func (q *Queue) SetLimits(l limits.Limits) { q.limits = l }
 
+// SetQuotaEnforcer verknuepft die Queue mit GOV-04s Kostenkontrolle; ohne
+// Aufruf bleibt Enqueue unveraendert unbegrenzt (Rueckwaertskompatibilitaet).
+func (q *Queue) SetQuotaEnforcer(e QuotaEnforcer) { q.quota = e }
+
 type EnqueueOptions struct {
 	IdempotencyKey string
 	MaxAttempts    int
+	TenantSlug     string
 }
 
 // Enqueue reiht einen Inferenz-Aufruf asynchron ein (AK1). Bei gesetztem
 // IdempotencyKey liefert ein erneuter Aufruf mit demselben Key die ID des
-// bereits vorhandenen Jobs statt eines Duplikats.
+// bereits vorhandenen Jobs statt eines Duplikats. Ist ein QuotaEnforcer
+// gesetzt und TenantSlug angegeben, wird die Tenant-Quota VOR dem Anlegen
+// des Jobs geprueft (GOV-04 AK2) — eine ueberschrittene Quota verhindert
+// zuverlaessig, dass ueberhaupt erst neue Arbeit entsteht.
 func (q *Queue) Enqueue(ctx context.Context, jobType Kind, payload any, opts EnqueueOptions) (string, error) {
 	payloadJSON, err := json.Marshal(payload)
 	if err != nil {
@@ -81,6 +99,11 @@ func (q *Queue) Enqueue(ctx context.Context, jobType Kind, payload any, opts Enq
 	if err := q.limits.CheckSize(len(payloadJSON)); err != nil {
 		return "", err
 	}
+	if q.quota != nil && opts.TenantSlug != "" {
+		if err := q.quota.Enforce(ctx, opts.TenantSlug); err != nil {
+			return "", err
+		}
+	}
 	maxAttempts := opts.MaxAttempts
 	if maxAttempts <= 0 {
 		maxAttempts = DefaultMaxAttempts
@@ -89,13 +112,17 @@ func (q *Queue) Enqueue(ctx context.Context, jobType Kind, payload any, opts Enq
 	if opts.IdempotencyKey != "" {
 		idempotencyKey = opts.IdempotencyKey
 	}
+	var tenantSlug any
+	if opts.TenantSlug != "" {
+		tenantSlug = opts.TenantSlug
+	}
 	var id string
 	err = q.pool.QueryRow(ctx, `
-		INSERT INTO inference_jobs (job_type, payload, idempotency_key, max_attempts)
-		VALUES ($1, $2, $3, $4)
+		INSERT INTO inference_jobs (job_type, payload, idempotency_key, max_attempts, tenant_slug)
+		VALUES ($1, $2, $3, $4, $5)
 		ON CONFLICT (idempotency_key) DO UPDATE SET job_type = inference_jobs.job_type
 		RETURNING id
-	`, string(jobType), payloadJSON, idempotencyKey, maxAttempts).Scan(&id)
+	`, string(jobType), payloadJSON, idempotencyKey, maxAttempts, tenantSlug).Scan(&id)
 	if err != nil {
 		return "", fmt.Errorf("queue: job einreihen: %w", err)
 	}
@@ -112,7 +139,7 @@ func (q *Queue) Dequeue(ctx context.Context, workerID string) (*Job, error) {
 	defer func() { _ = tx.Rollback(ctx) }()
 
 	row := tx.QueryRow(ctx, `
-		SELECT id, job_type, payload, status, attempts, max_attempts, last_error
+		SELECT id, job_type, payload, status, attempts, max_attempts, last_error, COALESCE(tenant_slug, '')
 		FROM inference_jobs
 		WHERE (
 			(status = 'pending' AND available_at <= now())
@@ -125,7 +152,7 @@ func (q *Queue) Dequeue(ctx context.Context, workerID string) (*Job, error) {
 
 	var j Job
 	var jobType string
-	if err := row.Scan(&j.ID, &jobType, &j.Payload, &j.Status, &j.Attempts, &j.MaxAttempts, &j.LastError); err != nil {
+	if err := row.Scan(&j.ID, &jobType, &j.Payload, &j.Status, &j.Attempts, &j.MaxAttempts, &j.LastError, &j.TenantSlug); err != nil {
 		if errors.Is(err, pgx.ErrNoRows) {
 			return nil, ErrNoJobAvailable
 		}
@@ -193,9 +220,9 @@ func (q *Queue) Status(ctx context.Context, jobID string) (*Job, error) {
 	var j Job
 	var jobType string
 	err := q.pool.QueryRow(ctx, `
-		SELECT id, job_type, payload, status, attempts, max_attempts, result, last_error
+		SELECT id, job_type, payload, status, attempts, max_attempts, result, last_error, COALESCE(tenant_slug, '')
 		FROM inference_jobs WHERE id = $1
-	`, jobID).Scan(&j.ID, &jobType, &j.Payload, &j.Status, &j.Attempts, &j.MaxAttempts, &j.Result, &j.LastError)
+	`, jobID).Scan(&j.ID, &jobType, &j.Payload, &j.Status, &j.Attempts, &j.MaxAttempts, &j.Result, &j.LastError, &j.TenantSlug)
 	if err != nil {
 		if errors.Is(err, pgx.ErrNoRows) {
 			return nil, ErrNotFound

ai/internal/queue/quota_integration_test.go

@@ -0,0 +1,99 @@
+package queue
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"gitea.perlbach24.de/scripte/nexarch/ai/internal/provider"
+	"gitea.perlbach24.de/scripte/nexarch/ai/internal/provider/mock"
+	"gitea.perlbach24.de/scripte/nexarch/ai/internal/quota"
+)
+
+// GOV-04 Pruefung 1 + 3: eine reale, ueber die Queue (ENG-05) ausgefuehrte
+// Inferenz-Last erhoeht den Tenant-Zaehler exakt um die Anzahl tatsaechlich
+// abgeschlossener Aufrufe, und eine erreichte Quota verhindert zuverlaessig,
+// dass weitere Jobs ueberhaupt erst eingereiht werden.
+func TestQuota_BlocksEnqueueAndTracksActualCompletions(t *testing.T) {
+	pool := newPool(t)
+	quotaStore := quota.NewStore(pool, nil)
+	ctx := context.Background()
+
+	const tenant = "tenant-integration"
+	const limit = 3
+	if err := quotaStore.SetQuota(ctx, tenant, quota.InferenceCallsMetric, limit); err != nil {
+		t.Fatalf("quota setzen: %v", err)
+	}
+
+	q := NewQueue(pool, time.Minute)
+	q.SetQuotaEnforcer(quotaStore)
+
+	p := &mock.Provider{ChatResponse: provider.ChatResponse{Message: provider.Message{Content: "ok"}}}
+	w := NewWorker(q, p, "worker-quota-test")
+	w.SetUsageRecorder(quotaStore)
+
+	// Genau `limit` Auftraege einreihen und abarbeiten -> danach ist die
+	// Quota erreicht (AK1: Zaehlerstand nachvollziehbar je Tenant).
+	for i := 0; i < limit; i++ {
+		if _, err := q.Enqueue(ctx, KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{TenantSlug: tenant}); err != nil {
+			t.Fatalf("enqueue %d haette erlaubt sein muessen: %v", i, err)
+		}
+		if err := w.ProcessOne(ctx); err != nil {
+			t.Fatalf("verarbeitung %d fehlgeschlagen: %v", i, err)
+		}
+	}
+
+	value, err := quotaStore.Get(ctx, tenant, quota.InferenceCallsMetric)
+	if err != nil {
+		t.Fatalf("zaehlerstand lesen: %v", err)
+	}
+	if value != limit {
+		t.Fatalf("erwartet zaehlerstand %d (= tatsaechlich ausgefuehrte aufrufe), bekommen %d", limit, value)
+	}
+
+	// AK2 / Pruefung 1: der naechste Enqueue-Versuch wird zuverlaessig
+	// abgelehnt, BEVOR ein weiterer Job entsteht.
+	_, err = q.Enqueue(ctx, KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{TenantSlug: tenant})
+	if !errors.Is(err, quota.ErrQuotaExceeded) {
+		t.Fatalf("enqueue nach erreichter quota haette blockiert werden muessen, bekommen: %v", err)
+	}
+
+	depth, err := q.Depth(ctx)
+	if err != nil {
+		t.Fatalf("warteschlangenstand lesen: %v", err)
+	}
+	if depth.Pending != 0 {
+		t.Fatalf("blockierter enqueue-versuch haette keinen job anlegen duerfen, pending=%d", depth.Pending)
+	}
+}
+
+// Ein Tenant ohne konfigurierte Quota bleibt unbeeinflusst von einem Tenant,
+// der seine Quota bereits erreicht hat (kein globaler, sondern strikt
+// tenant-gescopter Zustand).
+func TestQuota_TenantIsolation(t *testing.T) {
+	pool := newPool(t)
+	quotaStore := quota.NewStore(pool, nil)
+	ctx := context.Background()
+
+	if err := quotaStore.SetQuota(ctx, "tenant-blocked", quota.InferenceCallsMetric, 1); err != nil {
+		t.Fatalf("quota setzen: %v", err)
+	}
+
+	q := NewQueue(pool, time.Minute)
+	q.SetQuotaEnforcer(quotaStore)
+
+	if _, err := q.Enqueue(ctx, KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{TenantSlug: "tenant-blocked"}); err != nil {
+		t.Fatalf("erster enqueue haette erlaubt sein muessen: %v", err)
+	}
+	if err := quotaStore.RecordCall(ctx, "tenant-blocked"); err != nil {
+		t.Fatalf("call erfassen: %v", err)
+	}
+
+	if _, err := q.Enqueue(ctx, KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{TenantSlug: "tenant-blocked"}); !errors.Is(err, quota.ErrQuotaExceeded) {
+		t.Fatalf("tenant-blocked haette blockiert sein muessen: %v", err)
+	}
+	if _, err := q.Enqueue(ctx, KindChat, provider.ChatRequest{Model: "test"}, EnqueueOptions{TenantSlug: "tenant-unaffected"}); err != nil {
+		t.Fatalf("ein anderer tenant ohne quota-treffer sollte unbeeinflusst bleiben: %v", err)
+	}
+}

ai/internal/queue/worker.go

@@ -12,6 +12,14 @@ import (
 	"gitea.perlbach24.de/scripte/nexarch/ai/internal/provider"
 )
 
+// UsageRecorder ist die schmale Schnittstelle zu GOV-04 (ai/internal/quota):
+// zaehlt einen tatsaechlich ausgefuehrten Inferenz-Aufruf. Als Interface
+// hier statt Import, damit queue (ENG-05) nicht von quota abhaengt —
+// quota.Store erfuellt diese Schnittstelle strukturell.
+type UsageRecorder interface {
+	RecordCall(ctx context.Context, tenantSlug string) error
+}
+
 // Worker holt Jobs aus der Queue und fuehrt sie ueber den Provider aus.
 // Ein Ausfuehrungsfehler blockiert den Anfrage-Pfad nicht (AK1) — der
 // aufrufende Code hat laengst eine Job-ID zurueckbekommen und pollt oder
@@ -21,6 +29,7 @@ type Worker struct {
 	provider provider.Provider
 	id       string
 	limits   limits.Limits
+	usage    UsageRecorder
 }
 
 func NewWorker(q *Queue, p provider.Provider, workerID string) *Worker {
@@ -31,6 +40,10 @@ func NewWorker(q *Queue, p provider.Provider, workerID string) *Worker {
 // vor allem fuer Tests mit kleinen, deterministischen Werten.
 func (w *Worker) SetLimits(l limits.Limits) { w.limits = l }
 
+// SetUsageRecorder verknuepft den Worker mit GOV-04s Verbrauchszaehlung; ohne
+// Aufruf bleibt die Ausfuehrung unveraendert ungezaehlt.
+func (w *Worker) SetUsageRecorder(u UsageRecorder) { w.usage = u }
+
 // ProcessOne holt einen Job und fuehrt ihn aus. Liefert ErrNoJobAvailable,
 // wenn die Queue leer ist (kein Fehlerzustand).
 func (w *Worker) ProcessOne(ctx context.Context) error {
@@ -54,6 +67,16 @@ func (w *Worker) ProcessOne(ctx context.Context) error {
 	if completeErr := w.queue.Complete(ctx, job.ID, result); completeErr != nil {
 		return fmt.Errorf("queue: job %s abschliessen: %w", job.ID, completeErr)
 	}
+	// Zaehlung erst NACH erfolgreichem Abschluss (GOV-04 Pruefung 3): der
+	// Zaehlerstand bildet tatsaechlich ausgefuehrte Aufrufe ab, nicht
+	// blosse Versuche — ein Fehlschlag der Zaehlung faellt bereits
+	// erfolgreiche Inferenz nicht rueckwirkend zurueck, sondern wird nur
+	// geloggt.
+	if w.usage != nil && job.TenantSlug != "" {
+		if usageErr := w.usage.RecordCall(ctx, job.TenantSlug); usageErr != nil {
+			log.Printf("queue: verbrauchszaehlung fuer job %s (tenant %s) fehlgeschlagen: %v", job.ID, job.TenantSlug, usageErr)
+		}
+	}
 	return nil
 }
 

ai/internal/quota/quota.go

@@ -0,0 +1,183 @@
+// Package quota implementiert GOV-04: Verbrauchszaehlung und Obergrenzen fuer
+// KI-Inferenz je Tenant, gekoppelt an Core LIC-03 (Nutzungszaehler & Quotas).
+// AI ist ein eigenes Go-Modul (ai/go.mod) und kann Core internal/usage nicht
+// importieren (siehe ai/internal/migrate-Paketkommentar) — Store fuehrt
+// dieselbe Zaehler-/Quota-Semantik lokal und meldet jedes Delta zusaetzlich
+// an Core weiter (Reporter), damit ein Tenant genau einen konsistenten
+// Nutzungsstand fuer Cores Verwaltungsoberflaeche hat (dms/mail-Muster,
+// siehe dms/internal/storage.HTTPUsageReporter).
+package quota
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// InferenceCallsMetric ist der Metrikname fuer gezaehlte KI-Inferenz-Aufrufe.
+const InferenceCallsMetric = "ai_inference_calls"
+
+// ErrQuotaExceeded blockiert einen neuen Inferenz-Aufruf (Akzeptanzkriterium 2).
+var ErrQuotaExceeded = errors.New("quota: ki-inferenz-quota fuer diesen tenant ueberschritten")
+
+// Status ist die aus Zaehlerstand/Quota abgeleitete Reaktion.
+type Status string
+
+const (
+	StatusOK       Status = "ok"
+	StatusWarning  Status = "warning"  // Schwelle (80%) erreicht, noch nicht ueberschritten
+	StatusExceeded Status = "exceeded" // Quota erreicht/ueberschritten
+)
+
+const warningThreshold = 0.8
+
+// Reporter meldet ein angewendetes Delta an Core LIC-03 weiter. Schmale
+// Schnittstelle, damit Tests einen Fake statt eines echten HTTP-Aufrufs
+// einsetzen koennen.
+type Reporter interface {
+	Report(ctx context.Context, tenantSlug, metric string, delta int64) error
+}
+
+// WarnFunc wird bei StatusWarning aufgerufen (Akzeptanzkriterium 3), bevor
+// die Quota ueberschritten wird — Standard ist ein strukturiertes Log.
+type WarnFunc func(tenantSlug string, value, limit int64)
+
+type Store struct {
+	pool     *pgxpool.Pool
+	reporter Reporter
+	warn     WarnFunc
+}
+
+// NewStore verknuepft den lokalen Zaehler mit einem optionalen Reporter an
+// Core LIC-03 (reporter darf nil sein, z.B. in Tests ohne Core-Anbindung).
+func NewStore(pool *pgxpool.Pool, reporter Reporter) *Store {
+	return &Store{pool: pool, reporter: reporter, warn: defaultWarn}
+}
+
+func defaultWarn(tenantSlug string, value, limit int64) {
+	slog.Warn("ki-inferenz-quota naehert sich der grenze", "tenant", tenantSlug, "value", value, "limit", limit)
+}
+
+// SetWarnFunc ueberschreibt die Warnreaktion (Tests, andere Benachrichtigungskanaele).
+func (s *Store) SetWarnFunc(f WarnFunc) { s.warn = f }
+
+func (s *Store) increment(ctx context.Context, tenantSlug, metric string, delta int64) error {
+	_, err := s.pool.Exec(ctx, `
+		INSERT INTO ai_usage_counters (tenant_slug, metric, value, updated_at)
+		VALUES ($1, $2, $3, now())
+		ON CONFLICT (tenant_slug, metric) DO UPDATE
+			SET value = ai_usage_counters.value + $3, updated_at = now()
+	`, tenantSlug, metric, delta)
+	if err != nil {
+		return fmt.Errorf("quota: zaehler erhoehen: %w", err)
+	}
+	return nil
+}
+
+// Get liefert den aktuellen Zaehlerstand — 0, wenn noch nie erhoeht wurde,
+// strikt tenant-gescoped (Akzeptanzkriterium 1).
+func (s *Store) Get(ctx context.Context, tenantSlug, metric string) (int64, error) {
+	var value int64
+	err := s.pool.QueryRow(ctx, `
+		SELECT value FROM ai_usage_counters WHERE tenant_slug = $1 AND metric = $2
+	`, tenantSlug, metric).Scan(&value)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return 0, nil
+		}
+		return 0, fmt.Errorf("quota: zaehler lesen: %w", err)
+	}
+	return value, nil
+}
+
+// SetQuota legt die Obergrenze fuer (tenantSlug, metric) fest — Konfiguration,
+// kein Hardcode (Produkt-DNA: Datenresidenz-/Quota-Grenzen konfigurierbar).
+func (s *Store) SetQuota(ctx context.Context, tenantSlug, metric string, limit int64) error {
+	_, err := s.pool.Exec(ctx, `
+		INSERT INTO ai_usage_quotas (tenant_slug, metric, limit_value)
+		VALUES ($1, $2, $3)
+		ON CONFLICT (tenant_slug, metric) DO UPDATE SET limit_value = $3
+	`, tenantSlug, metric, limit)
+	if err != nil {
+		return fmt.Errorf("quota: obergrenze setzen: %w", err)
+	}
+	return nil
+}
+
+func (s *Store) getQuota(ctx context.Context, tenantSlug, metric string) (limit int64, configured bool, err error) {
+	err = s.pool.QueryRow(ctx, `
+		SELECT limit_value FROM ai_usage_quotas WHERE tenant_slug = $1 AND metric = $2
+	`, tenantSlug, metric).Scan(&limit)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return 0, false, nil
+		}
+		return 0, false, fmt.Errorf("quota: obergrenze lesen: %w", err)
+	}
+	return limit, true, nil
+}
+
+// Check liefert Zaehlerstand, konfigurierte Obergrenze und den daraus
+// abgeleiteten Status (Akzeptanzkriterium 1). Ohne konfigurierte Obergrenze
+// gilt die Metrik als unbegrenzt (StatusOK).
+func (s *Store) Check(ctx context.Context, tenantSlug, metric string) (value, limit int64, status Status, err error) {
+	value, err = s.Get(ctx, tenantSlug, metric)
+	if err != nil {
+		return 0, 0, "", err
+	}
+	limit, configured, err := s.getQuota(ctx, tenantSlug, metric)
+	if err != nil {
+		return 0, 0, "", err
+	}
+	if !configured {
+		return value, 0, StatusOK, nil
+	}
+	switch {
+	case value >= limit:
+		return value, limit, StatusExceeded, nil
+	case limit > 0 && float64(value) >= warningThreshold*float64(limit):
+		return value, limit, StatusWarning, nil
+	default:
+		return value, limit, StatusOK, nil
+	}
+}
+
+// Enforce blockiert einen neuen Inferenz-Aufruf, wenn die Quota fuer
+// tenantSlug bereits erreicht ist (Akzeptanzkriterium 2), und loest vorab
+// die Warnreaktion aus, wenn sich der Tenant der Grenze naehert
+// (Akzeptanzkriterium 3) — beides VOR dem eigentlichen Aufruf.
+func (s *Store) Enforce(ctx context.Context, tenantSlug string) error {
+	value, limit, status, err := s.Check(ctx, tenantSlug, InferenceCallsMetric)
+	if err != nil {
+		return fmt.Errorf("quota: pruefung fehlgeschlagen: %w", err)
+	}
+	if status == StatusExceeded {
+		return fmt.Errorf("%w (tenant %s)", ErrQuotaExceeded, tenantSlug)
+	}
+	if status == StatusWarning && s.warn != nil {
+		s.warn(tenantSlug, value, limit)
+	}
+	return nil
+}
+
+// RecordCall zaehlt genau einen tatsaechlich ausgefuehrten Inferenz-Aufruf
+// (Akzeptanzkriterium 1) und meldet das Delta an Core LIC-03 weiter. Der
+// Aufrufer ruft RecordCall NUR nach erfolgreicher Ausfuehrung auf, damit der
+// Zaehlerstand mit tatsaechlichen Aufrufen uebereinstimmt (Pruefung 3), nicht
+// mit blossen Versuchen. Eine fehlgeschlagene Core-Meldung wird geloggt,
+// blockiert aber nicht den bereits erfolgreich abgeschlossenen Aufruf.
+func (s *Store) RecordCall(ctx context.Context, tenantSlug string) error {
+	if err := s.increment(ctx, tenantSlug, InferenceCallsMetric, 1); err != nil {
+		return err
+	}
+	if s.reporter != nil {
+		if err := s.reporter.Report(ctx, tenantSlug, InferenceCallsMetric, 1); err != nil {
+			slog.Error("quota: meldung an core lic-03 fehlgeschlagen", "tenant", tenantSlug, "error", err)
+		}
+	}
+	return nil
+}

ai/internal/quota/quota_test.go

@@ -0,0 +1,184 @@
+package quota
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"gitea.perlbach24.de/scripte/nexarch/ai/internal/migrate"
+	"gitea.perlbach24.de/scripte/nexarch/ai/internal/testdb"
+)
+
+func newPool(t *testing.T) *pgxpool.Pool {
+	t.Helper()
+	dsn := os.Getenv("TEST_TENANT_DSN")
+	if dsn == "" {
+		t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen")
+	}
+	pool := testdb.Pool(t, dsn)
+
+	root, err := filepath.Abs("../../migrations")
+	if err != nil {
+		t.Fatalf("migrationsverzeichnis: %v", err)
+	}
+	migrations, err := migrate.Load(root)
+	if err != nil {
+		t.Fatalf("migrationen laden: %v", err)
+	}
+	if _, err := migrate.Up(context.Background(), pool, migrations); err != nil {
+		t.Fatalf("migrationen anwenden: %v", err)
+	}
+	return pool
+}
+
+// fakeReporter erfasst gemeldete Deltas, ohne einen echten HTTP-Aufruf an
+// Core zu machen — beweist, dass RecordCall die Meldung tatsaechlich ausloest
+// (Kopplung an Core LIC-03).
+type fakeReporter struct {
+	mu      sync.Mutex
+	reports []usageDeltaDTO
+}
+
+func (f *fakeReporter) Report(_ context.Context, tenantSlug, metric string, delta int64) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.reports = append(f.reports, usageDeltaDTO{TenantSlug: tenantSlug, Metric: metric, Delta: delta})
+	return nil
+}
+
+// AK1: Verbrauch ist je Tenant nachvollziehbar erfasst.
+func TestStore_CheckTracksPerTenant(t *testing.T) {
+	pool := newPool(t)
+	store := NewStore(pool, nil)
+	ctx := context.Background()
+
+	if err := store.SetQuota(ctx, "tenant-a", InferenceCallsMetric, 10); err != nil {
+		t.Fatalf("quota setzen: %v", err)
+	}
+	if err := store.RecordCall(ctx, "tenant-a"); err != nil {
+		t.Fatalf("call erfassen: %v", err)
+	}
+	if err := store.RecordCall(ctx, "tenant-a"); err != nil {
+		t.Fatalf("call erfassen: %v", err)
+	}
+
+	value, limit, status, err := store.Check(ctx, "tenant-a", InferenceCallsMetric)
+	if err != nil {
+		t.Fatalf("check: %v", err)
+	}
+	if value != 2 || limit != 10 || status != StatusOK {
+		t.Fatalf("erwartet value=2 limit=10 status=ok, bekommen value=%d limit=%d status=%s", value, limit, status)
+	}
+
+	// Ein anderer Tenant bleibt unbeeinflusst.
+	otherValue, err := store.Get(ctx, "tenant-b", InferenceCallsMetric)
+	if err != nil {
+		t.Fatalf("get tenant-b: %v", err)
+	}
+	if otherValue != 0 {
+		t.Fatalf("tenant-b sollte unbeeinflusst sein, bekommen %d", otherValue)
+	}
+}
+
+// AK2 / Pruefung 1: Quota-Ueberschreitung wird zuverlaessig verhindert.
+func TestStore_EnforceBlocksWhenExceeded(t *testing.T) {
+	pool := newPool(t)
+	store := NewStore(pool, nil)
+	ctx := context.Background()
+
+	if err := store.SetQuota(ctx, "tenant-exceed", InferenceCallsMetric, 2); err != nil {
+		t.Fatalf("quota setzen: %v", err)
+	}
+	for i := 0; i < 2; i++ {
+		if err := store.Enforce(ctx, "tenant-exceed"); err != nil {
+			t.Fatalf("call %d haette erlaubt sein muessen: %v", i, err)
+		}
+		if err := store.RecordCall(ctx, "tenant-exceed"); err != nil {
+			t.Fatalf("call %d erfassen: %v", i, err)
+		}
+	}
+
+	err := store.Enforce(ctx, "tenant-exceed")
+	if !errors.Is(err, ErrQuotaExceeded) {
+		t.Fatalf("dritter aufruf haette blockiert werden muessen, bekommen: %v", err)
+	}
+}
+
+// AK3 / Pruefung 2: Naeherung an die Grenze wird vorab gemeldet, BEVOR die
+// Quota ueberschritten ist.
+func TestStore_EnforceWarnsBeforeExceeding(t *testing.T) {
+	pool := newPool(t)
+	store := NewStore(pool, nil)
+	ctx := context.Background()
+
+	var mu sync.Mutex
+	var warnings []string
+	store.SetWarnFunc(func(tenantSlug string, value, limit int64) {
+		mu.Lock()
+		defer mu.Unlock()
+		warnings = append(warnings, tenantSlug)
+	})
+
+	if err := store.SetQuota(ctx, "tenant-warn", InferenceCallsMetric, 10); err != nil {
+		t.Fatalf("quota setzen: %v", err)
+	}
+	// 8 von 10 = 80% -> Warnschwelle erreicht, aber noch erlaubt.
+	for i := 0; i < 8; i++ {
+		if err := store.RecordCall(ctx, "tenant-warn"); err != nil {
+			t.Fatalf("call %d erfassen: %v", i, err)
+		}
+	}
+
+	if err := store.Enforce(ctx, "tenant-warn"); err != nil {
+		t.Fatalf("bei 80%% sollte der aufruf noch erlaubt sein: %v", err)
+	}
+
+	mu.Lock()
+	got := len(warnings)
+	mu.Unlock()
+	if got != 1 {
+		t.Fatalf("erwartet genau eine warnung vor ueberschreitung, bekommen %d", got)
+	}
+}
+
+// Pruefung 3: Verbrauchszaehlung wird an Core LIC-03 weitergemeldet (Kopplung).
+func TestStore_RecordCallReportsToCore(t *testing.T) {
+	pool := newPool(t)
+	reporter := &fakeReporter{}
+	store := NewStore(pool, reporter)
+	ctx := context.Background()
+
+	if err := store.RecordCall(ctx, "tenant-report"); err != nil {
+		t.Fatalf("call erfassen: %v", err)
+	}
+	if err := store.RecordCall(ctx, "tenant-report"); err != nil {
+		t.Fatalf("call erfassen: %v", err)
+	}
+
+	reporter.mu.Lock()
+	defer reporter.mu.Unlock()
+	if len(reporter.reports) != 2 {
+		t.Fatalf("erwartet 2 meldungen an core, bekommen %d", len(reporter.reports))
+	}
+	for _, r := range reporter.reports {
+		if r.TenantSlug != "tenant-report" || r.Metric != InferenceCallsMetric || r.Delta != 1 {
+			t.Fatalf("unerwartete meldung: %+v", r)
+		}
+	}
+}
+
+// Ohne konfigurierte Quota gilt eine Metrik als unbegrenzt.
+func TestStore_NoQuotaConfiguredMeansUnlimited(t *testing.T) {
+	pool := newPool(t)
+	store := NewStore(pool, nil)
+	ctx := context.Background()
+
+	if err := store.Enforce(ctx, "tenant-unlimited"); err != nil {
+		t.Fatalf("ohne konfigurierte quota sollte enforce nie blockieren: %v", err)
+	}
+}

ai/internal/quota/reporter.go

@@ -0,0 +1,62 @@
+package quota
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+)
+
+// usageDeltaDTO entspricht Core internal/resync.usageDeltaDTO (JSON-Vertrag:
+// tenant_slug/metric/delta) — hier gespiegelt wie im dms/mail-Muster
+// (dms/internal/storage.HTTPUsageReporter), da AI Cores internal/-Pakete
+// nicht importieren kann.
+type usageDeltaDTO struct {
+	TenantSlug string `json:"tenant_slug"`
+	Metric     string `json:"metric"`
+	Delta      int64  `json:"delta"`
+}
+
+// HTTPReporter meldet ueber Cores Resync-Nutzungs-Endpunkt
+// (internal/resync.Handler.UsageHandler, LIC-03/API-06) — authentifiziert
+// ueber dasselbe Service-Credential-Verfahren wie jeder andere Modul-Core-
+// Aufruf (API-02).
+type HTTPReporter struct {
+	endpointURL  string
+	clientID     string
+	clientSecret string
+	httpClient   *http.Client
+}
+
+func NewHTTPReporter(endpointURL, clientID, clientSecret string, httpClient *http.Client) *HTTPReporter {
+	if httpClient == nil {
+		httpClient = http.DefaultClient
+	}
+	return &HTTPReporter{endpointURL: endpointURL, clientID: clientID, clientSecret: clientSecret, httpClient: httpClient}
+}
+
+func (r *HTTPReporter) Report(ctx context.Context, tenantSlug, metric string, delta int64) error {
+	body, err := json.Marshal([]usageDeltaDTO{{TenantSlug: tenantSlug, Metric: metric, Delta: delta}})
+	if err != nil {
+		return fmt.Errorf("quota: nutzungsmeldung serialisieren: %w", err)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.endpointURL, bytes.NewReader(body))
+	if err != nil {
+		return fmt.Errorf("quota: nutzungsmeldungs-anfrage aufbauen: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Nexarch-Client-Id", r.clientID)
+	req.Header.Set("X-Nexarch-Client-Secret", r.clientSecret)
+
+	resp, err := r.httpClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("quota: nutzungsmeldung senden: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("quota: nutzungsmeldung von core abgelehnt: status %d", resp.StatusCode)
+	}
+	return nil
+}

ai/migrations/0007_ai_usage_quota.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE ai_usage_quotas;
+DROP TABLE ai_usage_counters;

ai/migrations/0007_ai_usage_quota.up.sql

@@ -0,0 +1,18 @@
+-- GOV-04: lokaler Nutzungszaehler/Quota-Spiegel je Tenant fuer KI-Inferenz.
+-- AI ist ein eigenes Go-Modul und kann Core internal/usage (LIC-03) nicht
+-- importieren; Struktur entspricht 1:1 Core migrations/0005_usage_counters,
+-- Deltas werden zusaetzlich an Core gemeldet (siehe ai/internal/quota).
+CREATE TABLE ai_usage_counters (
+    tenant_slug TEXT NOT NULL,
+    metric      TEXT NOT NULL,
+    value       BIGINT NOT NULL DEFAULT 0,
+    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (tenant_slug, metric)
+);
+
+CREATE TABLE ai_usage_quotas (
+    tenant_slug TEXT NOT NULL,
+    metric      TEXT NOT NULL,
+    limit_value BIGINT NOT NULL,
+    PRIMARY KEY (tenant_slug, metric)
+);

ai/migrations/0008_inference_jobs_tenant.down.sql

@@ -0,0 +1 @@
+ALTER TABLE inference_jobs DROP COLUMN tenant_slug;

ai/migrations/0008_inference_jobs_tenant.up.sql

@@ -0,0 +1,4 @@
+-- GOV-04: Tenant-Bezug je Inferenz-Job, damit Verbrauch/Quota (ai_usage_*)
+-- ENG-05-Jobs zuverlaessig zuordnen kann. Nullable: bestehende Jobs ohne
+-- Tenant-Angabe bleiben gueltig.
+ALTER TABLE inference_jobs ADD COLUMN tenant_slug TEXT;