L10b Betrieb & Aufräumen: Lotse-Betrieb – Aufbewahrung KI-Protokoll und Token-Kontingent

Aufräumpunkt k (Spec §31):
- Aufbewahrung: services/lotse/retention.ts leert input/output und createdById von
  AiGeneration-Einträgen älter als AI_GENERATION_RETENTION_DAYS (Default 180), Metadaten bleiben,
  Audit je Mandant. Queue/Processor ai-retention, täglicher BullMQ-Job-Scheduler beim Start des
  craftvia-worker.
- Kontingent: services/lotse/budget.ts (Tokens ein+aus je Kalendermonat, TenantSettings-Wert vor
  Env AI_MONTHLY_TOKEN_LIMIT, 0 = unbegrenzt). Lotse-Entwurf und Sprachnotiz-Zusammenfassung
  → blocked budget_exceeded mit Klartext; Import-Extraktion fällt auf manuelle Erfassung zurück
  (Hinweis ai_budget_exceeded). /settings/lotse: Kontingent setzen, Verbrauch anzeigen.
- scripts/test-betrieb-audit.ts: Audit nach Commit/Rollback/verschachtelt, Merge atomar und in
  äußerer Transaktion, Audit „read", Aufbewahrung (Frist, Metadaten, Idempotenz, Mandant B),
  Kontingent (Mandant/Env/Vormonat/unbegrenzt, Rollen, Audit).

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

src/server/actions/lotse-settings.ts

@@ -19,9 +19,12 @@ export async function saveLotseSettings(fd: FormData): Promise<void> {
   try {
     requirePermission(session, "tenant:manage"); // fast JWT check; requireApiContext re-checks against the DB
     const ctx = await requireApiContext(null, "tenant:manage");
+    // L10b: empty = platform default (null); invalid numbers are rejected by the service schema
+    const rawLimit = String(fd.get("monthlyTokenLimit") ?? "").trim();
     await updateLotseSettings(ctx, {
       enabled: fd.get("enabled") === "on",
       addressForm: (["sie", "du"].includes(String(fd.get("addressForm"))) ? String(fd.get("addressForm")) : "neutral") as "sie" | "du" | "neutral",
+      monthlyTokenLimit: rawLimit === "" ? null : Number(rawLimit),
     });
     revalidatePath("/settings/lotse");
     revalidatePath("/", "layout");

src/server/jobs/processors/ai-retention.ts

@@ -0,0 +1,11 @@
+import { purgeExpiredAiGenerations } from "@/server/services/lotse/retention";
+
+/**
+ * Daily retention job for the AI log (L10b, Spec §31). Scheduled by the craftvia worker
+ * (`scheduleRecurringJobs`); the payload carries no tenant — the service iterates all tenants and
+ * writes through dbForTenant.
+ */
+export async function process(): Promise<void> {
+  const res = await purgeExpiredAiGenerations();
+  console.info(`[ai-retention] ${res.pseudonymised} entries older than ${res.days} days pseudonymised (${res.tenants} tenants)`);
+}

src/server/jobs/processors/index.ts

@@ -12,6 +12,7 @@ export const PROCESSORS: Partial<Record<JobQueueName, () => Promise<JobProcessor
   transcription: () => import("./transcription").then((m) => m.process),
   "report-pdf": () => import(/* turbopackIgnore: true */ "./report-pdf").then((m) => m.process), // worker-only (react-dom/server + Chromium), kept out of the app bundle
   "image-derivatives": () => import("./image-derivatives").then((m) => m.process),
+  "ai-retention": () => import("./ai-retention").then((m) => m.process), // L10b: daily AI log retention (scheduled by the worker)
 };
 
 /** Inline fallback when no Redis is available (dev/demo). */

src/server/jobs/queues.ts

@@ -12,6 +12,7 @@ export const JOB_QUEUES = {
   transcription: "transcription",
   reportPdf: "report-pdf",
   imageDerivatives: "image-derivatives",
+  aiRetention: "ai-retention",
 } as const;
 
 export type JobQueueName = (typeof JOB_QUEUES)[keyof typeof JOB_QUEUES];
@@ -80,6 +81,24 @@ export async function enqueueJob(name: JobQueueName, payload: JobPayload): Promi
   return true;
 }
 
+/**
+ * Recurring jobs (L10b), registered once by the craftvia worker at start. BullMQ job schedulers
+ * are idempotent per id, so several worker replicas do not create duplicates.
+ * - ai-retention: daily pseudonymisation of AI log contents (services/lotse/retention.ts)
+ */
+export async function scheduleRecurringJobs(connection: Redis): Promise<void> {
+  const q = new Queue<JobPayload>(JOB_QUEUES.aiRetention, { connection });
+  try {
+    await q.upsertJobScheduler(
+      "ai-retention-daily",
+      { every: 24 * 60 * 60 * 1000 },
+      { name: JOB_QUEUES.aiRetention, data: { tenantId: "*", entityId: "retention" }, opts: { removeOnComplete: { count: 30 }, removeOnFail: { count: 30 } } },
+    );
+  } finally {
+    await q.close();
+  }
+}
+
 export async function closeJobQueues(): Promise<void> {
   await Promise.all([...queues.values()].map((q) => q.close()));
   queues.clear();

src/server/services/imports/process.ts

@@ -13,6 +13,7 @@ import { checkPlausibility } from "@/lib/imports/plausibility";
 // TODO(L3→L1): replace with "@/lib/customers/duplicates" after the L1 merge (same interface).
 import { findDuplicateCustomers } from "@/server/services/customers/duplicates";
 import { findSiteCandidates } from "./site-candidates";
+import { getTokenBudget } from "@/server/services/lotse/budget";
 
 export type ProcessDeps = {
   provider: DocumentExtractionProvider | null;
@@ -45,9 +46,11 @@ export async function processImport(ctx: ServiceCtx, importId: string, deps: Pro
     let providerName: string | null = null;
     let model: string | null = null;
 
-    if (!deps.provider) {
+    // L10b: monthly AI token budget used up → same graceful path as without provider
+    const budgetExceeded = deps.provider ? (await getTokenBudget(ctx, deps.now)).exceeded : false;
+    if (!deps.provider || budgetExceeded) {
       fields = emptyExtraction();
-      hints = [{ field: null, code: "manual_entry" }];
+      hints = budgetExceeded ? [{ field: null, code: "manual_entry" }, { field: null, code: "ai_budget_exceeded" }] : [{ field: null, code: "manual_entry" }];
     } else {
       const bytes = await deps.loadBytes({ storageKey: doc.storageKey });
       if (!bytes) throw new Error("file_unavailable");

src/server/services/lotse/budget.ts

@@ -0,0 +1,57 @@
+import { ServiceError, type ServiceCtx } from "@/server/services/context";
+
+/**
+ * Monthly AI token budget per tenant (Spec §31 „Kostenlimit", L10b).
+ * Limit = `TenantSettings.aiMonthlyTokenLimit` if set, otherwise env `AI_MONTHLY_TOKEN_LIMIT`
+ * (default 0). 0 = unlimited. Usage = sum of input + output tokens of all `AiGeneration`s of the
+ * tenant since the start of the current calendar month (UTC). Checked BEFORE a provider call —
+ * one call may overshoot the limit by its own size, the next one is refused.
+ * Transcription (audio) reports no tokens and is not counted.
+ */
+
+export function envMonthlyTokenLimit(): number {
+  const v = Number(process.env.AI_MONTHLY_TOKEN_LIMIT);
+  return Number.isInteger(v) && v >= 0 ? v : 0;
+}
+
+export type TokenBudget = {
+  /** effective limit, 0 = unlimited */
+  limit: number;
+  source: "tenant" | "env";
+  /** tenant override as stored (null = platform default) */
+  tenantLimit: number | null;
+  used: number;
+  periodStart: string;
+  exceeded: boolean;
+};
+
+export function monthStart(now: Date): Date {
+  return new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), 1));
+}
+
+export async function getTokenBudget(ctx: Pick<ServiceCtx, "db">, now: Date = new Date()): Promise<TokenBudget> {
+  const start = monthStart(now);
+  const [settings, usage] = await Promise.all([
+    ctx.db.tenantSettings.findFirst({ select: { aiMonthlyTokenLimit: true } }),
+    ctx.db.aiGeneration.aggregate({ where: { createdAt: { gte: start } }, _sum: { inputTokens: true, outputTokens: true } }),
+  ]);
+  const tenantLimit = settings?.aiMonthlyTokenLimit ?? null;
+  const limit = tenantLimit ?? envMonthlyTokenLimit();
+  const used = (usage._sum.inputTokens ?? 0) + (usage._sum.outputTokens ?? 0);
+  return {
+    limit,
+    source: tenantLimit === null ? "env" : "tenant",
+    tenantLimit,
+    used,
+    periodStart: start.toISOString(),
+    exceeded: limit > 0 && used >= limit,
+  };
+}
+
+/** Throws `blocked` (details.reason = "budget_exceeded") when the monthly budget is used up. */
+export async function assertTokenBudget(ctx: Pick<ServiceCtx, "db">, now: Date = new Date()): Promise<void> {
+  const budget = await getTokenBudget(ctx, now);
+  if (budget.exceeded) {
+    throw new ServiceError("blocked", "monthly AI token budget exceeded", { reason: "budget_exceeded", limit: budget.limit, used: budget.used });
+  }
+}

src/server/services/lotse/draft-report.ts

@@ -10,6 +10,7 @@ import { contentOf, requireVisibleReport } from "@/server/services/reports/commo
 import { buildReportDraftInput } from "./minimize";
 import { assertLotseEnabled, lotseVoice } from "./settings";
 import { loadDraftNotes, loadMinimizationContext } from "./sources";
+import { assertTokenBudget } from "./budget";
 
 export type LotseDeps = { provider: LotseAssistant | null; now?: () => Date };
 export const defaultLotseDeps = (): LotseDeps => ({ provider: getLotseProvider() });
@@ -45,6 +46,7 @@ export async function prepareDraftInput(ctx: ServiceCtx, report: Report, content
 export async function draftReportWithLotse(ctx: ServiceCtx, reportId: string, deps: LotseDeps = defaultLotseDeps()) {
   const report = await requireDraftableReport(ctx, reportId);
   if (!deps.provider) throw new ServiceError("invalid", "lotse provider not configured", { reason: "not_configured" });
+  await assertTokenBudget(ctx); // L10b: monthly token budget (blocked, reason budget_exceeded)
   const content = contentOf(report);
   const input = await prepareDraftInput(ctx, report, content);
 

src/server/services/lotse/retention.ts

@@ -0,0 +1,50 @@
+import { Prisma } from "@prisma/client";
+import { writeAuditLog } from "@/server/audit";
+import { dbForTenant, prisma } from "@/server/db";
+
+/**
+ * Retention of the AI log (Spec §31 „Aufbewahrung", L9 offener Punkt 4, L10b):
+ * after `AI_GENERATION_RETENTION_DAYS` (default 180) the CONTENT of an AiGeneration (minimised
+ * input sent to the provider, provider output) is deleted and the person reference is removed
+ * (`createdById = null`). Metadata (kind, provider, model, tokens, entity reference, time) stays
+ * for cost statistics and the reference from reports (`aiGenerationId`) keeps resolving.
+ * Runs daily in the craftvia worker (queue `ai-retention`); idempotent.
+ */
+
+export const DEFAULT_AI_GENERATION_RETENTION_DAYS = 180;
+
+export function aiGenerationRetentionDays(): number {
+  const v = Number(process.env.AI_GENERATION_RETENTION_DAYS);
+  return Number.isInteger(v) && v > 0 ? v : DEFAULT_AI_GENERATION_RETENTION_DAYS;
+}
+
+export type RetentionResult = { cutoff: string; days: number; tenants: number; pseudonymised: number };
+
+export async function purgeExpiredAiGenerations(opts: { now?: Date; days?: number; tenantIds?: string[] } = {}): Promise<RetentionResult> {
+  const days = opts.days ?? aiGenerationRetentionDays();
+  const cutoff = new Date((opts.now ?? new Date()).getTime() - days * 24 * 60 * 60 * 1000);
+  // tenant list = platform data (no tenant content); every content change runs through dbForTenant
+  const tenantIds = opts.tenantIds ?? (await prisma.tenant.findMany({ select: { id: true } })).map((t) => t.id);
+
+  let pseudonymised = 0;
+  for (const tenantId of tenantIds) {
+    const db = dbForTenant(tenantId);
+    const res = await db.aiGeneration.updateMany({
+      where: {
+        createdAt: { lt: cutoff },
+        OR: [{ input: { not: Prisma.DbNull } }, { output: { not: Prisma.DbNull } }, { createdById: { not: null } }],
+      },
+      data: { input: Prisma.DbNull, output: Prisma.DbNull, createdById: null },
+    });
+    if (res.count > 0) {
+      pseudonymised += res.count;
+      await writeAuditLog({
+        tenantId,
+        action: "delete",
+        entity: "ai_generation_retention",
+        after: { count: res.count, cutoff: cutoff.toISOString(), retentionDays: days },
+      });
+    }
+  }
+  return { cutoff: cutoff.toISOString(), days, tenants: tenantIds.length, pseudonymised };
+}

src/server/services/lotse/settings.ts

@@ -5,6 +5,7 @@ import { AI_MODEL, isAiConfigured } from "@/server/ai/client";
 import { transcriptionConfig } from "@/server/ai/transcription/openai-compatible";
 import { writeAuditLog } from "@/server/audit";
 import { assertCan, inTransaction, ServiceError, type ServiceCtx } from "@/server/services/context";
+import { envMonthlyTokenLimit, getTokenBudget } from "./budget";
 
 /**
  * Lotse settings per tenant (lane L9): on/off = module toggle `lotse` (TenantModule, missing row = on),
@@ -38,19 +39,26 @@ export async function lotseVoice(ctx: Pick<ServiceCtx, "db">): Promise<{ address
 
 export async function getLotseSettings(ctx: ServiceCtx) {
   assertCan(ctx, "tenant:manage");
-  const [enabled, s] = await Promise.all([isLotseEnabled(ctx), ctx.db.tenantSettings.findFirst({ select: { lotseAddressForm: true } })]);
+  const [enabled, s, budget] = await Promise.all([
+    isLotseEnabled(ctx),
+    ctx.db.tenantSettings.findFirst({ select: { lotseAddressForm: true } }),
+    getTokenBudget(ctx),
+  ]);
   const transcription = transcriptionConfig();
   return {
     enabled,
     addressForm: toAddressForm(s?.lotseAddressForm),
     draft: { configured: isAiConfigured(), provider: "Anthropic (Claude)", model: AI_MODEL },
     transcription,
+    budget: { ...budget, platformLimit: envMonthlyTokenLimit() },
   };
 }
 
 export const lotseSettingsSchema = z.object({
   enabled: z.boolean(),
   addressForm: z.enum(["sie", "du", "neutral"]),
+  /** L10b: tenant token budget per month; null = platform default, 0 = unlimited, undefined = unchanged */
+  monthlyTokenLimit: z.number().int().min(0).max(1_000_000_000).nullable().optional(),
 });
 export type LotseSettingsInput = z.input<typeof lotseSettingsSchema>;
 
@@ -59,10 +67,13 @@ export async function updateLotseSettings(ctx: ServiceCtx, raw: LotseSettingsInp
   const input = lotseSettingsSchema.parse(raw);
   const addressForm = input.addressForm === "neutral" ? null : input.addressForm;
 
+  const stored = await ctx.db.tenantSettings.findFirst({ select: { lotseAddressForm: true, aiMonthlyTokenLimit: true } });
   const before = {
     enabled: await isLotseEnabled(ctx),
-    addressForm: (await ctx.db.tenantSettings.findFirst({ select: { lotseAddressForm: true } }))?.lotseAddressForm ?? null,
+    addressForm: stored?.lotseAddressForm ?? null,
+    monthlyTokenLimit: stored?.aiMonthlyTokenLimit ?? null,
   };
+  const monthlyTokenLimit = input.monthlyTokenLimit === undefined ? before.monthlyTokenLimit : input.monthlyTokenLimit;
   await inTransaction(ctx, async (tx) => {
     await tx.db.tenantModule.upsert({
       where: { tenantId_moduleKey: { tenantId: tx.tenantId, moduleKey: LOTSE_MODULE_KEY } },
@@ -70,14 +81,15 @@ export async function updateLotseSettings(ctx: ServiceCtx, raw: LotseSettingsInp
       create: { tenantId: tx.tenantId, moduleKey: LOTSE_MODULE_KEY, enabled: input.enabled },
     });
     const existing = await tx.db.tenantSettings.findFirst({ select: { id: true } });
+    const data = { lotseAddressForm: addressForm, aiMonthlyTokenLimit: monthlyTokenLimit };
     if (existing) {
-      await tx.db.tenantSettings.update({ where: { id: existing.id }, data: { lotseAddressForm: addressForm } });
+      await tx.db.tenantSettings.update({ where: { id: existing.id }, data });
     } else {
       const tenant = await tx.db.tenant.findUnique({ where: { id: tx.tenantId }, select: { name: true } });
-      await tx.db.tenantSettings.create({ data: { tenantId: tx.tenantId, orgName: tenant?.name ?? "—", lotseAddressForm: addressForm } });
+      await tx.db.tenantSettings.create({ data: { tenantId: tx.tenantId, orgName: tenant?.name ?? "—", ...data } });
     }
   });
-  const after = { enabled: input.enabled, addressForm };
+  const after = { enabled: input.enabled, addressForm, monthlyTokenLimit };
   await writeAuditLog({ tenantId: ctx.tenantId, actorId: ctx.userId, action: "update", entity: "lotse_settings", entityId: ctx.tenantId, before, after });
   return after;
 }

src/server/services/lotse/voice.ts

@@ -7,6 +7,7 @@ import { createNote } from "@/server/services/field/notes";
 import { workOrderScope } from "@/server/services/work-orders/visibility";
 import { defaultLotseDeps, type LotseDeps } from "./draft-report";
 import { scrubText } from "./minimize";
+import { assertTokenBudget } from "./budget";
 import { assertLotseEnabled, lotseVoice } from "./settings";
 import { loadMinimizationContext } from "./sources";
 import { TRANSCRIPT_SEPARATOR } from "./transcription";
@@ -67,6 +68,7 @@ export async function summarizeVoiceNote(ctx: ServiceCtx, voiceNoteId: string, d
   await assertLotseEnabled(ctx);
   if (!voice.transcript?.trim()) throw new ServiceError("invalid", "voice note has no transcript", { reason: "no_transcript" });
   if (!deps.provider) throw new ServiceError("invalid", "lotse provider not configured", { reason: "not_configured" });
+  await assertTokenBudget(ctx); // L10b: monthly token budget
 
   const [minimization, lang] = await Promise.all([loadMinimizationContext(ctx, voice.workOrderId), lotseVoice(ctx)]);
   const transcript = scrubText(voice.transcript, minimization);