apps/web/prisma/schema.prisma

@@ -251,6 +251,11 @@ model Insight {
   severity              String
   status                String    @default("new")
   fingerprint           String
+  // Additional fingerprints semantically merged into this insight by the dedup
+  // gate: a new session whose own fingerprint differs but which the LLM judged
+  // to be the SAME underlying problem gets its fingerprint aliased here, so the
+  // next identical session fast-paths (no re-analysis, no duplicate row).
+  aliasFingerprints     String[]  @default([])
   title                 String
   body                  Json
   relatedSessionIds     String[]

apps/worker/package.json

@@ -5,7 +5,8 @@
   "scripts": {
     "dev": "tsx watch src/index.ts",
     "start": "tsx src/index.ts",
-    "typecheck": "tsc --noEmit"
+    "typecheck": "tsc --noEmit",
+    "test": "tsx src/lib/dedup.smoke.ts"
   },
   "dependencies": {
     "@panel/web": "workspace:*",

apps/worker/src/jobs/analyze.ts

@@ -1,10 +1,18 @@
 import { prisma } from "../db";
-import { callDeepSeek, extractJson, type Tier, DeepSeekError } from "../lib/deepseek";
+import { callDeepSeek, extractJson, type Tier, type CallResult, 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";
 import { alertP0Insight, alertBudgetCap } from "../lib/telegram";
+import {
+  buildInsightCatalog,
+  findSemanticMatch,
+  classifyExistingAction,
+  shouldAcceptMatch,
+  DEDUP_ENABLED,
+  DEDUP_THRESHOLD,
+} from "../lib/dedup";
 
 const PANEL_URL = process.env.PANEL_PUBLIC_URL ?? "https://sp.semih.ai";
 
@@ -22,6 +30,88 @@ export type AnalyzeResult = {
   budgetState: string;
 };
 
+type GroupSession = { id: string; startedAt: Date };
+
+type ExistingInsight = {
+  id: string;
+  status: string;
+  relatedSessionIds: string[];
+  lastSeenAt: Date;
+  updatedAt: Date;
+  aliasFingerprints: string[];
+};
+
+async function markAnalyzed(ids: string[]): Promise<void> {
+  await prisma.sessionMeta.updateMany({
+    where: { id: { in: ids } },
+    data: { status: "analyzed", processedAt: new Date() },
+  });
+}
+
+/**
+ * Attach a session group to an existing insight WITHOUT re-running the LLM.
+ * Returns true when fully handled (caller should `continue`). Returns false
+ * only for an active-but-stale insight when the caller allows a refresh
+ * re-analysis (Layer 1, same fingerprint) — the caller then falls through to
+ * the normal analysis/update path.
+ *
+ *  - dismissed / duplicate → suppress silently (respect the triage decision)
+ *  - validated / shipped    → flag regression (cheap, no re-analysis)
+ *  - active & recent        → attach occurrence
+ */
+async function attachToExisting(
+  existing: ExistingInsight,
+  group: GroupSession[],
+  opts: { allowStaleReanalyze: boolean; cutoff: Date },
+): Promise<boolean> {
+  const action = classifyExistingAction(
+    existing.status,
+    existing.updatedAt <= opts.cutoff,
+    opts.allowStaleReanalyze,
+  );
+  if (action === "reanalyze") return false; // active & stale → caller refreshes
+
+  const rel = Array.from(new Set([...existing.relatedSessionIds, ...group.map((g) => g.id)]));
+  const groupLast = group.reduce((a, b) => (a.startedAt > b.startedAt ? a : b)).startedAt;
+  const lastSeenAt = groupLast > existing.lastSeenAt ? groupLast : existing.lastSeenAt;
+
+  await prisma.insight.update({
+    where: { id: existing.id },
+    data: {
+      relatedSessionIds: rel,
+      occurrenceCount: rel.length,
+      lastSeenAt,
+      ...(action === "regress" ? { status: "regressed", regressionDetected: true } : {}),
+    },
+  });
+  await markAnalyzed(group.map((g) => g.id));
+  return true;
+}
+
+async function logDedupCost(sessionId: string, projectKey: string, raw: CallResult): Promise<void> {
+  await prisma.costLedger.create({
+    data: {
+      sessionId,
+      projectKey,
+      promptTag: "dedup_gate",
+      promptVersion: 0,
+      provider: "deepseek",
+      model: raw.model,
+      tier: "flash",
+      tokensInputCacheMiss: raw.usage.inputTokensMiss,
+      tokensInputCacheHit: raw.usage.inputTokensHit,
+      tokensOutput: raw.usage.outputTokens,
+      costInputCacheMissUsd: raw.cost.inputMissUsd,
+      costInputCacheHitUsd: raw.cost.inputHitUsd,
+      costOutputUsd: raw.cost.outputUsd,
+      costTotalUsd: raw.cost.totalUsd,
+      cacheHitRatio: raw.cost.cacheHitRatio,
+      callDurationMs: raw.durationMs,
+      errorCode: null,
+    },
+  });
+}
+
 export async function runAnalyze(): Promise<AnalyzeResult> {
   const budget = await checkBudget();
   if (!budget.allow) {
@@ -72,31 +162,73 @@ export async function runAnalyze(): Promise<AnalyzeResult> {
       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.
+    // ─── Layer 1: exact fingerprint (or aliased fingerprint) fast-path ───
+    // A session whose fingerprint already maps to an insight is attached without
+    // re-running the LLM: dismissed/duplicate SUPPRESS it (respect triage — no
+    // new row, no cost), validated/shipped FLAG a regression, active+recent just
+    // accumulate. Only active+stale falls through to a refresh re-analysis.
     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 } },
+    const existing = await prisma.insight.findFirst({
+      where: {
+        projectKey: PROJECT_KEY,
+        OR: [{ fingerprint: s.fingerprint }, { aliasFingerprints: { has: 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++;
+    if (existing) {
+      const handled = await attachToExisting(existing, group, { allowStaleReanalyze: true, cutoff });
+      if (handled) {
+        skipped++;
+        continue;
+      }
+      // active & stale → fall through to full re-analysis (existing stays set).
+    }
+
+    // Fetch timeline from MinIO — needed by both the dedup gate and the analysis.
+    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;
     }
 
+    // ─── Layer 2: semantic dedup gate (only for genuinely new fingerprints) ───
+    // Before spending the expensive analysis call, ask the flash model whether
+    // this session is the SAME underlying problem as an existing insight —
+    // including dismissed/duplicate ones we must NOT recreate. On a confident
+    // match we alias this fingerprint onto that insight and attach/suppress, so
+    // future identical sessions fast-path with no LLM and no duplicate row.
+    if (!existing && DEDUP_ENABLED) {
+      const catalog = await buildInsightCatalog(PROJECT_KEY);
+      const match = await findSemanticMatch({ timeline, tags: s.tags, catalog });
+      if (match.raw) {
+        costUsd += match.raw.cost.totalUsd;
+        await logDedupCost(s.id, PROJECT_KEY, match.raw);
+      }
+      const catalogIds = new Set(catalog.map((c) => c.id));
+      if (match.matchId && shouldAcceptMatch(match.matchId, match.confidence, catalogIds, DEDUP_THRESHOLD)) {
+        const matchId = match.matchId;
+        const matched = await prisma.insight.findUnique({ where: { id: matchId } });
+        if (matched) {
+          await prisma.insight.update({
+            where: { id: matched.id },
+            data: {
+              aliasFingerprints: Array.from(new Set([...matched.aliasFingerprints, s.fingerprint])),
+            },
+          });
+          await attachToExisting(matched, group, { allowStaleReanalyze: false, cutoff });
+          console.log(
+            `[analyze] dedup-gate matched ${s.fingerprint} → ${matched.id} ` +
+              `(${matched.status}) conf=${match.confidence.toFixed(2)} :: ${match.reason}`,
+          );
+          skipped++;
+          continue;
+        }
+      }
+      // no confident match → fall through and create a new insight below.
+    }
+
     const promptTag = isBundle ? "pattern_bundle" : pickPromptTag(s.tags);
     const template = await prisma.promptTemplate.findFirst({
       where: { tag: promptTag, active: true },
@@ -114,16 +246,6 @@ export async function runAnalyze(): Promise<AnalyzeResult> {
     else if (s.severity === "P2" || s.severity === "P3" || s.severity === "INFO") tier = "flash";
     if (budget.forceTier) tier = budget.forceTier;
 
-    // Fetch timeline from MinIO (primary)
-    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;
-    }
-
     // For bundles, append a summary of the other sessions in the group.
     if (isBundle) {
       const otherSessionIds = group.slice(1).map((x) => x.id);

apps/worker/src/lib/dedup.smoke.ts

@@ -0,0 +1,57 @@
+/**
+ * Smoke test for the insight dedup decision logic (pure helpers, no I/O).
+ * Run: pnpm --filter worker exec tsx src/lib/dedup.smoke.ts
+ *
+ * Covers the two new guarantees:
+ *   1. No duplicate insights — same-fingerprint / aliased sessions never create
+ *      a new row (attach/reanalyze), and a confident semantic match attaches.
+ *   2. No insights similar to dismissed ones — dismissed/duplicate fingerprints
+ *      are SUPPRESSED, never re-opened.
+ */
+import { classifyExistingAction, shouldAcceptMatch, type ExistingAction } from "./dedup";
+
+let pass = 0;
+let fail = 0;
+function eq<T>(label: string, got: T, want: T): void {
+  if (got === want) {
+    pass++;
+  } else {
+    fail++;
+    console.error(`✗ ${label}: got ${JSON.stringify(got)} want ${JSON.stringify(want)}`);
+  }
+}
+
+// ─── classifyExistingAction(status, isStale, allowStaleReanalyze) ───
+const cea = (s: string, stale: boolean, allow: boolean): ExistingAction =>
+  classifyExistingAction(s, stale, allow);
+
+// Suppressed themes are never re-opened, regardless of staleness/allow flag.
+eq("dismissed → suppress (fresh)", cea("dismissed", false, true), "suppress");
+eq("dismissed → suppress (stale)", cea("dismissed", true, true), "suppress");
+eq("dismissed → suppress (gate path)", cea("dismissed", true, false), "suppress");
+eq("duplicate → suppress", cea("duplicate", true, true), "suppress");
+
+// Fixed themes that recur are flagged as regressions.
+eq("validated → regress", cea("validated", false, true), "regress");
+eq("shipped → regress (stale)", cea("shipped", true, true), "regress");
+
+// Active themes accumulate; stale ones refresh only on the Layer-1 path.
+eq("new fresh → attach", cea("new", false, true), "attach");
+eq("new stale (Layer1) → reanalyze", cea("new", true, true), "reanalyze");
+eq("new stale (gate, no-reanalyze) → attach", cea("new", true, false), "attach");
+eq("in_backlog fresh → attach", cea("in_backlog", false, true), "attach");
+eq("in_progress stale (Layer1) → reanalyze", cea("in_progress", true, true), "reanalyze");
+eq("regressed fresh → attach", cea("regressed", false, true), "attach");
+eq("triaged stale (gate) → attach", cea("triaged", true, false), "attach");
+
+// ─── shouldAcceptMatch(matchId, confidence, catalogIds, threshold) ───
+const cat = new Set(["cm_a", "cm_b", "cm_c"]);
+const T = 0.72;
+eq("null match → reject", shouldAcceptMatch(null, 0.99, cat, T), false);
+eq("hallucinated id → reject", shouldAcceptMatch("cm_zzz", 0.99, cat, T), false);
+eq("in-catalog + above threshold → accept", shouldAcceptMatch("cm_a", 0.8, cat, T), true);
+eq("in-catalog + at threshold → accept", shouldAcceptMatch("cm_b", 0.72, cat, T), true);
+eq("in-catalog + below threshold → reject", shouldAcceptMatch("cm_c", 0.71, cat, T), false);
+
+console.log(`\ndedup smoke: ${pass} passed, ${fail} failed`);
+if (fail > 0) process.exit(1);

apps/worker/src/lib/dedup.ts

@@ -0,0 +1,160 @@
+import { prisma } from "../db";
+import { callDeepSeek, extractJson, type CallResult } from "./deepseek";
+
+// ─── Status families (kept in sync with analyze.ts) ───
+export const ACTIVE_STATUSES = ["new", "triaged", "in_backlog", "in_progress", "regressed"];
+export const SUPPRESSED_STATUSES = ["dismissed", "duplicate"];
+export const FIXED_STATUSES = ["validated", "shipped"];
+
+// ─── Tunables ───
+export const DEDUP_ENABLED = (process.env.INSIGHT_DEDUP_GATE ?? "true") !== "false";
+export const DEDUP_THRESHOLD = Number(process.env.INSIGHT_DEDUP_THRESHOLD ?? "0.72");
+const SUPPRESS_LOOKBACK_DAYS = Number(process.env.INSIGHT_DEDUP_SUPPRESS_DAYS ?? "90");
+const CATALOG_MAX = Number(process.env.INSIGHT_DEDUP_CATALOG_MAX ?? "120");
+
+export type CatalogEntry = { id: string; title: string; type: string; status: string };
+
+// ─── Pure decision helpers (unit-testable, no I/O) ───
+
+export type ExistingAction = "suppress" | "regress" | "attach" | "reanalyze";
+
+/**
+ * Decide what to do when a session's fingerprint already maps to an insight.
+ *   suppress  — dismissed/duplicate: bump counters, keep status (no LLM)
+ *   regress   — validated/shipped: flag regression (no LLM)
+ *   reanalyze — active but stale: caller re-runs the LLM to refresh
+ *   attach    — active and fresh: bump counters (no LLM)
+ */
+export function classifyExistingAction(
+  status: string,
+  isStale: boolean,
+  allowStaleReanalyze: boolean,
+): ExistingAction {
+  if (SUPPRESSED_STATUSES.includes(status)) return "suppress";
+  if (FIXED_STATUSES.includes(status)) return "regress";
+  if (allowStaleReanalyze && isStale) return "reanalyze";
+  return "attach";
+}
+
+/**
+ * Decide whether to accept a dedup-gate match: the id must exist in the catalog
+ * (anti-hallucination) and confidence must clear the threshold.
+ */
+export function shouldAcceptMatch(
+  matchId: string | null,
+  confidence: number,
+  catalogIds: Set<string>,
+  threshold: number,
+): boolean {
+  return !!matchId && catalogIds.has(matchId) && confidence >= threshold;
+}
+
+/**
+ * Build the dedup catalog the LLM gate compares a new session against:
+ *   - ALL currently-active insights (dedup target → attach).
+ *   - Recently-touched suppressed/fixed insights (dismissed / duplicate /
+ *     validated / shipped) within the lookback window (suppression target →
+ *     don't recreate a known non-issue / flag regression).
+ * Active entries are prioritised, then most-recent others, capped at CATALOG_MAX
+ * so the prompt stays bounded as the dismissed pile grows over time.
+ */
+export async function buildInsightCatalog(projectKey: string): Promise<CatalogEntry[]> {
+  const since = new Date(Date.now() - SUPPRESS_LOOKBACK_DAYS * 86_400_000);
+  const rows = await prisma.insight.findMany({
+    where: {
+      projectKey,
+      OR: [
+        { status: { in: ACTIVE_STATUSES } },
+        { status: { in: [...SUPPRESSED_STATUSES, ...FIXED_STATUSES] }, updatedAt: { gte: since } },
+      ],
+    },
+    select: { id: true, title: true, type: true, status: true },
+    orderBy: [{ updatedAt: "desc" }],
+  });
+  const active = rows.filter((r) => ACTIVE_STATUSES.includes(r.status));
+  const rest = rows.filter((r) => !ACTIVE_STATUSES.includes(r.status));
+  return [...active, ...rest].slice(0, CATALOG_MAX);
+}
+
+const SYSTEM_PROMPT = `You are a strict deduplication classifier for a product-analytics insight pipeline (Sase.tr — a Turkish car-parts catalog SaaS where users decode a VIN, browse a category tree / parts panel, and subscribe via Stripe).
+
+You receive ONE new user-session signal and a CATALOG of existing insights, each tagged with a status. Decide whether the new session describes the SAME underlying product problem as exactly one catalog entry.
+
+"Same problem" means: same root cause AND same user-facing failure on the same surface/flow — e.g. both are "category-tree navigation produces rage-clicks", or both are "VIN decode upstream provider timeout".
+NOT the same: merely sharing a page, a severity, or the generic signal "ux_friction". Distinct sub-problems on the same page are DIFFERENT (e.g. a broken schema IMAGE vs. a slow category LIST vs. a parts panel that won't SELECT are three different insights).
+
+Matching an entry whose status is "dismissed" or "duplicate" is expected and valuable — it means this is a known non-issue we must NOT recreate. Matching "validated"/"shipped" means a previously-fixed problem may have regressed.
+
+Return ONLY JSON: {"match_insight_id": "<id from catalog, or null>", "confidence": <0..1>, "reason": "<one short sentence>"}.
+Be conservative: when in doubt, return null with low confidence. Never invent an id that is not in the catalog.`;
+
+export type DedupResult = {
+  matchId: string | null;
+  confidence: number;
+  reason: string;
+  raw: CallResult | null;
+};
+
+/**
+ * Ask the flash model whether this session matches an existing insight.
+ * Returns matchId=null if the catalog is empty, parsing fails, or the model
+ * returns an id that is not actually in the catalog (anti-hallucination guard).
+ */
+export async function findSemanticMatch(input: {
+  timeline: string;
+  tags: string[];
+  catalog: CatalogEntry[];
+}): Promise<DedupResult> {
+  if (input.catalog.length === 0) {
+    return { matchId: null, confidence: 0, reason: "empty catalog", raw: null };
+  }
+  const catalogText = input.catalog
+    .map((c) => `- [${c.id}] (status=${c.status}, type=${c.type}) ${c.title}`)
+    .join("\n");
+  const snippet =
+    input.timeline.length > 2800 ? `${input.timeline.slice(0, 2800)}\n…(truncated)` : input.timeline;
+  const userPrompt = [
+    "=== NEW SESSION ===",
+    `tags: [${input.tags.join(", ")}]`,
+    "",
+    snippet,
+    "",
+    "=== EXISTING INSIGHTS CATALOG ===",
+    catalogText,
+    "",
+    'Return JSON only: {"match_insight_id": <id|null>, "confidence": 0..1, "reason": "..."}',
+  ].join("\n");
+
+  let raw: CallResult;
+  try {
+    raw = await callDeepSeek({
+      tier: "flash",
+      systemPrompt: SYSTEM_PROMPT,
+      userPrompt,
+      maxOutputTokens: 200,
+      temperature: 0,
+    });
+  } catch (e) {
+    return { matchId: null, confidence: 0, reason: `dedup call failed: ${(e as Error).message}`, raw: null };
+  }
+
+  try {
+    const p = JSON.parse(extractJson(raw.text)) as {
+      match_insight_id?: string | null;
+      confidence?: number;
+      reason?: string;
+    };
+    let matchId = p.match_insight_id ?? null;
+    let confidence = typeof p.confidence === "number" ? p.confidence : 0;
+    let reason = String(p.reason ?? "");
+    // Anti-hallucination: only accept ids that are actually in the catalog.
+    if (matchId && !input.catalog.some((c) => c.id === matchId)) {
+      reason = `rejected non-catalog id ${matchId}; ${reason}`;
+      matchId = null;
+      confidence = 0;
+    }
+    return { matchId, confidence, reason, raw };
+  } catch (e) {
+    return { matchId: null, confidence: 0, reason: `parse error: ${(e as Error).message}`, raw };
+  }
+}