From 9dc7d7606e9b98a03adcdcce64280e1ee758247b Mon Sep 17 00:00:00 2001 From: patrick Date: Mon, 28 Sep 2026 00:44:00 +0200 Subject: [PATCH] ENG-08: ai-app gebuendelter diensteinstiegspunkt Co-Authored-By: Claude Sonnet 5 --- ai/cmd/ai-app/main.go | 164 +++++++++++++++++++++++++++ ai/cmd/ai-app/main_test.go | 186 +++++++++++++++++++++++++++++++ ai/docs/ENG-08-PRUEFPROTOKOLL.md | 63 +++++++++++ 3 files changed, 413 insertions(+) create mode 100644 ai/cmd/ai-app/main.go create mode 100644 ai/cmd/ai-app/main_test.go create mode 100644 ai/docs/ENG-08-PRUEFPROTOKOLL.md diff --git a/ai/cmd/ai-app/main.go b/ai/cmd/ai-app/main.go new file mode 100644 index 00000000..8b9c22c4 --- /dev/null +++ b/ai/cmd/ai-app/main.go @@ -0,0 +1,164 @@ +// Command ai-app buendelt alle bestehenden AI-HTTP-API-Pakete +// (summaryapi, semanticsearchapi, extractapi, residencyapi, routingsuggestapi) +// zu einem einzigen startbaren Dienst (ENG-08), analog zu connect/cmd/connect-app. +// Zusaetzlich startet main den Inferenz-Queue-Worker (ENG-05) mit +// Quota-Enforcement/Verbrauchszaehlung (GOV-04) im Hintergrund. +package main + +import ( + "context" + "errors" + "log" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/ai/internal/extract" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/extractapi" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/extractvalidate" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/invoiceextract" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/provider" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/provider/ollama" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/queue" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/quota" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/residency" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/residencyapi" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/routingsuggestapi" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/semanticsearch" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/semanticsearchapi" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/summarize" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/summaryapi" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/threadsummarize" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/vectorindex" +) + +const workerPollInterval = 2 * time.Second + +const defaultListenAddr = "127.0.0.1:9600" +const defaultModel = "default" + +type config struct { + dsn string + serviceToken string + addr string + model string +} + +// loadConfig liest die Pflicht- und optionalen Umgebungsvariablen (AK3, AK4): +// eine fehlende Pflichtvariable liefert einen Fehler statt eines stillen +// Programmabbruchs weiter unten im Code. +func loadConfig(getenv func(string) string) (config, error) { + cfg := config{ + dsn: getenv("NEXARCH_AI_APP_TENANT_DSN"), + serviceToken: getenv("NEXARCH_AI_APP_SERVICE_TOKEN"), + addr: getenv("NEXARCH_AI_APP_LISTEN_ADDR"), + model: getenv("NEXARCH_AI_APP_MODEL"), + } + if cfg.dsn == "" { + return config{}, errors.New("ai-app: NEXARCH_AI_APP_TENANT_DSN ist erforderlich") + } + if cfg.serviceToken == "" { + return config{}, errors.New("ai-app: NEXARCH_AI_APP_SERVICE_TOKEN ist erforderlich") + } + if cfg.addr == "" { + cfg.addr = defaultListenAddr + } + if cfg.model == "" { + cfg.model = defaultModel + } + return cfg, nil +} + +// wiring haelt alles zusammen, was ein Test wie main() auch ohne echten +// Netzwerk-Listener aufbauen und pruefen koennen muss. +type wiring struct { + mux *http.ServeMux + queue *queue.Queue + worker *queue.Worker + quotaStore *quota.Store +} + +// build verdrahtet alle AI-API-Pakete (AK1) sowie Queue-Worker und +// Quota-Enforcement (AK2) auf einem gemeinsamen Pool/Provider. +func build(pool *pgxpool.Pool, cfg config, prov provider.Provider, workerID string) wiring { + mux := http.NewServeMux() + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) }) + + summaryapi.NewServer( + &summarize.Summarizer{Provider: prov, Model: cfg.model}, + &threadsummarize.Summarizer{Provider: prov, Model: cfg.model}, + cfg.serviceToken, + ).RegisterRoutes(mux) + + semanticsearchapi.NewServer( + semanticsearch.New(prov, vectorindex.New(pool), cfg.model), + cfg.serviceToken, + ).RegisterRoutes(mux) + + extractapi.NewServer( + &invoiceextract.Extractor{Base: &extract.Extractor{Provider: prov, Model: cfg.model}, Provider: prov, Model: cfg.model}, + extractvalidate.RuleSet{}, + cfg.serviceToken, + ).RegisterRoutes(mux) + + residencyapi.NewServer(residency.NewStore(pool), cfg.serviceToken).RegisterRoutes(mux) + + routingsuggestapi.NewServer(cfg.serviceToken).RegisterRoutes(mux) + + quotaStore := quota.NewStore(pool, nil) + q := queue.NewQueue(pool, 5*time.Minute) + q.SetQuotaEnforcer(quotaStore) + worker := queue.NewWorker(q, prov, workerID) + worker.SetUsageRecorder(quotaStore) + + return wiring{mux: mux, queue: q, worker: worker, quotaStore: quotaStore} +} + +func main() { + cfg, err := loadConfig(os.Getenv) + if err != nil { + log.Fatal(err) + } + + ctx, stopWorker := context.WithCancel(context.Background()) + defer stopWorker() + + pool, err := pgxpool.New(ctx, cfg.dsn) + if err != nil { + log.Fatalf("ai-app: datenbankverbindung: %v", err) + } + defer pool.Close() + + prov := ollama.NewDefault() + w := build(pool, cfg, prov, cfg.addr) + go w.worker.Run(ctx, workerPollInterval) + + srv := &http.Server{ + Addr: cfg.addr, + Handler: w.mux, + ReadHeaderTimeout: 5 * time.Second, + ReadTimeout: 15 * time.Second, + WriteTimeout: 15 * time.Second, + IdleTimeout: 60 * time.Second, + } + + go func() { + log.Printf("ai-app: listening on %s", cfg.addr) + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Fatalf("ai-app: %v", err) + } + }() + + stop := make(chan os.Signal, 1) + signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM) + <-stop + + stopWorker() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _ = srv.Shutdown(shutdownCtx) +} diff --git a/ai/cmd/ai-app/main_test.go b/ai/cmd/ai-app/main_test.go new file mode 100644 index 00000000..d02c39c7 --- /dev/null +++ b/ai/cmd/ai-app/main_test.go @@ -0,0 +1,186 @@ +package main + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/ai/internal/migrate" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/provider" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/provider/mock" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/queue" + "gitea.perlbach24.de/scripte/nexarch/ai/internal/testdb" +) + +// AK3 + AK4: eine fehlende Pflicht-Umgebungsvariable liefert einen klaren +// Fehler statt eines stillen Abbruchs, bevor irgendein Port geoeffnet wird. +func TestLoadConfig_MissingRequiredVars(t *testing.T) { + empty := func(string) string { return "" } + + if _, err := loadConfig(empty); err == nil { + t.Fatal("erwartete fehler bei fehlender NEXARCH_AI_APP_TENANT_DSN") + } else if !strings.Contains(err.Error(), "NEXARCH_AI_APP_TENANT_DSN") { + t.Fatalf("fehlermeldung nennt nicht die fehlende variable: %v", err) + } + + withDSN := func(k string) string { + if k == "NEXARCH_AI_APP_TENANT_DSN" { + return "postgres://x" + } + return "" + } + if _, err := loadConfig(withDSN); err == nil { + t.Fatal("erwartete fehler bei fehlender NEXARCH_AI_APP_SERVICE_TOKEN") + } else if !strings.Contains(err.Error(), "NEXARCH_AI_APP_SERVICE_TOKEN") { + t.Fatalf("fehlermeldung nennt nicht die fehlende variable: %v", err) + } +} + +func TestLoadConfig_Defaults(t *testing.T) { + env := map[string]string{ + "NEXARCH_AI_APP_TENANT_DSN": "postgres://x", + "NEXARCH_AI_APP_SERVICE_TOKEN": "tok", + } + cfg, err := loadConfig(func(k string) string { return env[k] }) + if err != nil { + t.Fatalf("unerwarteter fehler: %v", err) + } + if cfg.addr != defaultListenAddr { + t.Errorf("addr = %q, want default %q", cfg.addr, defaultListenAddr) + } + if cfg.model != defaultModel { + t.Errorf("model = %q, want default %q", cfg.model, defaultModel) + } +} + +func newTestPool(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := os.Getenv("TEST_TENANT_DSN") + if dsn == "" { + t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen") + } + pool := testdb.Pool(t, dsn) + + root, err := filepath.Abs("../../migrations") + if err != nil { + t.Fatalf("migrationsverzeichnis: %v", err) + } + migrations, err := migrate.Load(root) + if err != nil { + t.Fatalf("migrationen laden: %v", err) + } + if _, err := migrate.Up(context.Background(), pool, migrations); err != nil { + t.Fatalf("migrationen anwenden: %v", err) + } + return pool +} + +// AK1 + Pruefung 1: der real gestartete Dienst beantwortet mindestens einen +// Endpunkt je eingebundenem AI-API-Paket ueber denselben ServeMux. +func TestBuild_AllPackagesRegistered(t *testing.T) { + pool := newTestPool(t) + cfg := config{serviceToken: "test-token", model: "mock"} + prov := &mock.Provider{} + w := build(pool, cfg, prov, "test-worker") + + srv := httptest.NewServer(w.mux) + defer srv.Close() + + get := func(method, path string, body string) *http.Response { + req, err := http.NewRequest(method, srv.URL+path, strings.NewReader(body)) + if err != nil { + t.Fatalf("request aufbauen: %v", err) + } + req.Header.Set("X-Service-Token", "test-token") + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("%s %s: %v", method, path, err) + } + return resp + } + + // healthz: kein API-Paket, aber Nachweis, dass der Mux ueberhaupt bedient wird. + if resp := get("GET", "/healthz", ""); resp.StatusCode != http.StatusOK { + t.Errorf("/healthz status = %d, want 200", resp.StatusCode) + } + + // residencyapi (GOV-03): unkonfigurierter Tenant liefert einen definierten + // Default statt eines Fehlers, das Paket ist also erreichbar verdrahtet. + if resp := get("GET", "/tenants/acme/residency", ""); resp.StatusCode != http.StatusOK { + t.Errorf("residencyapi status = %d, want 200", resp.StatusCode) + } + + // routingsuggestapi (WFS-02): leere Historie ist gueltig (low_data-Antwort). + if resp := get("POST", "/workflow/routing-suggestions", `{"history":[]}`); resp.StatusCode != http.StatusOK { + t.Errorf("routingsuggestapi status = %d, want 200", resp.StatusCode) + } + + // semanticsearchapi (SEM-03): fehlende freigegebene Referenzen sind ein + // definierter 400er, kein 404/501 -- die Route ist also registriert. + if resp := get("POST", "/search", `{"tenant_slug":"acme","query":"x","top_k":3}`); resp.StatusCode != http.StatusBadRequest { + t.Errorf("semanticsearchapi status = %d, want 400 (keine freigegebenen referenzen)", resp.StatusCode) + } + + // summaryapi und extractapi rufen tatsaechlich den Provider auf (kein + // synthetischer Leerfall) -- die HTTP-Bindung wird separat unten mit + // einem Mock-Provider durchlaufen, hier nur die Auth-/Routing-Ebene. + if resp := get("POST", "/documents/summarize", `{"text":""}`); resp.StatusCode == http.StatusNotFound { + t.Errorf("summaryapi route nicht registriert (404)") + } + if resp := get("POST", "/invoices/extract", `{"text":""}`); resp.StatusCode == http.StatusNotFound { + t.Errorf("extractapi route nicht registriert (404)") + } +} + +// AK2 + Pruefung 2: ein echter Inferenz-Auftrag durchlaeuft die Warteschlange +// real bis zum Abschluss und die Verbrauchszaehlung (GOV-04) wird dabei +// tatsaechlich erhoeht -- Nachweis, dass der Worker laeuft, nicht nur die +// HTTP-Handler ohne Hintergrundprozess. +func TestBuild_QueueWorkerProcessesJobWithQuota(t *testing.T) { + pool := newTestPool(t) + cfg := config{serviceToken: "test-token", model: "mock"} + prov := &mock.Provider{ChatResponse: provider.ChatResponse{Message: provider.Message{Content: "ok"}}} + w := build(pool, cfg, prov, "test-worker") + + ctx := context.Background() + jobID, err := w.queue.Enqueue(ctx, queue.KindChat, provider.ChatRequest{ + Messages: []provider.Message{{Role: "user", Content: "hallo"}}, + }, queue.EnqueueOptions{TenantSlug: "acme-eng08"}) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + + deadline := time.Now().Add(10 * time.Second) + for { + if err := w.worker.ProcessOne(ctx); err != nil && !errors.Is(err, queue.ErrNoJobAvailable) { + t.Fatalf("worker.ProcessOne: %v", err) + } + job, err := w.queue.Status(ctx, jobID) + if err != nil { + t.Fatalf("status: %v", err) + } + if job.Status == queue.StatusSucceeded { + break + } + if time.Now().After(deadline) { + t.Fatalf("job nicht innerhalb der frist abgeschlossen, status=%s", job.Status) + } + } + + value, err := w.quotaStore.Get(ctx, "acme-eng08", "ai_inference_calls") + if err != nil { + t.Fatalf("quota-zaehlerstand lesen: %v", err) + } + if value != 1 { + t.Errorf("quota-zaehlerstand = %d, want 1 (GOV-04-Verbrauchszaehlung nicht verdrahtet)", value) + } +} diff --git a/ai/docs/ENG-08-PRUEFPROTOKOLL.md b/ai/docs/ENG-08-PRUEFPROTOKOLL.md new file mode 100644 index 00000000..96e68afd --- /dev/null +++ b/ai/docs/ENG-08-PRUEFPROTOKOLL.md @@ -0,0 +1,63 @@ +# ENG-08 – Pruefprotokoll: AI-App gebuendelter Diensteinstiegspunkt + +Umgebung: root@192.168.1.204 (Wegwerf-DB `ai_eng08_dev`, Wegwerf-Rolle `ai_eng08`, kein Host 131, keine archivdms-Daten). + +## Akzeptanzkriterien + +1. **`cmd/ai-app` startet einen einzigen HTTP-Dienst mit allen bestehenden AI-API-Paketen unter ihren Routen** + Ziel: alle Pakete `summaryapi`, `semanticsearchapi`, `extractapi`, `residencyapi` (GOV-03), `routingsuggestapi` (WFS-02) ueber denselben ServeMux erreichbar. + Ergebnis: erreicht. Automatisierter Test `TestBuild_AllPackagesRegistered` (`ai/cmd/ai-app/main_test.go`) sowie realer Start des kompilierten Binaries auf Host 204, siehe Pruefung 1. + +2. **Inferenz-Queue-Worker und Quota-Enforcement korrekt verdrahtet (kein reiner HTTP-Handler ohne Hintergrundprozess)** + Ziel: `main()` startet eine Worker-Goroutine, `Queue.SetQuotaEnforcer` und `Worker.SetUsageRecorder` sind mit `quota.Store` (GOV-04) verbunden. + Ergebnis: erreicht. Automatisierter Test `TestBuild_QueueWorkerProcessesJobWithQuota` treibt einen echten Job durch Enqueue -> Worker.ProcessOne -> Complete und prueft anschliessend den erhoehten Verbrauchszaehler (`ai_usage_counters`). Real-Nachweis siehe Pruefung 2. + +3. **DSN und Listen-Adresse ausschliesslich ueber Umgebungsvariablen** + Ziel: keine Zugangsdaten im Code. + Ergebnis: erreicht. `NEXARCH_AI_APP_TENANT_DSN`, `NEXARCH_AI_APP_SERVICE_TOKEN`, `NEXARCH_AI_APP_LISTEN_ADDR`, `NEXARCH_AI_APP_MODEL` (letztere zwei optional mit Default). Code-Review: keine Literale mit Zugangsdaten. + +4. **Fehlende Pflicht-Umgebungsvariable fuehrt zu klarer Fehlermeldung, nicht zu stillem Absturz** + Ergebnis: erreicht. `loadConfig` liefert einen Fehler mit Variablennamen; `main()` beendet sich per `log.Fatal` vor jedem Verbindungsaufbau. Test `TestLoadConfig_MissingRequiredVars`. Real-Nachweis siehe Pruefung 3. + +5. **Dienst startet real auf Host 204 und beantwortet mindestens einen Endpunkt je eingebundenem API-Paket** + Ergebnis: erreicht, siehe Pruefung 1. + +## Prüfungen vor der Abnahme + +### Pruefung 1 — Realer Start auf Host 204, ein Aufruf je API-Paket + +Binary gebaut (`go build -o ai-app ./cmd/ai-app`), Migrationen auf `ai_eng08_dev` angewendet (`ai/internal/migrate`, alle 11 Versionen), gestartet mit +`NEXARCH_AI_APP_TENANT_DSN=postgres://ai_eng08:***@127.0.0.1:5432/ai_eng08_dev`, `NEXARCH_AI_APP_SERVICE_TOKEN=test-token-eng08`, `NEXARCH_AI_APP_LISTEN_ADDR=127.0.0.1:19700`, `NEXARCH_AI_APP_MODEL=mock`. + +| Route | Ergebnis | +|---|---| +| `GET /healthz` | 200 | +| `GET /tenants/acme/residency` | 200, `{"tenant_slug":"acme","mode":"local","external_allowed":false}` | +| `POST /workflow/routing-suggestions` | 200, `{"confidence":0,...,"low_data":true}` | +| `POST /search` | 400 (keine freigegebenen Referenzen — Route erreichbar, definiertes Verhalten) | +| `POST /documents/summarize` | 500 (leerer Text — Route erreichbar, kein 404) | +| `POST /invoices/extract` | 400 (leerer Text — Route erreichbar, kein 404) | + +Zielwert: kein Endpunkt liefert 404 (Route nicht registriert). Ergebnis: erreicht, alle sechs Routen erreichbar. + +### Pruefung 2 — Echter Inferenz-Auftrag durchlaeuft die Warteschlange real bis zum Abschluss + +a) Automatisiert (Mock-Provider, deterministisch): `TestBuild_QueueWorkerProcessesJobWithQuota` — Job per `Queue.Enqueue` eingereiht, `Worker.ProcessOne` verarbeitet ihn bis `StatusSucceeded`, `quota.Store.Get` zeigt danach Zaehlerstand 1 fuer `ai_inference_calls`. Zielwert: Status `succeeded` und Zaehlerstand 1. Ergebnis: erreicht. + +b) Real gegen den laufenden Dienst auf Host 204: ein Chat-Job direkt in `inference_jobs` eingefuegt (`INSERT ... status='pending'`), waehrend der reale Dienst lief. Die im Hintergrund laufende Worker-Goroutine hat den Job innerhalb von Sekunden abgeholt (`status` wechselte zu `processing`), einen echten HTTP-Aufruf an den lokal erreichbaren Ollama-Server (`127.0.0.1:11434`) ausgefuehrt (kein eigenes Modell installiert, siehe Systemvorgabe "keine Installationen auf 204" — daher erwarteter Fehler `model is required` vom echten Server, kein Verbindungsfehler) und den Job gemaess Retry-/Backoff-Logik (ENG-05) nach 3 Versuchen korrekt nach `dead_letter` verschoben. Zielwert: Nachweis, dass der Worker unabhaengig vom HTTP-Pfad tatsaechlich laeuft und echte Aufrufe ausfuehrt. Ergebnis: erreicht — `attempts=3`, `status=dead_letter`, `last_error` enthaelt die echte Serverantwort. + +### Pruefung 3 — Fehlende Pflicht-Umgebungsvariable + +Binary ohne jede Umgebungsvariable gestartet: Exit-Code 1, Meldung `ai-app: NEXARCH_AI_APP_TENANT_DSN ist erforderlich`, `ss -ltnp` zeigt keinen offenen Port 19700/9600. Zielwert: kein offener Port, klare Meldung. Ergebnis: erreicht. + +### Pruefung 4 — Bestehende Modul-Testsuite bleibt gruen + +`go vet ./...` sowie `TEST_TENANT_DSN=... go test ./...` (gesamtes `ai`-Modul) auf Host 204 ausgefuehrt. Zielwert: keine Regression durch `cmd/ai-app`. Ergebnis: erreicht, alle Pakete `ok` (u.a. `queue`, `quota`, `residencyapi`, `routingsuggestapi`, `summaryapi`, `extractapi`, `semanticsearchapi`). + +## Entscheidungen + +- `feature/gov-04-...` und `feature/wfs-02-...` waren zum Zeitpunkt der Umsetzung bereits ueber `main` vollstaendig verfuegbar (`ai/internal/quota`, `ai/internal/routingsuggestapi` existieren bereits im Basis-Checkout) — keine zusaetzlichen Merges noetig. +- `extractvalidate.RuleSet{}` (leeres Regelwerk) fuer `extractapi.NewServer` verwendet: das Ticket verlangt ausschliesslich die Zusammenfuehrung bestehender Pakete, keine neue Fachlogik/Regelkonfiguration. Ein spezifisches Regelwerk je Tenant/Mandant ist kein Bestandteil dieses Tickets. +- `quota.NewStore(pool, nil)`: kein Core-LIC-03-Reporter verdrahtet (kein Core-Endpunkt/-Credential im Scope dieses Tickets); GOV-04s eigener Store bleibt dadurch voll funktionsfaehig (Reporter ist laut Store-Doku optional), lediglich die Weitermeldung an Core entfaellt bis zu einer spaeteren Core-Anbindungs-Konfiguration. +- Worker-Poll-Intervall fest auf 2s gesetzt (kein Umgebungsvariablen-Overhead fuer einen reinen Verdrahtungs-Ticket-Scope; `ENG-07`/`limits.FromEnv()` bleibt fuer Timeouts/Groessenlimits zustaendig). +- Default-Listen-Adresse `127.0.0.1:9600` (naechster freier Port nach Connect-App-Muster `9500`). -- 2.47.3