feat(jobs,extraction): process requests in the worker with retries and visible errors

[Fluory]

Warum

Fixes #7 · Teil von Epic #2 · gestapelt auf #33 (Basis claude/feat-ai-service-6).

Arbeitsstand

• Ziel: Der Worker verarbeitet Aufträge: Dokumente an den KI-Dienst, Lauf + Felder + Belege speichern und die Anfrage in einer Transaktion auf REVIEW setzen; Fehler werden mit Backoff wiederholt und enden sichtbar in ERROR.
• Nicht-Ziele: Vercel-Trigger für drain() (Showcase-Epic), Export (feat(export): export approved requests exactly once to the ERP mock #9), Review-UI (feat(review): review fields beside their source, correct, approve or reject #8), Liste mit Fehlern (feat(requests): request list with errors, retries and reprocessing #26).
• Erledigt: Statusmaschine (test-first), KI-Client mit Timeout, Fehlerklassen und Vertragsinvarianten, Feld-Zusammenführung (test-first), Tabellen für Läufe/Segmente/Felder, Handler, drain(), Dead-Letter → ERROR, Reprocess, Worker-Schleife mit Zeitbudget-Prüfung, Compose-Profil ai; frischer Review eingearbeitet (Kommentar); pnpm verify lokal grün; Migrationen auf frischer DB getestet.
• Offen: menschliche Freigabe (Migrationen 0006–0008, pg-boss-Grants).
• Annahmen: siehe Impact Manifest.
• Nächster kleinster Schritt: Review durch den Orchestrator nach dem Merge von feat(ai-service): stateless extraction service with grounding verifier #33.

Was ist passiert (Klartext)

Hochgeladene Anfragen werden jetzt im Hintergrund verarbeitet. Der Worker holt sich den Auftrag, schickt jedes Dokument an den KI-Dienst und speichert die gefundenen Angaben mit ihrer Fundstelle. Erst wenn alles gespeichert ist, springt die Anfrage in einem Schritt auf „zur Prüfung". Kommt derselbe Auftrag doppelt an (das kann bei Warteschlangen passieren), entsteht trotzdem nur ein Ergebnis.

Ist der KI-Dienst kurz nicht erreichbar, versucht das System es später erneut, mit wachsendem Abstand. Klappt es nach mehreren Versuchen nicht, steht die Anfrage auf „Fehler" – mit einer verständlichen Ursache, der Zahl der Versuche und ohne technische Details. Mitarbeitende können sie dann neu anstoßen. Fehler, die ein Wiederholen nicht behebt (z. B. ein falsch eingerichteter KI-Dienst), gehen sofort auf „Fehler". Ein einzelnes unlesbares Dokument bremst die anderen Dokumente derselben Anfrage nicht aus. Stürzt der Worker mitten in der Arbeit ab, wird der Auftrag nach Ablauf automatisch erneut ausgeführt.

Plan-Pflicht (SYSTEM.md §4)

• Kein Auslöser – keine Modulgrenze, öffentliche API, Migration, Auth/Rechte, kein Zahlungs-/Daten-/Infrapfad, höchstens zwei Module, keine Architekturvarianten, umkehrbar
• Auslöser zutreffend – Impact Manifest ausgefüllt (Plan vor Code)

Impact Manifest

• Betroffene Module: jobs (Handler, drain(), Dead-Letter, Reprocess, Zeitbudget, Worker), extraction (KI-Client, Merge, Repository), requests (Statusmaschine, Sperre, Übergänge, Fehlerzustand), observability (logEvent nur mit IDs), db (Migrationen 0006–0008, pg-boss-Client-Optionen und Grants), config (AI_SERVICE_*), compose.yaml (Profil ai).
• Schnittstellen / Datenänderungen: Aufruf POST /v1/extract des KI-Dienstes (Vertrag contracts/ai-service.openapi.yaml, Bearer, X-Request-Id); Tabellen app.extraction_runs (unique job_id), app.extraction_segments, app.extracted_fields (Check: found nur mit Zitat; Beleg muss auf ein gespeichertes Segment zeigen) – alle FORCE RLS + Composite-FKs; app.requests + error_stage, error_message, attempts, next_retry_at.
• Akzeptanzkriterien: alle aus feat(jobs,extraction): process requests in the worker with retries and visible errors #7 (Nachweis unten).
• Testplan: Unit test-first: Statusmaschine, Feld-Merge, KI-Client gegen HTTP-Stub (Retry-/Permanent-Klassen, Timeout, Vertragsprüfung), Zeitbudget; Integration gegen echtes Postgres + SeaweedFS + pg-boss mit KI-Stub an der HTTP-Grenze und eigenen Test-Queues (schnelle Retries, kurzer Ablauf).
• Verifizierte Fakten: pg-boss 12.33.6 – Wartung explizit über boss.supervise(name?) (ADR-Offenpunkt D4): gedrosselt über monitorIntervalSeconds/maintenanceIntervalSeconds; als app_rw braucht sie UPDATE auf pgboss.version; persistQueueStats muss aus sein (legt täglich Partitionen per DDL an). Ein abgelaufener, nicht abgeschlossener Job geht über supervise zurück in retry (Integrationstest). Job-Ablauf 1 h, beim Worker-Start gegen AI_SERVICE_TIMEOUT_MS × UPLOAD_MAX_FILES geprüft.
• Offene Annahmen: Retry-Parameter (5 Versuche, 30 s … 30 min) – mit dem Kunden abzustimmen.
• Nicht-Ziele: Vercel-drain()-Trigger, Export, UI.
• Risiken und Rollback: Dokumentinhalt geht an den KI-Dienst (nur EU-Endpunkt, nur synthetisch lokal); Logs nur mit IDs (Test); Migrationen additiv; Rollback per Revert.

Akzeptanzkriterien → Nachweis

Kriterium	Nachweis
pg-boss 12, transaktionales Enqueue (fromDrizzle), singletonKey = requestId, Retry mit Backoff, Dead-Letter	Queue-Definition jobs/queues.ts; Enqueue aus #5 (sendInTransaction); processing.test.ts („retries … then dead-letters"), Reprocess-Test prüft den Job mit singletonKey
Handler idempotent: doppelte Zustellung → genau ein Lauf	processing.test.ts: nacheinander und gleichzeitig (Promise.all) → ein Lauf, einmal REVIEW; Wiederanlauf nach Worker-Absturz (Ablauf → supervise → Retry) → ein Lauf; Job mit fremder companyId wird übersprungen
KI-Timeout/5xx → Retry; nach Limit → ERROR mit Ursache und Versuchen; Reprocess stellt neu ein	processing.test.ts (503 → Retry → Dead-Letter → ERROR „Der KI-Dienst ist nicht erreichbar.", attempts 2; Timeout retrybar; 401 sofort ERROR; Reprocess ERROR → NEW + Job + Audit; fremde Firma abgelehnt). Anzeige in der Liste: #26
drain({ maxMs }) bis zum Budget; Worker-Schleife nutzt es	jobs/drain.ts, src/worker.ts; alle Integrationstests laufen über drain
Logs mit requestId, jobId, companyId, ohne Dokumentinhalt	processing.test.ts („logs IDs only"): Log-Zeilen enthalten die requestId, aber keinen Firmennamen, keine Namen, keine E-Mail-Adressen, kein Datum

Geändert

• src/features/requests/status.ts (+Test), repository.ts (Sperre, Übergänge, Fehlerzustand).
• src/features/extraction/ai-client.ts (+Test), merge.ts (+Test), repository.ts, types.ts, fixtures.ts.
• src/features/jobs/process-request.ts, drain.ts, reprocess.ts, budget.ts (+Test); src/worker.ts.
• src/features/observability/log.ts.
• src/db/schema/app.ts, Migrationen 0006_processing.sql, 0007_processing_force_rls.sql, 0008_processing_evidence_fks.sql; src/db/job-queue-client.ts (persistQueueStats: false, UPDATE auf version, monitorIntervalSeconds für Tests).
• src/config/env.ts (+Test), .env.example, compose.yaml (Worker-Env, Profil ai), .gitignore (.secrets/).
• Tests: tests/integration/processing.test.ts.
• Doku: data-model.md, Architekturkarte, operations.md (Worker, KI-Profil, Wiederanlauf, bekannte Lücke), CHANGELOG.

Nachweis (SYSTEM.md §11)

• verify:changed: grün
• verify: grün lokal – lint, typecheck, 100 Unit + 56 Integrationstests, depcruise 0 Verstöße, build, audit 0 high; Migrationen 0000–0008 auf frischer DB, 56/56 grün. CI: siehe Checks dieses PRs.
• verify:full / E2E-Spec: nicht betroffen
• Manueller Prüfnachweis: pnpm worker startet, verarbeitet, stoppt sauber auf SIGTERM; ohne AI_SERVICE_TOKEN bricht der Start mit Variablennamen ab. Echter KI-Dienst mit Vertex: nicht geprüft (keine Credentials).
• Frischer Review (P1 vor Ready-for-review; Architektur/API/DB immer): feat(jobs,extraction): process requests in the worker with retries and visible errors #34 (comment) – alle Befunde behoben

Doku-Entscheidung (genau eine)

• Keine langlebige Doku betroffen – Begründung: –
• Doku betroffen und im selben PR aktualisiert:
  • Produktdoku (P0: README): –
  • Technische Doku: docs/technical/data-model.md, docs/technical/operations.md
  • Architekturkarte: docs/technical/architecture.md
  • ADR: –
  • CHANGELOG [Unreleased] (sichtbares Feature oder Verhalten – im selben PR, nie „später")

Entferntes oder Umbenanntes: nichts entfernt

Dateigrößen und neue Bausteine (SYSTEM.md §7)

Dateien über 500 Zeilen im Diff (Ausnahmen: generierter Code, Lockfiles, Fixtures, Migrationen, Schemas, Ressourcen, Doku, Konfiguration):

• keine
• bewusst belassen – Begründung: –
• im selben PR nach fachlicher Verantwortung geteilt
• Folge-Issue –

Über 800 Zeilen mit neuer Fachlogik oder über 1000 Zeilen (P1/P2): nicht betroffen

Neue Shared-Komponente, Utility-Datei, Adapter oder fachlicher Service:

• nein
• ja – gesucht nach: vorhandenem HTTP-Client/Logger/Retry-Code in src/; gefunden: nur observability/health – KI-Client (extraction) und logEvent (observability) sind in der Architekturkarte vorgesehen; pino folgt mit feat(observability): correlated structured logs and full health #28

Subagent-Einsätze

• Frischer Review: unabhängiger Subagent (read-only) – 5 Important, 3 Notes, alle bearbeitet.

Risiken / offene Punkte

• Live-Aufruf des KI-Dienstes gegen Vertex AI unverifiziert.
• .xlsx/.docx/.msg werden bis feat(ai-service): XLSX, DOCX, MSG and scanned PDFs #23 übersprungen (Original bleibt, Hinweis im Lauf).
• Anfragen, deren Job komplett verschwindet, bleiben in NEW/PROCESSING – firmenübergreifender Sweep folgt mit feat(requests): request list with errors, retries and reprocessing #26/feat(observability): correlated structured logs and full health #28 (operations.md).

---

🤖 Generated with Claude Code

https://claude.ai/code/session_01DJ5vaKvTYiMvdngT4d3xo1

[Fluory]

Frischer Review (unabhängiger Subagent) – Befunde und Auflösung

Review nach review-pr + Migrations-/Test-/AI-/Security-Regeln, read-only; pg-boss-12.33.6-Quellen gegengelesen (Dead-Letter behält data; fail/getJobById/Zustandsnamen korrekt; Wartung als app_rw baut keine Indizes neu). Auflösung in den Commits nach 346ff1c.

#	Befund	Auflösung
Important	Job-Ablauf (10 min) kürzer als der mögliche Verarbeitungsweg (10 Dokumente × 120 s) → Neuzustellung während der Arbeit	behoben – Ablauf 1 h; assertProcessingBudget prüft beim Worker-Start AI_SERVICE_TIMEOUT_MS × UPLOAD_MAX_FILES + Reserve < Ablauf (fail-closed, Unit-Test)
Important	recordProcessingFailure schrieb ohne Statusprüfung (Fehlertext auf REVIEW-Anfrage möglich)	behoben – nur bei PROCESSING
Important	Antwort mit found ohne Beleg, unbekanntem Segment, doppelten Segment-IDs oder fremder documentId lief in 5 teure Retries	behoben – Vertragsinvarianten im Client → contract_violation (permanent, 4 neue Tests); Constraint-Verletzungen beim Speichern → permanent; DB erzwingt jetzt: Beleg zeigt auf gespeichertes Segment (Composite-FK, Migration 0008)
Important	Wiederanlauf nach Worker-Absturz (DR4) unbewiesen	behoben – Integrationstest: Job geholt, Anfrage geclaimt, „Absturz", Ablauf, supervise() als app_rw → retry → drain → REVIEW, attempts 2, genau ein Lauf
Important	Dead-Letter nur bei leerer Queue; Fehler beim Markieren → Anfrage bleibt still in PROCESSING	behoben – jede Runde ein Dead-Letter-Job; die Dead-Letter-Queue wiederholt ihren eigenen Handler (10×, Backoff). Verbleibende Lücke (Job ganz verschwunden) dokumentiert in operations.md, Sweep folgt mit #26/#28
Note	400/404/405 als Dokumentfehler eingestuft	behoben – Dienst-Fehler (falsche URL/Client-Bug), Tests
Note	Nebenläufigkeitstest behauptete mehr als er prüft; Reprocess-Test ließ Job in der geteilten Queue	behoben – Test umbenannt und prüft KI-Aufrufe (≤ 2) = attempts; Aufräumen der geteilten Queue; neuer Test: Job mit fremder companyId wird übersprungen
Note	data-model-Tabelle zerbrochen; extracted_fields.request_id ohne FK	behoben – Tabelle repariert; Composite-FK (request_id, company_id) (Migration 0008, auf frischer DB getestet)

Verdict des Reviewers: nach Behebung mergebar, menschliche Freigabe nötig (Migrationen 0006–0008, pg-boss-Grants).

---

Generated by Claude Code