feat(phase6b): LLM analysis layer — DeepSeek + insights + cost ledger + budget guard

Schema: - insights (fingerprint dedup, occurrence_count, related_session_ids[], priority_score) - cost_ledger (per-call, cache_hit/miss/output token split + USD) - prompt_templates (versioned, per-tag, with JSON schema + tier + temperature) - budget_settings (global key/value: monthly_hard_cap_usd, daily_soft/hard, per_call_max, analysis_paused) Worker: - lib/deepseek.ts: Anthropic-compat endpoint client (deepseek-v4-flash | deepseek-v4-pro), usage→USD with V4 promo pricing, cache_read_input_tokens awareness, extractJson() helper - lib/budget.ts: checkBudget() returns active|soft_throttled|hard_paused|monthly_paused, forces flash tier on soft cap, halts on hard/monthly cap or analysis_paused - lib/prompts.ts: 5 seed templates (bug_triage P, ux_friction F, payment_issue P, onboarding_stuck F, provider_quality F) with embedded JSON schemas + Sase.tr context; pickPromptTag() maps session tags → prompt - lib/json-validate.ts: lightweight schema validator (no ajv dep) - lib/seed-runtime.ts: idempotent upsert of prompts + default budget settings on worker boot - lib/minio.ts: +getText() for compressed timeline fetch - jobs/analyze.ts: budget guard → 6h fingerprint cache (attach session to existing insight) → template lookup → severity-based tier override → DeepSeek call → JSON parse + validate → insert insight (or aggregate occurrence) → write cost_ledger - scheduler: analyze@*/4min on insight-pipeline queue UI: - /insights (was pipeline view) → now Insight Inbox: priority-sorted list w/ KPI strip (new/in_backlog/shipped/today $/month $), severity badges, link to detail - /insights/i/[id]: insight detail with structured body render, related sessions, per-session cost breakdown, raw JSON collapsible - /insights/costs: KPI cards (today, month, avg, cache hit), daily 30d bar table, by-model + by-prompt breakdowns, top 10 expensive, recent errors - /insights/pipeline: moved old session-pipeline view here - /insights/sessions/[id]: unchanged session timeline viewer Defaults: - monthly cap $30, daily soft $1.50 / hard $3, per-call $0.20, analysis_paused=false - Severity→Tier: P0/P1=pro, P2/P3/INFO=flash; budget soft-cap forces flash Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-13 23:01:55 +00:00
parent 978b71d818
commit 95c7e8e321
15 changed files with 1723 additions and 61 deletions
--- a/apps/worker/src/index.ts
+++ b/apps/worker/src/index.ts
@@ -1,6 +1,7 @@
 import { startEventBus } from "./consumers/event-bus";
 import { startScheduledJobs } from "./schedulers/nightly";
 import { startInsightPipeline } from "./schedulers/pipeline";
+import { upsertSeedData } from "./lib/seed-runtime";
 import { redis } from "./redis";
 import { prisma } from "./db";

@@ -11,6 +12,8 @@ async function main() {
  await prisma.$queryRaw`SELECT 1`;
  console.log("[worker] panel-db ok");

+  await upsertSeedData().catch((e) => console.warn("[seed] failed:", e.message));
+
  await startScheduledJobs();
  await startInsightPipeline();
  await startEventBus();
--- a/apps/worker/src/jobs/analyze.ts
+++ b/apps/worker/src/jobs/analyze.ts
@@ -0,0 +1,273 @@
+import { prisma } from "../db";
+import { callDeepSeek, extractJson, type Tier, DeepSeekError } from "../lib/deepseek";
+import { checkBudget } from "../lib/budget";
+import { pickPromptTag } from "../lib/prompts";
+import { validate } from "../lib/json-validate";
+import { getText } from "../lib/minio";
+
+const ANALYZE_BATCH = Number(process.env.INSIGHT_ANALYZE_BATCH ?? "8");
+const CACHE_TTL_HOURS = Number(process.env.INSIGHT_INSIGHT_CACHE_HOURS ?? "6");
+const PROJECT_KEY = process.env.INSIGHT_PROJECT_KEY ?? "sase";
+const COMPRESSION_BUCKET = process.env.INSIGHT_COMPRESSED_BUCKET ?? "insight-compressed";
+
+export type AnalyzeResult = {
+  analyzed: number;
+  skipped: number;
+  failed: number;
+  costUsd: number;
+  budgetState: string;
+};
+
+export async function runAnalyze(): Promise<AnalyzeResult> {
+  const budget = await checkBudget();
+  if (!budget.allow) {
+    console.log(`[analyze] budget=${budget.state} ${budget.reason}`);
+    return { analyzed: 0, skipped: 0, failed: 0, costUsd: 0, budgetState: budget.state };
+  }
+
+  const compressed = await prisma.sessionMeta.findMany({
+    where: { status: "compressed", projectKey: PROJECT_KEY },
+    include: { compressed: true },
+    orderBy: { startedAt: "desc" },
+    take: ANALYZE_BATCH,
+  });
+  if (compressed.length === 0) {
+    return { analyzed: 0, skipped: 0, failed: 0, costUsd: 0, budgetState: budget.state };
+  }
+
+  let analyzed = 0;
+  let skipped = 0;
+  let failed = 0;
+  let costUsd = 0;
+
+  for (const s of compressed) {
+    if (!s.compressed || !s.fingerprint) {
+      skipped++;
+      await prisma.sessionMeta.update({ where: { id: s.id }, data: { status: "discarded" } });
+      continue;
+    }
+
+    // Application-level fingerprint cache: if an active insight with same fingerprint
+    // exists and is younger than CACHE_TTL_HOURS, just attach this session to it.
+    const cutoff = new Date(Date.now() - CACHE_TTL_HOURS * 3600_000);
+    const existing = await prisma.insight.findUnique({
+      where: { projectKey_fingerprint: { projectKey: PROJECT_KEY, fingerprint: s.fingerprint } },
+    });
+    if (existing && existing.updatedAt > cutoff && !["dismissed", "validated"].includes(existing.status)) {
+      // Aggregate this session into the existing insight
+      const rel = Array.from(new Set([...existing.relatedSessionIds, s.id]));
+      await prisma.insight.update({
+        where: { id: existing.id },
+        data: {
+          relatedSessionIds: rel,
+          occurrenceCount: rel.length,
+          lastSeenAt: s.startedAt > existing.lastSeenAt ? s.startedAt : existing.lastSeenAt,
+        },
+      });
+      await prisma.sessionMeta.update({
+        where: { id: s.id },
+        data: { status: "analyzed", processedAt: new Date() },
+      });
+      skipped++;
+      continue;
+    }
+
+    const promptTag = pickPromptTag(s.tags);
+    const template = await prisma.promptTemplate.findFirst({
+      where: { tag: promptTag, active: true },
+      orderBy: { version: "desc" },
+    });
+    if (!template) {
+      console.warn(`[analyze] no prompt for tag=${promptTag}`);
+      skipped++;
+      continue;
+    }
+
+    // Tier selection: severity-based override of template default
+    let tier: Tier = template.modelTier as Tier;
+    if (s.severity === "P0" || s.severity === "P1") tier = "pro";
+    else if (s.severity === "P2" || s.severity === "P3" || s.severity === "INFO") tier = "flash";
+    if (budget.forceTier) tier = budget.forceTier;
+
+    // Fetch timeline from MinIO
+    let timeline: string;
+    try {
+      timeline = await getText(COMPRESSION_BUCKET, s.compressed.semanticTimelineMinioKey);
+    } catch (e) {
+      console.warn(`[analyze] timeline fetch failed ${s.id}: ${(e as Error).message}`);
+      failed++;
+      continue;
+    }
+
+    const userPrompt = template.userPromptTemplate.replace("{{timeline}}", timeline);
+
+    let result;
+    try {
+      result = await callDeepSeek({
+        tier,
+        systemPrompt: template.systemPrompt,
+        userPrompt,
+        maxOutputTokens: template.maxOutputTokens,
+        temperature: template.temperature,
+      });
+    } catch (e) {
+      const status = e instanceof DeepSeekError ? e.status : 0;
+      const msg = (e as Error).message;
+      console.warn(`[analyze] deepseek error ${s.id}: ${status} ${msg}`);
+      await prisma.costLedger.create({
+        data: {
+          sessionId: s.id,
+          projectKey: PROJECT_KEY,
+          promptTag,
+          promptVersion: template.version,
+          provider: "deepseek",
+          model: tier === "pro" ? "deepseek-v4-pro" : "deepseek-v4-flash",
+          tier,
+          tokensInputCacheMiss: 0,
+          tokensInputCacheHit: 0,
+          tokensOutput: 0,
+          costInputCacheMissUsd: 0,
+          costInputCacheHitUsd: 0,
+          costOutputUsd: 0,
+          costTotalUsd: 0,
+          cacheHitRatio: 0,
+          errorCode: `${status}`,
+        },
+      });
+      failed++;
+      continue;
+    }
+
+    // Per-call cost cap
+    if (result.cost.totalUsd > budget.limits.perCallMax) {
+      console.warn(`[analyze] over per-call cap ${result.cost.totalUsd} for ${s.id}`);
+    }
+
+    // Parse + validate JSON
+    let parsed: any;
+    let validationErrors = "";
+    try {
+      parsed = JSON.parse(extractJson(result.text));
+      const errs = validate(parsed, template.outputSchemaJson as any);
+      if (errs.length) validationErrors = errs.map((e) => `${e.path}: ${e.message}`).join("; ");
+    } catch (e) {
+      validationErrors = `json parse: ${(e as Error).message}`;
+    }
+
+    costUsd += result.cost.totalUsd;
+    await prisma.costLedger.create({
+      data: {
+        sessionId: s.id,
+        projectKey: PROJECT_KEY,
+        promptTag,
+        promptVersion: template.version,
+        provider: "deepseek",
+        model: result.model,
+        tier,
+        tokensInputCacheMiss: result.usage.inputTokensMiss,
+        tokensInputCacheHit: result.usage.inputTokensHit,
+        tokensOutput: result.usage.outputTokens,
+        costInputCacheMissUsd: result.cost.inputMissUsd,
+        costInputCacheHitUsd: result.cost.inputHitUsd,
+        costOutputUsd: result.cost.outputUsd,
+        costTotalUsd: result.cost.totalUsd,
+        cacheHitRatio: result.cost.cacheHitRatio,
+        callDurationMs: result.durationMs,
+        errorCode: validationErrors ? "validation_failed" : null,
+      },
+    });
+
+    if (validationErrors) {
+      console.warn(`[analyze] validation failed ${s.id}: ${validationErrors.slice(0, 200)}`);
+      failed++;
+      continue;
+    }
+
+    // Compute severity from parsed (LLM may override) but fall back to session severity
+    const sev = String(parsed.severity ?? s.severity ?? "P3");
+    const conf = typeof parsed.confidence === "number" ? parsed.confidence : 0.5;
+    const title = String(parsed.title ?? s.tags.join(", ") ?? "Insight");
+    const type = String(parsed.type ?? promptTag);
+
+    // Priority score (10.3 in PRD)
+    const priorityScore = computePriorityScore({
+      severity: sev,
+      occurrenceCount: existing ? existing.occurrenceCount + 1 : 1,
+      ageHours: 0,
+      confidence: conf,
+    });
+
+    if (existing) {
+      const rel = Array.from(new Set([...existing.relatedSessionIds, s.id]));
+      // Existing insight in dismissed/validated state → mark new occurrence as potential regression
+      const isRegression = ["validated", "shipped"].includes(existing.status);
+      await prisma.insight.update({
+        where: { id: existing.id },
+        data: {
+          body: parsed,
+          title,
+          severity: sev,
+          status: isRegression ? "regressed" : existing.status,
+          relatedSessionIds: rel,
+          occurrenceCount: rel.length,
+          lastSeenAt: s.startedAt > existing.lastSeenAt ? s.startedAt : existing.lastSeenAt,
+          confidence: conf,
+          priorityScore,
+          sourceModel: result.model,
+          sourceCostUsd: existing.sourceCostUsd + result.cost.totalUsd,
+        },
+      });
+    } else {
+      await prisma.insight.create({
+        data: {
+          projectKey: PROJECT_KEY,
+          type,
+          severity: sev,
+          status: "new",
+          fingerprint: s.fingerprint,
+          title,
+          body: parsed,
+          relatedSessionIds: [s.id],
+          occurrenceCount: 1,
+          uniqueUserCount: 1,
+          firstSeenAt: s.startedAt,
+          lastSeenAt: s.startedAt,
+          confidence: conf,
+          priorityScore,
+          sourcePromptTag: promptTag,
+          sourcePromptVersion: template.version,
+          sourceModel: result.model,
+          sourceCostUsd: result.cost.totalUsd,
+        },
+      });
+    }
+
+    await prisma.sessionMeta.update({
+      where: { id: s.id },
+      data: { status: "analyzed", processedAt: new Date() },
+    });
+    analyzed++;
+
+    // Stop if budget became hard-capped mid-batch
+    const recheck = await checkBudget();
+    if (!recheck.allow) {
+      console.log(`[analyze] budget exhausted mid-batch (${recheck.state})`);
+      break;
+    }
+  }
+
+  return { analyzed, skipped, failed, costUsd, budgetState: budget.state };
+}
+
+function computePriorityScore(input: {
+  severity: string;
+  occurrenceCount: number;
+  ageHours: number;
+  confidence: number;
+}): number {
+  const severityWeight: Record<string, number> = { P0: 1, P1: 0.8, P2: 0.5, P3: 0.3, INFO: 0.1 };
+  const sw = severityWeight[input.severity] ?? 0.3;
+  const occ = Math.min(1.0, Math.log10(input.occurrenceCount + 1) / 2);
+  const recency = input.ageHours < 1 ? 1.0 : input.ageHours < 24 ? 0.7 : 0.4;
+  return Math.round(sw * 30 + occ * 25 + recency * 15 + input.confidence * 10);
+}
--- a/apps/worker/src/lib/budget.ts
+++ b/apps/worker/src/lib/budget.ts
@@ -0,0 +1,120 @@
+import { prisma } from "../db";
+
+export type BudgetState = "active" | "soft_throttled" | "hard_paused" | "monthly_paused";
+
+export type BudgetDecision = {
+  allow: boolean;
+  state: BudgetState;
+  reason: string;
+  forceTier?: "flash"; // when soft cap hit, force pro→flash downgrade
+  todayUsd: number;
+  monthUsd: number;
+  limits: {
+    monthlyHardCap: number;
+    dailySoftCap: number;
+    dailyHardCap: number;
+    perCallMax: number;
+  };
+};
+
+async function getNumber(key: string, fallback: number): Promise<number> {
+  const row = await prisma.budgetSetting.findFirst({
+    where: { projectKey: null, settingKey: key },
+  });
+  const v = row?.settingValue;
+  return typeof v === "number" ? v : fallback;
+}
+
+async function getBool(key: string, fallback: boolean): Promise<boolean> {
+  const row = await prisma.budgetSetting.findFirst({
+    where: { projectKey: null, settingKey: key },
+  });
+  const v = row?.settingValue;
+  return typeof v === "boolean" ? v : fallback;
+}
+
+function startOfDayUtc(d: Date): Date {
+  return new Date(Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), d.getUTCDate()));
+}
+function startOfMonthUtc(d: Date): Date {
+  return new Date(Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), 1));
+}
+
+export async function checkBudget(): Promise<BudgetDecision> {
+  const monthlyHardCap = await getNumber("monthly_hard_cap_usd", 30);
+  const dailySoftCap = await getNumber("daily_soft_cap_usd", 1.5);
+  const dailyHardCap = await getNumber("daily_hard_cap_usd", 3);
+  const perCallMax = await getNumber("per_call_max_usd", 0.2);
+  const paused = await getBool("analysis_paused", false);
+
+  const limits = { monthlyHardCap, dailySoftCap, dailyHardCap, perCallMax };
+
+  if (paused) {
+    return {
+      allow: false,
+      state: "hard_paused",
+      reason: "analysis_paused setting is true",
+      todayUsd: 0,
+      monthUsd: 0,
+      limits,
+    };
+  }
+
+  const now = new Date();
+  const todayStart = startOfDayUtc(now);
+  const monthStart = startOfMonthUtc(now);
+
+  const [todayAgg, monthAgg] = await Promise.all([
+    prisma.costLedger.aggregate({
+      where: { createdAt: { gte: todayStart } },
+      _sum: { costTotalUsd: true },
+    }),
+    prisma.costLedger.aggregate({
+      where: { createdAt: { gte: monthStart } },
+      _sum: { costTotalUsd: true },
+    }),
+  ]);
+  const todayUsd = Number(todayAgg._sum.costTotalUsd ?? 0);
+  const monthUsd = Number(monthAgg._sum.costTotalUsd ?? 0);
+
+  if (monthUsd >= monthlyHardCap) {
+    return {
+      allow: false,
+      state: "monthly_paused",
+      reason: `month spend $${monthUsd.toFixed(4)} >= monthly cap $${monthlyHardCap}`,
+      todayUsd,
+      monthUsd,
+      limits,
+    };
+  }
+  if (todayUsd >= dailyHardCap) {
+    return {
+      allow: false,
+      state: "hard_paused",
+      reason: `today spend $${todayUsd.toFixed(4)} >= daily hard cap $${dailyHardCap}`,
+      todayUsd,
+      monthUsd,
+      limits,
+    };
+  }
+  if (todayUsd >= dailySoftCap) {
+    return {
+      allow: true,
+      state: "soft_throttled",
+      reason: `today spend $${todayUsd.toFixed(4)} >= daily soft cap $${dailySoftCap}, force flash tier`,
+      forceTier: "flash",
+      todayUsd,
+      monthUsd,
+      limits,
+    };
+  }
+
+  return {
+    allow: true,
+    state: "active",
+    reason: "ok",
+    todayUsd,
+    monthUsd,
+    limits,
+  };
+}
--- a/apps/worker/src/lib/deepseek.ts
+++ b/apps/worker/src/lib/deepseek.ts
@@ -0,0 +1,138 @@
+// DeepSeek client using Anthropic-compatible endpoint.
+// https://api.deepseek.com/anthropic/v1/messages
+
+const BASE = process.env.DEEPSEEK_BASE_URL ?? "https://api.deepseek.com/anthropic";
+const KEY = process.env.DEEPSEEK_API_KEY ?? "";
+
+export type Tier = "flash" | "pro";
+
+// Pricing per million tokens (USD). V4 Pro is 75% off until 2026-05-31.
+// Override via DEEPSEEK_PRICING env (JSON) if rates change.
+const DEFAULT_PRICING: Record<Tier, { in_miss: number; in_hit: number; out: number }> = {
+  flash: { in_miss: 0.14, in_hit: 0.0028, out: 0.28 },
+  pro: { in_miss: 0.435, in_hit: 0.003625, out: 0.87 },
+};
+
+function pricing(): typeof DEFAULT_PRICING {
+  const raw = process.env.DEEPSEEK_PRICING;
+  if (!raw) return DEFAULT_PRICING;
+  try {
+    return JSON.parse(raw);
+  } catch {
+    return DEFAULT_PRICING;
+  }
+}
+
+export function modelForTier(tier: Tier): string {
+  if (tier === "pro") return process.env.DEEPSEEK_MODEL_PRO ?? "deepseek-v4-pro";
+  return process.env.DEEPSEEK_MODEL_FLASH ?? "deepseek-v4-flash";
+}
+
+export type CallResult = {
+  text: string;
+  usage: {
+    inputTokensMiss: number;
+    inputTokensHit: number;
+    outputTokens: number;
+  };
+  cost: {
+    inputMissUsd: number;
+    inputHitUsd: number;
+    outputUsd: number;
+    totalUsd: number;
+    cacheHitRatio: number;
+  };
+  model: string;
+  durationMs: number;
+};
+
+export class DeepSeekError extends Error {
+  constructor(public status: number, message: string) {
+    super(message);
+  }
+}
+
+export async function callDeepSeek(opts: {
+  tier: Tier;
+  systemPrompt: string;
+  userPrompt: string;
+  maxOutputTokens: number;
+  temperature?: number;
+  // When true, mark the system prompt for caching (DeepSeek auto-caches stable prefixes).
+}): Promise<CallResult> {
+  if (!KEY) throw new DeepSeekError(0, "DEEPSEEK_API_KEY not set");
+
+  const model = modelForTier(opts.tier);
+  const url = `${BASE}/v1/messages`;
+  const body = {
+    model,
+    max_tokens: opts.maxOutputTokens,
+    temperature: opts.temperature ?? 0.3,
+    system: opts.systemPrompt,
+    messages: [{ role: "user" as const, content: opts.userPrompt }],
+  };
+
+  const start = Date.now();
+  let res: Response;
+  try {
+    res = await fetch(url, {
+      method: "POST",
+      headers: {
+        "x-api-key": KEY,
+        "anthropic-version": "2023-06-01",
+        "content-type": "application/json",
+      },
+      body: JSON.stringify(body),
+    });
+  } catch (e) {
+    throw new DeepSeekError(0, `network: ${(e as Error).message}`);
+  }
+  const durationMs = Date.now() - start;
+  const text = await res.text();
+  if (!res.ok) throw new DeepSeekError(res.status, `${res.status}: ${text.slice(0, 200)}`);
+
+  let parsed: any;
+  try {
+    parsed = JSON.parse(text);
+  } catch {
+    throw new DeepSeekError(res.status, `non-json: ${text.slice(0, 200)}`);
+  }
+  const content = parsed?.content?.[0]?.text;
+  if (typeof content !== "string") {
+    throw new DeepSeekError(res.status, `no content text: ${text.slice(0, 200)}`);
+  }
+
+  const usage = parsed.usage ?? {};
+  const inputTokensMiss = Number(usage.input_tokens ?? 0) - Number(usage.cache_read_input_tokens ?? 0);
+  const inputTokensHit = Number(usage.cache_read_input_tokens ?? 0);
+  const outputTokens = Number(usage.output_tokens ?? 0);
+
+  const p = pricing()[opts.tier];
+  const inputMissUsd = (inputTokensMiss * p.in_miss) / 1_000_000;
+  const inputHitUsd = (inputTokensHit * p.in_hit) / 1_000_000;
+  const outputUsd = (outputTokens * p.out) / 1_000_000;
+  const totalUsd = inputMissUsd + inputHitUsd + outputUsd;
+  const totalIn = Math.max(1, inputTokensMiss + inputTokensHit);
+
+  return {
+    text: content,
+    usage: { inputTokensMiss, inputTokensHit, outputTokens },
+    cost: {
+      inputMissUsd,
+      inputHitUsd,
+      outputUsd,
+      totalUsd,
+      cacheHitRatio: inputTokensHit / totalIn,
+    },
+    model,
+    durationMs,
+  };
+}
+
+// Strip code fences if the model wraps JSON in ```json ... ```
+export function extractJson(raw: string): string {
+  const trimmed = raw.trim();
+  const fence = trimmed.match(/^```(?:json)?\s*\n([\s\S]*?)\n```\s*$/);
+  if (fence) return fence[1].trim();
+  return trimmed;
+}
--- a/apps/worker/src/lib/json-validate.ts
+++ b/apps/worker/src/lib/json-validate.ts
@@ -0,0 +1,73 @@
+// Lightweight JSON schema validator (subset used in prompt schemas).
+// Avoids pulling ajv into the worker bundle.
+
+type Schema = {
+  type?: string;
+  const?: unknown;
+  enum?: unknown[];
+  required?: string[];
+  properties?: Record<string, Schema>;
+  items?: Schema;
+  minItems?: number;
+  maxItems?: number;
+  minimum?: number;
+  maximum?: number;
+  maxLength?: number;
+};
+
+export type ValidationError = { path: string; message: string };
+
+export function validate(value: unknown, schema: Schema, path = "$"): ValidationError[] {
+  const errs: ValidationError[] = [];
+
+  if (schema.const !== undefined) {
+    if (value !== schema.const) errs.push({ path, message: `expected const ${JSON.stringify(schema.const)}` });
+    return errs;
+  }
+  if (schema.enum) {
+    if (!schema.enum.includes(value as any)) {
+      errs.push({ path, message: `not in enum ${JSON.stringify(schema.enum)}` });
+    }
+    return errs;
+  }
+  if (schema.type) {
+    const t = schema.type;
+    const actual =
+      value === null ? "null" : Array.isArray(value) ? "array" : typeof value;
+    if (t === "integer") {
+      if (actual !== "number" || !Number.isInteger(value)) errs.push({ path, message: "not integer" });
+    } else if (actual !== t) {
+      errs.push({ path, message: `expected ${t}, got ${actual}` });
+      return errs;
+    }
+  }
+  if (typeof value === "string" && schema.maxLength && value.length > schema.maxLength) {
+    errs.push({ path, message: `string > maxLength ${schema.maxLength}` });
+  }
+  if (typeof value === "number") {
+    if (schema.minimum !== undefined && value < schema.minimum) errs.push({ path, message: `< minimum` });
+    if (schema.maximum !== undefined && value > schema.maximum) errs.push({ path, message: `> maximum` });
+  }
+  if (Array.isArray(value)) {
+    if (schema.minItems !== undefined && value.length < schema.minItems)
+      errs.push({ path, message: `array < minItems ${schema.minItems}` });
+    if (schema.maxItems !== undefined && value.length > schema.maxItems)
+      errs.push({ path, message: `array > maxItems ${schema.maxItems}` });
+    if (schema.items) {
+      value.forEach((v, i) => errs.push(...validate(v, schema.items!, `${path}[${i}]`)));
+    }
+  }
+  if (schema.properties && value && typeof value === "object" && !Array.isArray(value)) {
+    const obj = value as Record<string, unknown>;
+    for (const [k, sub] of Object.entries(schema.properties)) {
+      if (k in obj) errs.push(...validate(obj[k], sub, `${path}.${k}`));
+    }
+    if (schema.required) {
+      for (const r of schema.required) {
+        if (!(r in obj)) errs.push({ path: `${path}.${r}`, message: "required" });
+      }
+    }
+  }
+
+  return errs;
+}
--- a/apps/worker/src/lib/minio.ts
+++ b/apps/worker/src/lib/minio.ts
@@ -40,3 +40,15 @@ export async function putText(
  const buf = Buffer.from(body, "utf-8");
  await c.putObject(bucket, key, buf, buf.length, { "Content-Type": contentType });
 }
+
+export async function getText(bucket: string, key: string): Promise<string> {
+  const c = getMinio();
+  if (!c) throw new Error("minio_not_configured");
+  const stream = await c.getObject(bucket, key);
+  const chunks: Buffer[] = [];
+  return await new Promise<string>((resolve, reject) => {
+    stream.on("data", (d) => chunks.push(d as Buffer));
+    stream.on("end", () => resolve(Buffer.concat(chunks).toString("utf-8")));
+    stream.on("error", reject);
+  });
+}
--- a/apps/worker/src/lib/prompts.ts
+++ b/apps/worker/src/lib/prompts.ts
@@ -0,0 +1,253 @@
+// Prompt templates with embedded JSON schemas (zod-like, validated by ajv-style logic).
+// Loaded from DB at runtime; this module also provides the seed catalog.
+
+export type PromptTier = "flash" | "pro";
+
+export type PromptTemplate = {
+  tag: string;
+  version: number;
+  name: string;
+  systemPrompt: string;
+  userPromptTemplate: string;
+  outputSchemaJson: object;
+  modelTier: PromptTier;
+  maxOutputTokens: number;
+  temperature: number;
+};
+
+const SASE_CONTEXT = `Sase.tr context:
+- B2B SaaS — VIN lookup + parts compatibility for Turkish auto-parts dealers and service shops.
+- Stack: NestJS backend, Next.js frontend, PostgreSQL/Prisma.
+- 4 upstream providers: PL24 (Partslink24), PCAT, RMEX, TecDoc. Each can timeout/fail independently.
+- Auth: JWT, sticky session routing.
+- Subscription tiers: starter, brand_specific, full. Trial flow exists.
+- Common URL paths: /vin-lookup, /dashboard/search, /dashboard/catalog/<brand>/<vehicleId>, /api-keys, /subscription, /upgrade, /pricing.
+
+Output rules:
+- Return ONLY valid JSON matching the schema. No markdown, no code fences, no preamble.
+- If you are uncertain, lower the confidence; do not invent specifics.
+- All string fields are concise (titles ≤120 chars, hypothesis ≤500 chars).`;
+
+const BUG_TRIAGE_SCHEMA = {
+  type: "object",
+  required: ["type", "severity", "title", "reproduce_steps", "hypothesis", "confidence"],
+  properties: {
+    type: { const: "bug" },
+    severity: { enum: ["P0", "P1", "P2", "P3"] },
+    title: { type: "string", maxLength: 120 },
+    reproduce_steps: { type: "array", items: { type: "string" }, minItems: 2, maxItems: 10 },
+    affected_route: { type: "string" },
+    affected_component_hypothesis: { type: "string" },
+    error_signature: { type: "string" },
+    hypothesis: { type: "string", maxLength: 500 },
+    suggested_investigation: { type: "array", items: { type: "string" } },
+    suggested_fix_effort: { enum: ["S", "M", "L"] },
+    confidence: { type: "number", minimum: 0, maximum: 1 },
+    user_impact_estimate: { type: "string" },
+    is_likely_provider_issue: { type: "boolean" },
+    implicated_provider: { enum: ["PL24", "PCAT", "RMEX", "TecDoc", null] },
+  },
+};
+
+const UX_FRICTION_SCHEMA = {
+  type: "object",
+  required: ["type", "severity", "title", "intent_hypothesis", "friction_point", "quick_fix", "confidence"],
+  properties: {
+    type: { const: "ux_friction" },
+    severity: { enum: ["P1", "P2", "P3"] },
+    title: { type: "string", maxLength: 120 },
+    intent_hypothesis: { type: "string", maxLength: 300 },
+    friction_point: { type: "string", maxLength: 300 },
+    friction_type: { enum: ["copy", "layout", "affordance", "performance", "terminology", "discoverability"] },
+    quick_fix: { type: "string", maxLength: 300 },
+    long_term_fix: { type: "string", maxLength: 500 },
+    affected_user_segment: { type: "string" },
+    confidence: { type: "number", minimum: 0, maximum: 1 },
+  },
+};
+
+const PAYMENT_SCHEMA = {
+  type: "object",
+  required: ["type", "severity", "title", "failure_mode", "hypothesis", "confidence"],
+  properties: {
+    type: { const: "payment" },
+    severity: { enum: ["P0", "P1", "P2"] },
+    title: { type: "string", maxLength: 120 },
+    failure_mode: { enum: ["ui_silent", "backend_decline", "user_abandoned", "checkout_friction", "validation_failed", "trial_to_paid_lost"] },
+    hypothesis: { type: "string", maxLength: 500 },
+    payment_method_hint: { type: "string" },
+    suggested_investigation: { type: "array", items: { type: "string" } },
+    business_impact_estimate: { type: "string" },
+    confidence: { type: "number", minimum: 0, maximum: 1 },
+  },
+};
+
+const ONBOARDING_SCHEMA = {
+  type: "object",
+  required: ["type", "severity", "title", "block_point", "hypothesis", "confidence"],
+  properties: {
+    type: { const: "onboarding" },
+    severity: { enum: ["P1", "P2", "P3"] },
+    title: { type: "string", maxLength: 120 },
+    block_point: { type: "string", maxLength: 200 },
+    unclear_concept: { type: "string" },
+    documentation_gap: { type: "string" },
+    suggested_in_app_help: { type: "string" },
+    hypothesis: { type: "string", maxLength: 500 },
+    confidence: { type: "number", minimum: 0, maximum: 1 },
+  },
+};
+
+const PROVIDER_SCHEMA = {
+  type: "object",
+  required: ["type", "severity", "title", "affected_provider", "hypothesis", "confidence"],
+  properties: {
+    type: { const: "provider_quality" },
+    severity: { enum: ["P1", "P2", "P3"] },
+    title: { type: "string", maxLength: 120 },
+    affected_provider: { enum: ["PL24", "PCAT", "RMEX", "TecDoc", "multi"] },
+    failure_mode: { enum: ["timeout", "incomplete_data", "wrong_data", "auth_error", "ratelimit", "unknown"] },
+    hypothesis: { type: "string", maxLength: 500 },
+    suggested_action: { type: "string", maxLength: 300 },
+    confidence: { type: "number", minimum: 0, maximum: 1 },
+  },
+};
+
+export const SEED_PROMPTS: PromptTemplate[] = [
+  {
+    tag: "bug_triage",
+    version: 1,
+    name: "Bug Triage v1",
+    systemPrompt: `You analyze bugs found in user sessions of a B2B SaaS. Extract reproducible steps, locate the failing component, and propose a hypothesis.
+
+${SASE_CONTEXT}
+
+Schema (return JSON conforming exactly):
+${JSON.stringify(BUG_TRIAGE_SCHEMA)}`,
+    userPromptTemplate: `{{timeline}}
+
+This session contains a bug. Return JSON per the schema.`,
+    outputSchemaJson: BUG_TRIAGE_SCHEMA,
+    modelTier: "pro",
+    maxOutputTokens: 1200,
+    temperature: 0.2,
+  },
+  {
+    tag: "ux_friction",
+    version: 1,
+    name: "UX Friction v1",
+    systemPrompt: `You are a senior UX engineer. Given a session timeline, identify where the user got stuck and propose a quick fix and a long-term fix.
+
+${SASE_CONTEXT}
+
+Schema:
+${JSON.stringify(UX_FRICTION_SCHEMA)}`,
+    userPromptTemplate: `{{timeline}}
+
+Return JSON per the schema.`,
+    outputSchemaJson: UX_FRICTION_SCHEMA,
+    modelTier: "flash",
+    maxOutputTokens: 1000,
+    temperature: 0.3,
+  },
+  {
+    tag: "payment_issue",
+    version: 1,
+    name: "Payment Issue v1",
+    systemPrompt: `You analyze payment/conversion failures in a B2B SaaS. Identify failure mode and likely cause.
+
+${SASE_CONTEXT}
+
+Schema:
+${JSON.stringify(PAYMENT_SCHEMA)}`,
+    userPromptTemplate: `{{timeline}}
+
+Return JSON per the schema.`,
+    outputSchemaJson: PAYMENT_SCHEMA,
+    modelTier: "pro",
+    maxOutputTokens: 1000,
+    temperature: 0.2,
+  },
+  {
+    tag: "onboarding_stuck",
+    version: 1,
+    name: "Onboarding Stuck v1",
+    systemPrompt: `You analyze new-user onboarding sessions where the user did not reach first value. Propose where they got stuck and what help is missing.
+
+${SASE_CONTEXT}
+
+Schema:
+${JSON.stringify(ONBOARDING_SCHEMA)}`,
+    userPromptTemplate: `{{timeline}}
+
+Return JSON per the schema.`,
+    outputSchemaJson: ONBOARDING_SCHEMA,
+    modelTier: "flash",
+    maxOutputTokens: 900,
+    temperature: 0.3,
+  },
+  {
+    tag: "provider_quality",
+    version: 1,
+    name: "Provider Quality v1",
+    systemPrompt: `You analyze upstream provider failures (PL24/PCAT/RMEX/TecDoc) impacting Sase.tr users. Identify which provider failed and propose action.
+
+${SASE_CONTEXT}
+
+Schema:
+${JSON.stringify(PROVIDER_SCHEMA)}`,
+    userPromptTemplate: `{{timeline}}
+
+Return JSON per the schema.`,
+    outputSchemaJson: PROVIDER_SCHEMA,
+    modelTier: "flash",
+    maxOutputTokens: 800,
+    temperature: 0.2,
+  },
+];
+
+// Maps session tags → which prompt template to use. First match wins.
+// Severity routing (P0/P1 → pro, P2/P3/INFO → flash) is layered on top of the template's modelTier.
+export function pickPromptTag(tags: string[]): string {
+  const set = new Set(tags);
+  // Payment first (highest business impact)
+  if (
+    set.has("payment_ui_silent_failure") ||
+    set.has("payment_friction") ||
+    set.has("payment_failed_session") ||
+    set.has("checkout_abandonment") ||
+    set.has("downgrade_pending")
+  )
+    return "payment_issue";
+
+  // Provider issues
+  if (
+    set.has("provider_reliability_issue") ||
+    set.has("provider_mismatch") ||
+    set.has("vin_decode_fail_pattern") ||
+    set.has("vin_decode_repeated_failure")
+  )
+    return "provider_quality";
+
+  // Bugs
+  if (set.has("bug_suspected") || set.has("server_error_impact")) return "bug_triage";
+
+  // Onboarding
+  if (set.has("onboarding_stuck")) return "onboarding_stuck";
+
+  // UX friction
+  if (
+    set.has("ux_friction") ||
+    set.has("upgrade_hesitation") ||
+    set.has("search_validation_friction") ||
+    set.has("api_key_friction") ||
+    set.has("webhook_setup_struggle") ||
+    set.has("compatibility_quality_gap") ||
+    set.has("parts_export_abandoned") ||
+    set.has("vin_decode_no_outcome") ||
+    set.has("vin_decode_failed_single")
+  )
+    return "ux_friction";
+
+  return "ux_friction"; // safe default
+}
--- a/apps/worker/src/lib/seed-runtime.ts
+++ b/apps/worker/src/lib/seed-runtime.ts
@@ -0,0 +1,55 @@
+import { prisma } from "../db";
+import { SEED_PROMPTS } from "./prompts";
+
+const DEFAULT_BUDGETS: Array<{ key: string; value: unknown }> = [
+  { key: "monthly_hard_cap_usd", value: 30 },
+  { key: "daily_soft_cap_usd", value: 1.5 },
+  { key: "daily_hard_cap_usd", value: 3 },
+  { key: "per_call_max_usd", value: 0.2 },
+  { key: "min_score_for_analysis", value: 30 },
+  { key: "cache_ttl_hours", value: 6 },
+  { key: "analysis_paused", value: false },
+];
+
+export async function upsertSeedData(): Promise<void> {
+  // Prompt templates — only insert if (tag, version) doesn't exist.
+  for (const p of SEED_PROMPTS) {
+    const existing = await prisma.promptTemplate.findUnique({
+      where: { tag_version: { tag: p.tag, version: p.version } },
+    });
+    if (!existing) {
+      await prisma.promptTemplate.create({
+        data: {
+          tag: p.tag,
+          version: p.version,
+          name: p.name,
+          systemPrompt: p.systemPrompt,
+          userPromptTemplate: p.userPromptTemplate,
+          outputSchemaJson: p.outputSchemaJson as object,
+          modelTier: p.modelTier,
+          maxOutputTokens: p.maxOutputTokens,
+          temperature: p.temperature,
+          active: true,
+        },
+      });
+      console.log(`[seed] inserted prompt ${p.tag}@v${p.version}`);
+    }
+  }
+
+  // Budget settings — only insert if missing (don't overwrite user changes).
+  for (const b of DEFAULT_BUDGETS) {
+    const existing = await prisma.budgetSetting.findFirst({
+      where: { projectKey: null, settingKey: b.key },
+    });
+    if (!existing) {
+      await prisma.budgetSetting.create({
+        data: {
+          projectKey: null,
+          settingKey: b.key,
+          settingValue: b.value as object,
+        },
+      });
+      console.log(`[seed] inserted budget ${b.key}=${JSON.stringify(b.value)}`);
+    }
+  }
+}
--- a/apps/worker/src/schedulers/pipeline.ts
+++ b/apps/worker/src/schedulers/pipeline.ts
@@ -3,6 +3,7 @@ import { redis } from "../redis";
 import { runPostHogIngest } from "../jobs/posthog-ingest";
 import { runTagSessions } from "../jobs/tag-sessions";
 import { runCompressSessions } from "../jobs/compress-sessions";
+import { runAnalyze } from "../jobs/analyze";

 const QUEUE = "insight-pipeline";

@@ -27,6 +28,15 @@ async function runJob(job: Job) {
      if (res.compressed + res.failed > 0) console.log(`[pipeline] compress ok=${res.compressed} fail=${res.failed}`);
      return res;
    }
+    case "analyze": {
+      const res = await runAnalyze();
+      if (res.analyzed + res.skipped + res.failed > 0 || res.budgetState !== "active") {
+        console.log(
+          `[pipeline] analyze ok=${res.analyzed} skip=${res.skipped} fail=${res.failed} cost=$${res.costUsd.toFixed(4)} budget=${res.budgetState}`,
+        );
+      }
+      return res;
+    }
    default:
      return { ok: false, error: `unknown job ${job.name}` };
  }
@@ -48,6 +58,11 @@ export async function startInsightPipeline() {
    { pattern: "*/3 * * * *" },
    { name: "compress-sessions", data: {}, opts: { removeOnComplete: 50, removeOnFail: 25 } },
  );
+  await queue.upsertJobScheduler(
+    "analyze",
+    { pattern: "*/4 * * * *" },
+    { name: "analyze", data: {}, opts: { removeOnComplete: 50, removeOnFail: 25 } },
+  );

  new Worker(QUEUE, runJob, {
    connection: redis,
@@ -55,5 +70,7 @@ export async function startInsightPipeline() {
    lockDuration: 5 * 60_000,
    stalledInterval: 60_000,
  });
-  console.log("[pipeline] armed: posthog-ingest@*/5min, tag-sessions@*/2min, compress-sessions@*/3min");
+  console.log(
+    "[pipeline] armed: posthog-ingest@*/5min, tag-sessions@*/2min, compress-sessions@*/3min, analyze@*/4min",
+  );
 }