feat(PROJ-52): Vollständigkeits-Reconciliation (Zähl-Report Mailserver vs. Archiv)

Täglicher Cron-Job (archivmail reconcile) berechnet pro Tenant/Quelle
(SMTP-Journal, IMAP-Konto, POP3-Konto, Datei-Import) archivierte Mail-Zahlen,
für IMAP zusätzlich einen Soll/Ist-Vergleich via UID-Tracking. Abweichungen
über Schwellenwert erzeugen Audit-Log-Warnung. Neue Admin-Dashboard-Kachel
"Vollständigkeits-Check" (letzte 7 Tage, Warn-Badge, CSV-Export).

Schließt die "teilweise erfüllt"-Lücke bei Vollständigkeit im
GoBD/DSGVO-Compliance-Check (VOI-Grundsatz 2).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

cmd/archivmail/cmd_import.go

@@ -253,6 +253,11 @@ func importMessage(mailStore *storage.Store, idxMgr index.TenantIndexer, raw []b
 		return "error"
 	}
 
+	// PROJ-52: mark bulk-imported mails as source 'import' for reconciliation.
+	if err := mailStore.TagSource(context.Background(), id, "import", nil); err != nil {
+		fmt.Fprintf(os.Stderr, "warning: tag source failed: %v\n", err)
+	}
+
 	var attachNames []string
 	for _, a := range pm.Attachments {
 		attachNames = append(attachNames, a.Filename)

cmd/archivmail/cmd_reconcile.go

@@ -0,0 +1,95 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"log/slog"
+	"os"
+	"time"
+
+	"archivmail/config"
+	"archivmail/internal/audit"
+	"archivmail/internal/reconciliation"
+)
+
+// runReconcile computes the daily completeness reconciliation report (PROJ-52).
+// It is designed to be driven by cron once per day (e.g. shortly after
+// midnight) and, by default, reconciles the *previous* full calendar day so a
+// day is only counted once it is complete.
+//
+// Usage:
+//
+//	archivmail reconcile --config /etc/archivmail/config.yml
+//	archivmail reconcile --date 2026-07-01
+//	archivmail reconcile --days 7          # backfill: reconcile the last 7 days
+func runReconcile(args []string) {
+	fs := flag.NewFlagSet("reconcile", flag.ExitOnError)
+	configPath := fs.String("config", "/etc/archivmail/config.yml", "path to config file")
+	dateFlag := fs.String("date", "", "day to reconcile (YYYY-MM-DD, UTC); default: yesterday")
+	daysFlag := fs.Int("days", 1, "number of days to reconcile, ending at --date (backfill)")
+	fs.Parse(args)
+
+	logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
+
+	cfg, err := config.Load(*configPath)
+	if err != nil {
+		logger.Error("failed to load config", "err", err)
+		os.Exit(1)
+	}
+
+	// Determine the target (end) date.
+	var end time.Time
+	if *dateFlag != "" {
+		end, err = time.ParseInLocation("2006-01-02", *dateFlag, time.UTC)
+		if err != nil {
+			logger.Error("invalid --date (expected YYYY-MM-DD)", "err", err)
+			os.Exit(1)
+		}
+	} else {
+		end = time.Now().UTC().AddDate(0, 0, -1).Truncate(24 * time.Hour)
+	}
+
+	days := *daysFlag
+	if days < 1 {
+		days = 1
+	}
+
+	dsn := cfg.Database.DSN()
+	reconStore, err := reconciliation.New(dsn, logger)
+	if err != nil {
+		logger.Error("reconciliation store init failed", "err", err)
+		os.Exit(1)
+	}
+	defer reconStore.Close()
+
+	// Wire audit logging so anomalies are persisted as tenant-visible entries.
+	audlog, err := audit.New(dsn, cfg.Audit.ResolvedLogPath(), logger)
+	if err != nil {
+		logger.Warn("audit init failed — anomalies will only be logged", "err", err)
+	} else {
+		defer audlog.Close()
+		reconStore.SetAuditLogger(audlog)
+	}
+
+	thresholdPct := cfg.Reconciliation.ResolvedThresholdPct()
+	ctx := context.Background()
+
+	totalAnomalies := 0
+	// Reconcile oldest → newest so trailing-average history is populated in order.
+	for i := days - 1; i >= 0; i-- {
+		day := end.AddDate(0, 0, -i)
+		anomalies, err := reconStore.ComputeForDate(ctx, day, thresholdPct)
+		if err != nil {
+			// Per the spec: on failure the day is left WITHOUT a report row
+			// (no false zeros). Exit non-zero so cron surfaces the failure.
+			logger.Error("reconcile: compute failed", "date", day.Format("2006-01-02"), "err", err)
+			os.Exit(1)
+		}
+		totalAnomalies += len(anomalies)
+		logger.Info("reconcile: day complete",
+			"date", day.Format("2006-01-02"), "anomalies", len(anomalies))
+	}
+
+	logger.Info("reconcile: complete", "days", days, "threshold_pct", thresholdPct,
+		"anomalies_total", totalAnomalies)
+}

cmd/archivmail/main.go

@@ -31,6 +31,7 @@ import (
 	"archivmail/internal/mailer"
 	"archivmail/internal/ocr"
 	pop3store "archivmail/internal/pop3"
+	"archivmail/internal/reconciliation"
 	"archivmail/internal/smtpoutconfig"
 	"archivmail/internal/smtpd"
 	"archivmail/internal/storage"
@@ -73,6 +74,9 @@ func main() {
 		case "index-pending":
 			runIndexPending(os.Args[2:])
 			return
+		case "reconcile":
+			runReconcile(os.Args[2:])
+			return
 		case "update":
 			runUpdate(os.Args[2:])
 			return
@@ -313,11 +317,24 @@ func main() {
 	srv.SetGlobalRetentionDays(cfg.Storage.RetentionDays)
 	srv.SetMetrics(cfg.Metrics)
 
+	// PROJ-52: completeness reconciliation store — powers the dashboard +
+	// CSV-export endpoints. The daily computation itself is driven by cron via
+	// the `archivmail reconcile` subcommand (analog PROJ-58 batch jobs), so the
+	// daemon only needs read access here.
+	reconStore, err := reconciliation.New(cfg.Database.DSN(), logger)
+	if err != nil {
+		logger.Error("reconciliation store init failed", "err", err)
+		os.Exit(1)
+	}
+	defer reconStore.Close()
+	reconStore.SetAuditLogger(audlog)
+	srv.SetReconciliation(reconStore, cfg.Reconciliation.ResolvedThresholdPct())
+
 	// PROJ-28: Self-Service Onboarding — mailer + token store + FQDN
 	mlr := mailer.New(cfg.SMTPOut)
 
 	// SMTP-Out config store — load from DB, overrides config.yml if present
-	smtpOutSt, err := smtpoutconfig.New(cfg.Database.DSN(), cfg.API.Secret)
+	smtpOutSt, err := smtpoutconfig.New(cfg.Database.DSN(), aesKey)
 	if err != nil {
 		logger.Error("smtp-out config store init failed", "err", err)
 		os.Exit(1)

config.test.yml

@@ -50,3 +50,12 @@ audit:
   # Default falls leer: /var/log/archivmail/audit.log
   log_path: /tmp/archivmail-test/audit.log
   retention_days: 0
+
+# PROJ-52: Vollständigkeits-Reconciliation (Zähl-Report Mailserver vs. Archiv).
+# Der Job wird per Cron gestartet: `archivmail reconcile` (z.B. täglich 00:30).
+reconciliation:
+  # Relativer Rückgang (in Prozent) unter den 7-Tage-Durchschnitt einer Quelle,
+  # ab dem ein reconciliation_anomaly Audit-Eintrag geschrieben wird.
+  # Weglassen = Default 50. Hinweis: SMTP-Journale schwanken stark
+  # (Wochenende/Feiertage) — bei False-Positives Schwellenwert erhöhen.
+  alert_threshold_pct: 50

config/config.docker.yml.example

@@ -91,5 +91,12 @@ audit:
   log_path:       /var/archivmail/audit.log
   retention_days: 365
 
+# PROJ-52: Vollständigkeits-Reconciliation (Zähl-Report Mailserver vs. Archiv).
+# Täglich per Cron ausführen: `archivmail reconcile` (rechnet den Vortag ab).
+# alert_threshold_pct = relativer Rückgang (%) unter den 7-Tage-Durchschnitt,
+# ab dem ein reconciliation_anomaly Audit-Eintrag entsteht (Default 50).
+# reconciliation:
+#   alert_threshold_pct: 50
+
 logging:
   level: info

config/config.go

@@ -42,6 +42,39 @@ type Config struct {
 	// PROJ-56: load-spreading for background jobs.
 	OCR           OCRConfig           `yaml:"ocr"`
 	IMAPScheduler IMAPSchedulerConfig `yaml:"imap_scheduler"`
+	// PROJ-52: Vollständigkeits-Reconciliation (Zähl-Report Mailserver vs. Archiv).
+	Reconciliation ReconciliationConfig `yaml:"reconciliation"`
+}
+
+// ReconciliationConfig holds settings for the daily completeness reconciliation
+// job (PROJ-52). The job counts newly archived mails per source and per day and
+// flags significant drops against the trailing 7-day average.
+type ReconciliationConfig struct {
+	// AlertThresholdPct is the relative drop (in percent, below the trailing
+	// 7-day average) that triggers a `reconciliation_anomaly` audit entry.
+	// A pointer so an explicit 0 (alert on any drop) is distinguishable from an
+	// unset value (use the default).
+	//   nil -> DefaultReconciliationThresholdPct (50%)
+	//   0   -> alert whenever today's count is below the average
+	//   1..100 -> alert when today's count is more than this % below the average
+	AlertThresholdPct *int `yaml:"alert_threshold_pct,omitempty"`
+}
+
+// DefaultReconciliationThresholdPct is the default drop threshold (50% below the
+// trailing 7-day average) applied when reconciliation.alert_threshold_pct is
+// omitted from the config.
+const DefaultReconciliationThresholdPct = 50
+
+// ResolvedThresholdPct returns the effective alert threshold in percent.
+// A nil or out-of-range value falls back to the default.
+func (c ReconciliationConfig) ResolvedThresholdPct() int {
+	if c.AlertThresholdPct == nil {
+		return DefaultReconciliationThresholdPct
+	}
+	if *c.AlertThresholdPct < 0 || *c.AlertThresholdPct > 100 {
+		return DefaultReconciliationThresholdPct
+	}
+	return *c.AlertThresholdPct
 }
 
 // OCRConfig holds settings for the background OCR worker (PROJ-56).

features/INDEX.md

@@ -68,7 +68,7 @@
 | PROJ-49 | Verschlüsselungspflicht at-rest (Healthcheck & Warnung) | Deployed | [PROJ-49](PROJ-49-verschluesselungspflicht.md) | 2026-06-13 |
 | PROJ-50 | DSGVO-Löschersuchen für Mail-Inhalte (GoBD-Vorrang) | Deployed | [PROJ-50](PROJ-50-dsgvo-loeschersuchen.md) | 2026-06-13 |
 | PROJ-51 | Aufbewahrungsfristen nach Dokumentenart (Retention-Kategorien) | Deployed | [PROJ-51](PROJ-51-retention-kategorien.md) | 2026-06-13 |
-| PROJ-52 | Vollständigkeits-Reconciliation (Zähl-Report) | Planned | [PROJ-52](PROJ-52-vollstaendigkeits-reconciliation.md) | 2026-06-13 |
+| PROJ-52 | Vollständigkeits-Reconciliation (Zähl-Report) | In Review | [PROJ-52](PROJ-52-vollstaendigkeits-reconciliation.md) | 2026-06-13 |
 | PROJ-53 | Konfigurierbare Listenanzahl pro Seite | Deployed | [PROJ-53](PROJ-53-konfigurierbare-listenanzahl.md) | 2026-06-14 |
 | PROJ-54 | Fix Listenansicht/Pagination für Rolle "user" (Nachbesserung PROJ-6/PROJ-21) | Deployed | [PROJ-54](PROJ-54-fix-listenansicht-total.md) | 2026-06-14 |
 | PROJ-55 | Fix Tenant-Isolation für Rolle "auditor" + Audit-Log (Sicherheitsbug, DSGVO-relevant) | Deployed | [PROJ-55](PROJ-55-fix-auditor-tenant-isolation.md) | 2026-06-21 |

features/PROJ-52-vollstaendigkeits-reconciliation.md

@@ -0,0 +1,224 @@
+# PROJ-52: Vollständigkeits-Reconciliation (Zähl-Report Mailserver vs. Archiv)
+
+## Status: In Review
+**Created:** 2026-06-13
+**Last Updated:** 2026-07-03
+
+## Hintergrund
+Der GoBD/DSGVO-Compliance-Check (`docs/GOBD_DSGVO_CHECKLIST.md`, Punkt 1) bewertet
+"Vollständigkeit" nur als "Teilweise erfüllt": SMTP-BCC-Journaling (PROJ-4) und IMAP/POP3-Import
+(PROJ-3/8/14/45) sind robust (z.B. `452`-Retry bei Storage-Fehlern), aber es gibt keinen
+zentralen Mechanismus, der zeigt, ob tatsächlich ALLE erwarteten E-Mails archiviert wurden
+(VOI-Grundsatz 2: "kein Dokument darf auf dem Weg ins Archiv oder im Archiv selbst verloren
+gehen"). Diese Spec ergänzt einen täglichen Zähl-Report pro Quelle.
+
+## Dependencies
+- Requires: PROJ-4 (SMTP-Import), PROJ-3/PROJ-14 (IMAP/POP3-Import), PROJ-45
+  (IMAP Per-Folder UID-Tracking)
+- Requires: PROJ-17 (Admin Dashboard) – Anzeige des Reports
+- Requires: PROJ-11/PROJ-48 (Audit-Log) – Auffälligkeiten werden protokolliert
+
+## User Stories
+- Als Admin möchte ich täglich sehen, wie viele E-Mails pro Quelle (SMTP-Journal, IMAP-Konto,
+  POP3-Konto) archiviert wurden, damit ich Ausreißer (plötzlich 0 Mails) erkenne.
+- Als Admin möchte ich für IMAP/POP3-Quellen einen Soll/Ist-Vergleich sehen: Anzahl Mails im
+  Quell-Postfach (laut letztem Sync) vs. Anzahl archivierter Mails für diese Quelle.
+- Als Auditor möchte ich nachvollziehen können, ob es Tage mit auffälligen Abweichungen gab
+  (z.B. SMTP-Dienst war down).
+- Als Admin möchte ich bei einer signifikanten Abweichung (z.B. >50% Rückgang ggü.
+  Durchschnitt der letzten 7 Tage) eine Warnung im Dashboard sehen.
+
+## Acceptance Criteria
+- [ ] Täglicher Job (Cron, analog PROJ-8-Scheduler) berechnet pro Tag und Quelle
+      (`source_type`: `smtp`, `imap:<account_id>`, `pop3:<account_id>`, `import`) die Anzahl
+      neu archivierter E-Mails (`received_at`/`imported_at` am jeweiligen Tag)
+- [ ] Für IMAP-Konten (PROJ-45 UID-Tracking): zusätzlicher Soll/Ist-Vergleich – Anzahl Mails im
+      Quell-Ordner laut letztem `UIDVALIDITY`/UID-Stand vs. Anzahl im Archiv für diesen Ordner
+- [ ] Ergebnisse werden in Tabelle `reconciliation_reports`
+      (date, tenant_id, source_type, source_id, expected_count, archived_count, delta)
+      persistiert
+- [ ] Admin-Dashboard (PROJ-17) zeigt eine neue Kachel/Tabelle "Vollständigkeits-Check" mit
+      den letzten 7 Tagen pro Quelle
+- [ ] Abweichung > konfigurierbarem Schwellenwert (Default: 50% unter 7-Tage-Durchschnitt,
+      `reconciliation.alert_threshold_pct` in `config.yml`) → Warn-Badge im Dashboard +
+      Audit-Log-Eintrag (`event_type: reconciliation_anomaly`)
+- [ ] Report ist als CSV exportierbar (analog Audit-Log-Export aus PROJ-11)
+- [ ] Tage ohne Aktivität (0 Mails) werden explizit als `0` ausgewiesen, nicht als fehlender
+      Datensatz (damit Lücken im Cron-Lauf selbst erkennbar sind)
+
+## Edge Cases
+- Quelle wurde erst kürzlich angelegt (kein 7-Tage-Durchschnitt vorhanden) → kein Alert,
+  Anzeige "Noch nicht genug Daten"
+- SMTP-Journal hat naturgemäß starke Schwankungen (Wochenende vs. Wochentag) → Schwellenwert
+  ist konfigurierbar, Doku weist auf mögliche False-Positives an Wochenenden/Feiertagen hin
+- IMAP-Quell-Postfach wurde vom Nutzer geleert (Mails dort gelöscht, aber bereits archiviert) →
+  `expected_count` sinkt, `archived_count` bleibt hoch → `delta` negativ in "gute" Richtung,
+  kein Alert (nur Rückgang von `archived_count` selbst ist relevant)
+- Reconciliation-Job selbst schlägt fehl (z.B. DB-Timeout) → Fehler wird geloggt, vorheriger
+  Tag bleibt ohne Report-Eintrag, Dashboard zeigt "Daten fehlen für <Datum>" statt falscher
+  Nullwerte
+- Multi-Tenant: Reports sind pro Tenant; Tenant-Admins sehen nur eigene Quellen, Super-Admin
+  sieht alle
+
+## Technical Requirements
+- Neue Tabelle `reconciliation_reports` (siehe AC), Index auf `(tenant_id, date, source_type)`
+- Cron-Job-Registrierung analog bestehendem IMAP-Sync-Scheduler (PROJ-8)
+- Wiederverwendung von `internal/imap`-Funktionen zur Ermittlung der Quell-Postfach-Anzahl
+  (sofern bereits durch UID-Tracking verfügbar, kein zusätzlicher IMAP-Login nötig wenn
+  vermeidbar)
+
+---
+<!-- Sections below are added by subsequent skills -->
+
+## Implementation Notes (Backend, 2026-07-03)
+
+### Neues Package `internal/reconciliation/`
+- `reconciliation.go`: `Store` (eigener pgxpool), `initSchema()`, Tabelle
+  `reconciliation_reports (id, date, tenant_id, source_type, source_id,
+  expected_count, archived_count, delta, created_at)`. Upsert-Key ist ein
+  COALESCE-Ausdrucksindex `(date, COALESCE(tenant_id,-1), source_type,
+  COALESCE(source_id,-1))`, weil `tenant_id`/`source_id` NULL-fähig sind und
+  Postgres NULLs in einem normalen UNIQUE-Index als verschieden behandelt (sonst
+  Doppelzeilen für smtp/import/tenant-lose Buckets). Zusätzlicher Lookup-Index
+  `(tenant_id, date, source_type)` laut AC.
+- `compute.go`: `ComputeForDate(ctx, day, thresholdPct)` — Read-Phase (archived
+  pro Bucket für den Tag, known-buckets aus `emails` DISTINCT, IMAP-Snapshot),
+  dann Upsert in **einer Transaktion**. Bei Query-Fehler wird der Tag NICHT
+  geschrieben (kein falscher 0-Eintrag; Dashboard zeigt "fehlt"). 0-Mail-Tage
+  werden für jeden bekannten Bucket **explizit als 0** persistiert. Alert:
+  7-Tage-Durchschnitt (`trailing_average`, NULL-safe `IS NOT DISTINCT FROM`);
+  Alert nur bei ≥7 Vortages-Datensätzen; `archived < avg*(1-pct/100)` →
+  Audit-Eintrag `reconciliation_anomaly` (event via `audit.EventReconciliationAnomaly`).
+- `query.go`: `DashboardData()` (letzte N Tage pro Quelle, tenant-gescoped,
+  Alert-Flag + `enough_data`) und `ExportRows()` (CSV).
+
+### Source-Tracking (nötige Ergänzung — emails hatte keine Herkunftsspalte)
+`emails` bekam via `storage.initSchema()` zwei Spalten `source_type TEXT`,
+`source_id BIGINT` + Index `(received_at, source_type, source_id, tenant_id)`.
+Neue Methode `storage.Store.TagSource(ctx, id, sourceType, sourceID)` schreibt
+nur solange `source_type IS NULL` (first-write-wins → dedupte Mehrfach-Mails
+werden nicht doppelt gezählt). Verdrahtet in allen Ingestion-Pfaden:
+- `internal/smtpd/smtpd.go` → `smtp`, nil
+- `internal/imap/importer.go` + `internal/imap/scheduler.go` → `imap`,
+  account_id (accountID durch `fetchBatch`/`fetchSyncBatch`/`storeAndIndex`
+  durchgereicht)
+- `internal/pop3/importer.go` → `pop3`, account_id
+- `cmd/archivmail/cmd_import.go` + `internal/api/upload.go` → `import`, nil
+
+### IMAP Soll/Ist (Abweichung von der Spec — dokumentiert)
+Das PROJ-45 UID-Tracking (`imap_folder_state`) speichert nur `last_uid`, KEINE
+Nachrichtenzahl. Daher wird `expected_count` für IMAP-Quellen als Proxy aus
+`SUM(last_uid)` je Konto gebildet (kein zusätzlicher IMAP-Login) und
+`delta = kumulativ_archiviert(Konto) − expected`. `archived_count` bleibt
+konsistent für ALLE Quellen die **pro-Tag** neu archivierte Zahl (steuert das
+Alerting). Postfach-Leerung → expected sinkt, delta positiv → kein Alert
+(Alert nur bei Rückgang von `archived_count`), wie in Edge Cases gefordert.
+
+### Cron statt Dauer-Goroutine
+Analog PROJ-58 als CLI-Subcommand `archivmail reconcile` (`cmd_reconcile.go`),
+Default = Vortag; `--date`, `--days N` (Backfill, älteste→neueste Reihenfolge
+für konsistente Trailing-Average-Historie). Registriert in `main.go`. Der Daemon
+(`serve`) verdrahtet den Store nur lesend für die API (`SetReconciliation`).
+
+### Config
+`config.ReconciliationConfig.AlertThresholdPct *int` (`reconciliation.alert_threshold_pct`),
+Default 50 via `ResolvedThresholdPct()`. Beispiele in `config.test.yml` und
+`config/config.docker.yml.example`.
+
+### API-Endpoints (tenant-gescoped, `authAdmin` = domain_admin+)
+- `GET /api/admin/reconciliation?days=7` (max 90)
+  ```json
+  {
+    "days": 7,
+    "threshold_pct": 50,
+    "sources": [
+      {
+        "source_type": "smtp",
+        "source_id": null,
+        "source_key": "smtp",
+        "tenant_id": null,
+        "points": [
+          {"date":"2026-06-27","archived_count":42,"expected_count":null,"delta":null,"missing":false},
+          {"date":"2026-06-28","archived_count":null,"expected_count":null,"delta":null,"missing":true}
+        ],
+        "avg_7d": 40.5,
+        "enough_data": true,
+        "alert": false
+      }
+    ]
+  }
+  ```
+  `source_key`: `smtp` | `import` | `imap:<id>` | `pop3:<id>`. `missing:true` =
+  kein Report-Datensatz (Cron nicht gelaufen), ≠ `archived_count:0`.
+  `enough_data:false` → UI zeigt "Noch nicht genug Daten". `alert:true` →
+  Warn-Badge.
+- `GET /api/admin/reconciliation/export.csv?days=30` (max 366) — CSV
+  `date,tenant_id,source,expected_count,archived_count,delta`, Audit-Eintrag
+  `export`. Tenant-Scope: domain_admin nur eigener Tenant, superadmin alle.
+
+### Tenant-Isolation
+`reconTenantScope()` filtert wie `handleMailTimeseries`: Session mit `tenant_id`
+→ nur eigener Tenant, superadmin (nil) → alle. Kein `{id}`-Pfadparameter, daher
+kein IDOR-Vektor; Filter erfolgt in der SQL-WHERE.
+
+### Manticore
+Keine Index-Änderung nötig (reine PostgreSQL-Aggregation).
+
+### Offene Punkte / Handoff
+- Cron-Eintrag für `archivmail reconcile` muss in `install.sh`/`update.sh`
+  ergänzt werden (devops-deploy).
+- Frontend: Dashboard-Kachel + TS-Typen in `src/lib/api/`.
+- Kein lokaler `go build` möglich — Build/QA separat auf Testserver.
+
+## Implementation Notes (Frontend, 2026-07-03)
+
+### Neue API-Schicht `src/lib/api/reconciliation.ts`
+- Typen `ReconciliationPoint`, `ReconciliationSource`, `ReconciliationResponse`
+  (1:1 zur Backend-Response; `archived_count`/`expected_count`/`delta` sind
+  `number | null`, `missing: boolean`).
+- `getReconciliation(days = 7)` → `GET /api/admin/reconciliation?days=7` über den
+  bestehenden `request<T>`-Wrapper (`core.ts`, credentials/401-Handling inklusive).
+- `exportReconciliationCSV(days = 30)` → direkter `fetch` auf
+  `/api/admin/reconciliation/export.csv` (Blob-Download, Content-Disposition-
+  Dateiname-Parsing, analog `downloadMailAttachment`/`exportDSGVORequestPDF`).
+- Re-Exports in `src/lib/api/index.ts` ergänzt.
+
+### Neue Kachel `src/components/admin/tabs/ReconciliationCard.tsx`
+- Self-fetching Client-Komponente (`useEffect` beim Mount), Loading-/Error-/
+  Empty-States implementiert.
+- Tabelle (shadcn `Table`): eine Zeile pro Quelle, Label hübsch formatiert
+  ("SMTP-Journal", "Datei-Import", "IMAP-Konto #3", "POP3-Konto #X").
+- Spalten = letzte 7 Tage; Zelle zeigt `archived_count`, bei IMAP zusätzlich
+  "Soll <expected> (delta)". `missing:true` → "—" (klar unterschieden von "0",
+  mit Tooltip "Cron nicht gelaufen"). `enough_data:false` → Zeile zeigt
+  "Noch nicht genug Daten" (colSpan). `alert:true` → rotes Badge "Auffällig",
+  sonst "OK". Zusätzliche Spalten "Ø 7 Tage" und "Status".
+- CSV-Export-Button (30 Tage) und Aktualisieren-Button im Kachel-Header.
+- Horizontal scrollbar (`overflow-x-auto`) für mobile Breiten.
+
+### Einbindung / Rollen-Sichtbarkeit
+- Gerendert innerhalb `DashboardTab` (vor "Benutzerübersicht"). Der gesamte
+  Admin-Bereich (`src/app/admin/page.tsx`) ist bereits via
+  `useAuth("domain_admin", "/admin/login")` auf domain_admin+ beschränkt →
+  keine zusätzliche Client-Gate nötig, normale User erreichen den Tab nicht.
+  Backend bleibt maßgebliche Sicherheitsgrenze (tenant-gescoped, `authAdmin`);
+  superadmin sieht alle Quellen, domain_admin nur eigene.
+
+### Verifikation
+- `npx tsc --noEmit` fehlerfrei. Kein direktes `fetch()` in Komponenten außer
+  dem zentralen Blob-Download-Helper in der API-Schicht.
+
+### Geänderte/neue Dateien
+- neu: `src/lib/api/reconciliation.ts`
+- neu: `src/components/admin/tabs/ReconciliationCard.tsx`
+- geändert: `src/lib/api/index.ts` (Re-Exports)
+- geändert: `src/components/admin/tabs/DashboardTab.tsx` (Kachel eingebunden)
+
+## Tech Design (Solution Architect)
+_To be added by /architecture_
+
+## QA Test Results
+_To be added by /qa_
+
+## Deployment
+_To be added by /deploy_

internal/api/reconciliation_handlers.go

@@ -0,0 +1,124 @@
+package api
+
+import (
+	"encoding/csv"
+	"fmt"
+	"net/http"
+	"strconv"
+
+	"archivmail/internal/audit"
+)
+
+// tenantScope returns the tenant filter for reconciliation queries: a
+// domain_admin (and any other tenant-scoped role) is restricted to its own
+// tenant, while superadmin (sess.TenantID == nil) sees all tenants. This
+// mirrors handleMailTimeseries and prevents cross-tenant leakage of source
+// figures (PROJ-55/61 tenant-isolation discipline).
+func (s *Server) reconTenantScope(r *http.Request) *int64 {
+	sess := sessionFromCtx(r.Context())
+	if sess.TenantID != nil {
+		return tenantFromCtx(r.Context())
+	}
+	return nil
+}
+
+// handleReconciliation returns the last N days (default 7) of completeness
+// figures per source, tenant-scoped.
+// GET /api/admin/reconciliation?days=7
+func (s *Server) handleReconciliation(w http.ResponseWriter, r *http.Request) {
+	if s.reconStore == nil {
+		writeError(w, http.StatusServiceUnavailable, "reconciliation not enabled")
+		return
+	}
+
+	days := 7
+	if v := r.URL.Query().Get("days"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 90 {
+			days = n
+		}
+	}
+
+	tid := s.reconTenantScope(r)
+	sources, err := s.reconStore.DashboardData(r.Context(), tid, days, s.reconThresholdPct)
+	if err != nil {
+		s.logger.Error("reconciliation dashboard query failed", "err", err)
+		writeError(w, http.StatusInternalServerError, "reconciliation query failed")
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]interface{}{
+		"days":          days,
+		"threshold_pct": s.reconThresholdPct,
+		"sources":       sources,
+	})
+}
+
+// handleReconciliationExport streams the reconciliation report as CSV,
+// tenant-scoped (analog PROJ-11 audit export).
+// GET /api/admin/reconciliation/export.csv?days=30
+func (s *Server) handleReconciliationExport(w http.ResponseWriter, r *http.Request) {
+	if s.reconStore == nil {
+		writeError(w, http.StatusServiceUnavailable, "reconciliation not enabled")
+		return
+	}
+
+	days := 30
+	if v := r.URL.Query().Get("days"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 366 {
+			days = n
+		}
+	}
+
+	tid := s.reconTenantScope(r)
+	rows, err := s.reconStore.ExportRows(r.Context(), tid, days)
+	if err != nil {
+		s.logger.Error("reconciliation export query failed", "err", err)
+		writeError(w, http.StatusInternalServerError, "reconciliation query failed")
+		return
+	}
+
+	sess := sessionFromCtx(r.Context())
+
+	w.Header().Set("Content-Type", "text/csv; charset=utf-8")
+	w.Header().Set("Content-Disposition", `attachment; filename="reconciliation.csv"`)
+	w.WriteHeader(http.StatusOK)
+
+	cw := csv.NewWriter(w)
+	cw.Write([]string{"date", "tenant_id", "source", "expected_count", "archived_count", "delta"}) //nolint:errcheck
+	for _, row := range rows {
+		cw.Write([]string{ //nolint:errcheck
+			row.Date.UTC().Format("2006-01-02"),
+			nullableInt(row.TenantID),
+			reconSourceKey(row.SourceType, row.SourceID),
+			nullableInt(row.ExpectedCount),
+			strconv.FormatInt(row.ArchivedCount, 10),
+			nullableInt(row.Delta),
+		})
+	}
+	cw.Flush()
+
+	s.audlog.Log(audit.Entry{
+		EventType: audit.EventExport,
+		Username:  sess.Username,
+		TenantID:  sess.TenantID,
+		IPAddress: s.remoteIP(r),
+		Detail:    fmt.Sprintf("reconciliation csv: %d days, %d rows", days, len(rows)),
+		Success:   true,
+	})
+}
+
+// nullableInt formats a *int64 for CSV, emitting an empty string for nil.
+func nullableInt(v *int64) string {
+	if v == nil {
+		return ""
+	}
+	return strconv.FormatInt(*v, 10)
+}
+
+// reconSourceKey mirrors reconciliation.SourceKey without importing the package
+// into the CSV hot path (kept local and tiny).
+func reconSourceKey(sourceType string, sourceID *int64) string {
+	if sourceID != nil && (sourceType == "imap" || sourceType == "pop3") {
+		return fmt.Sprintf("%s:%d", sourceType, *sourceID)
+	}
+	return sourceType
+}

internal/api/server.go

@@ -20,6 +20,7 @@ import (
 	ldapcfg "archivmail/internal/ldapconfig"
 	"archivmail/internal/mailer"
 	pop3store "archivmail/internal/pop3"
+	"archivmail/internal/reconciliation"
 	"archivmail/internal/smtpoutconfig"
 	"archivmail/internal/smtpd"
 	"archivmail/internal/storage"
@@ -90,6 +91,8 @@ type Server struct {
 	fqdn                 string // from server.fqdn config (PROJ-28)
 	smtpOutStore         *smtpoutconfig.Store
 	apiKeyMw             *auth.APIKeyMiddleware // PROJ-13: external API auth
+	reconStore           *reconciliation.Store  // PROJ-52: completeness reconciliation
+	reconThresholdPct    int                    // PROJ-52: alert threshold (percent below 7-day avg)
 }
 
 // SetSMTPDaemon wires the SMTP daemon into the API server after construction.
@@ -151,6 +154,14 @@ func (s *Server) SetSMTPOutStore(store *smtpoutconfig.Store) {
 	s.smtpOutStore = store
 }
 
+// SetReconciliation wires the completeness-reconciliation store and the alert
+// threshold (percent below the trailing 7-day average) into the API server
+// (PROJ-52).
+func (s *Server) SetReconciliation(store *reconciliation.Store, thresholdPct int) {
+	s.reconStore = store
+	s.reconThresholdPct = thresholdPct
+}
+
 // New creates and wires up a new API server.
 func New(
 	cfg config.APIConfig,
@@ -222,6 +233,9 @@ func (s *Server) routes() {
 
 	s.mux.HandleFunc("GET /api/admin/system/stats", s.authAdmin(s.handleSystemStats))
 	s.mux.HandleFunc("GET /api/admin/stats/timeseries", s.authAdmin(s.handleMailTimeseries))
+	// PROJ-52: Vollständigkeits-Reconciliation (Dashboard + CSV-Export) — admin, tenant-scoped.
+	s.mux.HandleFunc("GET /api/admin/reconciliation", s.authAdmin(s.handleReconciliation))
+	s.mux.HandleFunc("GET /api/admin/reconciliation/export.csv", s.authAdmin(s.handleReconciliationExport))
 	s.mux.HandleFunc("GET /api/admin/security/audit", s.authAdmin(s.handleSecurityAudit))
 	// SEC-17: Security fix actions require superadmin, not just domain_admin.
 	s.mux.HandleFunc("POST /api/admin/security/fix", s.auth(s.requireRole(userstore.RoleSuperAdmin, s.handleSecurityFix)))

internal/api/upload.go

@@ -167,6 +167,11 @@ func (s *Server) importRawMessage(ctx context.Context, raw []byte, tenantID *int
 		return "error"
 	}
 
+	// PROJ-52: uploaded mails count as source 'import' for reconciliation.
+	if err := s.store.TagSource(ctx, id, "import", nil); err != nil {
+		s.logger.Warn("upload: tag source failed", "id", id, "err", err)
+	}
+
 	// Check dedup: storage.Save returns same id for duplicate content.
 	// If already indexed, skip indexing.
 	if already, _ := s.store.IsIndexed(ctx, id); already {

internal/audit/audit.go

@@ -24,6 +24,10 @@ const (
 	EventUserMgmt    = "user_mgmt"
 	EventOCRDownload = "mail:ocr_download" // PROJ-44: extracted OCR text downloaded
 	EventDSGVORequest = "dsgvo_request"    // PROJ-50: DSGVO Löschersuchen erfasst/bearbeitet
+	// EventReconciliationAnomaly (PROJ-52): a source's newly-archived count for a
+	// day dropped significantly below its trailing 7-day average, or the IMAP
+	// soll/ist comparison revealed a shortfall.
+	EventReconciliationAnomaly = "reconciliation_anomaly"
 )
 
 // Entry is a single audit log record.

internal/imap/importer.go

@@ -170,7 +170,7 @@ func (imp *Importer) doImport(ctx context.Context, acc *Account, password string
 
 			// Set per-batch deadline to prevent indefinite blocking on stalled connections.
 			c.SetFetchDeadline()
-			count, err := imp.fetchBatch(ctx, c.Client, batch, acc.TenantID, log)
+			count, err := imp.fetchBatch(ctx, c.Client, batch, acc.TenantID, acc.ID, log)
 			c.ClearDeadline()
 			if err != nil {
 				log.Error("batch fetch error — aborting import", "folder", folder, "offset", i, "err", err)
@@ -188,7 +188,7 @@ func (imp *Importer) doImport(ctx context.Context, acc *Account, password string
 }
 
 // fetchBatch fetches and stores a batch of messages by UID.
-func (imp *Importer) fetchBatch(ctx context.Context, c *imapclient.Client, uids []imapv2.UID, tenantID *int64, log *slog.Logger) (int, error) {
+func (imp *Importer) fetchBatch(ctx context.Context, c *imapclient.Client, uids []imapv2.UID, tenantID *int64, accountID int64, log *slog.Logger) (int, error) {
 	if len(uids) == 0 {
 		return 0, nil
 	}
@@ -223,7 +223,7 @@ func (imp *Importer) fetchBatch(ctx context.Context, c *imapclient.Client, uids
 					continue
 				}
 
-				if err := imp.storeAndIndex(raw, tenantID, log); err != nil {
+				if err := imp.storeAndIndex(raw, tenantID, accountID, log); err != nil {
 					log.Warn("failed to store/index message", "err", err)
 					continue
 				}
@@ -240,7 +240,8 @@ func (imp *Importer) fetchBatch(ctx context.Context, c *imapclient.Client, uids
 }
 
 // storeAndIndex saves a raw email to storage and indexes it.
-func (imp *Importer) storeAndIndex(raw []byte, tenantID *int64, log *slog.Logger) error {
+// accountID identifies the IMAP account for PROJ-52 source tracking.
+func (imp *Importer) storeAndIndex(raw []byte, tenantID *int64, accountID int64, log *slog.Logger) error {
 	ctx := context.Background()
 	// Save to file storage (deduplicates by SHA256 automatically)
 	id, err := imp.mailStore.Save(ctx, raw, time.Now(), tenantID)
@@ -248,6 +249,12 @@ func (imp *Importer) storeAndIndex(raw []byte, tenantID *int64, log *slog.Logger
 		return fmt.Errorf("save: %w", err)
 	}
 
+	// PROJ-52: record ingestion source (imap:<account_id>). Non-fatal.
+	accID := accountID
+	if err := imp.mailStore.TagSource(ctx, id, "imap", &accID); err != nil {
+		log.Warn("failed to tag source", "id", id, "err", err)
+	}
+
 	// Parse for indexing
 	pm, err := mailparser.Parse(raw)
 	if err != nil {

internal/imap/scheduler.go

@@ -438,7 +438,7 @@ func (s *Scheduler) syncFolder(
 		batch := uids[i:end]
 
 		c.SetFetchDeadline()
-		count, batchMaxUID, err := s.fetchSyncBatch(c.Client, batch, acc.TenantID, log)
+		count, batchMaxUID, err := s.fetchSyncBatch(c.Client, batch, acc.TenantID, acc.ID, log)
 		c.ClearDeadline()
 		if err != nil {
 			log.Warn("imap scheduler: batch error, continuing",
@@ -468,6 +468,7 @@ func (s *Scheduler) fetchSyncBatch(
 	c *imapclient.Client,
 	uids []imapv2.UID,
 	tenantID *int64,
+	accountID int64,
 	log *slog.Logger,
 ) (int, uint32, error) {
 	if len(uids) == 0 {
@@ -515,7 +516,7 @@ func (s *Scheduler) fetchSyncBatch(
 			}
 
 			if len(raw) > 0 {
-				if err := s.importer.storeAndIndex(raw, tenantID, log); err != nil {
+				if err := s.importer.storeAndIndex(raw, tenantID, accountID, log); err != nil {
 					log.Warn("imap scheduler: store/index failed", "err", err)
 				} else {
 					imported++

internal/pop3/importer.go

@@ -119,7 +119,7 @@ func (imp *Importer) doImport(ctx context.Context, acc *Account, password string
 			continue
 		}
 
-		if err := imp.storeAndIndex(raw, log); err != nil {
+		if err := imp.storeAndIndex(raw, acc.ID, log); err != nil {
 			log.Warn("failed to store/index message, skipping", "msg_num", num, "err", err)
 		} else {
 			imported++
@@ -133,7 +133,8 @@ func (imp *Importer) doImport(ctx context.Context, acc *Account, password string
 }
 
 // storeAndIndex saves a raw email to storage and indexes it.
-func (imp *Importer) storeAndIndex(raw []byte, log *slog.Logger) error {
+// accountID identifies the POP3 account for PROJ-52 source tracking.
+func (imp *Importer) storeAndIndex(raw []byte, accountID int64, log *slog.Logger) error {
 	ctx := context.Background()
 	// Save to file storage (deduplicates by SHA256 automatically)
 	id, err := imp.mailStore.Save(ctx, raw, time.Now(), imp.TenantID)
@@ -141,6 +142,12 @@ func (imp *Importer) storeAndIndex(raw []byte, log *slog.Logger) error {
 		return fmt.Errorf("pop3 save: %w", err)
 	}
 
+	// PROJ-52: record ingestion source (pop3:<account_id>). Non-fatal.
+	accID := accountID
+	if err := imp.mailStore.TagSource(ctx, id, "pop3", &accID); err != nil {
+		log.Warn("failed to tag source", "id", id, "err", err)
+	}
+
 	// Parse for indexing
 	pm, err := mailparser.Parse(raw)
 	if err != nil {

internal/reconciliation/compute.go

@@ -0,0 +1,337 @@
+package reconciliation
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"archivmail/internal/audit"
+)
+
+// bucketKey identifies one reconciliation bucket in memory. Nil tenant/source
+// IDs are encoded as -1 so they can be used as map keys.
+type bucketKey struct {
+	tenant     int64
+	sourceType string
+	sourceID   int64
+}
+
+func keyOf(tenant, sourceID *int64, sourceType string) bucketKey {
+	t := int64(-1)
+	if tenant != nil {
+		t = *tenant
+	}
+	sid := int64(-1)
+	if sourceID != nil {
+		sid = *sourceID
+	}
+	return bucketKey{tenant: t, sourceType: sourceType, sourceID: sid}
+}
+
+func ptr(v int64) *int64 { return &v }
+
+func nilIfNeg(v int64) *int64 {
+	if v < 0 {
+		return nil
+	}
+	return ptr(v)
+}
+
+// imapExpected holds the IMAP soll/ist snapshot for one account.
+type imapExpected struct {
+	tenant    *int64
+	expected  int64 // sum of per-folder last_uid high-water marks (source proxy)
+	cumulArch int64 // cumulative archived mails for this account
+}
+
+// Anomaly describes a detected significant drop for one source/day.
+type Anomaly struct {
+	Date       time.Time
+	TenantID   *int64
+	SourceKey  string
+	Archived   int64
+	Average    float64
+	ThresholdPct int
+}
+
+// ComputeForDate reconciles a single calendar day (UTC) and upserts one row per
+// known source bucket. Rows are only written after every read query has
+// succeeded, so a mid-job DB failure leaves the day WITHOUT a report row
+// (dashboard shows "data missing") instead of a misleading all-zero report.
+//
+// After persisting, it evaluates the trailing 7-day average per source and
+// writes a `reconciliation_anomaly` audit entry when today's archived count has
+// dropped more than thresholdPct percent below that average. Sources with fewer
+// than 7 prior daily records are skipped ("not enough data yet").
+//
+// Returns the anomalies detected (also useful for the CLI summary/tests).
+func (s *Store) ComputeForDate(ctx context.Context, day time.Time, thresholdPct int) ([]Anomaly, error) {
+	dayStart := time.Date(day.Year(), day.Month(), day.Day(), 0, 0, 0, 0, time.UTC)
+	dayEnd := dayStart.Add(24 * time.Hour)
+
+	// ── Read phase (all-or-nothing) ────────────────────────────────────────
+	archived, err := s.archivedForDay(ctx, dayStart, dayEnd)
+	if err != nil {
+		return nil, err
+	}
+	known, err := s.knownBuckets(ctx)
+	if err != nil {
+		return nil, err
+	}
+	imap, err := s.imapExpectedSnapshot(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	// ── Build rows ─────────────────────────────────────────────────────────
+	rows := make([]Report, 0, len(known))
+	for k := range known {
+		r := Report{
+			Date:          dayStart,
+			TenantID:      nilIfNeg(k.tenant),
+			SourceType:    k.sourceType,
+			SourceID:      nilIfNeg(k.sourceID),
+			ArchivedCount: archived[k], // 0 when the source had no mail that day
+		}
+		// IMAP soll/ist: expected = source mailbox size proxy (sum of last_uid),
+		// delta = cumulative archived for the account − expected. delta going
+		// negative because the user emptied the source mailbox is a "good"
+		// direction and never alerts (alerting keys off archived_count only).
+		if k.sourceType == "imap" && k.sourceID >= 0 {
+			if ie, ok := imap[k.sourceID]; ok {
+				r.ExpectedCount = ptr(ie.expected)
+				r.Delta = ptr(ie.cumulArch - ie.expected)
+			}
+		}
+		rows = append(rows, r)
+	}
+
+	// ── Write phase ────────────────────────────────────────────────────────
+	if err := s.upsertRows(ctx, rows); err != nil {
+		return nil, err
+	}
+
+	// ── Alert phase ────────────────────────────────────────────────────────
+	var anomalies []Anomaly
+	for _, r := range rows {
+		avg, n, err := s.trailingAverage(ctx, r, dayStart)
+		if err != nil {
+			s.logger.Warn("reconciliation: trailing average failed",
+				"source", SourceKey(r.SourceType, r.SourceID), "err", err)
+			continue
+		}
+		if n < 7 {
+			continue // not enough history yet
+		}
+		limit := avg * (1 - float64(thresholdPct)/100.0)
+		if float64(r.ArchivedCount) < limit {
+			a := Anomaly{
+				Date:         dayStart,
+				TenantID:     r.TenantID,
+				SourceKey:    SourceKey(r.SourceType, r.SourceID),
+				Archived:     r.ArchivedCount,
+				Average:      avg,
+				ThresholdPct: thresholdPct,
+			}
+			anomalies = append(anomalies, a)
+			s.logAnomaly(a)
+		}
+	}
+
+	return anomalies, nil
+}
+
+// archivedForDay returns the count of newly archived mails per source bucket for
+// the given day window. Mails with NULL source_type (legacy) bucket as 'import'.
+func (s *Store) archivedForDay(ctx context.Context, start, end time.Time) (map[bucketKey]int64, error) {
+	rows, err := s.pool.Query(ctx, `
+		SELECT tenant_id, COALESCE(source_type, 'import') AS st, source_id, COUNT(*)
+		FROM emails
+		WHERE received_at >= $1 AND received_at < $2
+		GROUP BY tenant_id, st, source_id
+	`, start, end)
+	if err != nil {
+		return nil, fmt.Errorf("reconciliation: archived-for-day query: %w", err)
+	}
+	defer rows.Close()
+
+	out := make(map[bucketKey]int64)
+	for rows.Next() {
+		var tenant, sourceID *int64
+		var st string
+		var cnt int64
+		if err := rows.Scan(&tenant, &st, &sourceID, &cnt); err != nil {
+			return nil, fmt.Errorf("reconciliation: archived-for-day scan: %w", err)
+		}
+		out[keyOf(tenant, sourceID, st)] = cnt
+	}
+	return out, rows.Err()
+}
+
+// knownBuckets returns every (tenant, source_type, source_id) combination that
+// has ever produced an archived mail. These are the buckets for which a report
+// row is written every day — including explicit 0 on inactive days so gaps in
+// the cron run are distinguishable from genuine zero-activity days.
+func (s *Store) knownBuckets(ctx context.Context) (map[bucketKey]struct{}, error) {
+	rows, err := s.pool.Query(ctx, `
+		SELECT DISTINCT tenant_id, COALESCE(source_type, 'import') AS st, source_id
+		FROM emails
+	`)
+	if err != nil {
+		return nil, fmt.Errorf("reconciliation: known-buckets query: %w", err)
+	}
+	defer rows.Close()
+
+	out := make(map[bucketKey]struct{})
+	for rows.Next() {
+		var tenant, sourceID *int64
+		var st string
+		if err := rows.Scan(&tenant, &st, &sourceID); err != nil {
+			return nil, fmt.Errorf("reconciliation: known-buckets scan: %w", err)
+		}
+		out[keyOf(tenant, sourceID, st)] = struct{}{}
+	}
+	return out, rows.Err()
+}
+
+// imapExpectedSnapshot returns the IMAP soll/ist snapshot keyed by account ID.
+// expected reuses the per-folder UID high-water marks from imap_folder_state
+// (PROJ-45) — no additional IMAP login. cumulArch is the cumulative number of
+// archived mails attributed to the account.
+func (s *Store) imapExpectedSnapshot(ctx context.Context) (map[int64]imapExpected, error) {
+	out := make(map[int64]imapExpected)
+
+	// Expected proxy + tenant per account. LEFT JOIN so accounts without any
+	// synced folder yet still appear (expected 0).
+	rows, err := s.pool.Query(ctx, `
+		SELECT a.id, a.tenant_id, COALESCE(SUM(fs.last_uid), 0)
+		FROM imap_accounts a
+		LEFT JOIN imap_folder_state fs ON fs.account_id = a.id
+		GROUP BY a.id, a.tenant_id
+	`)
+	if err != nil {
+		return nil, fmt.Errorf("reconciliation: imap expected query: %w", err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var id int64
+		var tenant *int64
+		var expected int64
+		if err := rows.Scan(&id, &tenant, &expected); err != nil {
+			return nil, fmt.Errorf("reconciliation: imap expected scan: %w", err)
+		}
+		out[id] = imapExpected{tenant: tenant, expected: expected}
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	// Cumulative archived count per IMAP account.
+	crows, err := s.pool.Query(ctx, `
+		SELECT source_id, COUNT(*)
+		FROM emails
+		WHERE source_type = 'imap' AND source_id IS NOT NULL
+		GROUP BY source_id
+	`)
+	if err != nil {
+		return nil, fmt.Errorf("reconciliation: imap archived query: %w", err)
+	}
+	defer crows.Close()
+	for crows.Next() {
+		var id, cnt int64
+		if err := crows.Scan(&id, &cnt); err != nil {
+			return nil, fmt.Errorf("reconciliation: imap archived scan: %w", err)
+		}
+		ie := out[id]
+		ie.cumulArch = cnt
+		out[id] = ie
+	}
+	return out, crows.Err()
+}
+
+// upsertRows writes all report rows in a single transaction. On any error the
+// transaction is rolled back so no partial day is persisted.
+func (s *Store) upsertRows(ctx context.Context, rows []Report) error {
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("reconciliation: begin tx: %w", err)
+	}
+	defer tx.Rollback(ctx)
+
+	for _, r := range rows {
+		_, err := tx.Exec(ctx, `
+			INSERT INTO reconciliation_reports
+				(date, tenant_id, source_type, source_id, expected_count, archived_count, delta)
+			VALUES ($1, $2, $3, $4, $5, $6, $7)
+			ON CONFLICT (date, COALESCE(tenant_id, -1), source_type, COALESCE(source_id, -1))
+			DO UPDATE SET
+				expected_count = EXCLUDED.expected_count,
+				archived_count = EXCLUDED.archived_count,
+				delta          = EXCLUDED.delta,
+				created_at     = NOW()
+		`, r.Date, r.TenantID, r.SourceType, r.SourceID, r.ExpectedCount, r.ArchivedCount, r.Delta)
+		if err != nil {
+			return fmt.Errorf("reconciliation: upsert row: %w", err)
+		}
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("reconciliation: commit: %w", err)
+	}
+	return nil
+}
+
+// trailingAverage returns the mean archived_count of the up-to-7 report rows
+// immediately preceding the given day for the same source bucket, plus how many
+// prior day rows were found. NULL-safe matching on tenant_id / source_id.
+func (s *Store) trailingAverage(ctx context.Context, r Report, day time.Time) (float64, int, error) {
+	rows, err := s.pool.Query(ctx, `
+		SELECT archived_count
+		FROM reconciliation_reports
+		WHERE source_type = $1
+		  AND tenant_id IS NOT DISTINCT FROM $2
+		  AND source_id IS NOT DISTINCT FROM $3
+		  AND date < $4
+		ORDER BY date DESC
+		LIMIT 7
+	`, r.SourceType, r.TenantID, r.SourceID, day)
+	if err != nil {
+		return 0, 0, fmt.Errorf("reconciliation: trailing average query: %w", err)
+	}
+	defer rows.Close()
+
+	var sum int64
+	var n int
+	for rows.Next() {
+		var c int64
+		if err := rows.Scan(&c); err != nil {
+			return 0, 0, err
+		}
+		sum += c
+		n++
+	}
+	if err := rows.Err(); err != nil {
+		return 0, 0, err
+	}
+	if n == 0 {
+		return 0, 0, nil
+	}
+	return float64(sum) / float64(n), n, nil
+}
+
+// logAnomaly emits a structured log line and, when wired, a tenant-visible
+// audit entry for a detected drop.
+func (s *Store) logAnomaly(a Anomaly) {
+	detail := fmt.Sprintf("source=%s date=%s archived=%d avg_7d=%.1f threshold=%d%%",
+		a.SourceKey, a.Date.Format("2006-01-02"), a.Archived, a.Average, a.ThresholdPct)
+	s.logger.Warn("reconciliation: anomaly detected", "detail", detail)
+	if s.audlog != nil {
+		s.audlog.Log(audit.Entry{
+			EventType: audit.EventReconciliationAnomaly,
+			Username:  "system",
+			TenantID:  a.TenantID,
+			Success:   false,
+			Detail:    detail,
+		})
+	}
+}

internal/reconciliation/query.go

@@ -0,0 +1,177 @@
+package reconciliation
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+// DayPoint is one day's figures for a source in the dashboard response.
+// ArchivedCount is nil (and Missing true) when no report row exists for that
+// date — i.e. the cron job did not run — which is distinct from an archived
+// count of 0 on a genuine zero-activity day.
+type DayPoint struct {
+	Date          string `json:"date"`
+	ArchivedCount *int64 `json:"archived_count"`
+	ExpectedCount *int64 `json:"expected_count"`
+	Delta         *int64 `json:"delta"`
+	Missing       bool   `json:"missing"`
+}
+
+// SourceSummary aggregates the trailing days plus alert state for one source.
+type SourceSummary struct {
+	SourceType string     `json:"source_type"`
+	SourceID   *int64     `json:"source_id"`
+	SourceKey  string     `json:"source_key"`
+	TenantID   *int64     `json:"tenant_id"`
+	Points     []DayPoint `json:"points"`
+	Avg7d      float64    `json:"avg_7d"`
+	EnoughData bool       `json:"enough_data"`
+	Alert      bool       `json:"alert"`
+}
+
+// DashboardData returns the last `days` calendar days of reconciliation figures
+// per source, tenant-scoped. When tenantID is nil (superadmin) all tenants are
+// included; otherwise only rows for that tenant are returned. thresholdPct is
+// used to compute the per-source Alert flag consistently with the cron job.
+func (s *Store) DashboardData(ctx context.Context, tenantID *int64, days, thresholdPct int) ([]SourceSummary, error) {
+	if days <= 0 {
+		days = 7
+	}
+	today := time.Now().UTC().Truncate(24 * time.Hour)
+	start := today.AddDate(0, 0, -(days - 1))
+
+	rows, err := s.queryRange(ctx, tenantID, start, today.Add(24*time.Hour))
+	if err != nil {
+		return nil, err
+	}
+
+	// Ordered date labels for the window.
+	dateLabels := make([]string, days)
+	for i := 0; i < days; i++ {
+		dateLabels[i] = start.AddDate(0, 0, i).Format("2006-01-02")
+	}
+
+	type srcAgg struct {
+		meta   Report
+		byDate map[string]Report
+	}
+	agg := map[string]*srcAgg{}
+	for _, r := range rows {
+		key := SourceKey(r.SourceType, r.SourceID)
+		// Distinguish sources of different tenants sharing a key.
+		if r.TenantID != nil {
+			key = fmt.Sprintf("t%d/%s", *r.TenantID, key)
+		}
+		a, ok := agg[key]
+		if !ok {
+			a = &srcAgg{meta: r, byDate: map[string]Report{}}
+			agg[key] = a
+		}
+		a.byDate[r.Date.UTC().Format("2006-01-02")] = r
+	}
+
+	summaries := make([]SourceSummary, 0, len(agg))
+	for _, a := range agg {
+		sum := SourceSummary{
+			SourceType: a.meta.SourceType,
+			SourceID:   a.meta.SourceID,
+			SourceKey:  SourceKey(a.meta.SourceType, a.meta.SourceID),
+			TenantID:   a.meta.TenantID,
+			Points:     make([]DayPoint, 0, days),
+		}
+		for _, d := range dateLabels {
+			if r, ok := a.byDate[d]; ok {
+				c := r.ArchivedCount
+				sum.Points = append(sum.Points, DayPoint{
+					Date:          d,
+					ArchivedCount: &c,
+					ExpectedCount: r.ExpectedCount,
+					Delta:         r.Delta,
+					Missing:       false,
+				})
+			} else {
+				sum.Points = append(sum.Points, DayPoint{Date: d, Missing: true})
+			}
+		}
+
+		// Alert against the trailing 7-day average of the most recent day that
+		// actually has a report row (mirrors the cron job's evaluation).
+		latest, latestDate, hasLatest := latestPresent(a.byDate, dateLabels)
+		if hasLatest {
+			avg, n, err := s.trailingAverage(ctx, a.meta, latestDate)
+			if err == nil && n >= 7 {
+				sum.Avg7d = avg
+				sum.EnoughData = true
+				limit := avg * (1 - float64(thresholdPct)/100.0)
+				if float64(latest.ArchivedCount) < limit {
+					sum.Alert = true
+				}
+			}
+		}
+		summaries = append(summaries, sum)
+	}
+	return summaries, nil
+}
+
+// latestPresent returns the most recent report row within the window that has a
+// stored row, along with its date.
+func latestPresent(byDate map[string]Report, dateLabels []string) (Report, time.Time, bool) {
+	for i := len(dateLabels) - 1; i >= 0; i-- {
+		if r, ok := byDate[dateLabels[i]]; ok {
+			d, _ := time.Parse("2006-01-02", dateLabels[i])
+			return r, d, true
+		}
+	}
+	return Report{}, time.Time{}, false
+}
+
+// ExportRows returns raw report rows for CSV export, tenant-scoped, for the
+// last `days` days, ordered by date descending then source.
+func (s *Store) ExportRows(ctx context.Context, tenantID *int64, days int) ([]Report, error) {
+	if days <= 0 {
+		days = 30
+	}
+	today := time.Now().UTC().Truncate(24 * time.Hour)
+	start := today.AddDate(0, 0, -(days - 1))
+	return s.queryRange(ctx, tenantID, start, today.Add(24*time.Hour))
+}
+
+// queryRange loads report rows in [start, end) filtered by tenant. tenantID nil
+// returns all tenants (superadmin scope).
+func (s *Store) queryRange(ctx context.Context, tenantID *int64, start, end time.Time) ([]Report, error) {
+	var (
+		sql  string
+		args []interface{}
+	)
+	if tenantID == nil {
+		sql = `SELECT date, tenant_id, source_type, source_id, expected_count, archived_count, delta
+			FROM reconciliation_reports
+			WHERE date >= $1 AND date < $2
+			ORDER BY date DESC, source_type, source_id`
+		args = []interface{}{start, end}
+	} else {
+		sql = `SELECT date, tenant_id, source_type, source_id, expected_count, archived_count, delta
+			FROM reconciliation_reports
+			WHERE date >= $1 AND date < $2 AND tenant_id = $3
+			ORDER BY date DESC, source_type, source_id`
+		args = []interface{}{start, end, *tenantID}
+	}
+
+	rows, err := s.pool.Query(ctx, sql, args...)
+	if err != nil {
+		return nil, fmt.Errorf("reconciliation: query range: %w", err)
+	}
+	defer rows.Close()
+
+	var out []Report
+	for rows.Next() {
+		var r Report
+		if err := rows.Scan(&r.Date, &r.TenantID, &r.SourceType, &r.SourceID,
+			&r.ExpectedCount, &r.ArchivedCount, &r.Delta); err != nil {
+			return nil, fmt.Errorf("reconciliation: scan range: %w", err)
+		}
+		out = append(out, r)
+	}
+	return out, rows.Err()
+}

internal/reconciliation/reconciliation.go

@@ -0,0 +1,120 @@
+// Package reconciliation implements the daily completeness reconciliation
+// report (PROJ-52). It counts newly archived mails per source (SMTP journal,
+// IMAP account, POP3 account, bulk import) and per day, persists the counts in
+// the reconciliation_reports table, and flags significant drops against the
+// trailing 7-day average via the audit log.
+//
+// The reconciliation is deliberately a read-only observer of the emails table
+// plus the IMAP UID-tracking state (imap_folder_state, PROJ-45). It never
+// mutates archived mail content and issues no additional IMAP logins — the
+// IMAP soll/ist comparison reuses the UID high-water marks already persisted by
+// the sync scheduler.
+package reconciliation
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"archivmail/internal/audit"
+)
+
+// Store owns the reconciliation_reports table and the reconciliation logic.
+// It uses its own connection pool so the CLI cron command and the daemon can
+// both operate it independently.
+type Store struct {
+	pool   *pgxpool.Pool
+	logger *slog.Logger
+	audlog *audit.Logger // optional; when nil, anomalies are only logged
+}
+
+// Report is a single persisted reconciliation row for one day and one source.
+type Report struct {
+	Date          time.Time `json:"date"`
+	TenantID      *int64    `json:"tenant_id"`
+	SourceType    string    `json:"source_type"`
+	SourceID      *int64    `json:"source_id"`
+	ExpectedCount *int64    `json:"expected_count"`
+	ArchivedCount int64     `json:"archived_count"`
+	Delta         *int64    `json:"delta"`
+}
+
+// New connects to PostgreSQL and initialises the reconciliation schema.
+func New(dsn string, logger *slog.Logger) (*Store, error) {
+	pool, err := pgxpool.New(context.Background(), dsn)
+	if err != nil {
+		return nil, fmt.Errorf("reconciliation: connect: %w", err)
+	}
+	s := &Store{pool: pool, logger: logger}
+	if err := s.initSchema(context.Background()); err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("reconciliation: init schema: %w", err)
+	}
+	return s, nil
+}
+
+// SetAuditLogger wires an audit.Logger so anomalies are persisted as
+// tenant-visible `reconciliation_anomaly` audit entries. Optional.
+func (s *Store) SetAuditLogger(a *audit.Logger) { s.audlog = a }
+
+// Close releases the connection pool.
+func (s *Store) Close() {
+	if s.pool != nil {
+		s.pool.Close()
+	}
+}
+
+// initSchema creates the reconciliation_reports table and its indexes.
+// Idempotent and safe on existing databases (CREATE ... IF NOT EXISTS).
+func (s *Store) initSchema(ctx context.Context) error {
+	if _, err := s.pool.Exec(ctx, `
+		CREATE TABLE IF NOT EXISTS reconciliation_reports (
+			id             BIGSERIAL   PRIMARY KEY,
+			date           DATE        NOT NULL,
+			tenant_id      BIGINT,
+			source_type    TEXT        NOT NULL,
+			source_id      BIGINT,
+			expected_count BIGINT,
+			archived_count BIGINT      NOT NULL DEFAULT 0,
+			delta          BIGINT,
+			created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
+		);
+	`); err != nil {
+		return err
+	}
+
+	// Upsert key. tenant_id and source_id are nullable and PostgreSQL treats
+	// NULLs as distinct in a plain UNIQUE index, which would allow duplicate
+	// rows for the (tenant-less / smtp) buckets. A COALESCE-based expression
+	// index gives a single deterministic key per (date, tenant, source_type,
+	// source_id). -1 is a safe sentinel because real tenant/account IDs are
+	// positive BIGSERIALs.
+	if _, err := s.pool.Exec(ctx, `
+		CREATE UNIQUE INDEX IF NOT EXISTS idx_recon_reports_key
+			ON reconciliation_reports (date, COALESCE(tenant_id, -1), source_type, COALESCE(source_id, -1));
+	`); err != nil {
+		return err
+	}
+
+	// Lookup index for the dashboard / CSV queries (per AC).
+	if _, err := s.pool.Exec(ctx, `
+		CREATE INDEX IF NOT EXISTS idx_recon_reports_lookup
+			ON reconciliation_reports (tenant_id, date, source_type);
+	`); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// SourceKey returns the canonical source identifier used in the API / CSV:
+// "smtp", "import", "imap:<account_id>", "pop3:<account_id>".
+func SourceKey(sourceType string, sourceID *int64) string {
+	if sourceID != nil && (sourceType == "imap" || sourceType == "pop3") {
+		return fmt.Sprintf("%s:%d", sourceType, *sourceID)
+	}
+	return sourceType
+}

internal/smtpd/smtpd.go

@@ -346,6 +346,12 @@ func (s *session) Data(r io.Reader) error {
 		}
 	}
 
+	// PROJ-52: record the ingestion source for the reconciliation report.
+	// Non-fatal — a failed metadata write must not reject an already-stored mail.
+	if err := s.daemon.store.TagSource(context.Background(), id, "smtp", nil); err != nil {
+		s.daemon.logger.Warn("SMTP: tag source failed", "id", id, "err", err)
+	}
+
 	s.daemon.stats.Received.Add(1)
 	s.daemon.stats.LastMailAt.Store(time.Now())
 	s.daemon.logger.Info("SMTP: mail stored", "id", id, "from", s.from,

internal/storage/storage.go

@@ -359,7 +359,46 @@ func (s *Store) initSchema(ctx context.Context) error {
 		ALTER TABLE emails ADD COLUMN IF NOT EXISTS marked_for_deletion_at TIMESTAMPTZ;
 		CREATE INDEX IF NOT EXISTS idx_emails_marked_for_deletion ON emails (marked_for_deletion) WHERE marked_for_deletion = TRUE;
 	`)
+	if err != nil {
 		return err
+	}
+
+	// PROJ-52: ingestion source tracking for the completeness reconciliation
+	// report. source_type is one of 'smtp', 'imap', 'pop3', 'import'; source_id
+	// holds the IMAP/POP3 account ID (NULL for smtp/import). Both are NULL for
+	// legacy mails archived before this migration — the reconciliation job
+	// buckets those as 'import'. The composite index accelerates the per-day,
+	// per-source GROUP BY the reconciliation job runs.
+	_, err = s.db.Exec(ctx, `
+		ALTER TABLE emails ADD COLUMN IF NOT EXISTS source_type TEXT;
+		ALTER TABLE emails ADD COLUMN IF NOT EXISTS source_id BIGINT;
+		CREATE INDEX IF NOT EXISTS idx_emails_source_recon ON emails (received_at, source_type, source_id, tenant_id);
+	`)
+	return err
+}
+
+// TagSource records the ingestion channel of an archived mail (PROJ-52).
+// sourceType is one of 'smtp', 'imap', 'pop3', 'import'; sourceID holds the
+// IMAP/POP3 account ID (nil for smtp/import).
+//
+// First-write-wins: the update only sets the columns while source_type IS NULL.
+// A mail deduplicated across channels (SHA-256 / Message-ID dedup in Save) keeps
+// the source of its first ingestion, so the reconciliation counts never
+// double-count a re-delivered mail. Errors are non-fatal for the intake path —
+// callers log and continue so a reconciliation-metadata write never blocks
+// archival (GoBD completeness of the mail itself takes precedence).
+func (s *Store) TagSource(ctx context.Context, id, sourceType string, sourceID *int64) error {
+	if s.db == nil {
+		return nil
+	}
+	_, err := s.db.Exec(ctx, `
+		UPDATE emails SET source_type = $2, source_id = $3
+		WHERE id = $1 AND source_type IS NULL
+	`, id, sourceType, sourceID)
+	if err != nil {
+		return fmt.Errorf("storage: tag source: %w", err)
+	}
+	return nil
 }
 
 // ── Core operations ───────────────────────────────────────────────────────

src/components/admin/tabs/DashboardTab.tsx

@@ -15,6 +15,7 @@ import { Badge } from "@/components/ui/badge";
 import { Skeleton } from "@/components/ui/skeleton";
 import { Separator } from "@/components/ui/separator";
 import { Alert, AlertDescription } from "@/components/ui/alert";
+import { ReconciliationCard } from "@/components/admin/tabs/ReconciliationCard";
 
 function formatBytes(bytes: number): string {
   if (bytes < 1024) return `${bytes} B`;
@@ -491,6 +492,9 @@ export function DashboardTab({
             </Card>
           )}
 
+          {/* Vollständigkeits-Check (PROJ-52) — tenant-gescoped, domain_admin+ */}
+          <ReconciliationCard />
+
           {/* Benutzerübersicht */}
           <Card>
             <CardContent className="pt-6 space-y-2">

src/components/admin/tabs/ReconciliationCard.tsx

@@ -0,0 +1,271 @@
+"use client";
+
+import { useCallback, useEffect, useState } from "react";
+import {
+  exportReconciliationCSV,
+  getReconciliation,
+  type ReconciliationResponse,
+  type ReconciliationSource,
+} from "@/lib/api";
+import { Button } from "@/components/ui/button";
+import { Card, CardContent } from "@/components/ui/card";
+import { Badge } from "@/components/ui/badge";
+import { Skeleton } from "@/components/ui/skeleton";
+import { Separator } from "@/components/ui/separator";
+import { Alert, AlertDescription } from "@/components/ui/alert";
+import {
+  Table,
+  TableBody,
+  TableCell,
+  TableHead,
+  TableHeader,
+  TableRow,
+} from "@/components/ui/table";
+
+/** Formatiert einen source_key zu einem lesbaren Label. */
+function formatSourceLabel(source: ReconciliationSource): string {
+  switch (source.source_type) {
+    case "smtp":
+      return "SMTP-Journal";
+    case "import":
+      return "Datei-Import";
+    case "imap":
+      return `IMAP-Konto #${source.source_id ?? "?"}`;
+    case "pop3":
+      return `POP3-Konto #${source.source_id ?? "?"}`;
+    default:
+      return source.source_key;
+  }
+}
+
+function formatDayLabel(iso: string): string {
+  const d = new Date(iso);
+  return d.toLocaleDateString("de-DE", { day: "2-digit", month: "2-digit" });
+}
+
+/** Zellinhalt für einen Tagespunkt: "—" bei fehlendem Datensatz, sonst Werte. */
+function DayCell({
+  archived,
+  expected,
+  delta,
+  missing,
+  isImap,
+}: {
+  archived: number | null;
+  expected: number | null;
+  delta: number | null;
+  missing: boolean;
+  isImap: boolean;
+}) {
+  if (missing) {
+    return (
+      <span
+        className="text-muted-foreground"
+        title="Kein Report-Datensatz für diesen Tag (Cron nicht gelaufen)"
+      >
+        —
+      </span>
+    );
+  }
+  return (
+    <span className="inline-flex flex-col items-end leading-tight">
+      <span className="font-medium tabular-nums">
+        {(archived ?? 0).toLocaleString("de-DE")}
+      </span>
+      {isImap && expected != null && (
+        <span className="text-[10px] text-muted-foreground tabular-nums">
+          Soll {expected.toLocaleString("de-DE")}
+          {delta != null && (
+            <span
+              className={
+                delta < 0 ? "ml-1 text-destructive" : "ml-1 text-green-600"
+              }
+            >
+              ({delta > 0 ? "+" : ""}
+              {delta.toLocaleString("de-DE")})
+            </span>
+          )}
+        </span>
+      )}
+    </span>
+  );
+}
+
+export function ReconciliationCard() {
+  const [data, setData] = useState<ReconciliationResponse | null>(null);
+  const [loading, setLoading] = useState(true);
+  const [error, setError] = useState<string | null>(null);
+  const [exporting, setExporting] = useState(false);
+
+  const load = useCallback(async () => {
+    setLoading(true);
+    setError(null);
+    try {
+      const res = await getReconciliation(7);
+      setData(res);
+    } catch (e: unknown) {
+      setError(e instanceof Error ? e.message : "Laden fehlgeschlagen");
+    } finally {
+      setLoading(false);
+    }
+  }, []);
+
+  useEffect(() => {
+    void load();
+  }, [load]);
+
+  const handleExport = async () => {
+    setExporting(true);
+    try {
+      const { blob, filename } = await exportReconciliationCSV(30);
+      const url = URL.createObjectURL(blob);
+      const a = document.createElement("a");
+      a.href = url;
+      a.download = filename;
+      document.body.appendChild(a);
+      a.click();
+      a.remove();
+      URL.revokeObjectURL(url);
+    } catch (e: unknown) {
+      setError(e instanceof Error ? e.message : "Export fehlgeschlagen");
+    } finally {
+      setExporting(false);
+    }
+  };
+
+  // Tages-Header aus dem ersten Quell-Eintrag ableiten (alle Quellen haben
+  // dieselben Tage in gleicher Reihenfolge).
+  const dayHeaders = data?.sources[0]?.points.map((p) => p.date) ?? [];
+
+  return (
+    <Card>
+      <CardContent className="pt-6 space-y-3">
+        <div className="flex items-center justify-between gap-3">
+          <div className="flex items-center gap-2">
+            <span className="text-sm font-medium text-muted-foreground">
+              Vollständigkeits-Check
+            </span>
+            {data && (
+              <span className="text-xs text-muted-foreground">
+                letzte {data.days} Tage · Schwelle {data.threshold_pct}%
+              </span>
+            )}
+          </div>
+          <div className="flex items-center gap-2">
+            <Button
+              variant="outline"
+              size="sm"
+              onClick={load}
+              disabled={loading}
+            >
+              {loading ? "..." : "Aktualisieren"}
+            </Button>
+            <Button
+              variant="outline"
+              size="sm"
+              onClick={handleExport}
+              disabled={exporting}
+            >
+              {exporting ? "Export..." : "CSV-Export (30 Tage)"}
+            </Button>
+          </div>
+        </div>
+        <Separator />
+
+        {loading ? (
+          <div className="space-y-2">
+            {Array.from({ length: 3 }).map((_, i) => (
+              <Skeleton key={i} className="h-8 w-full" />
+            ))}
+          </div>
+        ) : error ? (
+          <Alert variant="destructive">
+            <AlertDescription>
+              Vollständigkeits-Report konnte nicht geladen werden: {error}
+            </AlertDescription>
+          </Alert>
+        ) : !data || data.sources.length === 0 ? (
+          <p className="text-sm text-muted-foreground">
+            Noch keine Reconciliation-Daten vorhanden. Der tägliche Zähl-Job
+            (<code className="font-mono">archivmail reconcile</code>) hat noch
+            keine Datensätze erzeugt.
+          </p>
+        ) : (
+          <div className="overflow-x-auto">
+            <Table>
+              <TableHeader>
+                <TableRow>
+                  <TableHead className="whitespace-nowrap">Quelle</TableHead>
+                  {dayHeaders.map((d) => (
+                    <TableHead
+                      key={d}
+                      className="text-right whitespace-nowrap"
+                    >
+                      {formatDayLabel(d)}
+                    </TableHead>
+                  ))}
+                  <TableHead className="text-right whitespace-nowrap">
+                    Ø 7 Tage
+                  </TableHead>
+                  <TableHead className="text-right whitespace-nowrap">
+                    Status
+                  </TableHead>
+                </TableRow>
+              </TableHeader>
+              <TableBody>
+                {data.sources.map((s) => {
+                  const isImap = s.source_type === "imap";
+                  return (
+                    <TableRow key={s.source_key}>
+                      <TableCell className="font-medium whitespace-nowrap">
+                        {formatSourceLabel(s)}
+                      </TableCell>
+                      {!s.enough_data ? (
+                        <TableCell
+                          colSpan={dayHeaders.length + 1}
+                          className="text-center text-sm text-muted-foreground"
+                        >
+                          Noch nicht genug Daten
+                        </TableCell>
+                      ) : (
+                        <>
+                          {s.points.map((p) => (
+                            <TableCell key={p.date} className="text-right">
+                              <DayCell
+                                archived={p.archived_count}
+                                expected={p.expected_count}
+                                delta={p.delta}
+                                missing={p.missing}
+                                isImap={isImap}
+                              />
+                            </TableCell>
+                          ))}
+                          <TableCell className="text-right tabular-nums text-muted-foreground">
+                            {s.avg_7d != null
+                              ? s.avg_7d.toLocaleString("de-DE", {
+                                  maximumFractionDigits: 1,
+                                })
+                              : "—"}
+                          </TableCell>
+                        </>
+                      )}
+                      <TableCell className="text-right">
+                        {s.alert ? (
+                          <Badge variant="destructive">Auffällig</Badge>
+                        ) : s.enough_data ? (
+                          <Badge variant="secondary">OK</Badge>
+                        ) : (
+                          <Badge variant="outline">—</Badge>
+                        )}
+                      </TableCell>
+                    </TableRow>
+                  );
+                })}
+              </TableBody>
+            </Table>
+          </div>
+        )}
+      </CardContent>
+    </Card>
+  );
+}

src/lib/api/index.ts

@@ -178,6 +178,16 @@ export {
   deleteArchivingRule,
 } from "./archiving_rules";
 
+export type {
+  ReconciliationPoint,
+  ReconciliationSource,
+  ReconciliationResponse,
+} from "./reconciliation";
+export {
+  getReconciliation,
+  exportReconciliationCSV,
+} from "./reconciliation";
+
 export type { SavedSearch } from "./saved_searches";
 export {
   listSavedSearches,

src/lib/api/reconciliation.ts

@@ -0,0 +1,51 @@
+import { API_BASE, request } from "./core";
+
+// ── Types ────────────────────────────────────────────────────────────────────
+
+export interface ReconciliationPoint {
+  date: string; // "2026-06-27"
+  archived_count: number | null;
+  expected_count: number | null;
+  delta: number | null;
+  missing: boolean; // true = kein Report-Datensatz (Cron nicht gelaufen) ≠ archived_count:0
+}
+
+export interface ReconciliationSource {
+  source_type: string; // "smtp" | "imap" | "pop3" | "import"
+  source_id: number | null;
+  source_key: string; // "smtp" | "import" | "imap:<id>" | "pop3:<id>"
+  tenant_id: number | null;
+  points: ReconciliationPoint[];
+  avg_7d: number | null;
+  enough_data: boolean;
+  alert: boolean;
+}
+
+export interface ReconciliationResponse {
+  days: number;
+  threshold_pct: number;
+  sources: ReconciliationSource[];
+}
+
+// ── API ──────────────────────────────────────────────────────────────────────
+
+export async function getReconciliation(days = 7): Promise<ReconciliationResponse> {
+  return request<ReconciliationResponse>(`/api/admin/reconciliation?days=${days}`);
+}
+
+/** Lädt den Reconciliation-Report als CSV-Datei herunter. */
+export async function exportReconciliationCSV(
+  days = 30
+): Promise<{ blob: Blob; filename: string }> {
+  const res = await fetch(
+    `${API_BASE}/api/admin/reconciliation/export.csv?days=${days}`,
+    { credentials: "include" }
+  );
+  if (!res.ok) throw new Error(`Export fehlgeschlagen: ${res.status}`);
+  const disposition = res.headers.get("Content-Disposition") || "";
+  const match = disposition.match(/filename[^;=\n]*=((['"]).*?\2|[^;\n]*)/);
+  const filename = match
+    ? match[1].replace(/['"]/g, "")
+    : `reconciliation-${days}d.csv`;
+  return { blob: await res.blob(), filename };
+}