Add summarizer observability and LLM retries

packages/summarizer/index.ts

@@ -1,4 +1,5 @@
 import { basename, join } from "node:path";
+import { mkdir } from "node:fs/promises";
 import { listCompanies, summarizeCompany, writeSummary, type LlmCallStatus, type LlmCallTrace } from "./summarizer";
 import { dataPath } from "../../deployPaths";
 import { allowedModelFromEnv, isSafeStem, jsonError, validateRequestedModel, withAuth } from "../../security";
@@ -10,6 +11,9 @@ const PAGE_PATH = join(import.meta.dir, "index.html");
 const MAX_PARALLEL_SUMMARIES = 2;
 const MAX_ACTIVE_JOBS = Number(process.env.BMP_MAX_ACTIVE_SUMMARY_JOBS ?? 10);
 const EXPECTED_LLM_CALLS_PER_COMPANY = 13;
+const OBSERVABILITY_DIR =
+  process.env.BMP_OBSERVABILITY_DIR ??
+  dataPath(join(import.meta.dir, "outputs", "_observability"), "summarizer", "outputs", "_observability");
 
 type JobStatus = "queued" | "running" | "done" | "error" | "cancelled";
 
@@ -25,6 +29,8 @@ interface SummaryJob {
   downloads: SummaryDownload[];
   errors: string[];
   llmCalls: SafeLlmCallTrace[];
+  observability: ObservabilityLog;
+  observabilityWrite: Promise<void>;
   usage: UsageSummary;
   statusMessage?: string;
   createdAt: string;
@@ -35,9 +41,47 @@ type SafeLlmCallTrace = Pick<
   LlmCallTrace,
   "id" | "operation" | "label" | "model" | "status" | "startedAt" | "finishedAt" | "durationMs" | "error"
 > & {
+  attempt: number;
+  maxAttempts: number;
   usage?: unknown;
 };
 
+interface ObservabilityEvent {
+  at: string;
+  type: string;
+  message?: string;
+  data?: unknown;
+}
+
+interface ObservabilityLog {
+  version: 1;
+  jobId: string;
+  stem: string;
+  model: string;
+  status: JobStatus;
+  createdAt: string;
+  finishedAt?: string;
+  total: number;
+  completed: number;
+  current?: string;
+  companies: string[];
+  files: string[];
+  downloads: SummaryDownload[];
+  errors: SerializedError[];
+  inputs: Record<string, unknown>;
+  outputs: Record<string, unknown>;
+  llmCalls: LlmCallTrace[];
+  events: ObservabilityEvent[];
+  logPath: string;
+}
+
+interface SerializedError {
+  name?: string;
+  message: string;
+  stack?: string;
+  cause?: unknown;
+}
+
 interface SummaryDownload {
   company: string;
   label: string;
@@ -77,6 +121,8 @@ function sanitizeTrace(trace: LlmCallTrace): SafeLlmCallTrace {
     operation: trace.operation,
     label: trace.label,
     model: trace.model,
+    attempt: trace.attempt,
+    maxAttempts: trace.maxAttempts,
     status: trace.status,
     startedAt: trace.startedAt,
     finishedAt: trace.finishedAt,
@@ -209,9 +255,83 @@ function isCancelled(job: SummaryJob): boolean {
 
 const jobs = new Map<string, SummaryJob>();
 
+function serializeError(error: unknown): SerializedError {
+  if (error instanceof Error) {
+    return {
+      name: error.name,
+      message: error.message,
+      stack: error.stack,
+      cause: error.cause,
+    };
+  }
+  return { message: String(error) };
+}
+
+function createObservabilityLog(jobId: string, stem: string, model: string, total: number, createdAt: string): ObservabilityLog {
+  return {
+    version: 1,
+    jobId,
+    stem,
+    model,
+    status: "queued",
+    createdAt,
+    total,
+    completed: 0,
+    companies: [],
+    files: [],
+    downloads: [],
+    errors: [],
+    inputs: {},
+    outputs: {},
+    llmCalls: [],
+    events: [
+      {
+        at: createdAt,
+        type: "job.created",
+        message: "Summary job was queued.",
+      },
+    ],
+    logPath: join(OBSERVABILITY_DIR, `${createdAt.replaceAll(":", "-")}-${jobId}.json`),
+  };
+}
+
+function recordEvent(job: SummaryJob, type: string, message?: string, data?: unknown): void {
+  job.observability.events.push({
+    at: new Date().toISOString(),
+    type,
+    message,
+    data,
+  });
+}
+
+function syncObservabilityState(job: SummaryJob): void {
+  job.observability.status = job.status;
+  job.observability.finishedAt = job.finishedAt;
+  job.observability.completed = job.completed;
+  job.observability.current = job.current;
+  job.observability.files = [...job.files];
+  job.observability.downloads = [...job.downloads];
+}
+
+function persistObservability(job: SummaryJob): Promise<void> {
+  syncObservabilityState(job);
+  job.observabilityWrite = job.observabilityWrite
+    .catch(() => undefined)
+    .then(async () => {
+      await mkdir(OBSERVABILITY_DIR, { recursive: true });
+      await Bun.write(job.observability.logPath, JSON.stringify(job.observability, null, 2));
+    })
+    .catch((error) => {
+      console.error("Failed to write summarizer observability log:", error);
+    });
+  return job.observabilityWrite;
+}
+
 function createJob(stem: string, model: string, total: number): SummaryJob {
+  const id = crypto.randomUUID();
+  const createdAt = new Date().toISOString();
   const job: SummaryJob = {
-    id: crypto.randomUUID(),
+    id,
     status: "queued",
     stem,
     model,
@@ -221,10 +341,13 @@ function createJob(stem: string, model: string, total: number): SummaryJob {
     downloads: [],
     errors: [],
     llmCalls: [],
+    observability: createObservabilityLog(id, stem, model, total, createdAt),
+    observabilityWrite: Promise.resolve(),
     usage: emptyUsageSummary(),
-    createdAt: new Date().toISOString(),
+    createdAt,
   };
   jobs.set(job.id, job);
+  void persistObservability(job);
   return job;
 }
 
@@ -269,14 +392,31 @@ async function executeJob(job: SummaryJob): Promise<void> {
   if (isCancelled(job)) return;
   job.status = "running";
   job.statusMessage = "Auswertung wird vorbereitet.";
+  recordEvent(job, "job.started", job.statusMessage);
+  await persistObservability(job);
   const companies = job.stem === "__all__" ? await listCompanies() : [job.stem];
+  job.observability.companies = [...companies];
+  recordEvent(job, "companies.selected", undefined, { companies });
+  await persistObservability(job);
 
   try {
     await runWithConcurrency(companies, MAX_PARALLEL_SUMMARIES, async (company) => {
       if (isCancelled(job)) return;
       job.current = company;
       job.statusMessage = `LLM-Auswertung für ${company} läuft.`;
+      recordEvent(job, "company.started", job.statusMessage, { company });
+      await persistObservability(job);
       const summary = await summarizeCompany(company, job.model, {
+        onCompanyInput: (companyStem, input) => {
+          job.observability.inputs[companyStem] = input;
+          recordEvent(job, "company.input.loaded", undefined, { company: companyStem });
+          void persistObservability(job);
+        },
+        onCompanyOutput: (companyStem, output) => {
+          job.observability.outputs[companyStem] = output;
+          recordEvent(job, "company.output.created", undefined, { company: companyStem });
+          void persistObservability(job);
+        },
         onLlmCall: (trace) => {
           const existingIndex = job.llmCalls.findIndex((call) => call.id === trace.id);
           const safeTrace = sanitizeTrace(trace);
@@ -285,24 +425,48 @@ async function executeJob(job: SummaryJob): Promise<void> {
           } else {
             job.llmCalls.push(safeTrace);
           }
+          const fullTraceIndex = job.observability.llmCalls.findIndex((call) => call.id === trace.id);
+          if (fullTraceIndex >= 0) {
+            job.observability.llmCalls[fullTraceIndex] = trace;
+          } else {
+            job.observability.llmCalls.push(trace);
+          }
           job.usage = summarizeUsage(job.llmCalls);
           job.statusMessage = trace.status === "running"
-            ? `${trace.label} wird verarbeitet.`
+            ? `${trace.label} wird verarbeitet (Versuch ${trace.attempt}/${trace.maxAttempts}).`
-            : `${trace.label} abgeschlossen.`;
+            : trace.status === "error"
+              ? `${trace.label} fehlgeschlagen (Versuch ${trace.attempt}/${trace.maxAttempts}).`
+              : `${trace.label} abgeschlossen.`;
+          recordEvent(job, `llm.${trace.status}`, job.statusMessage, {
+            id: trace.id,
+            operation: trace.operation,
+            label: trace.label,
+            attempt: trace.attempt,
+            maxAttempts: trace.maxAttempts,
+          });
+          void persistObservability(job);
         },
       });
       job.statusMessage = `Downloads für ${company} werden vorbereitet.`;
+      recordEvent(job, "company.summary.created", job.statusMessage, { company });
       if (isCancelled(job)) return;
       const outputPaths = await writeSummary(company, summary, OUTPUTS_DIR);
       job.files.push(...outputPaths.map((path) => basename(path)));
       job.downloads.push(...outputPaths.map((path) => buildDownload(company, path)));
       job.completed += 1;
+      recordEvent(job, "company.finished", `Auswertung für ${company} fertig.`, {
+        company,
+        outputFiles: outputPaths.map((path) => basename(path)),
+      });
+      await persistObservability(job);
     });
 
     if (isCancelled(job)) {
       job.current = undefined;
       job.statusMessage = "Auswertung abgebrochen.";
       job.finishedAt = new Date().toISOString();
+      recordEvent(job, "job.cancelled", job.statusMessage);
+      await persistObservability(job);
       return;
     }
 
@@ -310,17 +474,25 @@ async function executeJob(job: SummaryJob): Promise<void> {
     job.current = undefined;
     job.statusMessage = "Auswertung fertig.";
     job.finishedAt = new Date().toISOString();
+    recordEvent(job, "job.done", job.statusMessage);
+    await persistObservability(job);
   } catch (error) {
     if (isCancelled(job)) {
       job.current = undefined;
       job.statusMessage = "Auswertung abgebrochen.";
       job.finishedAt = new Date().toISOString();
+      recordEvent(job, "job.cancelled", job.statusMessage);
+      await persistObservability(job);
       return;
     }
+    const serializedError = serializeError(error);
     job.status = "error";
-    job.errors.push(String(error));
+    job.errors.push(serializedError.message);
-    job.statusMessage = "Auswertung fehlgeschlagen.";
+    job.observability.errors.push(serializedError);
+    job.statusMessage = `Auswertung fehlgeschlagen: ${serializedError.message}`;
     job.finishedAt = new Date().toISOString();
+    recordEvent(job, "job.error", job.statusMessage, serializedError);
+    await persistObservability(job);
   }
 }
 
@@ -432,7 +604,10 @@ export const routes = {
       return Response.json(publicJob(job));
     }),
     DELETE: withAuth(async (req: Request) => {
-      const id = (req as Request & { params: Record<string, string> }).params.id;
+      const id = (req as Request & { params: Record<string, string | undefined> }).params.id;
+      if (!id) {
+        return Response.json({ error: "Job not found" }, { status: 404 });
+      }
       const job = jobs.get(id);
       if (!job) {
         return Response.json({ error: "Job not found" }, { status: 404 });

packages/summarizer/summarizer.ts

@@ -90,6 +90,8 @@ export interface LlmCallTrace {
   operation: "segmentierung" | "zusammenfassung" | "scoring" | "swot";
   label: string;
   model: string;
+  attempt: number;
+  maxAttempts: number;
   status: LlmCallStatus;
   startedAt: string;
   finishedAt?: string;
@@ -101,6 +103,8 @@ export interface LlmCallTrace {
 
 export interface SummarizeCompanyOptions {
   onLlmCall?: (trace: LlmCallTrace) => void;
+  onCompanyInput?: (stem: string, input: ExtractedData) => void;
+  onCompanyOutput?: (stem: string, output: SummaryData) => void;
 }
 
 const OUTPUTS_DIR =
@@ -111,6 +115,8 @@ const OPENROUTER_PRIVACY_PROVIDER = {
   data_collection: "deny",
   zdr: true,
 } as const;
+const MAX_LLM_RETRY_ATTEMPTS = Math.max(1, Number(process.env.BMP_LLM_RETRY_ATTEMPTS ?? 3));
+const LLM_RETRY_BASE_DELAY_MS = Math.max(0, Number(process.env.BMP_LLM_RETRY_BASE_DELAY_MS ?? 750));
 
 function createClient(): OpenAI {
   const apiKey = process.env.OPENROUTER_API_KEY;
@@ -169,12 +175,16 @@ function beginLlmTrace(
   label: string,
   model: string,
   request: Record<string, unknown>,
+  attempt: number,
+  maxAttempts: number,
 ): LlmCallTrace {
   const trace: LlmCallTrace = {
     id: crypto.randomUUID(),
     operation,
     label,
     model,
+    attempt,
+    maxAttempts,
     status: "running",
     startedAt: new Date().toISOString(),
     request,
@@ -200,6 +210,66 @@ function finishLlmTrace(
   });
 }
 
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+function retryDelayMs(attempt: number): number {
+  if (LLM_RETRY_BASE_DELAY_MS <= 0) return 0;
+  const jitter = Math.floor(Math.random() * Math.min(250, LLM_RETRY_BASE_DELAY_MS));
+  return LLM_RETRY_BASE_DELAY_MS * 2 ** Math.max(0, attempt - 1) + jitter;
+}
+
+function shouldRetryLlmError(error: unknown): boolean {
+  const status = (error as { status?: unknown; code?: unknown })?.status;
+  if (typeof status === "number") {
+    return status === 408 || status === 409 || status === 425 || status === 429 || status >= 500;
+  }
+  const message = String(error).toLowerCase();
+  return [
+    "timeout",
+    "timed out",
+    "rate limit",
+    "temporarily",
+    "econnreset",
+    "fetch failed",
+    "socket",
+    "overloaded",
+    "provider returned error",
+  ].some((needle) => message.includes(needle));
+}
+
+async function createCompletionWithRetry(
+  client: OpenAI,
+  request: Record<string, unknown>,
+  trace: {
+    operation: LlmCallTrace["operation"];
+    label: string;
+    model: string;
+    onLlmCall?: SummarizeCompanyOptions["onLlmCall"];
+  } | undefined,
+) {
+  let lastError: unknown;
+  for (let attempt = 1; attempt <= MAX_LLM_RETRY_ATTEMPTS; attempt += 1) {
+    const llmTrace = trace
+      ? beginLlmTrace(trace.onLlmCall, trace.operation, trace.label, trace.model, request, attempt, MAX_LLM_RETRY_ATTEMPTS)
+      : undefined;
+    try {
+      const completion = await client.chat.completions.create(request as never);
+      if (llmTrace) finishLlmTrace(trace?.onLlmCall, llmTrace, { response: completion });
+      return completion;
+    } catch (error) {
+      lastError = error;
+      if (llmTrace) finishLlmTrace(trace?.onLlmCall, llmTrace, { error });
+      if (attempt >= MAX_LLM_RETRY_ATTEMPTS || !shouldRetryLlmError(error)) {
+        throw error;
+      }
+      await sleep(retryDelayMs(attempt));
+    }
+  }
+  throw lastError;
+}
+
 async function createStructuredCompletion(
   client: OpenAI,
   model: string,
@@ -222,19 +292,16 @@ async function createStructuredCompletion(
     session_id: sessionId,
   };
   const primaryTrace = trace
-    ? beginLlmTrace(trace.onLlmCall, trace.operation, trace.label, model, structuredRequest as Record<string, unknown>)
+    ? { ...trace, model }
     : undefined;
   try {
-    const completion = await client.chat.completions.create(structuredRequest as never);
+    const completion = await createCompletionWithRetry(client, structuredRequest as Record<string, unknown>, primaryTrace);
-    if (primaryTrace) finishLlmTrace(trace?.onLlmCall, primaryTrace, { response: completion });
     return completion;
   } catch (error) {
     const message = String(error);
     if (!message.includes("400") && !message.toLowerCase().includes("provider returned error")) {
-      if (primaryTrace) finishLlmTrace(trace?.onLlmCall, primaryTrace, { error });
       throw error;
     }
-    if (primaryTrace) finishLlmTrace(trace?.onLlmCall, primaryTrace, { error });
 
     const fallbackRequest = {
       model,
@@ -245,16 +312,9 @@ async function createStructuredCompletion(
       session_id: sessionId,
     };
     const fallbackTrace = trace
-      ? beginLlmTrace(trace.onLlmCall, trace.operation, `${trace.label} (JSON fallback)`, model, fallbackRequest as Record<string, unknown>)
+      ? { ...trace, label: `${trace.label} (JSON fallback)`, model }
       : undefined;
-    try {
+    return createCompletionWithRetry(client, fallbackRequest as Record<string, unknown>, fallbackTrace);
-      const completion = await client.chat.completions.create(fallbackRequest as never);
-      if (fallbackTrace) finishLlmTrace(trace?.onLlmCall, fallbackTrace, { response: completion });
-      return completion;
-    } catch (fallbackError) {
-      if (fallbackTrace) finishLlmTrace(trace?.onLlmCall, fallbackTrace, { error: fallbackError });
-      throw fallbackError;
-    }
   }
 }
 
@@ -287,16 +347,9 @@ async function summarizeText(
     session_id: sessionId,
   };
   const llmTrace = trace
-    ? beginLlmTrace(trace.onLlmCall, "zusammenfassung", trace.label, model, request as Record<string, unknown>)
+    ? { operation: "zusammenfassung" as const, label: trace.label, model, onLlmCall: trace.onLlmCall }
     : undefined;
-  let completion: Awaited<ReturnType<typeof client.chat.completions.create>>;
+  const completion = await createCompletionWithRetry(client, request as Record<string, unknown>, llmTrace);
-  try {
-    completion = await client.chat.completions.create(request as never);
-    if (llmTrace) finishLlmTrace(trace?.onLlmCall, llmTrace, { response: completion });
-  } catch (error) {
-    if (llmTrace) finishLlmTrace(trace?.onLlmCall, llmTrace, { error });
-    throw error;
-  }
 
   const msg = completion.choices[0]?.message as Record<string, unknown> | undefined;
   const content = (msg?.content ?? msg?.reasoning) as string | null | undefined;
@@ -1142,6 +1195,7 @@ export async function summarizeCompany(
   }
 
   const data: ExtractedData = await file.json();
+  options.onCompanyInput?.(stem, data);
   const client = createClient();
   const output: SummaryData = structuredClone(data);
   const privacy: Pseudonymizer = createPseudonymizer(data);
@@ -1292,6 +1346,7 @@ export async function summarizeCompany(
   output._privacy = privacy.audit();
   output._schemaVersion = 3;
   output._summarizedAt = new Date().toISOString();
+  options.onCompanyOutput?.(stem, output);
   return output;
 }