fix(phase6a-v1.2): PostHog event filter client-side + backfill distinct_id + power_user_path rrweb fallback

- listSessionEvents: PostHog only honors single event= param; switched to client-side filter
- tag-sessions: backfill posthogDistinctId from getRecording when null (for rows pre-column)
- tagger: power_user_path v1 fallback (auth + clicks>=20 + no errors), vin_decode_no_outcome P3
- ingest upsert: always update posthogDistinctId (idempotent)

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

apps/worker/src/jobs/posthog-ingest.ts

@@ -87,7 +87,10 @@ export async function runPostHogIngest(): Promise<IngestResult> {
       try {
         const created = await prisma.sessionMeta.upsert({
           where: { id: rec.id },
-          update: {},
+          update: {
+            // Backfill the raw distinct_id for older rows where this column was added later.
+            posthogDistinctId: rec.distinct_id,
+          },
           create: {
             id: rec.id,
             projectKey: PROJECT_KEY,

apps/worker/src/jobs/tag-sessions.ts

@@ -2,6 +2,7 @@ import { prisma } from "../db";
 import { tagSession, scoreSession } from "../lib/tagger";
 import { buildEnrichment } from "../lib/enrich";
 import { customEventPromoteReasons } from "../lib/heuristic";
+import { getRecording } from "../lib/posthog";
 
 const MIN_SCORE_FOR_COMPRESSION = Number(process.env.INSIGHT_MIN_SCORE ?? "30");
 const TAG_BATCH_SIZE = Number(process.env.INSIGHT_TAG_BATCH ?? "40");
@@ -18,13 +19,30 @@ export async function runTagSessions(): Promise<{ tagged: number; discarded: num
   let discarded = 0;
 
   for (const s of pending) {
+    // Backfill posthogDistinctId for rows created before this column existed.
+    let distinctId = s.posthogDistinctId;
+    if (!distinctId) {
+      try {
+        const rec = await getRecording(s.id);
+        if (rec?.distinct_id) {
+          distinctId = rec.distinct_id;
+          await prisma.sessionMeta.update({
+            where: { id: s.id },
+            data: { posthogDistinctId: distinctId },
+          });
+        }
+      } catch {
+        // continue without distinct_id
+      }
+    }
+
     // Enrichment is best-effort — degrade gracefully if PostHog events endpoint is unavailable.
     const endedAt = new Date(s.startedAt.getTime() + s.durationMs);
     let enrichment;
     try {
       enrichment = await buildEnrichment({
         sessionId: s.id,
-        distinctId: s.posthogDistinctId,
+        distinctId,
         startedAt: s.startedAt,
         endedAt,
       });

apps/worker/src/lib/posthog.ts

@@ -109,7 +109,8 @@ export type PHCustomEvent = {
 };
 
 // Returns events for a single session, sorted ascending by timestamp.
-// `eventNames` filters server-side via PostHog event= param when supported.
+// PostHog `event=` query param only accepts a single value (last-wins on dupes),
+// so we fetch all events for the session and filter client-side via `eventNames`.
 export async function listSessionEvents(opts: {
   sessionId: string;
   dateFrom: string;
@@ -124,14 +125,17 @@ export async function listSessionEvents(opts: {
     "properties",
     JSON.stringify([{ key: "$session_id", value: opts.sessionId, operator: "exact" }]),
   );
-  if (opts.eventNames && opts.eventNames.length) {
-    // PostHog accepts repeated event= or comma-joined event=
-    for (const n of opts.eventNames) url.searchParams.append("event", n);
-  }
   const res = await fetch(url, { headers: headers() });
   if (!res.ok) throw new Error(`posthog events ${res.status}`);
   const data = (await res.json()) as { results?: PHCustomEvent[] };
-  const out = data.results ?? [];
+  let out = data.results ?? [];
+  if (opts.eventNames && opts.eventNames.length) {
+    const allow = new Set(opts.eventNames);
+    out = out.filter((e) => allow.has(e.event));
+  } else {
+    // By default drop PostHog implicit events that aren't useful for tagging.
+    out = out.filter((e) => !e.event.startsWith("$"));
+  }
   out.sort((a, b) => new Date(a.timestamp).getTime() - new Date(b.timestamp).getTime());
   return out;
 }

apps/worker/src/lib/tagger.ts

@@ -166,6 +166,7 @@ export function tagSession(s: SessionMeta, ctx: EnrichmentCtx | null): TagResult
   }
 
   // ─── Power user path ───
+  // v2 (custom events): 10+ vin_decode_succeeded + multi-provider + parts_export_initiated
   const vinSuccesses = events.filter((e) => e.name === "vin_decode_succeeded");
   if (vinSuccesses.length >= 10) {
     const providers = new Set(
@@ -173,9 +174,24 @@ export function tagSession(s: SessionMeta, ctx: EnrichmentCtx | null): TagResult
     );
     if (providers.size >= 2 && has(events, "parts_export_initiated")) {
       tags.push("power_user_path");
-      // severity stays INFO
     }
   }
+  // v1 fallback (rrweb-only): authenticated + many clicks + no errors and no other tag yet
+  if (
+    !tags.length &&
+    s.isAuthenticated &&
+    s.clickCount >= 20 &&
+    s.errorCount === 0
+  ) {
+    tags.push("power_user_path");
+  }
+
+  // ─── vin_decode lifecycle (v1.0 graceful) ───
+  // If we see at least one vin_decode_initiated/decoded but no succeeded, that's friction.
+  if (has(events, "vin_decode_initiated") && !has(events, "vin_decode_succeeded") && !has(events, "vin_decode_failed")) {
+    tags.push("vin_decode_no_outcome");
+    severity = bump(severity, "P3");
+  }
 
   // ─── At-risk active session (group analytics + user props) ───
   const subStatus = pick<string>(userProps, "subscription_status");