From 5d0dae369e93b04af428c8deb281c8bdd66998bd Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Thu, 10 Sep 2026 13:43:23 +0000 Subject: [PATCH] Send past the held cards, say what holds them, and put a face on each lead Autopilot sent six messages a day against a queue of 597 and a cap of 200. The candidate window was `cap - sent today` rows in priority order, and the top of that order was a hundred and twenty prospects who all resolve to a handful of shared company inboxes already inside their weekly limit. Every tick read the same window, held every card in it, and stopped, while 172 cards with a personal address sat just past it and were never looked at. - The window is now a ceiling that only keeps a run finite; the loop stops at the cap, not at the window. Address gates run from a per-run memo before the per-person query and before any model draft, so a held card costs a comparison rather than a query and a draft nobody will send. - A hold is written to the live feed when it is new or changes, not every tick: one day produced 18,532 identical rows for 120 people, enough to make the day's events unreadable. `HoldLedger` remembers, and feeds the digest. - The digest splits "Awaiting approval: 1117" into what autopilot will send (and how many of those a limit is holding, and why) and what only a person can act on, by network. - An address given in a consented import is the person's own. The sender, the approve path, the queue badge and the recommender all read only `social_identities`; 15,917 imported contacts were invisible to all four. - An active campaign whose seed has not been read in a week is asked again when it has no work outstanding. Every campaign was seeded exactly once, and the digest had read "sites read 0" for ten days. - A picture for each lead: Gravatar's avatar is kept instead of dropped, and a ValueSERP image search accepts a result only when its page is the person's LinkedIn profile or their company's own domain, titled with their name. The profile URL is recorded as research. Shown in the digest, the prospect pages and the approval card; capped per tick and per day. Migration 0033 adds `people.avatar_url`, `avatar_source`, `photo_looked_up_at` and `campaigns.reseeded_at`. New env: `VALUESERP_API_KEY`, `PHOTO_LOOKUPS_PER_DAY`. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01234vh2fZQDqyXTEY8XBi4S --- .env.example | 8 + apps/api/src/app.ts | 3 +- apps/api/src/repository.ts | 25 +- apps/server/src/index.ts | 55 +++ apps/web/app/(app)/prospects/[id]/page.tsx | 16 +- apps/web/app/(app)/prospects/page.tsx | 19 +- apps/web/components/approval-card.tsx | 10 +- apps/web/components/avatar.tsx | 61 +++ apps/web/lib/types.ts | 3 + migrations/0033_avatars_and_reseed.sql | 29 ++ packages/email/src/notifications.ts | 90 +++- packages/pipeline/src/autopilot.test.ts | 190 ++++++++- packages/pipeline/src/autopilot.ts | 401 +++++++++++++----- packages/pipeline/src/enrich-contact.ts | 17 + packages/pipeline/src/index.ts | 12 + packages/pipeline/src/notify.test.ts | 79 ++++ packages/pipeline/src/notify.ts | 55 ++- packages/pipeline/src/photos.test.ts | 126 ++++++ packages/pipeline/src/photos.ts | 229 ++++++++++ packages/pipeline/src/pipeline.ts | 17 + packages/pipeline/src/reseed.test.ts | 113 +++++ packages/pipeline/src/reseed.ts | 112 +++++ packages/providers/src/index.ts | 11 + packages/providers/src/valueserp/index.ts | 199 +++++++++ .../providers/src/valueserp/valueserp.test.ts | 132 ++++++ 25 files changed, 1883 insertions(+), 129 deletions(-) create mode 100644 apps/web/components/avatar.tsx create mode 100644 migrations/0033_avatars_and_reseed.sql create mode 100644 packages/pipeline/src/photos.test.ts create mode 100644 packages/pipeline/src/photos.ts create mode 100644 packages/pipeline/src/reseed.test.ts create mode 100644 packages/pipeline/src/reseed.ts create mode 100644 packages/providers/src/valueserp/index.ts create mode 100644 packages/providers/src/valueserp/valueserp.test.ts diff --git a/.env.example b/.env.example index 92645a3..b109542 100644 --- a/.env.example +++ b/.env.example @@ -34,6 +34,14 @@ APOLLO_API_KEY= PDL_API_KEY= GITHUB_TOKEN= + +# Finds a lead's photo (and their LinkedIn profile URL, as research) by +# searching their name, title and company. One image search per person, +# roughly a quarter of a cent each. Unset, only Gravatar supplies pictures. +VALUESERP_API_KEY= +# Ceiling on photo lookups per workspace per day. Default 300. +PHOTO_LOOKUPS_PER_DAY= + BLUESKY_IDENTIFIER= BLUESKY_APP_PASSWORD= X_API_KEY= diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 3946782..260fe9d 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -1683,7 +1683,8 @@ export function createApp(options: AppOptions): Hono { const limit = clampLimit(c.req.query('limit')); const rows = await c.get('db').execute({ - sql: `SELECT p.id, p.display_name, p.current_title, p.identity_confidence, p.status, + sql: `SELECT p.id, p.display_name, p.current_title, p.avatar_url, p.identity_confidence, + p.status, co.name AS current_company, cp.status AS prospect_status, cp.interaction_state, s.opportunity, s.icp_fit, s.intent, s.reachability, diff --git a/apps/api/src/repository.ts b/apps/api/src/repository.ts index 2a190a8..6cf8b98 100644 --- a/apps/api/src/repository.ts +++ b/apps/api/src/repository.ts @@ -28,6 +28,7 @@ export interface PersonRow { current_title: string | null; current_company_id: string | null; location: string | null; + avatar_url?: string | null; identity_confidence: number; status: string; outreach_eligible: number; @@ -190,7 +191,7 @@ export async function listPendingRecommendations( return queryAll( db, - `SELECT r.*, p.display_name, p.current_title, p.identity_confidence, + `SELECT r.*, p.display_name, p.current_title, p.avatar_url, p.identity_confidence, s.summary AS signal_summary, s.source_url AS signal_url, s.source_timestamp AS signal_at, d.body AS draft_body, d.subject AS draft_subject, @@ -475,6 +476,15 @@ export async function resolveContactAddress( ); if (personal?.handle) return { address: personal.handle.trim().toLowerCase(), shared: false }; + // An imported address is the person's own too: they gave it to us, with + // consent recorded on the import. Personal, not shared. + const imported = await queryOne<{ address: string }>( + db, + `SELECT address FROM person_emails WHERE person_id = ? ORDER BY created_at LIMIT 1`, + [personId], + ); + if (imported?.address) return { address: imported.address.trim().toLowerCase(), shared: false }; + const company = await queryOne<{ contact_email: string }>( db, `SELECT co.contact_email FROM people p @@ -578,10 +588,15 @@ export async function pendingAddressUsage( `WITH candidate AS ( SELECT r.id AS recommendation_id, r.campaign_id AS campaign_id, EXISTS (SELECT 1 FROM drafts d WHERE d.recommendation_id = r.id) AS has_draft, - (SELECT lower(trim(si.handle)) FROM social_identities si - WHERE si.person_id = r.person_id AND si.network = 'email' - AND si.handle IS NOT NULL AND trim(si.handle) <> '' - ORDER BY si.confidence DESC LIMIT 1) AS personal, + COALESCE( + (SELECT lower(trim(si.handle)) FROM social_identities si + WHERE si.person_id = r.person_id AND si.network = 'email' + AND si.handle IS NOT NULL AND trim(si.handle) <> '' + ORDER BY si.confidence DESC LIMIT 1), + (SELECT lower(trim(pe.address)) FROM person_emails pe + WHERE pe.person_id = r.person_id AND pe.workspace_id = r.workspace_id + ORDER BY pe.created_at LIMIT 1) + ) AS personal, (SELECT lower(trim(co.contact_email)) FROM people p JOIN companies co ON co.id = p.current_company_id WHERE p.id = r.person_id AND co.contact_email IS NOT NULL diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 8123014..42430d9 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -47,7 +47,10 @@ import { pruneWorkflowEvents, regenerateRecommendations, rescoreProspect, + reseedIdleCampaigns, runAutopilot, + sweepProfilePhotos, + workspacesAwaitingPhotos, runCadences, runCrawlJob, runDiscoveryJob, @@ -67,6 +70,7 @@ import { RedditSource, RssSource, SiteProvider, + ValueSerpClient, type FeedSource, } from '@outreachgraph/providers'; @@ -237,6 +241,21 @@ if (!encryptionKey) { console.log('no SECRET_ENCRYPTION_KEY: workspaces cannot connect their own sending mailbox'); } +/** + * Pictures of leads, by search. + * + * Optional, like every provider: unset, Gravatar is the only source and most + * people stay as initials. Each lookup is a paid request, so the sweep is + * capped per workspace per day as well as per tick. + */ +const photoFinder = process.env.VALUESERP_API_KEY + ? new ValueSerpClient({ apiKey: process.env.VALUESERP_API_KEY }) + : undefined; + +const photoLookupsPerDay = Number(process.env.PHOTO_LOOKUPS_PER_DAY ?? 300); + +if (!photoFinder) console.log('no VALUESERP_API_KEY: lead photos come from Gravatar only'); + /** * The feed clients for one campaign's own targets. * @@ -682,6 +701,42 @@ async function tick(): Promise { } } + // A face for the people about to be written to. Bounded per tick and per + // day inside the sweep; this loop only decides who is looked at. + if (photoFinder) { + for (const workspaceId of await workspacesAwaitingPhotos(db)) { + try { + const swept = await sweepProfilePhotos( + { db, finder: photoFinder, dailyCap: photoLookupsPerDay }, + { workspaceId }, + ); + + if (swept.looked > 0) { + console.log( + `photos: looked up ${swept.looked} in ${workspaceId}, ${swept.found} found, ` + + `${swept.profiles} LinkedIn profile(s) recorded, ${swept.remainingToday} left today`, + ); + } + } catch (error) { + console.error(`photo sweep failed for ${workspaceId}`, error); + } + } + } + + // Campaigns whose seed has not been read in a week are asked again. This is + // the only thing that puts work back into an idle queue: once the first crop + // of research cards is cleared, nothing else ever asks the seed for more. + for (const workspace of workspaces) { + try { + const reseeded = await reseedIdleCampaigns(db, { workspaceId: workspace.id }); + if (reseeded.queued > 0) { + console.log(`reseeded ${reseeded.queued} idle campaign(s) in ${workspace.id}`); + } + } catch (error) { + console.error(`reseed failed for ${workspace.id}`, error); + } + } + // The queue drains before the send sweep, so anything discovered this tick // can go out on the same tick rather than waiting for the next one. const drained = await drainQueue(db, runJob, { isOutage: isBudgetExhausted }); diff --git a/apps/web/app/(app)/prospects/[id]/page.tsx b/apps/web/app/(app)/prospects/[id]/page.tsx index d697e94..d51c31a 100644 --- a/apps/web/app/(app)/prospects/[id]/page.tsx +++ b/apps/web/app/(app)/prospects/[id]/page.tsx @@ -1,5 +1,6 @@ import Link from 'next/link'; import { notFound, redirect } from 'next/navigation'; +import { Avatar } from '../../../../components/avatar'; import { EmailCandidates } from '../../../../components/email-candidates'; import { EnrolButton } from '../../../../components/enrol-button'; import { PageGuide } from '../../../../components/page-guide'; @@ -65,12 +66,15 @@ export default async function ProspectPage({ params }: { params: Promise<{ id: s ← Prospects -
-

{person.display_name}

-

{person.current_title ?? '—'}

-

- Identity confidence {Math.round((person.identity_confidence ?? 0) * 100)}% -

+
+ +
+

{person.display_name}

+

{person.current_title ?? '—'}

+

+ Identity confidence {Math.round((person.identity_confidence ?? 0) * 100)}% +

+
diff --git a/apps/web/app/(app)/prospects/page.tsx b/apps/web/app/(app)/prospects/page.tsx index 65f95b7..941f266 100644 --- a/apps/web/app/(app)/prospects/page.tsx +++ b/apps/web/app/(app)/prospects/page.tsx @@ -1,5 +1,6 @@ import Link from 'next/link'; import { redirect } from 'next/navigation'; +import { Avatar } from '../../../components/avatar'; import { PageGuide } from '../../../components/page-guide'; import { ApiUnavailableError, NotAuthenticatedError, fetchProspects } from '../../../lib/api'; import type { ProspectRow } from '../../../lib/types'; @@ -75,15 +76,19 @@ function ProspectItem({ person }: { person: ProspectRow }) { href={`/prospects/${person.id}`} className="border-border bg-surface-raised block rounded-2xl border p-4" > -
- {person.display_name} - - {person.opportunity ?? '—'} - +
+ +
+
+ {person.display_name} + + {person.opportunity ?? '—'} + +
+

{subtitle || '—'}

+
-

{subtitle || '—'}

-
Signals
diff --git a/apps/web/components/approval-card.tsx b/apps/web/components/approval-card.tsx index a96d7e2..98fde95 100644 --- a/apps/web/components/approval-card.tsx +++ b/apps/web/components/approval-card.tsx @@ -5,6 +5,7 @@ import { useState } from 'react'; // Imported from the server-safe modules, not lib/api: that pulls in // next/headers and cannot be bundled for the browser. import { relativeTime } from '../lib/format'; +import { Avatar } from './avatar'; import { ShareButtons } from './share-buttons'; import type { ApprovalCard as Card } from '../lib/types'; @@ -166,9 +167,12 @@ export function ApprovalCard({ card }: { card: Card }) { return (
-
-

{card.display_name}

-

{card.current_title ?? '—'}

+
+ +
+

{card.display_name}

+

{card.current_title ?? '—'}

+
diff --git a/apps/web/components/avatar.tsx b/apps/web/components/avatar.tsx new file mode 100644 index 0000000..66a9ab3 --- /dev/null +++ b/apps/web/components/avatar.tsx @@ -0,0 +1,61 @@ +/** + * A face, or the initials standing in for one. + * + * The picture is a URL the person published — Gravatar, their LinkedIn + * profile, their company's team page — and it is hot-linked rather than + * copied, so it can vanish. When it does, or when there never was one, the + * initials take the same space, so a list of people does not jump around + * depending on who has a photograph. + */ + +const SIZES = { + sm: 'h-8 w-8 text-xs', + md: 'h-10 w-10 text-sm', + lg: 'h-16 w-16 text-lg', +} as const; + +export function Avatar({ + name, + src, + size = 'md', + className = '', +}: { + name: string; + src?: string | null; + size?: keyof typeof SIZES; + className?: string; +}) { + const box = `${SIZES[size]} shrink-0 rounded-full ${className}`; + + if (src) { + return ( + // eslint-disable-next-line @next/next/no-img-element -- remote hosts are not known ahead of time + + ); + } + + return ( + + ); +} + +export function initials(name: string): string { + const parts = name + .trim() + .split(/\s+/) + .filter((part) => /[\p{L}\p{N}]/u.test(part)); + const first = parts[0]?.[0] ?? ''; + const last = parts.length > 1 ? (parts[parts.length - 1]?.[0] ?? '') : ''; + return `${first}${last}`.toUpperCase() || '?'; +} diff --git a/apps/web/lib/types.ts b/apps/web/lib/types.ts index d6edcaa..f2d4029 100644 --- a/apps/web/lib/types.ts +++ b/apps/web/lib/types.ts @@ -19,6 +19,7 @@ export interface ApprovalCard { person_id: string; display_name: string; current_title: string | null; + avatar_url?: string | null; action: string; network: string; priority: number; @@ -154,6 +155,7 @@ export interface ProspectRow { id: string; display_name: string; current_title: string | null; + avatar_url?: string | null; current_company: string | null; identity_confidence: number; prospect_status: string; @@ -196,6 +198,7 @@ export interface ProspectDetail { id: string; display_name: string; current_title: string | null; + avatar_url?: string | null; identity_confidence: number; status: string; }; diff --git a/migrations/0033_avatars_and_reseed.sql b/migrations/0033_avatars_and_reseed.sql new file mode 100644 index 0000000..00e255e --- /dev/null +++ b/migrations/0033_avatars_and_reseed.sql @@ -0,0 +1,29 @@ +-- 0033_avatars_and_reseed.sql +-- +-- Two things the digest could not show and one thing the pipeline could not do. +-- +-- A face. The digest and the queue list people by name and title, and a name +-- is a poor handle on a stranger. Gravatar already returned a `thumbnailUrl` +-- for every profile it found and the enrichment step threw it away; a search +-- adapter can find the rest. The URL is stored rather than the bytes: it is a +-- pointer to a picture the person published, not a copy of it, and it can be +-- dropped by clearing one column. +-- +-- `photo_looked_up_at` is a timestamp for the same reason `contact_enriched_at` +-- is: most lookups miss, and a boolean would make the sweep retry every miss +-- forever. A miss stamps the column; only a hit fills `avatar_url`. +-- +-- `reseeded_at` records the last time an active campaign's seed was read +-- again. Every campaign was crawled exactly once, at creation, and a directory +-- that gains members after that day never produced another lead: "sites read +-- 0, new people 0" for ten days straight while six campaigns sat active. + +ALTER TABLE people ADD COLUMN avatar_url TEXT; +-- gravatar | search | site — where the picture came from, for the audit trail. +ALTER TABLE people ADD COLUMN avatar_source TEXT; +ALTER TABLE people ADD COLUMN photo_looked_up_at TEXT; + +CREATE INDEX IF NOT EXISTS idx_people_photo_lookup + ON people(photo_looked_up_at) WHERE photo_looked_up_at IS NULL; + +ALTER TABLE campaigns ADD COLUMN reseeded_at TEXT; diff --git a/packages/email/src/notifications.ts b/packages/email/src/notifications.ts index 4836a38..fd70177 100644 --- a/packages/email/src/notifications.ts +++ b/packages/email/src/notifications.ts @@ -147,6 +147,8 @@ export interface DigestLead { readonly companyName?: string; readonly opportunity?: number; readonly sentTo?: string; + /** A picture they published, when one is known. HTML only; text reads fine without. */ + readonly avatarUrl?: string; } export interface DailyDigest { @@ -155,7 +157,22 @@ export interface DailyDigest { readonly sitesCrawled: number; readonly peopleFound: number; readonly messagesSent: number; + /** Every pending card, whoever will act on it. */ readonly awaitingApproval: number; + /** + * The pending cards autopilot will send itself, and how many of those are + * currently held by a limit. Absent when the workspace has no campaign on + * autopilot, in which case "awaiting approval" says everything. + */ + readonly autopilotQueue?: { readonly total: number; readonly held: number }; + /** + * Pending cards only a human can act on — LinkedIn, X, GitHub — where the + * product drafts and the person acts in the network's own interface. + */ + readonly needsYou?: { + readonly total: number; + readonly byNetwork: Readonly>; + }; readonly repliesReceived?: number; readonly leads: readonly DigestLead[]; /** Campaigns that produced nothing, and why, when the reason is knowable. */ @@ -178,11 +195,28 @@ export function dailyDigestEmail(to: string, digest: DailyDigest, appUrl: string : `${digest.peopleFound} new ${digest.peopleFound === 1 ? 'lead' : 'leads'}, ` + `${digest.messagesSent} sent · OutreachGraph`; + // "Awaiting approval: 1117" was the line that got this digest replied to + // with "why is it not sending?". Most of those cards were not waiting for + // anyone: autopilot was going to send them, and was holding them under a + // limit. The split says which is which, and the notes say what the limit is. + const queue = digest.autopilotQueue; + const needsYou = digest.needsYou; + const networks = needsYou + ? Object.entries(needsYou.byNetwork) + .filter(([, n]) => n > 0) + .map(([network, n]) => `${networkName(network)} ${n}`) + .join(', ') + : ''; + const counts = [ `Sites read: ${digest.sitesCrawled}`, `New people: ${digest.peopleFound}`, `Messages sent: ${digest.messagesSent}`, - `Awaiting approval: ${digest.awaitingApproval}`, + queue + ? `Autopilot queue: ${queue.total}${queue.held > 0 ? ` (${queue.held} held by a limit)` : ''}` + : '', + needsYou ? `Needs you: ${needsYou.total}${networks ? ` (${networks})` : ''}` : '', + !queue && !needsYou ? `Awaiting approval: ${digest.awaitingApproval}` : '', digest.repliesReceived !== undefined ? `Replies: ${digest.repliesReceived}` : '', ].filter(Boolean); @@ -219,19 +253,28 @@ export function dailyDigestEmail(to: string, digest: DailyDigest, appUrl: string row('Sites read', digest.sitesCrawled), row('New people', digest.peopleFound), row('Messages sent', digest.messagesSent), - row('Awaiting approval', digest.awaitingApproval), + queue + ? row( + 'Autopilot queue', + queue.total, + queue.held > 0 ? `${queue.held} held by a limit` : undefined, + ) + : '', + needsYou ? row('Needs you', needsYou.total, networks || undefined) : '', + !queue && !needsYou ? row('Awaiting approval', digest.awaitingApproval) : '', digest.repliesReceived !== undefined ? row('Replies', digest.repliesReceived) : '', '', quiet ? '

Nothing new came back today. Campaigns are still running.

' : digest.leads.length - ? `
    ${digest.leads + ? `
      ${digest.leads .map((lead) => { const where = [lead.title, lead.companyName].filter(Boolean).join(' at '); const score = lead.opportunity !== undefined ? ` [${lead.opportunity}]` : ''; const sent = lead.sentTo ? ' — written to' : ''; return ( - `
    • ` + + `
    • ${avatarHtml(lead)}` + + `` + `${escapeHtml(lead.personName)}` + `${where ? ` — ${escapeHtml(where)}` : ''}${score}${sent}
      • ` ); @@ -248,9 +291,44 @@ export function dailyDigestEmail(to: string, digest: DailyDigest, appUrl: string return { to, subject, text: text + foot.text, html }; } -function row(label: string, value: number): string { +function row(label: string, value: number, note?: string): string { return ( `${escapeHtml(label)}` + - `${value}` + `${value}` + + `${note ? ` ${escapeHtml(note)}` : ''}` + + '' ); } + +/** + * A small round picture before the name, when one is known. + * + * Inline styles and fixed dimensions because mail clients honour little else, + * and `alt=""` because the name follows immediately — a reader with images off + * loses nothing. + */ +function avatarHtml(lead: DigestLead): string { + if (!lead.avatarUrl) return ''; + return ( + `' + ); +} + +function networkName(network: string): string { + switch (network) { + case 'linkedin': + return 'LinkedIn'; + case 'x': + return 'X'; + case 'github': + return 'GitHub'; + case 'bluesky': + return 'Bluesky'; + case 'mastodon': + return 'Mastodon'; + default: + return network; + } +} diff --git a/packages/pipeline/src/autopilot.test.ts b/packages/pipeline/src/autopilot.test.ts index 9552bd2..222ea82 100644 --- a/packages/pipeline/src/autopilot.test.ts +++ b/packages/pipeline/src/autopilot.test.ts @@ -3,7 +3,7 @@ import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; import type { Mailer, Message, SendResult } from '@outreachgraph/email'; import type { GenerateResult, TextModel } from '@outreachgraph/ai'; import { seedDatabase, SEED, type SeededDatabase } from '../../../apps/api/src/test-seed'; -import { runAutopilot } from './autopilot'; +import { HoldLedger, describeHold, runAutopilot } from './autopilot'; let seeded: SeededDatabase | undefined; @@ -551,3 +551,191 @@ describe('one mailbox, several colleagues', () => { expect(result.skipped).toHaveLength(1); }); }); + +describe('a queue held at the top', () => { + /** + * The production shape that sent six a day: the highest-priority cards all + * resolve to one shared inbox that has already been written to, and the + * card that could actually go out sits behind them. + */ + async function addHeldColleague(db: Client, name: string, priority: number): Promise { + const stamp = now(); + const personId = `per_${name.toLowerCase()}`; + + await db.execute({ + sql: `INSERT INTO people (id, display_name, current_company_id, status, believed_minor, + outreach_eligible, identity_confidence, created_at, updated_at) + VALUES (?, ?, ?, 'active', 0, 1, 0.95, ?, ?)`, + args: [personId, name, SEED.companyId, stamp, stamp], + }); + + await db.execute({ + sql: `INSERT INTO recommendations (id, workspace_id, campaign_id, person_id, action, network, + priority, reason, policy_status, policy_version, status, created_at) + VALUES (?, ?, ?, ?, 'send_email', 'email', ?, 'same company', + 'allow', '2026-01-01', 'pending', ?)`, + args: [ + `rec_${name.toLowerCase()}`, + SEED.workspaceId, + SEED.campaignId, + personId, + priority, + stamp, + ], + }); + + await db.execute({ + sql: `INSERT INTO drafts (id, workspace_id, recommendation_id, subject, body, checks_json, + created_at, updated_at) + VALUES (?, ?, ?, 'Hello', 'A grounded message.', '[]', ?, ?)`, + args: [ + `drf_${name.toLowerCase()}`, + SEED.workspaceId, + `rec_${name.toLowerCase()}`, + stamp, + stamp, + ], + }); + } + + async function inboxAlreadyWrittenTo(db: Client, address: string): Promise { + await db.execute({ + sql: `INSERT INTO interactions (id, workspace_id, person_id, campaign_id, network, + direction, state, body, contact_address, shared_inbox, occurred_at, recorded_at) + VALUES ('int_earlier', ?, ?, ?, 'email', 'outbound', 'contacted', 'earlier', + ?, 1, ?, ?)`, + args: [SEED.workspaceId, SEED.personId, SEED.campaignId, address, now(), now()], + }); + } + + test('a sendable card behind the held ones is still reached', async () => { + seeded = await seedDatabase('autopilot-starvation'); + const { db } = seeded; + + // Jane can be written to, at the lowest priority in the queue. + await makeSendable(db, { personEmail: 'jane@acme.com', companyEmail: 'support@acme.com' }); + await db.execute({ + sql: 'UPDATE recommendations SET priority = 1 WHERE id = ?', + args: [SEED.recommendationId], + }); + + // Three colleagues ahead of her, all resolving to an inbox on cooldown. + await inboxAlreadyWrittenTo(db, 'support@acme.com'); + await addHeldColleague(db, 'Held1', 50); + await addHeldColleague(db, 'Held2', 50); + await addHeldColleague(db, 'Held3', 50); + + // A cap smaller than the number of held cards. The old window was + // `cap - sent today` rows, so this read two held cards and stopped. + const stamp = now(); + await db.execute({ + sql: `INSERT INTO workspace_settings (workspace_id, autopilot_daily_cap, created_at, updated_at) + VALUES (?, 2, ?, ?)`, + args: [SEED.workspaceId, stamp, stamp], + }); + + const { sent, mailer } = recordingMailer(); + const result = await runAutopilot( + { db, mailer, holdLedger: new HoldLedger() }, + SEED.workspaceId, + ); + + expect(sent.map((message) => message.to)).toEqual(['jane@acme.com']); + expect(result.skipped).toHaveLength(3); + expect(result.skipped.every((skip) => /address|company inbox/i.test(skip.reason))).toBe(true); + }); + + test('the same hold is written down once, not every tick', async () => { + seeded = await seedDatabase('autopilot-hold-once'); + const { db } = seeded; + + await makeSendable(db, { companyEmail: 'support@acme.com' }); + await inboxAlreadyWrittenTo(db, 'support@acme.com'); + + const ledger = new HoldLedger(); + const { mailer } = recordingMailer(); + + const first = await runAutopilot({ db, mailer, holdLedger: ledger }, SEED.workspaceId); + const second = await runAutopilot({ db, mailer, holdLedger: ledger }, SEED.workspaceId); + + // Reported on every run — the log still says what is happening. + expect(first.skipped).toHaveLength(1); + expect(second.skipped).toHaveLength(1); + + // Written to the live feed once. + const events = await queryAll<{ message: string }>( + db, + `SELECT message FROM workflow_events WHERE phase = 'send' AND level = 'warn'`, + ); + expect(events).toHaveLength(1); + + // And the digest can say what is holding the queue. + expect(ledger.summary(SEED.workspaceId)).toEqual([ + { label: 'the address was written to within the cooldown', count: 1 }, + ]); + }); + + test('an address they gave us in an import is their own, not a shared one', async () => { + seeded = await seedDatabase('autopilot-imported-address'); + const { db } = seeded; + + // No published identity, no company inbox — only the imported row. + await makeSendable(db); + await db.execute({ + sql: `INSERT INTO person_emails (id, workspace_id, person_id, address, dedupe_key, source, + verified, created_at) VALUES ('pem_jane', ?, ?, 'jane@home.example', + 'jane@home.example', 'import', 1, ?)`, + args: [SEED.workspaceId, SEED.personId, now()], + }); + + const { sent, mailer } = recordingMailer(); + const result = await runAutopilot( + { db, mailer, holdLedger: new HoldLedger() }, + SEED.workspaceId, + ); + + expect(sent.map((message) => message.to)).toEqual(['jane@home.example']); + expect(result.sent[0]?.toSharedInbox).toBe(false); + }); +}); + +describe('HoldLedger', () => { + test('a changed number is the same hold; a changed reason is a new one', () => { + const ledger = new HoldLedger(); + + expect(ledger.observe('wsp', 'rec', 'Only 9.85h since this address was last contacted')).toBe( + true, + ); + expect(ledger.observe('wsp', 'rec', 'Only 9.09h since this address was last contacted')).toBe( + false, + ); + expect(ledger.observe('wsp', 'rec', 'no drafted message')).toBe(true); + + ledger.release('wsp', 'rec'); + expect(ledger.observe('wsp', 'rec', 'no drafted message')).toBe(true); + }); + + test('forgets cards a full pass did not see', () => { + const ledger = new HoldLedger(); + ledger.observe('wsp', 'gone', 'no drafted message'); + ledger.observe('wsp', 'kept', 'no drafted message'); + + ledger.retain('wsp', new Set(['kept'])); + + expect(ledger.summary('wsp')).toEqual([{ label: 'no message written yet', count: 1 }]); + }); + + test('describes the engine’s refusals as short reasons', () => { + expect( + describeHold( + 'This prospect shares a company inbox that has already had 2 message(s) this week; the limit is 2.', + ), + ).toBe("the company inbox already had this week's messages"); + expect(describeHold('Weekly limit for this prospect reached (1/1).')).toBe( + 'already written to this week', + ); + expect(describeHold('something the engine has never said')).toBe( + 'something the engine has never said', + ); + }); +}); diff --git a/packages/pipeline/src/autopilot.ts b/packages/pipeline/src/autopilot.ts index e6beaed..5e8f8f3 100644 --- a/packages/pipeline/src/autopilot.ts +++ b/packages/pipeline/src/autopilot.ts @@ -31,7 +31,7 @@ import { INTERNAL_ACTION_KINDS, newId, type ActionKind, type Network } from '@outreachgraph/domain'; import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; -import { evaluatePolicy, isExecutable } from '@outreachgraph/policy'; +import { evaluateAddressLimits, evaluatePolicy, isExecutable } from '@outreachgraph/policy'; import type { Mailer } from '@outreachgraph/email'; import { draftForRecommendation, type TextModel } from '@outreachgraph/ai'; import { mailerForWorkspace } from './email-account'; @@ -85,6 +85,8 @@ export interface AutopilotDeps { */ readonly model?: TextModel; readonly now?: Date; + /** Where holds are remembered between runs. Defaults to the process-wide one. */ + readonly holdLedger?: HoldLedger; } export interface SentOutreach { @@ -141,11 +143,151 @@ interface Candidate { /** Attempts before a recommendation is left alone for a human to look at. */ const MAX_SEND_ATTEMPTS = 3; +/** + * How many pending cards one run will look at. + * + * This used to be `cap - today`, on the theory that a run can send at most + * that many and so need not read more. It was wrong in the one situation that + * matters: a queue where the top of the priority order cannot be sent. In + * production 597 email cards were pending against a cap of 200, and the 194 + * highest-priority ones all resolved to a handful of shared company inboxes + * already inside their weekly limit. Every tick read those 194, held every + * one, and stopped — while 172 cards with a personal address sat just past the + * window and were never looked at. Six messages a day out of a queue that + * could have sent a hundred. + * + * So the window is now bounded by a ceiling that exists only to keep one run + * finite, and the loop stops when the cap is spent rather than when the window + * is. A held card costs an in-memory check, not a slot. + */ +const CANDIDATE_CEILING = 2000; + +/** + * Why a card is currently held, remembered across runs. + * + * Autopilot runs every tick and a held card is held on every one of them, so + * writing the reason each time produced 18,532 identical "held back" rows in + * one day for 120 people — enough that reading the day's events timed out. + * The reason is written when it is new or when it changes, and released when + * the card is sent. Numbers are ignored when comparing, because "Only 9.85h + * since this address was last contacted" and "Only 9.09h" are the same hold. + * + * The ledger also answers "what is holding the queue up" for the digest, + * which is the question a queue that sends six a day most needs to answer. + */ +export class HoldLedger { + private readonly held = new Map>(); + + /** Records a hold. True when it is worth writing down — new, or changed. */ + observe(workspaceId: string, recommendationId: string, reason: string): boolean { + const key = holdKey(reason); + const entries = this.entriesFor(workspaceId); + const previous = entries.get(recommendationId); + entries.set(recommendationId, { reason, key }); + return previous?.key !== key; + } + + /** Forgets a card that was sent, or otherwise stopped being held. */ + release(workspaceId: string, recommendationId: string): void { + this.held.get(workspaceId)?.delete(recommendationId); + } + + /** + * Drops cards a completed run did not see — sent by hand, superseded, + * expired. Only called after a full pass, so a run that stopped at the cap + * does not forget the cards it never reached. + */ + retain(workspaceId: string, seen: ReadonlySet): void { + const entries = this.held.get(workspaceId); + if (!entries) return; + for (const id of [...entries.keys()]) { + if (!seen.has(id)) entries.delete(id); + } + } + + /** The holds in force, grouped by reason, largest group first. */ + summary(workspaceId: string): readonly HeldGroup[] { + const groups = new Map(); + + for (const entry of this.held.get(workspaceId)?.values() ?? []) { + const group = groups.get(entry.key); + if (group) group.count += 1; + else groups.set(entry.key, { label: describeHold(entry.reason), count: 1 }); + } + + return [...groups.values()].sort((a, b) => b.count - a.count); + } + + private entriesFor(workspaceId: string): Map { + let entries = this.held.get(workspaceId); + if (!entries) { + entries = new Map(); + this.held.set(workspaceId, entries); + } + return entries; + } +} + +interface HoldEntry { + readonly reason: string; + readonly key: string; +} + +export interface HeldGroup { + /** The reason as a short noun phrase, the same for one card or a hundred. */ + readonly label: string; + readonly count: number; +} + +/** The process-wide ledger; one container, one replica, one worker loop. */ +export const holdLedger = new HoldLedger(); + +/** What is holding a workspace's queue up right now. */ +export function heldSummary( + workspaceId: string, + ledger: HoldLedger = holdLedger, +): readonly HeldGroup[] { + return ledger.summary(workspaceId); +} + +function holdKey(reason: string): string { + return reason.replace(/\d+(?:\.\d+)?/g, '#'); +} + +/** + * The engine's refusal, reworded to follow a count. + * + * "This prospect shares a company inbox that has already had 2 message(s) + * this week" is right for one card and wrong for a hundred and twenty. Unknown + * reasons are passed through rather than dropped. + */ +export function describeHold(reason: string): string { + if (/shares a company inbox/i.test(reason)) { + return "the company inbox already had this week's messages"; + } + if (/weekly limit for this address/i.test(reason)) { + return "the address already had this week's messages"; + } + if (/the cooldown is/i.test(reason)) return 'the address was written to within the cooldown'; + if (/weekly limit for this prospect/i.test(reason)) return 'already written to this week'; + if (/no address published/i.test(reason)) return 'no address to write to'; + if (/no drafted message/i.test(reason)) return 'no message written yet'; + if (/quality checks/i.test(reason)) return 'the draft failed its quality checks'; + if (/giving up after/i.test(reason)) return 'sending failed repeatedly'; + if (/requires human approval/i.test(reason)) return 'waiting for your approval'; + if (/no mailbox is connected/i.test(reason)) return 'no mailbox connected'; + if (/budget/i.test(reason)) return "over the plan's monthly allowance"; + return reason; +} + /** * Sends everything due for one workspace. * * Ordered by priority so a daily cap spends itself on the best leads rather - * than on whichever rows the planner happened to return first. + * than on whichever rows the planner happened to return first. A personal + * address sorts ahead of a shared inbox at equal priority: the shared one is + * the more likely to be held, and the cap is better spent on a message that + * reaches the person it names. */ export async function runAutopilot( deps: AutopilotDeps, @@ -153,6 +295,7 @@ export async function runAutopilot( ): Promise { const { db } = deps; const at = deps.now ?? new Date(); + const ledger = deps.holdLedger ?? holdLedger; const sent: SentOutreach[] = []; const skipped: SkippedOutreach[] = []; @@ -174,6 +317,11 @@ export async function runAutopilot( return { sent, skipped, failed }; } + // The address is the person's own when one is known — published on a page + // they control, or given to us in an import they consented to — and the + // employer's shared inbox otherwise. Imported addresses live in + // `person_emails`, and until now this query only read `social_identities`, + // so sixteen thousand consented mailboxes were invisible to the sender. const candidates = await queryAll( db, `SELECT r.id AS recommendation_id, r.campaign_id, r.person_id, r.action, r.network, @@ -183,9 +331,15 @@ export async function runAutopilot( w.min_outreach_confidence, d.id AS draft_id, d.subject, d.body, d.checks_json, co.name AS company_name, co.contact_email AS company_contact_email, - (SELECT si.handle FROM social_identities si - WHERE si.person_id = p.id AND si.network = 'email' - ORDER BY si.confidence DESC LIMIT 1) AS person_email, + COALESCE( + (SELECT si.handle FROM social_identities si + WHERE si.person_id = p.id AND si.network = 'email' + AND si.handle IS NOT NULL AND trim(si.handle) <> '' + ORDER BY si.confidence DESC LIMIT 1), + (SELECT pe.address FROM person_emails pe + WHERE pe.person_id = p.id AND pe.workspace_id = r.workspace_id + ORDER BY pe.created_at LIMIT 1) + ) AS person_email, (SELECT COUNT(*) FROM actions a WHERE a.recommendation_id = r.id AND a.status = 'failed') AS failed_attempts FROM recommendations r @@ -201,20 +355,40 @@ export async function runAutopilot( AND r.action = 'send_email' AND r.network = 'email' AND p.status = 'active' - ORDER BY r.priority DESC, r.created_at ASC + ORDER BY r.priority DESC, (person_email IS NULL) ASC, r.created_at ASC LIMIT ?`, - [workspaceId, Math.max(cap - today, 0)], + [workspaceId, CANDIDATE_CEILING], ); + // What each mailbox has already had, read once per address per run. A + // hundred colleagues behind one `support@` cost one query and a hundred + // comparisons, not a hundred queries — and after a send the entry is + // updated in place so the next colleague sees the message that just left. + const addressUsage = new Map(); + + // Read once and refreshed after each send rather than once per candidate: a + // workspace can cross its monthly allowance partway through a sweep, and only + // a send can move it. + let budgetState = await budgetStatus(db, workspaceId, at); + + const seen = new Set(); + let completed = true; + for (const row of candidates) { - if (today >= cap) break; + if (today >= cap) { + completed = false; + break; + } + seen.add(row.recommendation_id); // Skips are reported, not swallowed. // // "No address published for this person" and "still requires human // approval" are the two reasons a campaign sits at a stage looking broken, // and neither is an error anywhere else in the system — so if they are not - // surfaced here they are not surfaced at all. + // surfaced here they are not surfaced at all. The event is written when + // the reason is new; the same hold on the next tick is remembered, not + // repeated. const note = async (reason: string): Promise => { skipped.push({ recommendationId: row.recommendation_id, @@ -222,6 +396,8 @@ export async function runAutopilot( reason, }); + if (!ledger.observe(workspaceId, row.recommendation_id, reason)) return; + await emitEvent(db, { workspaceId, campaignId: row.campaign_id, @@ -246,6 +422,111 @@ export async function runAutopilot( continue; } + const recipient = pickEmailRecipient(row); + if (!recipient) { + await note('no address published for this person or their company'); + continue; + } + + const budget = safeJson(row.budget_json); + const cooldown = + typeof budget.minHoursBetweenActions === 'number' + ? { minHoursBetweenActions: budget.minHoursBetweenActions } + : {}; + + // ------------------------------------------------------ address gates + // + // Counted against the mailbox as well as the person, and checked first, + // before anything that costs a query or a model call. This is the gate + // that holds most of a real queue: a prospect with no personal address + // falls back to their employer's shared inbox, so N colleagues are N + // separate people, each comfortably inside its own weekly limit, while one + // `support@` receives N messages. The engine grew these gates in #34 and + // only the human approval route fed them until #48; they are the same + // arithmetic the queue's badge uses, so what is shown as held is held. + const address = recipient.address.trim().toLowerCase(); + let usage = addressUsage.get(address); + if (!usage) { + usage = await addressCounts(db, workspaceId, recipient.address, at); + addressUsage.set(address, usage); + } + + const breaches = evaluateAddressLimits({ + actionsThisWeek: usage.thisWeek, + maxPerWeek: numberOr(budget.maxActionsPerAddressPerWeek, 1), + shared: recipient.shared, + ...(usage.hoursSinceLast !== undefined ? { hoursSinceLast: usage.hoursSinceLast } : {}), + ...(typeof budget.minHoursBetweenActions === 'number' + ? { cooldownHours: budget.minHoursBetweenActions } + : {}), + }); + + // The engine reports the last breach when several fire; so does this. + const breach = breaches[breaches.length - 1]; + if (breach) { + await note(breach.reason); + continue; + } + + // ------------------------------------------------------------- policy + // + // Re-evaluated from live rows, never from the stored snapshot. + const counts = await actionCounts(db, workspaceId, row.person_id, at); + + const decision = evaluatePolicy({ + network: row.network as Network, + action: row.action as ActionKind, + approvalMode: row.approval_mode as 'trusted_automation', + hasConnectedAccount: sender !== undefined, + personSuppressed: row.person_status === 'suppressed' || row.outreach_eligible === 0, + personBelievedMinor: row.believed_minor === 1, + personDeleted: row.person_status === 'deleted', + identityConfidence: row.identity_confidence, + minIdentityConfidence: row.min_outreach_confidence, + actionsToday: today, + maxActionsPerDay: Math.min(numberOr(budget.maxActionsPerDay, 50), cap), + actionsToThisProspectThisWeek: counts.thisProspect, + maxActionsPerProspectPerWeek: numberOr(budget.maxActionsPerProspectPerWeek, 1), + // The cooldown the campaign configured, not only the engine default. + ...cooldown, + ...(counts.hoursSinceLast !== undefined + ? { hoursSinceLastActionToProspect: counts.hoursSinceLast } + : {}), + // Already known to pass; supplied so the engine's answer is complete. + actionsToThisAddressThisWeek: usage.thisWeek, + maxActionsPerAddressPerWeek: numberOr(budget.maxActionsPerAddressPerWeek, 1), + addressShared: recipient.shared, + ...(usage.hoursSinceLast !== undefined + ? { hoursSinceLastActionToAddress: usage.hoursSinceLast } + : {}), + budgetExhausted: budgetState.exhausted, + }); + + // `approved: false` is the whole point. Autopilot holds no approval, so + // only a decision of plain `allow` gets through — `allow_with_approval` + // means a human still has to look at it, and reaching here with that would + // mean the capability matrix no longer marks email customer-managed. + // Sending anyway would be exactly what the approval default prevents. + if (!isExecutable(decision.decision, false)) { + await note( + decision.decision === 'allow_with_approval' + ? 'this action still requires human approval' + : decision.reason, + ); + continue; + } + + if (!sender) { + await note('no mailbox is connected, so nothing can be sent'); + continue; + } + + // -------------------------------------------------------------- draft + // + // Only now, once the card is known to be sendable. Drafting is a model + // call, and writing a message for a card the address gate was about to + // hold anyway paid for a hundred drafts a tick that went nowhere. + // // A recommendation with no draft has nothing to send — but "the composer // declined to write one" and "nobody ever tried" are different states, and // until now they produced the same warning on every tick forever. Drafting @@ -310,93 +591,6 @@ export async function runAutopilot( continue; } - const recipient = pickEmailRecipient(row); - if (!recipient) { - await note('no address published for this person or their company'); - continue; - } - - // ------------------------------------------------------------- policy - // - // Re-evaluated from live rows, never from the stored snapshot. - const counts = await actionCounts(db, workspaceId, row.person_id, at); - - // Counted against the mailbox as well as the person. - // - // Both limits are needed and neither substitutes for the other. The - // per-person limit answers "how often do we contact this human"; this one - // answers "how much mail does this mailbox get", and a prospect with no - // personal address falls back to their employer's shared inbox — so N - // colleagues are N separate people, each comfortably inside its own weekly - // limit, while one `support@` receives N messages. - // - // The policy engine grew these gates in #34, but only the human approval - // route in `app.ts` ever filled them in. They are optional inputs, so - // omitting them does not fail loudly — it silently disables them, and this - // is the unattended path that sends at volume. In production the manual - // route was protected and autopilot was not, which is how an address that - // had already been written to that afternoon was written to again hours - // after the fix shipped. - const addressUsage = await addressCounts(db, workspaceId, recipient.address, at); - const budget = safeJson(row.budget_json); - - // Read inside the loop rather than once per run: a workspace can cross its - // monthly allowance partway through a sweep, and a snapshot taken before - // the first send would let the rest of the batch through on a stale count. - const budgetState = await budgetStatus(db, workspaceId, at); - - const decision = evaluatePolicy({ - network: row.network as Network, - action: row.action as ActionKind, - approvalMode: row.approval_mode as 'trusted_automation', - hasConnectedAccount: sender !== undefined, - personSuppressed: row.person_status === 'suppressed' || row.outreach_eligible === 0, - personBelievedMinor: row.believed_minor === 1, - personDeleted: row.person_status === 'deleted', - identityConfidence: row.identity_confidence, - minIdentityConfidence: row.min_outreach_confidence, - actionsToday: today, - maxActionsPerDay: Math.min(numberOr(budget.maxActionsPerDay, 50), cap), - actionsToThisProspectThisWeek: counts.thisProspect, - maxActionsPerProspectPerWeek: numberOr(budget.maxActionsPerProspectPerWeek, 1), - // The cooldown the campaign configured, not only the engine default. - // `evaluateAddressLimits` already honoured this for the queue's badge, - // so leaving it out here made the preview and the refusal disagree about - // the same card. - ...(typeof budget.minHoursBetweenActions === 'number' - ? { minHoursBetweenActions: budget.minHoursBetweenActions } - : {}), - ...(counts.hoursSinceLast !== undefined - ? { hoursSinceLastActionToProspect: counts.hoursSinceLast } - : {}), - actionsToThisAddressThisWeek: addressUsage.thisWeek, - maxActionsPerAddressPerWeek: numberOr(budget.maxActionsPerAddressPerWeek, 1), - addressShared: recipient.shared, - ...(addressUsage.hoursSinceLast !== undefined - ? { hoursSinceLastActionToAddress: addressUsage.hoursSinceLast } - : {}), - budgetExhausted: budgetState.exhausted, - }); - - // `approved: false` is the whole point. Autopilot holds no approval, so - // only a decision of plain `allow` gets through — `allow_with_approval` - // means a human still has to look at it, and reaching here with that would - // mean the capability matrix no longer marks email customer-managed. - // Sending anyway would be exactly what the approval default prevents. - if (!isExecutable(decision.decision, false)) { - await note( - decision.decision === 'allow_with_approval' - ? 'this action still requires human approval' - : decision.reason, - ); - continue; - } - - if (!sender) { - await note('no mailbox is connected, so nothing can be sent'); - continue; - } - // --------------------------------------------------------------- send const actionId = newId('action'); const stamp = now(); @@ -485,6 +679,12 @@ export async function runAutopilot( }); today += 1; + ledger.release(workspaceId, row.recommendation_id); + + // The mailbox just received one; colleagues behind it on this run must + // see that without re-reading the table. + addressUsage.set(address, { thisWeek: usage.thisWeek + 1, hoursSinceLast: 0 }); + budgetState = await budgetStatus(db, workspaceId, at); } catch (error) { const message = error instanceof Error ? error.message : String(error); failed += 1; @@ -520,9 +720,16 @@ export async function runAutopilot( } } + if (completed) ledger.retain(workspaceId, seen); + return { sent, skipped, failed }; } +interface AddressUsage { + readonly thisWeek: number; + readonly hoursSinceLast?: number; +} + /** True when any recorded quality gate failed. Unparseable checks fail closed. */ function hasFailingCheck(checksJson: string | null): boolean { if (!checksJson) return false; @@ -612,7 +819,7 @@ async function addressCounts( workspaceId: string, address: string, at: Date, -): Promise<{ thisWeek: number; hoursSinceLast?: number }> { +): Promise { const weekAgo = new Date(at.getTime() - 7 * 24 * 3_600_000).toISOString(); const row = await queryOne<{ n: number; last_at: string | null }>( diff --git a/packages/pipeline/src/enrich-contact.ts b/packages/pipeline/src/enrich-contact.ts index c53899a..7c06025 100644 --- a/packages/pipeline/src/enrich-contact.ts +++ b/packages/pipeline/src/enrich-contact.ts @@ -252,6 +252,18 @@ export async function enrichContact( const filledName = await fillGaps(db, input.personId, profile); + // The picture they chose for exactly this purpose. Gravatar's whole point is + // an avatar looked up by address, and it was being fetched and thrown away. + // Never overwrites one already held; a later, better source wins by being + // first, not by being later. + if (profile.avatarUrl) { + await db.execute({ + sql: `UPDATE people SET avatar_url = ?, avatar_source = 'gravatar', updated_at = ? + WHERE id = ? AND avatar_url IS NULL`, + args: [sizedGravatar(profile.avatarUrl), now(), input.personId], + }); + } + return { personId: input.personId, found: true, @@ -261,6 +273,11 @@ export async function enrichContact( }; } +/** Gravatar serves 80px unless asked; the digest and the queue want more. */ +function sizedGravatar(url: string): string { + return url.includes('?') ? url : `${url}?s=256`; +} + /** * Stores one published account, once. * diff --git a/packages/pipeline/src/index.ts b/packages/pipeline/src/index.ts index 8deba1e..31630ca 100644 --- a/packages/pipeline/src/index.ts +++ b/packages/pipeline/src/index.ts @@ -16,11 +16,23 @@ export { export { runDiscoveryJob, type DiscoveryJobDeps, type DiscoveryJobResult } from './discovery'; export { runAutopilot, + HoldLedger, + holdLedger, + heldSummary, + describeHold, type AutopilotDeps, type AutopilotResult, + type HeldGroup, type SentOutreach, type SkippedOutreach, } from './autopilot'; +export { + sweepProfilePhotos, + workspacesAwaitingPhotos, + type PhotoSweepDeps, + type PhotoSweepResult, +} from './photos'; +export { reseedIdleCampaigns, type ReseedResult } from './reseed'; export { runListening, listeningCampaigns, diff --git a/packages/pipeline/src/notify.test.ts b/packages/pipeline/src/notify.test.ts index 7b365bb..4328970 100644 --- a/packages/pipeline/src/notify.test.ts +++ b/packages/pipeline/src/notify.test.ts @@ -3,6 +3,8 @@ import { now, queryAll, type Client } from '@outreachgraph/db'; import type { Mailer, Message, SendResult } from '@outreachgraph/email'; import { seedDatabase, SEED, type SeededDatabase } from '../../../apps/api/src/test-seed'; import { notifyAddress, loadNotifySettings, sendDailyDigest, sendLeadAlerts } from './notify'; +import { HoldLedger } from './autopilot'; +import { recordDiscovered } from './stages'; let seeded: SeededDatabase | undefined; @@ -229,3 +231,80 @@ describe('sendDailyDigest', () => { expect(sent).toHaveLength(0); }); }); + +describe('what the digest says about the queue', () => { + test('splits the queue into what autopilot will send and what needs a human', async () => { + seeded = await seedDatabase('digest-queue'); + const { db } = seeded; + await setDigestHour(db, 9); + + // One email card autopilot owns, on a campaign that opted in. + await db.execute({ + sql: `UPDATE campaigns SET approval_mode = 'trusted_automation' WHERE id = ?`, + args: [SEED.campaignId], + }); + await db.execute({ + sql: `UPDATE recommendations SET action = 'send_email', network = 'email' WHERE id = ?`, + args: [SEED.recommendationId], + }); + + // One LinkedIn card only a person can act on. + await db.execute({ + sql: `INSERT INTO recommendations (id, workspace_id, campaign_id, person_id, action, network, + priority, reason, policy_status, policy_version, status, created_at) + VALUES ('rec_li', ?, ?, ?, 'reply', 'linkedin', 10, 'they posted', + 'manual_only', '2026-01-01', 'pending', ?)`, + args: [SEED.workspaceId, SEED.campaignId, SEED.personId, now()], + }); + + // Autopilot remembers holding the email card. + const ledger = new HoldLedger(); + ledger.observe( + SEED.workspaceId, + SEED.recommendationId, + 'This prospect shares a company inbox that has already had 2 message(s) this week; the limit is 2.', + ); + + // Jane has a picture and surfaced today. + await db.execute({ + sql: `UPDATE people SET avatar_url = 'https://pics.example/jane.jpg' WHERE id = ?`, + args: [SEED.personId], + }); + await recordDiscovered(db, { + workspaceId: SEED.workspaceId, + campaignId: SEED.campaignId, + personId: SEED.personId, + }); + + const { sent, mailer } = recordingMailer(); + await sendDailyDigest( + { db, mailer, appUrl: APP_URL, now: at(12), holdLedger: ledger }, + SEED.workspaceId, + ); + + const mail = sent[0]; + expect(mail?.text).toContain('Autopilot queue: 1 (1 held by a limit)'); + expect(mail?.text).toContain('Needs you: 1 (LinkedIn 1)'); + expect(mail?.text).not.toContain('Awaiting approval'); + expect(mail?.text).toContain( + "1 held in the autopilot queue: the company inbox already had this week's messages.", + ); + expect(mail?.html).toContain(' { + seeded = await seedDatabase('digest-plain'); + const { db } = seeded; + await setDigestHour(db, 9); + + const { sent, mailer } = recordingMailer(); + await sendDailyDigest( + { db, mailer, appUrl: APP_URL, now: at(12), holdLedger: new HoldLedger() }, + SEED.workspaceId, + ); + + // The seed's one pending card is a reply on a manual network. + expect(sent[0]?.text).toContain('Needs you: 1'); + expect(sent[0]?.text).not.toContain('Autopilot queue'); + }); +}); diff --git a/packages/pipeline/src/notify.ts b/packages/pipeline/src/notify.ts index 47b25e7..b893f70 100644 --- a/packages/pipeline/src/notify.ts +++ b/packages/pipeline/src/notify.ts @@ -19,8 +19,9 @@ * a crash mid-loop cannot produce a second copy. */ -import { newId } from '@outreachgraph/domain'; +import { INTERNAL_ACTION_KINDS, newId } from '@outreachgraph/domain'; import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; +import { heldSummary, type HoldLedger } from './autopilot'; import { dailyDigestEmail, leadAlertEmail, @@ -36,6 +37,8 @@ export interface NotifyDeps { /** Absolute base for links in the mail. */ readonly appUrl: string; readonly now?: Date; + /** Where autopilot remembers what it is holding. Defaults to the process-wide one. */ + readonly holdLedger?: HoldLedger; } export interface NotifySettings { @@ -291,7 +294,9 @@ export async function sendDailyDigest(deps: NotifyDeps, workspaceId: string): Pr const since = `${today}T00:00:00.000Z`; - const [crawled, found, sent, awaiting, leads] = await Promise.all([ + const internal = INTERNAL_ACTION_KINDS.map(() => '?').join(', '); + + const [crawled, found, sent, awaiting, queue, manual, leads] = await Promise.all([ countOne( deps.db, `SELECT COUNT(*) AS n FROM jobs @@ -317,16 +322,42 @@ export async function sendDailyDigest(deps: NotifyDeps, workspaceId: string): Pr WHERE workspace_id = ? AND status = 'pending'`, [workspaceId], ), + // What autopilot will send on its own: email cards in a campaign that + // opted in. The only network the matrix lets a machine act on. + countOne( + deps.db, + `SELECT COUNT(*) AS n FROM recommendations r + JOIN campaigns c ON c.id = r.campaign_id + WHERE r.workspace_id = ? AND r.status = 'pending' + AND r.action = 'send_email' AND r.network = 'email' + AND c.approval_mode = 'trusted_automation' AND c.status != 'archived'`, + [workspaceId], + ), + // What only a human can do: everything outbound that is not the above. + // Research cards are internal and clear themselves, so they are neither. + queryAll<{ network: string; n: number }>( + deps.db, + `SELECT r.network, COUNT(*) AS n FROM recommendations r + JOIN campaigns c ON c.id = r.campaign_id + WHERE r.workspace_id = ? AND r.status = 'pending' + AND r.action NOT IN (${internal}) + AND NOT (r.action = 'send_email' AND r.network = 'email' + AND c.approval_mode = 'trusted_automation' AND c.status != 'archived') + GROUP BY r.network + ORDER BY n DESC`, + [workspaceId, ...INTERNAL_ACTION_KINDS], + ), queryAll<{ person_id: string; display_name: string; current_title: string | null; company_name: string | null; opportunity: number | null; + avatar_url: string | null; sent_today: number; }>( deps.db, - `SELECT p.id AS person_id, p.display_name, p.current_title, + `SELECT p.id AS person_id, p.display_name, p.current_title, p.avatar_url, co.name AS company_name, s.opportunity, (SELECT COUNT(*) FROM actions a WHERE a.person_id = p.id AND a.workspace_id = e.workspace_id @@ -350,16 +381,34 @@ export async function sendDailyDigest(deps: NotifyDeps, workspaceId: string): Pr ...(lead.current_title ? { title: lead.current_title } : {}), ...(lead.company_name ? { companyName: lead.company_name } : {}), ...(lead.opportunity !== null ? { opportunity: Math.round(lead.opportunity) } : {}), + ...(lead.avatar_url ? { avatarUrl: lead.avatar_url } : {}), ...(lead.sent_today > 0 ? { sentTo: 'sent' } : {}), })); + // Why the queue is not moving, from the sweep's own memory of what it held + // on its last pass. The question a digest that says "6 sent" has to answer + // is what happened to the other five hundred, and until now it did not. + const held = heldSummary(workspaceId, deps.holdLedger); + const heldCount = held.reduce((sum, group) => sum + group.count, 0); + + const notes = held.map((group) => `${group.count} held in the autopilot queue: ${group.label}.`); + + const byNetwork: Record = {}; + for (const row of manual) byNetwork[row.network] = Number(row.n); + const manualTotal = Object.values(byNetwork).reduce((sum, n) => sum + n, 0); + const digest: DailyDigest = { date: today, sitesCrawled: crawled, peopleFound: found, messagesSent: sent, awaitingApproval: awaiting, + ...(queue > 0 || heldCount > 0 + ? { autopilotQueue: { total: queue, held: Math.min(heldCount, queue) } } + : {}), + ...(manualTotal > 0 ? { needsYou: { total: manualTotal, byNetwork } } : {}), leads: digestLeads, + ...(notes.length > 0 ? { notes } : {}), }; try { diff --git a/packages/pipeline/src/photos.test.ts b/packages/pipeline/src/photos.test.ts new file mode 100644 index 0000000..05c9b87 --- /dev/null +++ b/packages/pipeline/src/photos.test.ts @@ -0,0 +1,126 @@ +/** + * Finding a face for a lead. + * + * What matters here is the bookkeeping around the lookup, not the lookup: a + * miss must be remembered so it is never paid for twice, a hit must record the + * profile it came from as research, and the daily ceiling must hold. + */ + +import { afterEach, describe, expect, test } from 'bun:test'; +import { queryOne } from '@outreachgraph/db'; +import type { ProfilePhoto, ProfilePhotoFinder, ProfilePhotoQuery } from '@outreachgraph/providers'; +import { seedDatabase, SEED, type SeededDatabase } from '../../../apps/api/src/test-seed'; +import { sweepProfilePhotos, workspacesAwaitingPhotos } from './photos'; + +let seeded: SeededDatabase | undefined; + +afterEach(() => { + seeded?.cleanup(); + seeded = undefined; +}); + +function finder(answer: ProfilePhoto | undefined): { + asked: ProfilePhotoQuery[]; + finder: ProfilePhotoFinder; +} { + const asked: ProfilePhotoQuery[] = []; + return { + asked, + finder: { + findProfilePhoto: async (query) => { + asked.push(query); + return answer; + }, + }, + }; +} + +const LINKEDIN_HIT: ProfilePhoto = { + photoUrl: 'https://media.licdn.com/dms/image/jane.jpg', + pageUrl: 'https://www.linkedin.com/in/jane-smith-123/', + source: 'linkedin', +}; + +describe('sweepProfilePhotos', () => { + test('asks with everything known and records the picture and the profile', async () => { + seeded = await seedDatabase('photos-hit'); + const { db } = seeded; + const { asked, finder: found } = finder(LINKEDIN_HIT); + + const result = await sweepProfilePhotos( + { db, finder: found }, + { workspaceId: SEED.workspaceId }, + ); + + expect(result).toEqual({ looked: 1, found: 1, profiles: 1, remainingToday: 299 }); + expect(asked[0]).toEqual({ + name: 'Jane Smith', + title: 'VP Engineering', + company: 'Acme', + companyDomain: 'acme.com', + }); + + const person = await queryOne<{ avatar_url: string; avatar_source: string }>( + db, + 'SELECT avatar_url, avatar_source FROM people WHERE id = ?', + [SEED.personId], + ); + expect(person?.avatar_url).toBe(LINKEDIN_HIT.photoUrl); + expect(person?.avatar_source).toBe('search'); + + // Research, at a confidence no machine can act on. + const identity = await queryOne<{ handle: string; profile_url: string; confidence: number }>( + db, + `SELECT handle, profile_url, confidence FROM social_identities + WHERE person_id = ? AND network = 'linkedin'`, + [SEED.personId], + ); + expect(identity?.handle).toBe('jane-smith-123'); + expect(identity?.profile_url).toBe('https://www.linkedin.com/in/jane-smith-123/'); + expect(identity?.confidence).toBeLessThan(0.85); + + // Done: nothing left to look up. + expect(await workspacesAwaitingPhotos(db)).toEqual([]); + }); + + test('a miss is remembered and never paid for again', async () => { + seeded = await seedDatabase('photos-miss'); + const { db } = seeded; + const { asked, finder: missed } = finder(undefined); + + const first = await sweepProfilePhotos( + { db, finder: missed }, + { workspaceId: SEED.workspaceId }, + ); + const second = await sweepProfilePhotos( + { db, finder: missed }, + { workspaceId: SEED.workspaceId }, + ); + + expect(first.looked).toBe(1); + expect(second.looked).toBe(0); + expect(asked).toHaveLength(1); + + const person = await queryOne<{ avatar_url: string | null; photo_looked_up_at: string | null }>( + db, + 'SELECT avatar_url, photo_looked_up_at FROM people WHERE id = ?', + [SEED.personId], + ); + expect(person?.avatar_url).toBeNull(); + expect(person?.photo_looked_up_at).not.toBeNull(); + }); + + test('the daily ceiling stops the sweep before it asks', async () => { + seeded = await seedDatabase('photos-cap'); + const { db } = seeded; + const { asked, finder: found } = finder(LINKEDIN_HIT); + + const result = await sweepProfilePhotos( + { db, finder: found, dailyCap: 0 }, + { workspaceId: SEED.workspaceId }, + ); + + expect(result.looked).toBe(0); + expect(asked).toHaveLength(0); + }); +}); diff --git a/packages/pipeline/src/photos.ts b/packages/pipeline/src/photos.ts new file mode 100644 index 0000000..1d38665 --- /dev/null +++ b/packages/pipeline/src/photos.ts @@ -0,0 +1,229 @@ +/** + * A face for each lead. + * + * The digest and the queue list people by name and title, and a name is a + * poor handle on a stranger: "Mark Ramsey — Global Senior Pastor at Citipointe + * Church" reads as a line item, the same line with a photograph reads as a + * person. Gravatar supplies one for about a person in a hundred; for the rest + * a search finds the picture on their LinkedIn profile or their company's own + * team page, and only those — see `@outreachgraph/providers` for why the page + * matters more than the picture. + * + * Bounded twice, because every lookup costs money and most of them miss: + * + * - **Per tick.** A small batch, so a workspace that just enrolled a + * thousand people does not spend a thousand credits in one minute. + * - **Per day.** A ceiling on lookups per workspace per day, counted from + * the `photo_looked_up_at` stamps rather than from memory so a restart + * cannot reset it. + * + * And scoped to people the workspace is actually working — members of an + * active campaign — rather than everyone ever imported. A picture of someone + * nobody will write to is a credit spent on nothing. + */ + +import { newId } from '@outreachgraph/domain'; +import type { ProfilePhotoFinder } from '@outreachgraph/providers'; +import { isLinkedInProfile } from '@outreachgraph/providers'; +import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; + +/** Lookups per workspace per tick. */ +const SWEEP_SIZE = 20; +/** Lookups per workspace per day, absent a configured ceiling. */ +const DEFAULT_DAILY_CAP = 300; + +/** + * How much to believe a LinkedIn URL found by search. + * + * Below the outreach floor on purpose. The result carried the person's name + * and the company we already had for them, which is enough to show a human + * the profile as research and nowhere near enough to act on it. Confirmation + * comes from the human, in LinkedIn's own interface. + */ +const SEARCH_IDENTITY_CONFIDENCE = 0.6; + +export interface PhotoSweepDeps { + readonly db: Client; + readonly finder: ProfilePhotoFinder; + readonly limit?: number; + readonly dailyCap?: number; + readonly now?: Date; +} + +export interface PhotoSweepResult { + readonly looked: number; + readonly found: number; + /** LinkedIn profiles recorded as research identities along the way. */ + readonly profiles: number; + /** Lookups left under today's ceiling after this run. */ + readonly remainingToday: number; +} + +interface Subject { + readonly person_id: string; + readonly display_name: string; + readonly current_title: string | null; + readonly company_name: string | null; + readonly company_domain: string | null; +} + +/** + * Looks up the next batch of people in this workspace's campaigns who have no + * picture and have never been looked up. + * + * People with a message waiting to go out come first: they are the ones about + * to appear in the digest as "written to", and the ones a reviewer is about to + * open in the queue. + */ +export async function sweepProfilePhotos( + deps: PhotoSweepDeps, + input: { readonly workspaceId: string }, +): Promise { + const at = deps.now ?? new Date(); + const dailyCap = deps.dailyCap ?? DEFAULT_DAILY_CAP; + const dayStart = `${at.toISOString().slice(0, 10)}T00:00:00.000Z`; + + const spent = await queryOne<{ n: number }>( + deps.db, + `SELECT count(*) AS n FROM people p + WHERE p.photo_looked_up_at >= ? + AND EXISTS (SELECT 1 FROM campaign_people cp + WHERE cp.person_id = p.id AND cp.workspace_id = ?)`, + [dayStart, input.workspaceId], + ); + + const room = Math.max(dailyCap - Number(spent?.n ?? 0), 0); + const limit = Math.min(deps.limit ?? SWEEP_SIZE, room); + + if (limit === 0) return { looked: 0, found: 0, profiles: 0, remainingToday: 0 }; + + const subjects = await queryAll( + deps.db, + `SELECT p.id AS person_id, p.display_name, p.current_title, + co.name AS company_name, co.domain AS company_domain + FROM people p + JOIN campaign_people cp ON cp.person_id = p.id + JOIN campaigns c ON c.id = cp.campaign_id + LEFT JOIN companies co ON co.id = p.current_company_id + WHERE cp.workspace_id = ? AND c.status IN ('active', 'running') + AND p.status = 'active' + AND p.avatar_url IS NULL AND p.photo_looked_up_at IS NULL + GROUP BY p.id + ORDER BY EXISTS (SELECT 1 FROM recommendations r + WHERE r.person_id = p.id AND r.status = 'pending' + AND r.action NOT IN ('refresh_research', 'observe', 'wait')) DESC, + cp.updated_at DESC + LIMIT ?`, + [input.workspaceId, limit], + ); + + let found = 0; + let profiles = 0; + + for (const subject of subjects) { + const photo = await deps.finder + .findProfilePhoto({ + name: subject.display_name, + title: subject.current_title ?? undefined, + company: subject.company_name ?? undefined, + companyDomain: subject.company_domain ?? undefined, + }) + .catch(() => undefined); + + // A miss is stamped so it is not retried forever; only a hit fills the URL. + if (!photo) { + await deps.db.execute({ + sql: 'UPDATE people SET photo_looked_up_at = ? WHERE id = ?', + args: [now(), subject.person_id], + }); + continue; + } + + found += 1; + await deps.db.execute({ + sql: `UPDATE people SET avatar_url = ?, avatar_source = ?, photo_looked_up_at = ?, + updated_at = ? + WHERE id = ? AND avatar_url IS NULL`, + args: [ + photo.photoUrl, + photo.source === 'linkedin' ? 'search' : 'site', + now(), + now(), + subject.person_id, + ], + }); + + // The profile the picture came from is the more useful half of the + // answer: it is the page the human will open to act. Recorded as research, + // once, and never at a confidence that could let a machine use it. + if ( + photo.source === 'linkedin' && + (await recordProfile(deps.db, subject.person_id, photo.pageUrl)) + ) { + profiles += 1; + } + } + + return { + looked: subjects.length, + found, + profiles, + remainingToday: Math.max(room - subjects.length, 0), + }; +} + +/** Workspaces with campaign members still waiting for a picture. */ +export async function workspacesAwaitingPhotos(db: Client): Promise { + const rows = await queryAll<{ workspace_id: string }>( + db, + `SELECT DISTINCT cp.workspace_id + FROM campaign_people cp + JOIN campaigns c ON c.id = cp.campaign_id + JOIN people p ON p.id = cp.person_id + WHERE c.status IN ('active', 'running') AND p.status = 'active' + AND p.avatar_url IS NULL AND p.photo_looked_up_at IS NULL`, + ); + + return rows.map((row) => row.workspace_id); +} + +async function recordProfile(db: Client, personId: string, pageUrl: string): Promise { + let host: string; + let path: string; + try { + const url = new URL(pageUrl); + host = url.hostname.toLowerCase(); + path = url.pathname; + } catch { + return false; + } + if (!isLinkedInProfile(host, path)) return false; + + const existing = await queryOne<{ id: string }>( + db, + `SELECT id FROM social_identities WHERE person_id = ? AND network = 'linkedin' LIMIT 1`, + [personId], + ); + if (existing) return false; + + const handle = path.split('/').filter(Boolean)[1] ?? null; + const stamp = now(); + + await db.execute({ + sql: `INSERT INTO social_identities (id, person_id, network, handle, profile_url, confidence, + source_type, verified_by, first_seen_at, last_verified_at) + VALUES (?, ?, 'linkedin', ?, ?, ?, 'public_web', ?, ?, ?)`, + args: [ + newId('socialIdentity'), + personId, + handle, + `https://${host}${path}`, + SEARCH_IDENTITY_CONFIDENCE, + JSON.stringify(['search']), + stamp, + stamp, + ], + }); + + return true; +} diff --git a/packages/pipeline/src/pipeline.ts b/packages/pipeline/src/pipeline.ts index 0afeb74..bdb3fe7 100644 --- a/packages/pipeline/src/pipeline.ts +++ b/packages/pipeline/src/pipeline.ts @@ -926,6 +926,23 @@ async function createRecommendation( if (inbox?.contact_email) reachable.push('email'); } + // An address they gave us themselves counts before either of the above. + // + // Imported contacts keep their mailbox in `person_emails`, not in + // `social_identities`, and this list was built from the latter alone — so a + // consented, imported person with nothing else known about them was + // "unreachable" and dead-ended at research. Sixteen thousand of them, in + // production, enrolled in nothing and written to never. + if (!reachable.includes('email')) { + const imported = await queryOne<{ id: string }>( + db, + `SELECT id FROM person_emails WHERE person_id = ? AND workspace_id = ? LIMIT 1`, + [personId, workspaceId], + ); + + if (imported) reachable.push('email'); + } + const score = await queryOne<{ opportunity: number }>( db, 'SELECT opportunity FROM scores WHERE campaign_id = ? AND person_id = ?', diff --git a/packages/pipeline/src/reseed.test.ts b/packages/pipeline/src/reseed.test.ts new file mode 100644 index 0000000..52cdb72 --- /dev/null +++ b/packages/pipeline/src/reseed.test.ts @@ -0,0 +1,113 @@ +/** + * Asking a campaign's seed again. + * + * The cases are about restraint: only an idle campaign, only after the + * interval, only once per interval — and then the right kind of job for the + * kind of seed. + */ + +import { afterEach, describe, expect, test } from 'bun:test'; +import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; +import { seedDatabase, SEED, type SeededDatabase } from '../../../apps/api/src/test-seed'; +import { reseedIdleCampaigns } from './reseed'; + +let seeded: SeededDatabase | undefined; + +afterEach(() => { + seeded?.cleanup(); + seeded = undefined; +}); + +const TEN_DAYS_AGO = new Date(Date.now() - 10 * 86_400_000).toISOString(); + +async function seededCampaign( + db: Client, + options: { kind: 'url' | 'keyword'; value: string; startedAt?: string }, +): Promise { + await db.execute({ + sql: `UPDATE campaigns SET status = 'active', seed_kind = ?, seed_value = ?, started_at = ? + WHERE id = ?`, + args: [options.kind, options.value, options.startedAt ?? TEN_DAYS_AGO, SEED.campaignId], + }); +} + +async function jobs(db: Client): Promise<{ kind: string; payload_json: string; status: string }[]> { + return queryAll(db, 'SELECT kind, payload_json, status FROM jobs WHERE workspace_id = ?', [ + SEED.workspaceId, + ]); +} + +describe('reseedIdleCampaigns', () => { + test('reads a URL seed again after the interval, once', async () => { + seeded = await seedDatabase('reseed-url'); + const { db } = seeded; + await seededCampaign(db, { kind: 'url', value: 'acme.com/team' }); + + const first = await reseedIdleCampaigns(db, { workspaceId: SEED.workspaceId }); + expect(first).toEqual({ considered: 1, queued: 1 }); + + const queued = await jobs(db); + expect(queued).toHaveLength(1); + expect(queued[0]?.kind).toBe('crawl_site'); + expect(JSON.parse(queued[0]?.payload_json ?? '{}')).toEqual({ + url: 'https://acme.com/team', + campaignId: SEED.campaignId, + }); + + const campaign = await queryOne<{ reseeded_at: string | null }>( + db, + 'SELECT reseeded_at FROM campaigns WHERE id = ?', + [SEED.campaignId], + ); + expect(campaign?.reseeded_at).not.toBeNull(); + + // The stamp holds it for another interval. + const second = await reseedIdleCampaigns(db, { workspaceId: SEED.workspaceId }); + expect(second).toEqual({ considered: 0, queued: 0 }); + }); + + test('a keyword seed is discovered again rather than crawled', async () => { + seeded = await seedDatabase('reseed-keyword'); + const { db } = seeded; + await seededCampaign(db, { kind: 'keyword', value: 'dental practices in Austin' }); + + await reseedIdleCampaigns(db, { workspaceId: SEED.workspaceId }); + + const queued = await jobs(db); + expect(queued.map((job) => job.kind)).toEqual(['discover_domains']); + expect(JSON.parse(queued[0]?.payload_json ?? '{}').keyword).toBe('dental practices in Austin'); + }); + + test('a campaign seeded recently is left alone', async () => { + seeded = await seedDatabase('reseed-recent'); + const { db } = seeded; + await seededCampaign(db, { kind: 'url', value: 'acme.com', startedAt: now() }); + + const result = await reseedIdleCampaigns(db, { workspaceId: SEED.workspaceId }); + expect(result.considered).toBe(0); + expect(await jobs(db)).toHaveLength(0); + }); + + test('a campaign with work outstanding is not idle', async () => { + seeded = await seedDatabase('reseed-busy'); + const { db } = seeded; + await seededCampaign(db, { kind: 'url', value: 'acme.com' }); + + await db.execute({ + sql: `INSERT INTO jobs (id, workspace_id, kind, payload_json, status, attempts, max_attempts, + run_after, created_at, updated_at) + VALUES ('job_busy', ?, 'crawl_site', ?, 'pending', 0, 5, ?, ?, ?)`, + args: [ + SEED.workspaceId, + JSON.stringify({ url: 'https://acme.com/about', campaignId: SEED.campaignId }), + now(), + now(), + now(), + ], + }); + + const result = await reseedIdleCampaigns(db, { workspaceId: SEED.workspaceId }); + expect(result.considered).toBe(0); + expect(await jobs(db)).toHaveLength(1); + }); +}); diff --git a/packages/pipeline/src/reseed.ts b/packages/pipeline/src/reseed.ts new file mode 100644 index 0000000..41abb0c --- /dev/null +++ b/packages/pipeline/src/reseed.ts @@ -0,0 +1,112 @@ +/** + * Reading a campaign's seed again. + * + * A campaign is seeded once, at creation: the URL is crawled, or the keyword + * is expanded into companies and those are crawled. Nothing ever returns to + * the seed. So a directory that gains members, a team page that hires, a + * market that a second discovery run would name differently — none of it + * reaches the product, and once the initial crop of research cards has been + * cleared the intake has nothing left to run on. The digest reported "sites + * read 0, new people 0" for ten consecutive days while six campaigns sat + * active. Nothing was broken; nothing was being asked. + * + * This asks again, on a slow clock, for campaigns that are active and idle. + * Idle means no job of theirs is pending or running, so a campaign still + * working through its first crop is left alone. The crawl dedupe key is + * partial — it only blocks a duplicate while the earlier job is outstanding — + * so the same seed queues cleanly, and the pipeline already recognises a + * person it has read before, so a re-read finds only what is new. + */ + +import { now, queryAll, type Client } from '@outreachgraph/db'; +import { emitEvent } from './events'; +import { enqueue } from './queue'; + +/** How long a campaign is left alone after its seed was last read. */ +const DEFAULT_EVERY_DAYS = 7; + +export interface ReseedResult { + readonly considered: number; + readonly queued: number; +} + +interface IdleCampaign { + readonly id: string; + readonly name: string; + readonly seed_kind: string; + readonly seed_value: string; +} + +export async function reseedIdleCampaigns( + db: Client, + input: { readonly workspaceId: string; readonly everyDays?: number; readonly now?: Date }, +): Promise { + const at = input.now ?? new Date(); + const days = input.everyDays ?? DEFAULT_EVERY_DAYS; + const cutoff = new Date(at.getTime() - days * 86_400_000).toISOString(); + + const idle = await queryAll( + db, + `SELECT c.id, c.name, c.seed_kind, c.seed_value + FROM campaigns c + WHERE c.workspace_id = ? + AND c.status IN ('active', 'running') + AND c.seed_kind IN ('url', 'keyword') + AND c.seed_value IS NOT NULL AND trim(c.seed_value) <> '' + AND COALESCE(c.reseeded_at, c.started_at, c.created_at) < ? + AND NOT EXISTS ( + SELECT 1 FROM jobs j + WHERE j.workspace_id = c.workspace_id + AND j.status IN ('pending', 'running') + AND j.payload_json LIKE '%' || c.id || '%') + ORDER BY COALESCE(c.reseeded_at, c.started_at, c.created_at) ASC`, + [input.workspaceId, cutoff], + ); + + let queued = 0; + + for (const campaign of idle) { + const seed = campaign.seed_value.trim(); + + const result = + campaign.seed_kind === 'url' + ? await enqueue(db, { + workspaceId: input.workspaceId, + kind: 'crawl_site', + payload: { + url: /^https?:\/\//i.test(seed) ? seed : `https://${seed}`, + campaignId: campaign.id, + }, + dedupeKey: `crawl:${campaign.id}:${seed.replace(/^https?:\/\//i, '')}`, + }) + : await enqueue(db, { + workspaceId: input.workspaceId, + kind: 'discover_domains', + payload: { keyword: seed, campaignId: campaign.id }, + dedupeKey: `discover:${campaign.id}`, + }); + + // Stamped whether or not a job was queued: a dedupe hit means the work is + // already outstanding, and asking again next tick would not change that. + await db.execute({ + sql: 'UPDATE campaigns SET reseeded_at = ?, updated_at = ? WHERE id = ?', + args: [now(), now(), campaign.id], + }); + + if (!result.queued) continue; + queued += 1; + + await emitEvent(db, { + workspaceId: input.workspaceId, + campaignId: campaign.id, + phase: 'intake', + message: + campaign.seed_kind === 'url' + ? `Reading ${seed} again for anyone new` + : `Looking again for companies matching “${seed}”`, + detail: { seedKind: campaign.seed_kind, seed, everyDays: days }, + }); + } + + return { considered: idle.length, queued }; +} diff --git a/packages/providers/src/index.ts b/packages/providers/src/index.ts index 2f01791..a668b71 100644 --- a/packages/providers/src/index.ts +++ b/packages/providers/src/index.ts @@ -28,6 +28,17 @@ export { export { deriveEvidence, type EvidenceContext } from './evidence'; +export { + ValueSerpClient, + carriesName, + corroborate, + isLinkedInProfile, + type ProfilePhoto, + type ProfilePhotoFinder, + type ProfilePhotoQuery, + type ValueSerpOptions, +} from './valueserp'; + export { FixtureProvider, FIXTURE_CANDIDATES, diff --git a/packages/providers/src/valueserp/index.ts b/packages/providers/src/valueserp/index.ts new file mode 100644 index 0000000..7e4cefc --- /dev/null +++ b/packages/providers/src/valueserp/index.ts @@ -0,0 +1,199 @@ +/** + * ValueSERP: a picture of the person, found the way a human would find one. + * + * Given a name, a title and a company, a person searches Google Images and + * takes the headshot from the LinkedIn profile or the company's team page. + * This does the same through a search API, and it is deliberately no cleverer + * than that: one query, the first result whose page corroborates the person, + * nothing otherwise. + * + * What makes it precise rather than merely plausible is the page the image + * sits on, not the image. A picture is accepted only when it was published on + * a LinkedIn profile whose title carries the person's name, or on the + * company's own domain. "A face that came up for this name" is not evidence of + * anything; "the picture on the team page of the company we already know they + * work at" is. Identity precision beats recall (PRD §9), and a wrong face on a + * lead is worse than no face. + * + * This reads search results. It never fetches LinkedIn itself, never logs in, + * and never acts there — the human still does that in LinkedIn's own + * interface. What it records is research: a URL the person published under + * their own name. + */ + +export interface ProfilePhotoQuery { + readonly name: string; + readonly title?: string | undefined; + readonly company?: string | undefined; + /** The company's own domain; a picture published there is accepted. */ + readonly companyDomain?: string | undefined; +} + +export interface ProfilePhoto { + readonly photoUrl: string; + /** The page the picture was published on — the corroborating evidence. */ + readonly pageUrl: string; + readonly source: 'linkedin' | 'site'; +} + +export interface ProfilePhotoFinder { + findProfilePhoto(query: ProfilePhotoQuery): Promise; +} + +export interface ValueSerpOptions { + readonly apiKey: string; + readonly baseUrl?: string; + readonly fetchImpl?: typeof fetch; + readonly timeoutMs?: number; +} + +const DEFAULT_BASE = 'https://api.valueserp.com'; + +/** One image result, as much of it as this adapter reads. */ +interface ImageResult { + readonly title?: string; + readonly link?: string; + readonly image?: string; + readonly original?: string; + readonly thumbnail?: string; + readonly domain?: string; + readonly source?: string; +} + +export class ValueSerpClient implements ProfilePhotoFinder { + private readonly apiKey: string; + private readonly baseUrl: string; + private readonly fetchImpl: typeof fetch; + private readonly timeoutMs: number; + + constructor(options: ValueSerpOptions) { + this.apiKey = options.apiKey; + this.baseUrl = (options.baseUrl ?? DEFAULT_BASE).replace(/\/$/, ''); + this.fetchImpl = options.fetchImpl ?? fetch; + this.timeoutMs = options.timeoutMs ?? 10_000; + } + + /** + * One search, one answer, or none. + * + * A miss is the common case and costs a credit either way, which is why the + * caller stamps the person as looked up regardless. Network failures are a + * miss too: this runs in a sweep, and one flaky response should cost that + * person's picture, not the run. + */ + async findProfilePhoto(query: ProfilePhotoQuery): Promise { + const name = query.name.trim(); + if (!name) return undefined; + + const terms = [`"${name}"`, query.company?.trim(), query.title?.trim()].filter(Boolean); + + const url = new URL('/search', this.baseUrl); + url.searchParams.set('api_key', this.apiKey); + url.searchParams.set('search_type', 'images'); + url.searchParams.set('q', terms.join(' ')); + + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), this.timeoutMs); + + try { + const response = await this.fetchImpl(url, { + signal: controller.signal, + headers: { accept: 'application/json' }, + }); + if (!response.ok) return undefined; + + const body = (await response.json()) as { image_results?: ImageResult[] }; + const results = Array.isArray(body.image_results) ? body.image_results : []; + + for (const result of results) { + const match = corroborate(result, query); + if (match) return match; + } + + return undefined; + } catch { + return undefined; + } finally { + clearTimeout(timer); + } + } +} + +/** + * Whether one result is evidence enough. + * + * Exported for the tests and for anyone auditing what "corroborated" means: + * the page is a LinkedIn profile titled with the person's name, or the page is + * on the company's own domain and titled with their name. + */ +export function corroborate( + result: ImageResult, + query: ProfilePhotoQuery, +): ProfilePhoto | undefined { + const photoUrl = firstHttp(result.image, result.original, result.thumbnail); + const pageUrl = firstHttp(result.link, result.source); + if (!photoUrl || !pageUrl) return undefined; + + const title = result.title ?? ''; + if (!carriesName(title, query.name)) return undefined; + + let host: string; + let path: string; + try { + const page = new URL(pageUrl); + host = page.hostname.toLowerCase(); + path = page.pathname; + } catch { + return undefined; + } + + if (isLinkedInProfile(host, path)) return { photoUrl, pageUrl, source: 'linkedin' }; + + const domain = query.companyDomain + ?.trim() + .toLowerCase() + .replace(/^www\./, ''); + if (domain && (host === domain || host.endsWith(`.${domain}`))) { + return { photoUrl, pageUrl, source: 'site' }; + } + + return undefined; +} + +/** `linkedin.com/in/` on any LinkedIn host — a profile, not a company page. */ +export function isLinkedInProfile(host: string, path: string): boolean { + const onLinkedIn = host === 'linkedin.com' || host.endsWith('.linkedin.com'); + return onLinkedIn && /^\/in\/[^/]+/.test(path); +} + +/** + * Every part of the name that could be a name, present in the title. + * + * Initials and particles ("J", "de") are too short to mean anything and are + * not required; "Mark Ramsey" needs both `mark` and `ramsey`. Case and + * diacritics are folded so "Klaudia Majcher" matches "KLAUDIA MAJCHER" and + * "Stefan Wienold" matches a title that spells it with a different accent. + */ +export function carriesName(title: string, name: string): boolean { + const haystack = fold(title); + const parts = fold(name) + .split(/[\s,]+/) + .filter((part) => part.length >= 2); + + if (parts.length === 0) return false; + return parts.every((part) => haystack.includes(part)); +} + +function fold(value: string): string { + return value + .normalize('NFD') + .replace(/[\u0300-\u036f]/g, '') + .toLowerCase(); +} + +function firstHttp(...candidates: (string | undefined)[]): string | undefined { + for (const candidate of candidates) { + if (typeof candidate === 'string' && /^https?:\/\//i.test(candidate)) return candidate; + } + return undefined; +} diff --git a/packages/providers/src/valueserp/valueserp.test.ts b/packages/providers/src/valueserp/valueserp.test.ts new file mode 100644 index 0000000..d45d4b3 --- /dev/null +++ b/packages/providers/src/valueserp/valueserp.test.ts @@ -0,0 +1,132 @@ +/** + * What counts as the right face. + * + * The adapter's whole value is in what it refuses. A search for a common name + * returns a page of strangers; only a result whose page vouches for the person + * — their LinkedIn profile, their employer's site — is worth attaching to them. + */ + +import { describe, expect, test } from 'bun:test'; +import { ValueSerpClient, carriesName, corroborate, isLinkedInProfile } from './index'; + +const JANE = { + name: 'Jane Smith', + title: 'VP Engineering', + company: 'Acme', + companyDomain: 'acme.com', +}; + +describe('corroborate', () => { + test('accepts a LinkedIn profile titled with the name', () => { + const match = corroborate( + { + title: 'Jane Smith - VP Engineering - Acme | LinkedIn', + link: 'https://www.linkedin.com/in/janesmith', + image: 'https://media.licdn.com/dms/image/jane.jpg', + }, + JANE, + ); + expect(match).toEqual({ + photoUrl: 'https://media.licdn.com/dms/image/jane.jpg', + pageUrl: 'https://www.linkedin.com/in/janesmith', + source: 'linkedin', + }); + }); + + test('accepts the company’s own site', () => { + const match = corroborate( + { + title: 'Our team — Jane Smith', + link: 'https://www.acme.com/team', + image: 'https://www.acme.com/img/jane.jpg', + }, + JANE, + ); + expect(match?.source).toBe('site'); + }); + + test('rejects a LinkedIn company page, a stranger’s site, and a title without the name', () => { + const image = 'https://cdn.example/jane.jpg'; + + expect( + corroborate( + { title: 'Jane Smith', link: 'https://www.linkedin.com/company/acme', image }, + JANE, + ), + ).toBeUndefined(); + expect( + corroborate({ title: 'Jane Smith', link: 'https://someblog.example/jane', image }, JANE), + ).toBeUndefined(); + expect( + corroborate( + { title: 'Jane Doe - Acme | LinkedIn', link: 'https://linkedin.com/in/jd', image }, + JANE, + ), + ).toBeUndefined(); + }); + + test('needs both an image and a page', () => { + expect( + corroborate({ title: 'Jane Smith | LinkedIn', link: 'https://linkedin.com/in/js' }, JANE), + ).toBeUndefined(); + }); +}); + +describe('carriesName', () => { + test('folds case and accents and ignores initials', () => { + expect(carriesName('KLAUDIA MAJCHER – Właściciel', 'Klaudia Majcher')).toBe(true); + expect(carriesName('Stefan Wienold | Vertriebstrainer', 'Stefan Wienöld')).toBe(true); + expect(carriesName('Mark J. Ramsey — Citipointe', 'Mark J Ramsey')).toBe(true); + expect(carriesName('Mark Ramsay — Citipointe', 'Mark Ramsey')).toBe(false); + }); +}); + +describe('isLinkedInProfile', () => { + test('is a person, on any LinkedIn host', () => { + expect(isLinkedInProfile('www.linkedin.com', '/in/jane')).toBe(true); + expect(isLinkedInProfile('uk.linkedin.com', '/in/jane/')).toBe(true); + expect(isLinkedInProfile('linkedin.com', '/company/acme')).toBe(false); + expect(isLinkedInProfile('notlinkedin.com', '/in/jane')).toBe(false); + }); +}); + +describe('ValueSerpClient', () => { + function respond(body: unknown, status = 200): { calls: URL[]; fetchImpl: typeof fetch } { + const calls: URL[] = []; + const fetchImpl = (async (input: string | URL | Request) => { + calls.push(new URL(String(input instanceof Request ? input.url : input))); + return new Response(JSON.stringify(body), { + status, + headers: { 'content-type': 'application/json' }, + }); + }) as unknown as typeof fetch; + return { calls, fetchImpl }; + } + + test('searches images for the quoted name with title and company', async () => { + const { calls, fetchImpl } = respond({ + image_results: [ + { title: 'Someone Else', link: 'https://linkedin.com/in/else', image: 'https://x/1.jpg' }, + { + title: 'Jane Smith - Acme | LinkedIn', + link: 'https://linkedin.com/in/jane', + image: 'https://media.licdn.com/jane.jpg', + }, + ], + }); + + const client = new ValueSerpClient({ apiKey: 'k', fetchImpl }); + const photo = await client.findProfilePhoto(JANE); + + expect(photo?.photoUrl).toBe('https://media.licdn.com/jane.jpg'); + expect(calls[0]?.searchParams.get('search_type')).toBe('images'); + expect(calls[0]?.searchParams.get('q')).toBe('"Jane Smith" Acme VP Engineering'); + expect(calls[0]?.searchParams.get('api_key')).toBe('k'); + }); + + test('a refused request is a miss, not an error', async () => { + const { fetchImpl } = respond({ request_info: { success: false } }, 401); + const client = new ValueSerpClient({ apiKey: 'k', fetchImpl }); + expect(await client.findProfilePhoto(JANE)).toBeUndefined(); + }); +});