Buyer leads: no lead on wording alone when a model is there to judge it

migrations-pg/0060_community_leads_pending.sql

@@ -0,0 +1,2 @@
+-- 0060_community_leads_pending.sql (Postgres). See migrations/0060_community_leads_pending.sql.
+update community_leads set intent = 45 where judged = 0 and intent > 45;

migrations/0060_community_leads_pending.sql

@@ -0,0 +1,6 @@
+-- 0060_community_leads_pending.sql: hold unjudged buyer leads below the floor.
+--
+-- 0059 shipped scoring posts the model had no budget for by wording alone, and
+-- a game "demo" scored 72. Unjudged posts now wait at 45 or less until a scan
+-- judges them; this brings the rows written before that change into line.
+UPDATE community_leads SET intent = 45 WHERE judged = 0 AND intent > 45;

packages/ai/src/community-leads.ts

@@ -52,9 +52,11 @@ const JUDGE_SYSTEM = `You read public posts from Reddit, Hacker News and Bluesky
 
 Score 0-100:
 - 80-100: the author is actively looking for, comparing, or asking for recommendations on a tool/service in the brand's category, or is unhappy with a competitor and wants an alternative.
-- 60-79: the author describes a concrete problem the brand solves and is asking for help with it.
-- 30-59: on-topic discussion or question, but no sign they want a product.
-- 0-29: not a buyer. Vendors announcing or promoting their own product, changelogs, contests, challenges, news links, job posts, memes, the brand's own posts, or off-topic matches.
+- 60-79: the author has a concrete problem the brand solves, right now, and is asking for help with it.
+- 30-59: on-topic but no sign this author wants a product: news, opinions, tutorials, people already happily using a tool, learning or studying a topic, general discussion.
+- 0-29: not a buyer. Vendors announcing or promoting their own product, courses and training ads, changelogs, contests, challenges, games, job posts, memes, the brand's own posts, or posts where the keyword means something else.
+
+Be strict: 60 or more only when this author would plausibly welcome a product suggestion today. When unsure, score below 50.
 
 Return exactly one result for every post, in the order given.
 reason: one short sentence a salesperson would find useful ("asking for a SIEM alternative to Splunk for a 20-person team").

packages/pipeline/src/community-leads.test.ts

@@ -18,6 +18,7 @@ import { seedDatabase, SEED, type SeededDatabase } from '../../../apps/api/src/t
 import {
   buildLeadSources,
   containsKeyword,
+  judgeOrder,
   createLeadMonitor,
   draftCommunityLeadReply,
   listCommunityLeads,
@@ -227,7 +228,69 @@ describe('scanning', () => {
     );
     expect(r.judged).toBe(0);
     const [lead] = await listCommunityLeads(client, SEED.workspaceId);
-    expect(lead).toMatchObject({ externalId: 'r1', intent: 75, judged: false });
+    expect(lead).toMatchObject({ externalId: 'r1', intent: 70, judged: false });
+  });
+
+  test('with a model, an unjudged post waits below the floor and the next scan judges it', async () => {
+    const client = await db('leads-backlog');
+    const monitor = await createLeadMonitor({ db: client }, SEED.workspaceId, {
+      name: 'ThreatCrush',
+      keywords: ['siem', 'threat intel', 'intrusion detection'],
+      subreddits: ['devsecops'],
+    });
+    const sources = () => [fake('reddit', POSTS), fake('hackernews', HN)];
+    const down = {
+      generate: async () => {
+        throw new Error('429 insufficient_quota');
+      },
+    };
+
+    const first = await scanLeadMonitor(
+      { db: client, model: down, sources, now: NOW },
+      SEED.workspaceId,
+      monitor.id,
+    );
+    expect(first).toMatchObject({ stored: 3, judged: 0, leads: 0 });
+    expect(first.failures[0]).toMatchObject({ source: 'judge' });
+    // The wording guess for "Any recommendations for…" is 70, held at 45.
+    expect(await listCommunityLeads(client, SEED.workspaceId)).toEqual([]);
+
+    const second = await scanLeadMonitor(
+      {
+        db: client,
+        model: new StubModel(JUDGED),
+        sources,
+        now: new Date(NOW.getTime() + 3_600_000),
+      },
+      SEED.workspaceId,
+      monitor.id,
+    );
+    expect(second).toMatchObject({ stored: 0, judged: 3, leads: 2 });
+    expect((await listCommunityLeads(client, SEED.workspaceId)).map((l) => l.externalId)).toEqual([
+      '42',
+      'r1',
+    ]);
+  });
+
+  test('judging order: best wording first per source, sources take turns', () => {
+    const row = (source: string, pattern: number, hoursAgo: number) => ({
+      source,
+      pattern,
+      postedAt: ago(hoursAgo),
+      key: `${source}-${pattern}-${hoursAgo}`,
+    });
+    const rows = [
+      ...Array.from({ length: 5 }, (_, i) => row('bluesky', 15, i)),
+      row('bluesky', 70, 9),
+      row('reddit', 35, 1),
+      row('reddit', 70, 8),
+    ];
+    expect(judgeOrder(rows, 4).map((r) => r.key)).toEqual([
+      'bluesky-70-9',
+      'reddit-70-8',
+      'bluesky-15-0',
+      'reddit-35-1',
+    ]);
   });
 
   test('a failing source costs that source, and is reported', async () => {

packages/pipeline/src/community-leads.ts

@@ -62,18 +62,65 @@ const JUDGE_BATCH = 10;
 const DIGEST_LIMIT = 15;
 const DAY_MS = 86_400_000;
 
-/** The wording classifier's guess at intent, when there is no model. */
+/**
+ * The wording classifier's guess at intent, when there is no model. Only a
+ * request for recommendations clears the default floor of 60 on wording
+ * alone: "demo" or "evaluating" in a post is as often a game demo as a buyer.
+ */
 const PATTERN_INTENT: Partial<Record<SignalType, number>> = {
-  recommendation_request: 75,
-  purchase_intent: 72,
-  competitor_mention: 65,
-  public_complaint: 55,
-  pain: 55,
-  public_question: 40,
-  content_topic: 20,
-  hiring: 10,
+  recommendation_request: 70,
+  competitor_mention: 60,
+  purchase_intent: 55,
+  public_complaint: 50,
+  pain: 50,
+  public_question: 35,
+  content_topic: 15,
+  hiring: 5,
 };
 
+/**
+ * Where an unjudged post sits while it waits for the model: below any sensible
+ * floor, so a wording guess never reaches the digest when a model is there to
+ * be asked. The next scan judges the backlog first.
+ */
+const PENDING_INTENT_CAP = 45;
+/** How far back the backlog of unjudged posts is still worth judging. */
+const BACKLOG_DAYS = 7;
+
+export function patternIntent(text: string): number {
+  return PATTERN_INTENT[classifyPost(text).type] ?? 15;
+}
+
+/**
+ * Picks what the model judges this scan: best wording first within each
+ * source, then round-robin across sources, so one chatty source (Bluesky)
+ * cannot spend the whole budget while Reddit's few real asks wait.
+ */
+export function judgeOrder<T extends { source: string; pattern: number; postedAt: string }>(
+  rows: readonly T[],
+  limit: number,
+): T[] {
+  const bySource = new Map<string, T[]>();
+  for (const row of rows) {
+    const queue = bySource.get(row.source) ?? [];
+    queue.push(row);
+    bySource.set(row.source, queue);
+  }
+  for (const queue of bySource.values()) {
+    queue.sort((a, b) => b.pattern - a.pattern || Date.parse(b.postedAt) - Date.parse(a.postedAt));
+  }
+  const out: T[] = [];
+  const queues = [...bySource.values()];
+  while (out.length < limit && queues.some((q) => q.length > 0)) {
+    for (const queue of queues) {
+      const next = queue.shift();
+      if (next) out.push(next);
+      if (out.length >= limit) break;
+    }
+  }
+  return out;
+}
+
 /* ---------------------------------------------------------------- types -- */
 
 export interface LeadMonitor {
@@ -662,43 +709,21 @@ export async function scanLeadMonitor(
     }
   }
 
-  // Newest first, so the judged ones are the ones a reply can still land on.
-  fresh.sort((a, b) => Date.parse(b.post.postedAt) - Date.parse(a.post.postedAt));
-
-  const verdicts = new Map<string, LeadJudgement>();
-  if (deps.model && fresh.length > 0) {
-    const toJudge = fresh.slice(0, JUDGE_LIMIT);
-    for (let i = 0; i < toJudge.length; i += JUDGE_BATCH) {
-      const batch = toJudge.slice(i, i + JUDGE_BATCH).map(({ post }, j) => ({
-        id: String(i + j),
-        source: post.network,
-        title: post.title,
-        text: post.text,
-      }));
-      try {
-        for (const verdict of await judgeLeads(deps.model, brandOf(monitor), batch)) {
-          verdicts.set(verdict.id, verdict);
-        }
-      } catch (error) {
-        failures.push({ source: 'judge', reason: (error as Error).message });
-        break;
-      }
-    }
-  }
-
   const createdAt = at.toISOString();
-  let judged = 0;
   let leads = 0;
-  for (const [index, { post, term }] of fresh.entries()) {
-    const verdict = verdicts.get(String(index));
-    const intent = verdict ? verdict.intent : (PATTERN_INTENT[classifyPost(post.text).type] ?? 20);
-    if (verdict) judged += 1;
-    if (intent >= monitor.minIntent) leads += 1;
+
+  // Every new post is stored first. With a model, its wording guess is capped
+  // below the floor until the model has looked at it; without one, the guess
+  // is the score.
+  for (const { post, term } of fresh) {
+    const guess = patternIntent(post.text);
+    const intent = deps.model ? Math.min(guess, PENDING_INTENT_CAP) : guess;
+    if (!deps.model && intent >= monitor.minIntent) leads += 1;
     await db.execute({
       sql: `INSERT INTO community_leads (id, workspace_id, monitor_id, source, external_id, url, title, excerpt,
               author, author_url, container, posted_at, matched_term, intent, reason, judged, status,
               created_at, updated_at)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'new', ?, ?)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, 0, 'new', ?, ?)
             ON CONFLICT (monitor_id, source, external_id) DO NOTHING`,
       args: [
         newId('communityLead'),
@@ -715,14 +740,69 @@ export async function scanLeadMonitor(
         post.postedAt,
         term ?? null,
         intent,
-        verdict?.reason || null,
-        verdict ? 1 : 0,
         createdAt,
         createdAt,
       ],
     });
   }
 
+  // Then the model judges the backlog: this scan's posts and any earlier ones
+  // a previous scan had no budget (or no working model) for.
+  let judged = 0;
+  if (deps.model) {
+    const backlog = await queryAll<{
+      id: string;
+      source: string;
+      title: string | null;
+      excerpt: string;
+      posted_at: string;
+    }>(
+      db,
+      `SELECT id, source, title, excerpt, posted_at FROM community_leads
+        WHERE monitor_id = ? AND judged = 0 AND posted_at >= ?
+        ORDER BY posted_at DESC LIMIT 1000`,
+      [monitor.id, new Date(at.getTime() - BACKLOG_DAYS * DAY_MS).toISOString()],
+    );
+    const chosen = judgeOrder(
+      backlog.map((row) => ({
+        ...row,
+        postedAt: row.posted_at,
+        pattern: patternIntent(`${row.title ?? ''}\n${row.excerpt}`),
+      })),
+      JUDGE_LIMIT,
+    );
+    for (let i = 0; i < chosen.length; i += JUDGE_BATCH) {
+      const batch = chosen.slice(i, i + JUDGE_BATCH);
+      let verdicts: LeadJudgement[];
+      try {
+        verdicts = await judgeLeads(
+          deps.model,
+          brandOf(monitor),
+          batch.map((row, j) => ({
+            id: String(j),
+            source: row.source,
+            title: row.title ?? undefined,
+            text: row.excerpt,
+          })),
+        );
+      } catch (error) {
+        failures.push({ source: 'judge', reason: (error as Error).message });
+        break;
+      }
+      for (const verdict of verdicts) {
+        const row = batch[Number(verdict.id)];
+        if (!row) continue;
+        await db.execute({
+          sql: `UPDATE community_leads SET intent = ?, reason = ?, judged = 1, updated_at = ?
+                WHERE id = ? AND judged = 0`,
+          args: [verdict.intent, verdict.reason || null, createdAt, row.id],
+        });
+        judged += 1;
+        if (verdict.intent >= monitor.minIntent) leads += 1;
+      }
+    }
+  }
+
   const result: LeadScanResult = {
     monitorId: monitor.id,
     read,