diff --git a/migrations-pg/0060_community_leads_pending.sql b/migrations-pg/0060_community_leads_pending.sql new file mode 100644 index 0000000..ee68e49 --- /dev/null +++ b/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; diff --git a/migrations/0060_community_leads_pending.sql b/migrations/0060_community_leads_pending.sql new file mode 100644 index 0000000..0a3a2c0 --- /dev/null +++ b/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; diff --git a/packages/ai/src/community-leads.ts b/packages/ai/src/community-leads.ts index 0e67efc..2405ed4 100644 --- a/packages/ai/src/community-leads.ts +++ b/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"). diff --git a/packages/pipeline/src/community-leads.test.ts b/packages/pipeline/src/community-leads.test.ts index 6a15810..9775796 100644 --- a/packages/pipeline/src/community-leads.test.ts +++ b/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 () => { diff --git a/packages/pipeline/src/community-leads.ts b/packages/pipeline/src/community-leads.ts index c5ef210..1cce462 100644 --- a/packages/pipeline/src/community-leads.ts +++ b/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> = { - 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( + rows: readonly T[], + limit: number, +): T[] { + const bySource = new Map(); + 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(); - 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,