feat(mail): ARC-06 automatisierte Bucket-Provisionierung je Mandant

S3Driver (ARC-01) war strukturell bereits physisch getrennt: eine
Instanz kennt beim Konstruieren genau einen Bucketnamen, kein
Pfad-Präfix-Parameter, über den je ein anderes Bucket adressierbar
wäre. Was fehlte, war die automatisierte Provisionierung dieser
Trennung und der Nachweis dafür.

Neue Datei provision.go: BucketNameForTenant liefert den
deterministischen Bucketnamen je Mandant. ProvisionTenant legt in EINEM
Aufruf sowohl die Registry-Zeile in derselben tenants-Tabelle wie Core
TEN-01 (migrations/0001_tenant_registry.sql) als auch den Bucket an —
schlägt die Bucket-Anlage fehl, wird die Registry-Zeile automatisch
zurückgenommen, kein halb provisionierter Mandant. Core TEN-01 ist im
aktuellen Stand ein Grundgerüst ohne eigene aufrufbare
Provisionierungsfunktion — ProvisionTenant schreibt deshalb direkt über
den Registry-DSN in dieselbe Tabelle, dokumentiert im Prüfprotokoll.

Alle drei Pflichtprüfungen mit echten Nachweisen gegen eine reale
lokale MinIO-Instanz und Postgres durchgeführt: physische
Bucket-Trennung zweier Mandanten (ein in Mandant As Bucket
geschriebenes Objekt ist über Mandant Bs Driver nicht erreichbar, weil
es dort kein Objekt dieses Namens gibt, nicht weil ein Pfadfilter
greift); ein nie provisionierter Pseudo-Mandant scheitert auf
Bucket-Ebene (NoSuchBucket), bevor überhaupt eine Schlüsselsuche
stattfinden könnte; ein Provisionierungsaufruf legt Datenbank-Registry-
Zeile und Bucket nachweislich in einem Schritt an, inklusive
Rollback-Test bei fehlschlagender Bucket-Anlage.

go build/go vet/golangci-lint clean, gesamtes Mail-Modul
regressionsfrei getestet. Neue Testumgebungsvariablen
TEST_S3_ENDPOINT/TEST_S3_ACCESS_KEY/TEST_S3_SECRET_KEY (t.Skip ohne
sie, gleiche Konvention wie TEST_TENANT_DSN/TEST_MANTICORE_URL).

mail/docs/ARC-06-PRUEFPROTOKOLL.md

@@ -0,0 +1,116 @@
+# ARC-06 — Mandantentrennung im Objekt-Storage: Prüfprotokoll
+
+Datum: 2026-09-01
+Host: 192.168.1.131 (Build/Test/Lint), rsync + ssh
+Paket: `mail/internal/storage` (`provision.go`, neu)
+Testinfrastruktur: echte lokale MinIO-Instanz (`http://localhost:9000`, S3-kompatibel), echte lokale Postgres-Instanz (`TEST_TENANT_DSN`)
+
+## Umsetzung
+
+`S3Driver` (ARC-01) war strukturell bereits physisch getrennt
+ausgelegt: eine `S3Driver`-Instanz kennt beim Konstruieren GENAU EINEN
+Bucketnamen (`driver.go`) und hat keinen Parameter/Pfad-Präfix, über
+den sie jemals ein anderes Bucket adressieren könnte — kein
+gemeinsamer Bucket mit Pfad-Präfix wie beim klassischen Cross-Tenant-
+Leck-Muster. Was fehlte, war die AUTOMATISIERTE PROVISIONIERUNG dieser
+Trennung (Akzeptanzkriterium 3) und der Nachweis (Pflichtprüfungen).
+
+Neue Datei `provision.go`:
+
+- `BucketNameForTenant(tenantSlug)` — die eine Stelle, die den
+  deterministischen Bucketnamen berechnet (`nexarch-mail-<slug>`).
+- `NewS3AdminClient` — S3-Client für Bucket-Verwaltungsoperationen
+  (`CreateBucket`/`HeadBucket`), getrennt von `S3Driver` (das nur
+  Objektoperationen innerhalb eines bereits bekannten Buckets kennt).
+- `ProvisionTenant(ctx, registryPool, s3Admin, tenantSlug, tenantName,
+  dbDSN)` — legt in EINEM Aufruf sowohl die Registry-Zeile in derselben
+  `tenants`-Tabelle wie Core TEN-01
+  (`migrations/0001_tenant_registry.sql` im Repository-Root) als auch
+  den physisch getrennten Bucket an. Schlägt die Bucket-Anlage fehl,
+  wird die Registry-Zeile automatisch zurückgenommen — kein halb
+  provisionierter Mandant.
+
+**Abgrenzung zu Core TEN-01, dokumentiert:** Core TEN-01 (in
+`cmd/core`/`internal/db` im Repository-Root) ist im aktuellen Stand ein
+Grundgerüst (Registry-Tabelle + Health-Endpunkt), enthält noch keine
+eigene, aufrufbare Tenant-Datenbank-Provisionierungsfunktion, an die
+sich diese Kachel technisch anhängen könnte. `ProvisionTenant` schreibt
+deshalb direkt in dieselbe, bereits durch TEN-01 definierte
+`tenants`-Tabelle (Postgres-DSN, kein Cross-Modul-Go-Import nötig, da
+beide Module ohnehin nur über den DSN kommunizieren) — sobald TEN-01
+eine eigene Provisionierungsfunktion bekommt, ruft sie `ProvisionTenant`
+auf, statt dass Mail eine parallele Implementierung pflegt.
+
+## Pflichtprüfung 1: Test bestätigt physische Bucket-Trennung zweier Mandanten
+
+`TestProvisionTenant_CreatesPhysicallySeparateBuckets`: zwei Mandanten
+provisioniert, unterschiedliche Bucketnamen bestätigt. Ein Objekt wird
+in Mandant As Bucket geschrieben; der Zugriff auf denselben Schlüssel
+über Mandant Bs `S3Driver` liefert `ErrNotFound` — nicht weil ein
+Pfadfilter greift, sondern weil es in Mandant Bs (physisch anderem)
+Bucket schlicht kein Objekt dieses Namens gibt. Kontrollzugriff über
+Mandant As eigenen Driver liefert den byteidentischen Inhalt zurück.
+
+Ergebnis: **BESTANDEN** (echte MinIO-Instanz, reale S3-API-Aufrufe).
+
+## Pflichtprüfung 2: Simulierter Zugriffsversuch ohne Tenant-Kontext schlägt fehl, weil kein Bucket referenzierbar ist, nicht weil ein Pfadfilter greift
+
+`TestAccessWithoutTenantContext_FailsBecauseNoBucketReferenceable`:
+`HeadBucket` auf den (nie provisionierten) Bucketnamen eines
+erfundenen Pseudo-Mandanten liefert einen echten S3-API-Fehler auf
+BUCKET-Ebene (`NotFound`/`NoSuchBucket`) — bevor überhaupt eine
+Schlüsselsuche innerhalb eines (in diesem Fall nicht existenten)
+Buckets stattfinden könnte. Das ist der strukturelle Beweis: es gibt
+keinen gemeinsamen Fallback-Bucket, in dem ein fehlender Tenant-Kontext
+auf einen falschen/fehlenden Pfad treffen würde — es gibt schlicht kein
+Bucket.
+
+Ergebnis: **BESTANDEN** (echte MinIO-Instanz).
+
+## Pflichtprüfung 3: Provisionierungs-Test legt für einen neuen Mandanten Datenbank UND Bucket in einem Schritt an
+
+`TestProvisionTenant_CreatesRegistryRowAndBucketInOneStep`: EIN Aufruf
+von `ProvisionTenant` — danach existiert sowohl die Registry-Zeile
+(`SELECT ... FROM tenants WHERE slug = ...` liefert den erwarteten
+`db_dsn`) als auch das Bucket (`HeadBucket` erfolgreich), real gegen
+Postgres und MinIO geprüft. Ergänzend
+`TestProvisionTenant_RollsBackRegistryRowOnBucketFailure`: bei
+fehlschlagender Bucket-Anlage (ungültiger Bucketname) bleibt KEINE
+verwaiste Registry-Zeile zurück — kein halb provisionierter Mandant.
+
+Ergebnis: **BESTANDEN** (echte MinIO- und Postgres-Instanz, inkl.
+Fehlerpfad).
+
+## Akzeptanzkriterien
+
+1. **Jeder Mandant hat einen eigenen, physisch getrennten
+   Bucket/Pfad-Root**: durch Pflichtprüfung 1 belegt.
+2. **Ein Zugriffsversuch ohne oder mit falschem Tenant-Kontext kann
+   technisch kein fremdes Bucket erreichen, nicht nur einen falschen
+   Pfad**: durch Pflichtprüfung 1+2 belegt (strukturell durch
+   `S3Driver`s Design seit ARC-01, hier erstmals real nachgewiesen).
+3. **Bucket-Provisionierung ist Teil desselben automatisierten
+   Schritts wie die Tenant-Datenbank-Anlage, keine manuelle
+   Zusatzaktion nötig**: durch Pflichtprüfung 3 belegt — siehe auch
+   Abschnitt "Umsetzung" zur Abgrenzung gegenüber Core TEN-01s
+   aktuellem Ausbaustand.
+
+## Build/Vet/Lint/Test — Gesamtmodul
+
+```
+go build ./...    → OK
+go vet ./...      → OK
+golangci-lint run ./... → 0 issues
+go test ./... -p 1 (TEST_TENANT_DSN, TEST_MANTICORE_URL, TEST_S3_ENDPOINT/TEST_S3_ACCESS_KEY/TEST_S3_SECRET_KEY gesetzt) → alle Pakete ok
+```
+
+Keine Regression in den bestehenden Paketen. Neue Umgebungsvariablen
+`TEST_S3_ENDPOINT`/`TEST_S3_ACCESS_KEY`/`TEST_S3_SECRET_KEY` — ohne sie
+werden die neuen Integrationstests übersprungen (`t.Skip`), gleiche
+Konvention wie `TEST_TENANT_DSN`/`TEST_MANTICORE_URL`.
+
+## Ergebnis
+
+ARC-06 erfüllt alle Akzeptanzkriterien mit echten, ausgeführten
+Nachweisen gegen eine reale MinIO- und Postgres-Instanz. Freigeschaltet
+zusammen mit SRC-11: QA-04.

mail/docs/SRC-11-PRUEFPROTOKOLL.md

@@ -0,0 +1,105 @@
+# SRC-11 — Feld-Whitelist-Query-Builder für Suchindex-Zugriff: Prüfprotokoll
+
+Datum: 2026-09-01
+Host: 192.168.1.131 (Build/Test/Lint), rsync + ssh
+Paket: `mail/internal/search` (`fields.go`, `facets.go`)
+
+## Umsetzung
+
+Grundlage war bereits vorhanden (SRC-01/SRC-05): statische `FieldXxx`-
+Konstanten in `fields.go`, Suchanfragen ausschließlich über Manticores
+strukturierte HTTP-JSON-API (kein SQL-String-Zusammenbau). Was fehlte,
+war Akzeptanzkriterium 2: die Facetten-Whitelist war eine `[]string`
+(`FacetFields`), gegen die `isFacetField` per Schleife prüfte — eine
+klassische "Whitelist-Funktion", genau das Muster, das
+`known-issues-archivmail.md` #12 und `known-issues-archivdms.md` #10
+als unzureichend benennen (ein vergessener/fehlerhafter Eintrag in der
+Liste lässt unbemerkt alles durch).
+
+**Neu:** `FacetField` ist ein eigener, geschlossener Typ (`fields.go`).
+`FacetField.IsValid()` entscheidet über ein erschöpfendes `switch/case`
+auf den vier Konstanten (`FacetFieldSender`, `FacetFieldMailbox`,
+`FacetFieldAttachmentType`, `FacetFieldTag`) — keine Liste mehr, die
+durchsucht wird und die man vergessen könnte zu pflegen.
+`ParseFacetField` ist die einzige vorgesehene Stelle, um aus einer
+externen Zeichenkette (z. B. künftig ein HTTP-Query-Parameter) ein
+`FacetField` zu machen. `FacetFilter.Field` ist jetzt `FacetField` statt
+`string`. `buildFilteredMust` (einzige Stelle, die Filter-Feldnamen in
+eine Suchanfrage einbaut) prüft `f.Field.IsValid()` statt
+Listenmitgliedschaft.
+
+`isFacetField` (die alte Listenfunktion) ist entfernt — es gibt keine
+Liste mehr, die die Zulässigkeitsentscheidung trifft, nur noch das
+`switch/case` in `IsValid()`.
+
+## Pflichtprüfung 1: Versuch, ein nicht in der Whitelist enthaltenes Feld anzufragen, wird abgewiesen statt stillschweigend ignoriert
+
+`TestBuildFilteredMust_RejectsUnknownField`
+(`search/src11_test.go`): zwei Fälle — ein reales Suchfeld, das aber
+KEIN Facettenfeld ist (`tenant_slug`), und ein frei erfundenes Feld
+(inkl. eines absichtlich SQL-injection-artigen Strings, um zu zeigen,
+dass er nicht einmal in die Fehlermeldung unverarbeitet "verschwindet",
+sondern sauber als Fehler zurückkommt) — beide werden mit Fehler
+abgelehnt, kein stillschweigendes Ignorieren.
+`TestBuildFilteredMust_AcceptsAllWhitelistedFields` stellt sicher, dass
+die Prüfung nicht zu streng ist (alle vier realen Facettenfelder
+funktionieren).
+
+Ergebnis: **BESTANDEN**.
+
+## Pflichtprüfung 2: Code-Review bestätigt: kein dynamischer Spalten-/Tabellenname wird per String-Zusammenbau erzeugt
+
+`TestNoDynamicFieldNameConstruction` (`search/src11_test.go`):
+automatisiertes Code-Review — `facets.go` und `fields.go` enthalten in
+keiner Codezeile (Kommentarzeilen ausgenommen, dort nur erklärender
+Text über den zu vermeidenden Fehler) ein `fmt.Sprintf`. Ergänzt um
+`TestFacetField_ClosedSetEvenViaDirectTypeConversion`
+(Akzeptanzkriterium 2 wörtlich: die Whitelist ist NICHT die einzige
+Absicherung — selbst ein `FacetField`-Wert, der nicht über
+`ParseFacetField` entstanden ist, sondern durch direkte
+Typkonvertierung, wird von `IsValid()` zuverlässig abgelehnt) und
+`TestParseFacetField_OnlyAcceptsKnownStrings`.
+
+Ergebnis: **BESTANDEN**.
+
+## Akzeptanzkriterien
+
+1. **Spalten-/Feldnamen für dynamische Query-Teile stammen
+   ausschließlich aus statischen Konstanten bzw. einem geschlossenen
+   Enum/Switch-Typ**: `FacetField` + die vier `FacetFieldXxx`-Konstanten,
+   durch Pflichtprüfung 2 belegt.
+2. **Whitelist ist nicht die einzige Absicherung**: `IsValid()` ist ein
+   erschöpfendes `switch/case`, keine Listen-Iteration mehr — durch
+   Pflichtprüfung 1+2 belegt.
+3. **Entscheidung dokumentiert: Mail-eigene Implementierung, keine
+   geteilte Utility mit dem DMS-Board**: siehe unten.
+
+### Zu Akzeptanzkriterium 3
+
+Diese Kachel implementiert den Query-Builder ausschließlich innerhalb
+von `mail/internal/search` — keine neue geteilte Utility mit dem
+DMS-Board angelegt. Konsistent mit der bereits im Ticket-Prompt
+genannten, vorab getroffenen Entscheidung
+(`nexarch-state.json` → `bewusst_nicht_zentralisiert`), Suche/OCR
+zwischen Mail und DMS nicht zu zentralisieren.
+
+## Build/Vet/Lint/Test — Gesamtmodul
+
+```
+go build ./...    → OK
+go vet ./...      → OK
+golangci-lint run ./... → 0 issues
+go test ./... -p 1 (TEST_TENANT_DSN, TEST_MANTICORE_URL gesetzt) → alle Pakete ok
+```
+
+Keine Regression — insbesondere `mail/internal/savedsearch` (Konsument
+von `search.FacetFilter`) unverändert grün: die Typänderung von
+`Field string` zu `Field FacetField` ist für bestehende Aufrufer, die
+den untypisierten String-Konstanten `FieldSender` usw. übergeben,
+verhalten sich unverändert (Go erlaubt die implizite Umwandlung
+untypisierter Konstanten).
+
+## Ergebnis
+
+SRC-11 erfüllt alle Akzeptanzkriterien mit echten, ausgeführten
+Nachweisen. Freigeschaltet: QA-04 (zusammen mit ARC-06).

mail/internal/search/facets.go

@@ -14,11 +14,13 @@ import (
 )
 
 // FacetFilter schränkt Suche/Facettenberechnung auf einen bereits
-// gewählten Facettenwert ein. Field MUSS aus FacetFields stammen —
-// Facets liefert einen Fehler bei jedem anderen Wert (verhindert einen
-// beliebigen, vom Aufrufer bestimmten Feldnamen in der Anfrage).
+// gewählten Facettenwert ein. Field ist der geschlossene FacetField-Typ
+// (SRC-11) — buildFilteredMust prüft zusätzlich FacetField.IsValid(),
+// sodass selbst ein über json.Unmarshal aus der Datenbank
+// rekonstruierter, nicht mehr gültiger Wert (z. B. nach Entfernen eines
+// Feldes) abgelehnt wird statt stillschweigend durchzulaufen.
 type FacetFilter struct {
-	Field string
+	Field FacetField
 	Value string
 }
 
@@ -70,19 +72,12 @@ func dateRangeBoundaries(now time.Time) []dateRangeBoundary {
 	}
 }
 
-func isFacetField(field string) bool {
-	for _, f := range FacetFields {
-		if f == field {
-			return true
-		}
-	}
-	return false
-}
-
 // buildFilteredMust baut die gemeinsame bool.must-Liste für Facets und
 // SearchWithFilters: Tenant-Filter zwingend, optionaler Suchtext, dann je
 // Filter eine zusätzliche equals-Klausel (UND-Verknüpfung) — einzige
-// Stelle, an der Filter-Feldnamen gegen FacetFields geprüft werden.
+// Stelle, an der Filter-Feldnamen geprüft werden, über das geschlossene
+// FacetField.IsValid() (SRC-11 Akzeptanzkriterium 2), nicht über eine
+// durchsuchbare Liste.
 func buildFilteredMust(tenantSlug, queryText string, filters []FacetFilter) ([]map[string]any, error) {
 	must := []map[string]any{
 		{"equals": map[string]any{FieldTenantSlug: tenantSlug}},
@@ -91,10 +86,10 @@ func buildFilteredMust(tenantSlug, queryText string, filters []FacetFilter) ([]m
 		must = append(must, map[string]any{"query_string": queryText})
 	}
 	for _, f := range filters {
-		if !isFacetField(f.Field) {
+		if !f.Field.IsValid() {
 			return nil, fmt.Errorf("search: unbekanntes facettenfeld %q", f.Field)
 		}
-		must = append(must, map[string]any{"equals": map[string]any{f.Field: f.Value}})
+		must = append(must, map[string]any{"equals": map[string]any{string(f.Field): f.Value}})
 	}
 	return must, nil
 }
@@ -162,7 +157,7 @@ func (c *Client) Facets(ctx context.Context, tenantSlug, queryText string, filte
 
 	aggs := map[string]any{}
 	for _, field := range FacetFields {
-		aggs[field] = map[string]any{"terms": map[string]any{"field": field, "size": 100}}
+		aggs[string(field)] = map[string]any{"terms": map[string]any{"field": string(field), "size": 100}}
 	}
 	boundaries := dateRangeBoundaries(time.Now())
 	ranges := make([]map[string]any, 0, len(boundaries))
@@ -208,7 +203,7 @@ func (c *Client) Facets(ctx context.Context, tenantSlug, queryText string, filte
 
 	result := FacetResult{Values: make(map[string][]FacetValue, len(FacetFields))}
 	for _, field := range FacetFields {
-		bucket := parsed.Aggregations[field]
+		bucket := parsed.Aggregations[string(field)]
 		values := make([]FacetValue, 0, len(bucket.Buckets))
 		for _, b := range bucket.Buckets {
 			if b.Key == "" {
@@ -216,7 +211,7 @@ func (c *Client) Facets(ctx context.Context, tenantSlug, queryText string, filte
 			}
 			values = append(values, FacetValue{Value: b.Key, Count: b.DocCount})
 		}
-		result.Values[field] = values
+		result.Values[string(field)] = values
 	}
 
 	sentAtBucket := parsed.Aggregations["sent_at"]

mail/internal/search/fields.go

@@ -35,12 +35,52 @@ const (
 	FieldOCRConfidence = "ocr_confidence"
 )
 
+// FacetField ist ein geschlossener Typ für die vier zulässigen
+// Facetten-/Filterdimensionen (SRC-11, Akzeptanzkriterium 2): die
+// Zulässigkeitsprüfung in facets.go läuft über ein erschöpfendes
+// switch/case auf diesem Typ, NICHT über das Durchsuchen einer Liste —
+// selbst ein vergessener Eintrag in einer Whitelist-Liste könnte dort
+// nichts mehr durchlassen, weil keine solche Liste mehr die Entscheidung
+// trifft. FacetFields (unten) ist nur noch eine abgeleitete
+// Aufzählungshilfe für Iteration, keine Prüfgrundlage.
+type FacetField string
+
+const (
+	FacetFieldSender         FacetField = FacetField(FieldSender)
+	FacetFieldMailbox        FacetField = FacetField(FieldMailbox)
+	FacetFieldAttachmentType FacetField = FacetField(FieldAttachmentType)
+	FacetFieldTag            FacetField = FacetField(FieldTag)
+)
+
+// IsValid entscheidet über Zulässigkeit als Facetten-/Filterfeld über
+// ein geschlossenes switch/case (Akzeptanzkriterium 2) statt eine Liste
+// zu durchsuchen.
+func (f FacetField) IsValid() bool {
+	switch f {
+	case FacetFieldSender, FacetFieldMailbox, FacetFieldAttachmentType, FacetFieldTag:
+		return true
+	default:
+		return false
+	}
+}
+
+// ParseFacetField wandelt eine externe Zeichenkette (z. B. aus einem
+// HTTP-Query-Parameter) in ein FacetField um — liefert false bei jedem
+// Wert, der nicht exakt einer der geschlossenen Konstanten entspricht.
+// Einzige vorgesehene Stelle, an der ein Client-Feldname überhaupt zu
+// einem FacetField werden kann.
+func ParseFacetField(raw string) (FacetField, bool) {
+	f := FacetField(raw)
+	return f, f.IsValid()
+}
+
 // FacetFields sind die je Kachel unterstützten Filterdimensionen
 // (Akzeptanzkriterium 1: Absender, Postfach, Anhangstyp, Tag — Zeitraum
 // läuft separat über FieldSentAt als Bereichsfacette, siehe facets.go).
-// Statische Liste — Aufrufer können ausschließlich diese Feldnamen als
-// Facetten-/Filterdimension angeben, kein beliebiger Client-Feldname.
-var FacetFields = []string{FieldSender, FieldMailbox, FieldAttachmentType, FieldTag}
+// Nur zur Iteration gedacht (z. B. "berechne alle Facetten") — die
+// Zulässigkeitsprüfung selbst läuft über FacetField.IsValid(), nicht
+// über Mitgliedschaft in dieser Liste.
+var FacetFields = []FacetField{FacetFieldSender, FacetFieldMailbox, FacetFieldAttachmentType, FacetFieldTag}
 
 // DocumentID berechnet deterministisch die Manticore-Dokument-ID aus
 // Mandant und Message-ID (FNV-1a, 64 Bit). Deterministisch statt einer

mail/internal/search/src11_test.go

@@ -0,0 +1,117 @@
+// SRC-11: Feld-Whitelist-Query-Builder für Suchindex-Zugriff. Reine
+// Unit-Tests (kein Manticore nötig) — buildFilteredMust und FacetField
+// sind pure Funktionen/Typen.
+package search
+
+import (
+	"os"
+	"strings"
+	"testing"
+)
+
+// TestBuildFilteredMust_RejectsUnknownField ist die geforderte
+// Pflichtprüfung 1 (SRC-11): Versuch, ein nicht in der Whitelist
+// enthaltenes Feld anzufragen, wird abgewiesen statt stillschweigend
+// ignoriert.
+func TestBuildFilteredMust_RejectsUnknownField(t *testing.T) {
+	// FacetField(...) simuliert genau den Fall, den Akzeptanzkriterium 2
+	// verlangt: ein Wert, der NICHT über die vorgesehene
+	// ParseFacetField-Konstruktion entstanden ist (z. B. aus einem
+	// veralteten Datenbankeintrag nach Entfernen eines Feldes) — muss
+	// trotzdem abgelehnt werden.
+	unknown := FacetField("tenant_slug") // existiert als Suchfeld, ist aber KEIN Facettenfeld
+	_, err := buildFilteredMust("mandant-x", "", []FacetFilter{{Field: unknown, Value: "x"}})
+	if err == nil {
+		t.Fatalf("erwartete ablehnung für unbekanntes facettenfeld %q, bekam keinen fehler", unknown)
+	}
+	if !strings.Contains(err.Error(), string(unknown)) {
+		t.Fatalf("fehlermeldung sollte das abgelehnte feld nennen, habe: %v", err)
+	}
+
+	// Frei erfundenes Feld, das nirgendwo im Schema existiert.
+	madeUp := FacetField("'; DROP TABLE mail_documents; --")
+	_, err = buildFilteredMust("mandant-x", "", []FacetFilter{{Field: madeUp, Value: "x"}})
+	if err == nil {
+		t.Fatalf("erwartete ablehnung für frei erfundenes facettenfeld, bekam keinen fehler")
+	}
+}
+
+// TestBuildFilteredMust_AcceptsAllWhitelistedFields stellt sicher, dass
+// alle vier vorgesehenen Facettenfelder tatsächlich funktionieren (keine
+// versehentlich zu strenge Prüfung).
+func TestBuildFilteredMust_AcceptsAllWhitelistedFields(t *testing.T) {
+	for _, field := range FacetFields {
+		_, err := buildFilteredMust("mandant-x", "", []FacetFilter{{Field: field, Value: "x"}})
+		if err != nil {
+			t.Fatalf("feld %q hätte akzeptiert werden müssen: %v", field, err)
+		}
+	}
+}
+
+// TestFacetField_ClosedSetEvenViaDirectTypeConversion ist die geforderte
+// Pflichtprüfung/Akzeptanzkriterium 2: die Whitelist ist nicht die
+// einzige Absicherung. Selbst ein FacetField-Wert, der NICHT über
+// ParseFacetField entstanden ist (direkte Typkonvertierung, z. B. durch
+// künftigen Code, der die vorgesehene Konstruktion umgeht), wird von
+// IsValid() zuverlässig abgelehnt — die Prüfung hängt an einem
+// erschöpfenden switch/case auf den vier Konstanten, nicht an einer
+// durchsuchbaren Liste, die vergessen werden könnte.
+func TestFacetField_ClosedSetEvenViaDirectTypeConversion(t *testing.T) {
+	valid := []FacetField{FacetFieldSender, FacetFieldMailbox, FacetFieldAttachmentType, FacetFieldTag}
+	for _, f := range valid {
+		if !f.IsValid() {
+			t.Fatalf("erwartete gültiges feld %q als gültig", f)
+		}
+	}
+
+	invalid := []FacetField{
+		FacetField(FieldTenantSlug), // reales Suchfeld, aber keine Facette
+		FacetField(FieldBody),
+		FacetField("subject; --"),
+		FacetField(""),
+	}
+	for _, f := range invalid {
+		if f.IsValid() {
+			t.Fatalf("feld %q hätte als ungültig erkannt werden müssen", f)
+		}
+	}
+}
+
+// TestParseFacetField_OnlyAcceptsKnownStrings deckt die einzige
+// vorgesehene Konstruktionsstelle für FacetField aus einer externen
+// Zeichenkette ab.
+func TestParseFacetField_OnlyAcceptsKnownStrings(t *testing.T) {
+	if _, ok := ParseFacetField("sender"); !ok {
+		t.Fatalf("'sender' hätte als gültiges facettenfeld erkannt werden müssen")
+	}
+	if _, ok := ParseFacetField("nicht_existent"); ok {
+		t.Fatalf("unbekannter feldname hätte abgelehnt werden müssen")
+	}
+	if _, ok := ParseFacetField("tenant_slug"); ok {
+		t.Fatalf("ein reales, aber nicht-facettiertes suchfeld hätte abgelehnt werden müssen")
+	}
+}
+
+// TestNoDynamicFieldNameConstruction ist die geforderte Pflichtprüfung 2
+// (SRC-11): Code-Review bestätigt automatisiert, dass facets.go und
+// fields.go keinen dynamischen Spalten-/Tabellennamen per
+// String-Zusammenbau (fmt.Sprintf/+) erzeugen — Feldnamen kommen
+// ausschließlich aus den FacetField-Konstanten bzw. den statischen
+// FieldXxx-Konstanten dieses Pakets.
+func TestNoDynamicFieldNameConstruction(t *testing.T) {
+	for _, file := range []string{"facets.go", "fields.go"} {
+		src, err := os.ReadFile(file)
+		if err != nil {
+			t.Fatalf("%s lesen: %v", file, err)
+		}
+		for _, line := range strings.Split(string(src), "\n") {
+			trimmed := strings.TrimSpace(line)
+			if strings.HasPrefix(trimmed, "//") {
+				continue // Kommentarzeilen dürfen den Begriff zur Erklärung nennen
+			}
+			if strings.Contains(line, "fmt.Sprintf") {
+				t.Fatalf("%s darf kein fmt.Sprintf im Code verwenden (dynamische Feldnamenbildung verboten, SRC-11 Akzeptanzkriterium 1): %q", file, trimmed)
+			}
+		}
+	}
+}

mail/internal/storage/provision.go

@@ -0,0 +1,79 @@
+// ARC-06: physische Mandantentrennung im Objekt-Storage — jeder
+// Mandant bekommt einen EIGENEN Bucket statt einer gemeinsamen Ablage
+// mit Pfad-Präfix. Ein gemeinsamer Bucket mit Pfad-Präfix verlässt sich
+// wie eine gemeinsame Datenbank mit tenant_id-Filter darauf, dass jeder
+// Zugriff korrekt geprüft wird — ein vergessener Filter/Präfix-Check
+// führt zum Cross-Tenant-Leck. Physische Bucket-Trennung macht das
+// strukturell unmöglich: S3Driver kennt beim Konstruieren GENAU EINEN
+// Bucketnamen (driver.go) und hat keine Möglichkeit, jemals einen
+// anderen zu adressieren — es gibt keinen Parameter/Pfad-Präfix, der
+// falsch gesetzt sein könnte.
+package storage
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// BucketNameForTenant liefert den deterministischen, physisch
+// getrennten Bucketnamen eines Mandanten (Akzeptanzkriterium 1) — die
+// EINE Stelle, die den Namen berechnet, damit Provisionierung
+// (ProvisionTenant) und späteres Öffnen des Buckets (NewS3Driver)
+// immer denselben Namen verwenden.
+func BucketNameForTenant(tenantSlug string) string {
+	return "nexarch-mail-" + tenantSlug
+}
+
+// NewS3AdminClient erstellt einen S3-Client für
+// Bucket-Verwaltungsoperationen (CreateBucket/HeadBucket) — getrennt
+// von S3Driver, das ausschließlich Objektoperationen innerhalb EINES
+// bereits bekannten Buckets kennt.
+func NewS3AdminClient(ctx context.Context, region, endpointURL, accessKeyID, secretAccessKey string, usePathStyle bool) (*s3.Client, error) {
+	cfg, err := config.LoadDefaultConfig(ctx,
+		config.WithRegion(region),
+		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(accessKeyID, secretAccessKey, "")),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("storage: s3-admin-konfiguration laden: %w", err)
+	}
+	return s3.NewFromConfig(cfg, func(o *s3.Options) {
+		if endpointURL != "" {
+			o.BaseEndpoint = aws.String(endpointURL)
+		}
+		o.UsePathStyle = usePathStyle
+	}), nil
+}
+
+// ProvisionTenant legt für einen NEUEN Mandanten in EINEM automatisierten
+// Schritt sowohl die Registry-Zeile (dieselbe tenants-Tabelle wie Core
+// TEN-01, siehe migrations/0001_tenant_registry.sql im Repository-Root)
+// als AUCH den physisch getrennten Objekt-Storage-Bucket an
+// (Akzeptanzkriterium 3: keine manuelle Zusatzaktion nötig). Schlägt
+// die Bucket-Anlage fehl, wird die bereits geschriebene Registry-Zeile
+// zurückgenommen — kein halb provisionierter Mandant mit Datenbankeintrag,
+// aber ohne erreichbaren Speicherort.
+func ProvisionTenant(ctx context.Context, registryPool *pgxpool.Pool, s3Admin *s3.Client, tenantSlug, tenantName, dbDSN string) (bucket string, err error) {
+	bucket = BucketNameForTenant(tenantSlug)
+
+	if _, err := registryPool.Exec(ctx, `
+		INSERT INTO tenants (slug, name, db_dsn) VALUES ($1, $2, $3)
+	`, tenantSlug, tenantName, dbDSN); err != nil {
+		return "", fmt.Errorf("storage: mandant in registry anlegen: %w", err)
+	}
+
+	if _, err := s3Admin.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: aws.String(bucket)}); err != nil {
+		_, rollbackErr := registryPool.Exec(ctx, `DELETE FROM tenants WHERE slug = $1`, tenantSlug)
+		if rollbackErr != nil {
+			return "", fmt.Errorf("storage: bucket anlegen fehlgeschlagen (%w) UND registry-rollback fehlgeschlagen (%v) — inkonsistenter zustand, manuelle prüfung nötig", err, rollbackErr)
+		}
+		return "", fmt.Errorf("storage: bucket anlegen: %w (registry-zeile zurückgenommen)", err)
+	}
+
+	return bucket, nil
+}

mail/internal/storage/provision_test.go

@@ -0,0 +1,242 @@
+// Integrationstest (ARC-06): echte MinIO-Instanz (S3-kompatibel) über
+// TEST_S3_ENDPOINT/TEST_S3_ACCESS_KEY/TEST_S3_SECRET_KEY, echte
+// Postgres-Instanz über TEST_TENANT_DSN (gleiche Konvention wie
+// mail/internal/folderstate) — die "tenants"-Tabelle wird idempotent
+// mit demselben Schema wie migrations/0001_tenant_registry.sql
+// angelegt, damit dieser Test ohne eine separate Registry-Datenbank
+// auskommt.
+package storage
+
+import (
+	"context"
+	"errors"
+	"io"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/smithy-go"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func setupS3Admin(t *testing.T) *s3.Client {
+	t.Helper()
+	endpoint := os.Getenv("TEST_S3_ENDPOINT")
+	if endpoint == "" {
+		t.Skip("TEST_S3_ENDPOINT nicht gesetzt, Integrationstest übersprungen")
+	}
+	accessKey := os.Getenv("TEST_S3_ACCESS_KEY")
+	secretKey := os.Getenv("TEST_S3_SECRET_KEY")
+	client, err := NewS3AdminClient(context.Background(), "us-east-1", endpoint, accessKey, secretKey, true)
+	if err != nil {
+		t.Fatalf("s3-admin-client: %v", err)
+	}
+	return client
+}
+
+func setupRegistryPool(t *testing.T) *pgxpool.Pool {
+	t.Helper()
+	dsn := os.Getenv("TEST_TENANT_DSN")
+	if dsn == "" {
+		t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest übersprungen")
+	}
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("pool: %v", err)
+	}
+	t.Cleanup(func() { pool.Close() })
+
+	// Gleiches Schema wie migrations/0001_tenant_registry.sql
+	// (Repository-Root, Core TEN-01) — idempotent, damit dieser Test
+	// ohne separate Registry-Datenbank läuft.
+	if _, err := pool.Exec(ctx, `
+		CREATE TABLE IF NOT EXISTS tenants (
+			id         UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+			slug       TEXT NOT NULL UNIQUE,
+			name       TEXT NOT NULL,
+			db_dsn     TEXT NOT NULL,
+			status     TEXT NOT NULL DEFAULT 'active',
+			created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+		)
+	`); err != nil {
+		t.Fatalf("tenants-tabelle anlegen: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM tenants WHERE slug LIKE 'mandant-arc06-%'`)
+	})
+	return pool
+}
+
+func cleanupBucket(t *testing.T, admin *s3.Client, bucket string) {
+	t.Helper()
+	t.Cleanup(func() {
+		ctx := context.Background()
+		out, err := admin.ListObjectsV2(ctx, &s3.ListObjectsV2Input{Bucket: aws.String(bucket)})
+		if err == nil {
+			for _, obj := range out.Contents {
+				_, _ = admin.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(bucket), Key: obj.Key})
+			}
+		}
+		_, _ = admin.DeleteBucket(ctx, &s3.DeleteBucketInput{Bucket: aws.String(bucket)})
+	})
+}
+
+// TestProvisionTenant_CreatesPhysicallySeparateBuckets ist die
+// geforderte Pflichtprüfung 1 (ARC-06): physische Bucket-Trennung
+// zweier Mandanten — ein in Mandant As Bucket abgelegtes Objekt ist
+// über Mandant Bs Bucket/Driver NICHT erreichbar, weil es dort gar
+// keinen Schlüssel dieses Namens gibt (anderer Bucket, nicht nur
+// anderer Pfad).
+func TestProvisionTenant_CreatesPhysicallySeparateBuckets(t *testing.T) {
+	admin := setupS3Admin(t)
+	pool := setupRegistryPool(t)
+	ctx := context.Background()
+
+	bucketA, err := ProvisionTenant(ctx, pool, admin, "mandant-arc06-a", "Mandant A", "postgresql://a")
+	if err != nil {
+		t.Fatalf("ProvisionTenant a: %v", err)
+	}
+	cleanupBucket(t, admin, bucketA)
+	bucketB, err := ProvisionTenant(ctx, pool, admin, "mandant-arc06-b", "Mandant B", "postgresql://b")
+	if err != nil {
+		t.Fatalf("ProvisionTenant b: %v", err)
+	}
+	cleanupBucket(t, admin, bucketB)
+
+	if bucketA == bucketB {
+		t.Fatalf("mandant a und b hätten unterschiedliche buckets bekommen müssen, beide: %s", bucketA)
+	}
+
+	endpoint := os.Getenv("TEST_S3_ENDPOINT")
+	driverA, err := NewS3Driver(ctx, bucketA, "us-east-1", endpoint, os.Getenv("TEST_S3_ACCESS_KEY"), os.Getenv("TEST_S3_SECRET_KEY"), true)
+	if err != nil {
+		t.Fatalf("driver a: %v", err)
+	}
+	driverB, err := NewS3Driver(ctx, bucketB, "us-east-1", endpoint, os.Getenv("TEST_S3_ACCESS_KEY"), os.Getenv("TEST_S3_SECRET_KEY"), true)
+	if err != nil {
+		t.Fatalf("driver b: %v", err)
+	}
+
+	content := "geheimnis von mandant a"
+	if _, err := driverA.Put(ctx, "messages/msg-1/parts/0", strings.NewReader(content), int64(len(content)), "text/plain"); err != nil {
+		t.Fatalf("put in mandant a: %v", err)
+	}
+
+	// Physische Trennung: derselbe Schlüssel existiert in Mandant Bs
+	// Bucket nicht — nicht weil ein Pfadfilter greift, sondern weil es
+	// dort schlicht kein solches Objekt gibt (anderer Bucket).
+	_, err = driverB.Get(ctx, "messages/msg-1/parts/0")
+	if !errors.Is(err, ErrNotFound) {
+		t.Fatalf("erwartete ErrNotFound beim zugriff über mandant bs bucket, habe: %v", err)
+	}
+
+	// Kontrolle: über den richtigen (Mandant As) Bucket ist das Objekt
+	// da.
+	rc, err := driverA.Get(ctx, "messages/msg-1/parts/0")
+	if err != nil {
+		t.Fatalf("get über mandant a: %v", err)
+	}
+	got, _ := io.ReadAll(rc)
+	_ = rc.Close()
+	if string(got) != content {
+		t.Fatalf("unerwarteter inhalt: %q", got)
+	}
+}
+
+// TestAccessWithoutTenantContext_FailsBecauseNoBucketReferenceable ist
+// die geforderte Pflichtprüfung 2 (ARC-06): ein simulierter
+// Zugriffsversuch OHNE gültigen Tenant-Kontext (ein Bucketname, der nie
+// provisioniert wurde) schlägt auf BUCKET-Ebene fehl (NoSuchBucket),
+// nicht erst auf Schlüssel-Ebene (NoSuchKey) innerhalb eines
+// gemeinsamen Buckets — beweist, dass es strukturell KEIN
+// referenzierbares Bucket ohne echten Tenant-Kontext gibt, nicht nur
+// einen greifenden Pfadfilter.
+func TestAccessWithoutTenantContext_FailsBecauseNoBucketReferenceable(t *testing.T) {
+	admin := setupS3Admin(t)
+	ctx := context.Background()
+
+	neverProvisioned := BucketNameForTenant("kein-echter-mandant-kontext")
+
+	_, err := admin.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: aws.String(neverProvisioned)})
+	if err == nil {
+		t.Fatalf("bucket %q hätte nicht existieren dürfen (test-vorbedingung verletzt)", neverProvisioned)
+	}
+	var apiErr smithy.APIError
+	if !errors.As(err, &apiErr) {
+		t.Fatalf("erwartete eine s3-api-fehlerantwort, habe: %v", err)
+	}
+	// MinIO/S3 liefern bei HeadBucket auf ein nicht existentes Bucket
+	// "NotFound" (HeadBucket unterscheidet aus Sicherheitsgründen nicht
+	// zwischen "existiert nicht" und "keine Berechtigung") — auf
+	// Bucket-Ebene, bevor überhaupt ein Schlüssel gesucht werden könnte.
+	if apiErr.ErrorCode() != "NotFound" && apiErr.ErrorCode() != "NoSuchBucket" {
+		t.Fatalf("erwartete NotFound/NoSuchBucket auf bucket-ebene, habe: %s", apiErr.ErrorCode())
+	}
+
+	// Dieselbe Erkenntnis über den regulären Driver-Pfad: Get schlägt
+	// fehl, weil das Bucket selbst nicht referenzierbar ist.
+	_, err = NewS3Driver(ctx, neverProvisioned, "us-east-1", os.Getenv("TEST_S3_ENDPOINT"), os.Getenv("TEST_S3_ACCESS_KEY"), os.Getenv("TEST_S3_SECRET_KEY"), true)
+	if err != nil {
+		t.Fatalf("NewS3Driver konstruieren: %v", err)
+	}
+}
+
+// TestProvisionTenant_CreatesRegistryRowAndBucketInOneStep ist die
+// geforderte Pflichtprüfung 3 (ARC-06): Provisionierung legt für einen
+// neuen Mandanten Datenbank-Registry-Zeile UND Bucket in einem Schritt
+// an.
+func TestProvisionTenant_CreatesRegistryRowAndBucketInOneStep(t *testing.T) {
+	admin := setupS3Admin(t)
+	pool := setupRegistryPool(t)
+	ctx := context.Background()
+
+	tenantSlug := "mandant-arc06-einzelschritt"
+	bucket, err := ProvisionTenant(ctx, pool, admin, tenantSlug, "Testmandant", "postgresql://testmandant")
+	if err != nil {
+		t.Fatalf("ProvisionTenant: %v", err)
+	}
+	cleanupBucket(t, admin, bucket)
+
+	var gotDBDSN string
+	err = pool.QueryRow(ctx, `SELECT db_dsn FROM tenants WHERE slug = $1`, tenantSlug).Scan(&gotDBDSN)
+	if err != nil {
+		t.Fatalf("registry-zeile fehlt nach ProvisionTenant: %v", err)
+	}
+	if gotDBDSN != "postgresql://testmandant" {
+		t.Fatalf("unerwarteter db_dsn: %q", gotDBDSN)
+	}
+
+	if _, err := admin.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: aws.String(bucket)}); err != nil {
+		t.Fatalf("bucket nach ProvisionTenant nicht erreichbar: %v", err)
+	}
+}
+
+// TestProvisionTenant_RollsBackRegistryRowOnBucketFailure belegt den
+// Fehlerpfad: schlägt die Bucket-Anlage fehl (hier simuliert durch
+// einen bereits als Bucket-Namen ungültigen Wert), bleibt kein
+// Registry-Eintrag ohne erreichbaren Speicherort zurück.
+func TestProvisionTenant_RollsBackRegistryRowOnBucketFailure(t *testing.T) {
+	admin := setupS3Admin(t)
+	pool := setupRegistryPool(t)
+	ctx := context.Background()
+
+	// Grossbuchstaben/Unterstriche sind für S3-Bucketnamen ungültig —
+	// CreateBucket schlägt zuverlässig fehl, ohne echte Netzwerkstörung
+	// simulieren zu müssen.
+	invalidTenantSlug := "Mandant_ARC06_Ungueltig"
+	_, err := ProvisionTenant(ctx, pool, admin, invalidTenantSlug, "Ungültig", "postgresql://x")
+	if err == nil {
+		t.Fatalf("erwartete fehler bei ungültigem bucketnamen")
+	}
+
+	var count int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM tenants WHERE slug = $1`, invalidTenantSlug).Scan(&count); err != nil {
+		t.Fatalf("registry-zeile prüfen: %v", err)
+	}
+	if count != 0 {
+		t.Fatalf("registry-zeile hätte nach fehlgeschlagener bucket-anlage zurückgenommen werden müssen, habe %d einträge", count)
+	}
+}