From ad6a05807fac80697cee6b1a61ae78e003376b5e Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Wed, 7 Oct 2026 02:12:39 +0000 Subject: [PATCH 1/2] List quality: verify before sending, stop a campaign over 2% bounce, retire bounced addresses Hunter's outreach planner, run by the worker instead of remembered by a person: - Every address is verified before its first message and again after 90 days (MX, then an SMTP RCPT probe where port 25 allows; a blocked port sends on MX alone). Verdicts are cached in email_verifications, 25 fresh checks per run. - A bounce the delivery report pins to an address marks it invalid for good and stops the person's cadences. The sender query skips invalid addresses. - A campaign whose bounce rate passes 2% over 50+ sends pauses itself, re-verifies every queued address, drops the failures and resumes on a fresh window once a complete pass finds nothing waiting (campaign_list_health). - Accept-all addresses send only while the campaign bounces at or under 1%. - GET /autogtm/campaigns/:id/list-health, og leads health, MCP get_list_health. Migration 0053. Co-Authored-By: Claude Opus 5.5 --- apps/api/src/autogtm-docs.ts | 27 ++ apps/api/src/autogtm.ts | 25 ++ apps/cli/src/commands.ts | 25 +- apps/mcp/src/tools.ts | 18 ++ apps/server/src/index.ts | 3 + migrations-pg/0053_list_quality.sql | 20 ++ migrations/0053_list_quality.sql | 34 +++ packages/domain/src/index.ts | 1 + packages/domain/src/list-quality.ts | 118 ++++++++ packages/pipeline/src/autopilot.ts | 102 ++++++- packages/pipeline/src/index.ts | 13 + packages/pipeline/src/list-quality.test.ts | 286 ++++++++++++++++++ packages/pipeline/src/list-quality.ts | 332 +++++++++++++++++++++ packages/pipeline/src/receive-email.ts | 19 ++ 14 files changed, 1020 insertions(+), 3 deletions(-) create mode 100644 migrations-pg/0053_list_quality.sql create mode 100644 migrations/0053_list_quality.sql create mode 100644 packages/domain/src/list-quality.ts create mode 100644 packages/pipeline/src/list-quality.test.ts create mode 100644 packages/pipeline/src/list-quality.ts diff --git a/apps/api/src/autogtm-docs.ts b/apps/api/src/autogtm-docs.ts index 11bd9cc..7fb153a 100644 --- a/apps/api/src/autogtm-docs.ts +++ b/apps/api/src/autogtm-docs.ts @@ -646,6 +646,33 @@ export const OPERATIONS: readonly Operation[] = [ }, }, }, + { + method: 'get', + path: '/autogtm/campaigns/{campaign_id}/list-health', + id: 'getListHealth', + tag: 'Campaigns', + summary: 'Bounce rate, the 2% pause, and address verdicts', + description: + 'Every address is verified before its first message and again after 90 days (MX, then an ' + + 'SMTP RCPT probe where the host allows port 25). A campaign whose bounce rate passes 2% over ' + + '50+ sends pauses itself, re-verifies every queued address, drops the ones that fail and ' + + 'resumes on a fresh window. Accept-all addresses only send while the campaign bounces at or ' + + 'under 1%. A bounced address is never sent to again.', + params: [idParam('campaign_id', 'The campaign.')], + response: { + type: 'object', + properties: { + sends: { type: 'integer' }, + bounces: { type: 'integer' }, + bounce_rate: { type: 'number' }, + max_bounce_rate: { type: 'number' }, + window_start: { type: ['string', 'null'] }, + paused: { type: 'boolean' }, + paused_at: { type: ['string', 'null'] }, + addresses: { type: 'object' }, + }, + }, + }, { method: 'get', path: '/autogtm/campaigns/import/{task_id}', diff --git a/apps/api/src/autogtm.ts b/apps/api/src/autogtm.ts index 1877661..416301a 100644 --- a/apps/api/src/autogtm.ts +++ b/apps/api/src/autogtm.ts @@ -31,6 +31,7 @@ import { CONTACT_PRICE_USD, isConsumerMailDomain, mapHeaders, + MAX_BOUNCE_RATE, newId, parseCsv, replyRate, @@ -53,6 +54,7 @@ import { type LeadEnrichDeps, finishContactImport, importContactChunk, + listHealthReport, normaliseDomain, peopleMatchingKeys, recordDiscovered, @@ -971,6 +973,29 @@ export function autogtmRoutes(deps: AutogtmDeps): Hono { }); }); + /** + * The campaign's list quality: bounce rate on the current window, whether + * the 2% gate has it paused for re-verification, and the verdicts on the + * addresses it has sent to. + */ + r.get('/campaigns/:id/list-health', async (c) => { + const actor = c.get('actor'); + const db = c.get('db'); + const campaign = await ownedCampaign(db, actor.workspaceId, c.req.param('id')); + const report = await listHealthReport(db, campaign.id); + return c.json({ + campaign_id: campaign.id, + sends: report.sends, + bounces: report.bounces, + bounce_rate: report.bounceRate, + max_bounce_rate: MAX_BOUNCE_RATE, + window_start: report.windowStart, + paused: report.pausedAt !== null, + paused_at: report.pausedAt, + addresses: report.addresses, + }); + }); + r.get('/campaigns/:id/enrichment', async (c) => { const actor = c.get('actor'); const db = c.get('db'); diff --git a/apps/cli/src/commands.ts b/apps/cli/src/commands.ts index 58cafcd..973d565 100644 --- a/apps/cli/src/commands.ts +++ b/apps/cli/src/commands.ts @@ -306,6 +306,25 @@ async function runLeads({ client, args, flags }: CommandContext): Promise'); + const health = (await client.get( + `/autogtm/campaigns/${encodeURIComponent(target)}/list-health`, + )) as Record; + const addresses = (health.addresses ?? {}) as Record; + const rate = Number(health.bounce_rate ?? 0); + return [ + `${text(health, 'sends', '0')} sent, ${text(health, 'bounces', '0')} bounced ` + + `(${(rate * 100).toFixed(1)}%, stop line 2%)`, + health.paused + ? `PAUSED since ${text(health, 'paused_at')}: re-verifying every queued address, resumes by itself` + : 'sending', + `addresses: ${text(addresses, 'valid', '0')} valid, ${text(addresses, 'catch_all', '0')} accept-all, ` + + `${text(addresses, 'unverified', '0')} MX-only, ${text(addresses, 'invalid', '0')} invalid, ` + + `${text(addresses, 'unchecked', '0')} not yet checked`, + ].join('\n'); + } + if (verb === 'allow' || verb === 'hold') { if (!target) throw new Error(`usage: og leads ${verb} `); await client.post(`/autogtm/leads/${encodeURIComponent(target)}/screening`, { @@ -316,7 +335,9 @@ async function runLeads({ client, args, flags }: CommandContext): Promise { @@ -1695,7 +1716,7 @@ export const COMMANDS: readonly Command[] = [ { name: 'leads', usage: - 'og leads add --consent-source "" [--allow-flagged] [--keep-project-duplicates] [--report out.csv] | report [--csv] | screened [--all] | enrich [--max ] | enrichment | allow | hold ', + 'og leads add --consent-source "" [--allow-flagged] [--keep-project-duplicates] [--report out.csv] | report [--csv] | screened [--all] | enrich [--max ] | enrichment | health | allow | hold ', summary: 'Add a CSV of leads to a running campaign, with a per-row report and screening.', run: runLeads, }, diff --git a/apps/mcp/src/tools.ts b/apps/mcp/src/tools.ts index b1ea2fe..2759cfb 100644 --- a/apps/mcp/src/tools.ts +++ b/apps/mcp/src/tools.ts @@ -1050,6 +1050,24 @@ export const TOOLS: readonly ToolDefinition[] = [ `/autogtm/campaigns/${encodeURIComponent(require(args, 'campaignId'))}/enrichment`, ), }, + { + name: 'get_list_health', + title: 'A campaign’s bounce rate and address verdicts', + description: + 'Bounces against the 2% stop line, whether the campaign is paused re-verifying its queue ' + + '(it resumes by itself), and how many sent-to addresses are valid, accept-all, MX-only, ' + + 'invalid or unchecked.', + readOnly: true, + inputSchema: { + type: 'object', + properties: { campaignId: { type: 'string' } }, + required: ['campaignId'], + }, + run: (client, args) => + client.get( + `/autogtm/campaigns/${encodeURIComponent(require(args, 'campaignId'))}/list-health`, + ), + }, { name: 'list_inbox', title: 'List conversations', diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index efa0273..685a49c 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -1390,6 +1390,9 @@ async function tick(): Promise { // reported. Without this the sweep only ever says "no drafted // message", once per tick, forever. ...(model ? { model } : {}), + // Verify before sending: MX always, an SMTP RCPT probe while port 25 + // answers. A blocked port degrades to MX-only, never to a hold. + verifier: smtpProber && !smtpProber.blocked() ? { smtp: smtpProber } : {}, }, workspace.id, ); diff --git a/migrations-pg/0053_list_quality.sql b/migrations-pg/0053_list_quality.sql new file mode 100644 index 0000000..456fcd8 --- /dev/null +++ b/migrations-pg/0053_list_quality.sql @@ -0,0 +1,20 @@ +-- 0053_list_quality.sql (Postgres). See migrations/0053_list_quality.sql for the reasoning. +-- Two new, empty tables. Nothing existing is altered. + +create table if not exists email_verifications ( + address text PRIMARY KEY, + status text NOT NULL, + reason text, + mx text, + checked_at text NOT NULL +); + +create table if not exists campaign_list_health ( + campaign_id text PRIMARY KEY REFERENCES campaigns(id) ON DELETE CASCADE, + workspace_id text NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE, + window_start text NOT NULL, + paused_at text, + paused_rate double precision, + resumed_at text, + updated_at text NOT NULL +); diff --git a/migrations/0053_list_quality.sql b/migrations/0053_list_quality.sql new file mode 100644 index 0000000..ece9f3a --- /dev/null +++ b/migrations/0053_list_quality.sql @@ -0,0 +1,34 @@ +-- 0053_list_quality.sql +-- +-- Verify before sending, and stop a campaign that bounces (Hunter's outreach +-- planner, run by the worker rather than remembered by a person). +-- +-- email_verifications is the verdict on one address: valid, catch_all, +-- unverified (the domain takes mail but no server could be asked) or invalid. +-- Keyed by the lower-cased address and shared across workspaces, because +-- whether a mailbox exists is a fact about the mailbox. A bounce writes +-- `invalid` here, which is what keeps a bounced address from ever being sent to +-- again. checked_at is what the 90-day re-verify reads. +-- +-- campaign_list_health holds a campaign's bounce gate. window_start is where +-- its bounce rate is counted from; paused_at is set when the rate passed 2% +-- over 50+ sends, and cleared (with window_start moved forward) once every +-- queued address has been re-verified since the pause. + +CREATE TABLE IF NOT EXISTS email_verifications ( + address TEXT PRIMARY KEY, + status TEXT NOT NULL, + reason TEXT, + mx TEXT, + checked_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS campaign_list_health ( + campaign_id TEXT PRIMARY KEY REFERENCES campaigns(id) ON DELETE CASCADE, + workspace_id TEXT NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE, + window_start TEXT NOT NULL, + paused_at TEXT, + paused_rate REAL, + resumed_at TEXT, + updated_at TEXT NOT NULL +); diff --git a/packages/domain/src/index.ts b/packages/domain/src/index.ts index 64dac34..bb33ebb 100644 --- a/packages/domain/src/index.ts +++ b/packages/domain/src/index.ts @@ -38,4 +38,5 @@ export * from './sender-pool'; export * from './replies'; export * from './webhooks'; export * from './mailbox-health'; +export * from './list-quality'; export * from './warmup'; diff --git a/packages/domain/src/list-quality.ts b/packages/domain/src/list-quality.ts new file mode 100644 index 0000000..1b0ab0b --- /dev/null +++ b/packages/domain/src/list-quality.ts @@ -0,0 +1,118 @@ +/** + * List quality: the rules every send list has to meet before a message leaves. + * + * These are the standards from Hunter's outreach planner, applied by the + * machine instead of by a person remembering to: + * + * - **Verify before sending.** An address is checked (MX, then an SMTP RCPT + * probe where port 25 allows it) before its first message, and again once + * the check is older than `REVERIFY_AFTER_DAYS`: roughly a fifth of B2B + * addresses decay every year, so a list verified last spring is not. + * - **Under 2% bounce.** A campaign whose bounce rate passes + * `MAX_BOUNCE_RATE` over at least `MIN_BOUNCE_SAMPLE` sends stops, re-verifies + * everyone still queued in it, and starts again on a fresh window. Nobody + * has to notice, pause, re-verify and resume by hand. + * - **Accept-all addresses are a separate list.** A catch-all domain says yes + * to every recipient, so its addresses bounce more than verified ones. They + * are sent only while the campaign has room under the gate + * (`CATCH_ALL_MAX_RATE`), so they can never be the thing that trips it. + * + * Pure functions. Reading and writing the verification cache is the pipeline's. + */ + +export const MAX_BOUNCE_RATE = 0.02; +export const MIN_BOUNCE_SAMPLE = 50; +export const REVERIFY_AFTER_DAYS = 90; +/** Catch-all addresses send only while the campaign bounces at or under this. */ +export const CATCH_ALL_MAX_RATE = 0.01; + +/** + * What is known about one address. + * + * - `valid` — a server accepted it, on a domain that refuses made-up names. + * - `catch_all` — the domain accepts every recipient, so the yes proves nothing. + * - `unverified` — the domain takes mail but no server could be asked (port 25 + * blocked, greeting refused). Sendable: this is the most a cloud host can learn. + * - `invalid` — no mail exchanger, a server refused the recipient, or it bounced. + */ +export type AddressStatus = 'valid' | 'catch_all' | 'unverified' | 'invalid'; + +export interface AddressVerification { + readonly status: AddressStatus; + readonly checkedAt: string; + readonly reason?: string | null; +} + +/** True when the check is recent enough to trust without asking again. */ +export function verificationFresh( + checkedAt: string, + at: Date, + options: { readonly notBefore?: string | null } = {}, +): boolean { + const stamp = Date.parse(checkedAt); + if (Number.isNaN(stamp)) return false; + // A campaign that tripped the bounce gate wants every address re-checked + // from that moment on, however recent the last check was. + if (options.notBefore) { + const floor = Date.parse(options.notBefore); + if (!Number.isNaN(floor) && stamp < floor) return false; + } + return at.getTime() - stamp < REVERIFY_AFTER_DAYS * 86_400_000; +} + +/** What the verifier saw, reduced to the fields this decision needs. */ +export interface VerifierEvidence { + /** Mail exchangers found. Empty means the domain takes no mail. */ + readonly mx: readonly string[]; + readonly smtp: 'probed' | 'unavailable' | 'skipped'; + readonly catchAll?: boolean | undefined; + readonly verdict: 'accepted' | 'rejected' | 'unknown'; +} + +export function statusFromEvidence(evidence: VerifierEvidence): { + status: AddressStatus; + reason: string; +} { + if (evidence.mx.length === 0) return { status: 'invalid', reason: 'the domain takes no mail' }; + if (evidence.smtp !== 'probed') { + return { status: 'unverified', reason: 'the domain takes mail; no server could be asked' }; + } + if (evidence.catchAll === true) { + return { status: 'catch_all', reason: 'the domain accepts every address' }; + } + if (evidence.verdict === 'rejected') { + return { status: 'invalid', reason: 'the mail server refused this address' }; + } + if (evidence.verdict === 'accepted') + return { status: 'valid', reason: 'the mail server accepted it' }; + return { status: 'unverified', reason: 'the mail server would not say' }; +} + +export interface BounceWindow { + readonly sends: number; + readonly bounces: number; +} + +export function bounceRate(window: BounceWindow): number { + return window.sends > 0 ? window.bounces / window.sends : 0; +} + +/** + * True when the campaign must stop and re-verify. + * + * Below the sample size the rate is noise: one bounce in ten sends is 10% + * and says nothing. Above it, more than 2% is the planner's stop line. + */ +export function bounceGateTripped(window: BounceWindow): boolean { + return window.sends >= MIN_BOUNCE_SAMPLE && bounceRate(window) > MAX_BOUNCE_RATE; +} + +/** True when a catch-all address may go out without risking the gate. */ +export function catchAllAllowed(window: BounceWindow): boolean { + return bounceRate(window) <= CATCH_ALL_MAX_RATE; +} + +/** Percent with one decimal, for messages people read. */ +export function formatRate(rate: number): string { + return `${(Math.round(rate * 1000) / 10).toFixed(1)}%`; +} diff --git a/packages/pipeline/src/autopilot.ts b/packages/pipeline/src/autopilot.ts index 2d05687..5d28283 100644 --- a/packages/pipeline/src/autopilot.ts +++ b/packages/pipeline/src/autopilot.ts @@ -50,6 +50,14 @@ import { import { budgetStatus } from './metering'; import { holdReason } from './lead-screen'; import { ownDomains, refuseRecipient } from './recipient-guard'; +import { catchAllAllowed, formatRate } from '@outreachgraph/domain'; +import type { VerifierDeps } from '@outreachgraph/providers'; +import { + applyCampaignBounceGate, + resumeCampaign, + verifyAddress, + type CampaignBounceState, +} from './list-quality'; export interface AutopilotDeps { readonly db: Client; @@ -89,6 +97,14 @@ export interface AutopilotDeps { */ readonly model?: TextModel; readonly now?: Date; + /** + * How addresses are checked before their first message and every 90 days + * after (MX, then an SMTP RCPT probe where port 25 allows). Omit it and only + * verdicts already cached are enforced — a known bounce still never sends. + */ + readonly verifier?: VerifierDeps | undefined; + /** Fresh checks one run may make; the rest wait for the next tick. */ + readonly maxVerificationsPerRun?: number; /** Where holds are remembered between runs. Defaults to the process-wide one. */ readonly holdLedger?: HoldLedger; } @@ -168,6 +184,13 @@ const MAX_SEND_ATTEMPTS = 3; */ const CANDIDATE_CEILING = 2000; +/** + * Fresh address checks per run. Each is a DNS lookup and possibly an SMTP + * session of up to eight seconds, and a tick runs every minute; a 600-lead + * queue is verified in under half an hour without one tick stalling the rest. + */ +const DEFAULT_VERIFICATIONS_PER_RUN = 25; + /** * Why a card is currently held, remembered across runs. * @@ -293,6 +316,12 @@ export function describeHold(reason: string): string { if (/talking to this person is paused/i.test(reason)) { return 'the mailbox already talking to them is paused'; } + if (/campaign paused/i.test(reason)) return 'the campaign bounced over 2% and is re-verifying'; + if (/failed verification/i.test(reason)) return 'the address failed verification'; + if (/accept-all/i.test(reason)) return 'an accept-all address, held while bounces are above 1%'; + if (/address to be verified|could not be checked/i.test(reason)) { + return 'waiting for the address to be verified'; + } if (/budget/i.test(reason)) return "over the plan's monthly allowance"; return reason; } @@ -373,9 +402,13 @@ export async function runAutopilot( (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) <> '' + AND NOT EXISTS (SELECT 1 FROM email_verifications ev + WHERE ev.address = lower(trim(si.handle)) AND ev.status = 'invalid') 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 + AND NOT EXISTS (SELECT 1 FROM email_verifications ev + WHERE ev.address = lower(trim(pe.address)) AND ev.status = 'invalid') ORDER BY pe.created_at LIMIT 1) ) AS person_email, (SELECT COUNT(*) FROM actions a @@ -425,6 +458,14 @@ export async function runAutopilot( const seen = new Set(); let completed = true; + // List quality, per run: each campaign's bounce gate is read once, fresh + // address checks are budgeted, and a paused campaign counts what it still + // has waiting so a complete pass with nothing left can start it again. + const campaignStates = new Map(); + const awaitingRecheck = new Map(); + const droppedOnRecheck = new Map(); + let verificationsLeft = deps.maxVerificationsPerRun ?? DEFAULT_VERIFICATIONS_PER_RUN; + for (const row of candidates) { if (today >= cap) { completed = false; @@ -486,6 +527,55 @@ export async function runAutopilot( continue; } + // ------------------------------------------------------- list quality + // + // Verify before sending, under 2% bounce, accept-all kept apart. Checked + // before the address limits because a paused campaign spends this pass + // re-verifying its queue instead of sending, and a card whose address is + // dead should say so rather than "already written to this week". + let campaignState = campaignStates.get(row.campaign_id); + if (!campaignState) { + campaignState = await applyCampaignBounceGate(db, workspaceId, row.campaign_id, at); + campaignStates.set(row.campaign_id, campaignState); + } + + const verified = await verifyAddress(db, recipient.address, { + verifier: deps.verifier, + at, + notBefore: campaignState.pausedAt, + allowCheck: verificationsLeft > 0, + }); + if (verified.kind === 'known' && verified.checked) verificationsLeft -= 1; + + if (campaignState.pausedAt) { + if (verified.kind === 'pending') { + awaitingRecheck.set(row.campaign_id, (awaitingRecheck.get(row.campaign_id) ?? 0) + 1); + } else if (verified.verification.status === 'invalid') { + droppedOnRecheck.set(row.campaign_id, (droppedOnRecheck.get(row.campaign_id) ?? 0) + 1); + } + await note( + `campaign paused: ${formatRate(campaignState.rate)} of its messages bounced, ` + + 're-verifying its list before sending again', + ); + continue; + } + + if (verified.kind === 'pending') { + await note(verified.reason); + continue; + } + if (verified.verification.status === 'invalid') { + await note(`the address failed verification (${verified.verification.reason ?? 'invalid'})`); + continue; + } + if (verified.verification.status === 'catch_all' && !catchAllAllowed(campaignState.window)) { + await note( + `an accept-all address, held while this campaign's bounce rate ` + + `(${formatRate(campaignState.rate)}) is above 1%`, + ); + continue; + } + const budget = safeJson(row.budget_json); const cooldown = typeof budget.minHoursBetweenActions === 'number' @@ -824,7 +914,17 @@ export async function runAutopilot( } } - if (completed) ledger.retain(workspaceId, seen); + if (completed) { + ledger.retain(workspaceId, seen); + + // A paused campaign whose whole queue was re-checked on a complete pass + // starts again. An incomplete pass proves nothing about cards it never + // reached, so it never resumes anything. + for (const [campaignId, state] of campaignStates) { + if (!state.pausedAt || (awaitingRecheck.get(campaignId) ?? 0) > 0) continue; + await resumeCampaign(db, workspaceId, campaignId, at, droppedOnRecheck.get(campaignId) ?? 0); + } + } for (const transport of transports.values()) transport.close(); if (sender?.mailer instanceof SmtpMailer) sender.mailer.close(); diff --git a/packages/pipeline/src/index.ts b/packages/pipeline/src/index.ts index 1d11513..c37746b 100644 --- a/packages/pipeline/src/index.ts +++ b/packages/pipeline/src/index.ts @@ -551,3 +551,16 @@ export { type IdeaStatus, type IdeaSummary, } from './ideas'; +export { + applyCampaignBounceGate, + campaignBounceState, + listHealthReport, + markAddressBounced, + readVerification, + recordVerification, + resumeCampaign, + verifyAddress, + type CampaignBounceState, + type ListHealthReport, + type VerifyOutcome, +} from './list-quality'; diff --git a/packages/pipeline/src/list-quality.test.ts b/packages/pipeline/src/list-quality.test.ts new file mode 100644 index 0000000..a8da47c --- /dev/null +++ b/packages/pipeline/src/list-quality.test.ts @@ -0,0 +1,286 @@ +import { afterEach, describe, expect, test } from 'bun:test'; +import { now, queryOne, type Client } from '@outreachgraph/db'; +import type { Mailer, Message, SendResult } from '@outreachgraph/email'; +import type { SmtpProber, VerifierDeps } from '@outreachgraph/providers'; +import { + bounceGateTripped, + catchAllAllowed, + statusFromEvidence, + verificationFresh, +} from '@outreachgraph/domain'; +import { seedDatabase, SEED, type SeededDatabase } from '../../../apps/api/src/test-seed'; +import { HoldLedger, runAutopilot } from './autopilot'; +import { + campaignBounceState, + listHealthReport, + markAddressBounced, + readVerification, + recordVerification, +} from './list-quality'; + +let seeded: SeededDatabase | undefined; + +afterEach(() => { + seeded?.cleanup(); + seeded = undefined; +}); + +function recordingMailer(): { sent: Message[]; mailer: Mailer } { + const sent: Message[] = []; + return { + sent, + mailer: { + send: async (message): Promise => { + sent.push(message); + return { id: 'resend_1' }; + }, + }, + }; +} + +async function makeSendable(db: Client, personEmail = 'jane@acme.com'): Promise { + 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], + }); + await db.execute({ + sql: `INSERT INTO social_identities (id, person_id, network, handle, platform_user_id, + confidence, source_type, verified_by, first_seen_at) + VALUES ('sid_jane_email', ?, 'email', ?, ?, 0.88, 'public_web', '[]', ?)`, + args: [SEED.personId, personEmail, personEmail, now()], + }); +} + +/** Writes a campaign history of `sends` messages, `bounces` of which bounced. */ +async function history(db: Client, sends: number, bounces: number): Promise { + const stamp = new Date(Date.now() - 3_600_000).toISOString(); + for (let i = 0; i < sends; i += 1) { + await db.execute({ + sql: `INSERT INTO interactions (id, workspace_id, person_id, campaign_id, network, direction, + state, contact_address, occurred_at, recorded_at) + VALUES (?, ?, ?, ?, 'email', 'outbound', 'sent', ?, ?, ?)`, + args: [ + `int_s${i}`, + SEED.workspaceId, + SEED.personId, + SEED.campaignId, + `p${i}@x.com`, + stamp, + stamp, + ], + }); + } + for (let i = 0; i < bounces; i += 1) { + await db.execute({ + sql: `INSERT INTO interactions (id, workspace_id, person_id, campaign_id, network, direction, + state, contact_address, occurred_at, recorded_at) + VALUES (?, ?, ?, ?, 'email', 'automated', 'bounced', ?, ?, ?)`, + args: [ + `int_b${i}`, + SEED.workspaceId, + SEED.personId, + SEED.campaignId, + `p${i}@x.com`, + stamp, + stamp, + ], + }); + } +} + +/** A resolver and prober that answer from a table instead of the network. */ +function fakeVerifier(answers: { + mx?: readonly string[]; + code?: number; + catchAll?: boolean; + unreachable?: boolean; +}): VerifierDeps & { asked: string[] } { + const asked: string[] = []; + const smtp: SmtpProber = { + probe: async (_host, addresses) => { + asked.push(...addresses.slice(1)); + if (answers.unreachable) return { reachable: false, reason: 'connect_failed' }; + const codes = new Map(); + codes.set(addresses[0] as string, answers.catchAll ? 250 : 550); + for (const address of addresses.slice(1)) codes.set(address, answers.code ?? 250); + return { reachable: true, codes }; + }, + }; + return { + asked, + resolveMx: async () => + (answers.mx ?? ['mx.acme.com']).map((exchange) => ({ exchange, priority: 10 })), + smtp, + randomLocalPart: () => 'nobody-here', + }; +} + +describe('list quality rules', () => { + test('a check older than 90 days, or older than a pause, is stale', () => { + const at = new Date('2026-10-07T00:00:00Z'); + expect(verificationFresh('2026-09-01T00:00:00Z', at)).toBe(true); + expect(verificationFresh('2026-06-01T00:00:00Z', at)).toBe(false); + expect( + verificationFresh('2026-10-01T00:00:00Z', at, { notBefore: '2026-10-05T00:00:00Z' }), + ).toBe(false); + }); + + test('the gate needs 50 sends and more than 2%', () => { + expect(bounceGateTripped({ sends: 10, bounces: 5 })).toBe(false); + expect(bounceGateTripped({ sends: 100, bounces: 2 })).toBe(false); + expect(bounceGateTripped({ sends: 100, bounces: 3 })).toBe(true); + expect(catchAllAllowed({ sends: 100, bounces: 1 })).toBe(true); + expect(catchAllAllowed({ sends: 100, bounces: 2 })).toBe(false); + }); + + test('evidence maps to a status', () => { + expect(statusFromEvidence({ mx: [], smtp: 'skipped', verdict: 'unknown' }).status).toBe( + 'invalid', + ); + expect(statusFromEvidence({ mx: ['a'], smtp: 'unavailable', verdict: 'unknown' }).status).toBe( + 'unverified', + ); + expect( + statusFromEvidence({ mx: ['a'], smtp: 'probed', catchAll: true, verdict: 'unknown' }).status, + ).toBe('catch_all'); + expect(statusFromEvidence({ mx: ['a'], smtp: 'probed', verdict: 'rejected' }).status).toBe( + 'invalid', + ); + expect(statusFromEvidence({ mx: ['a'], smtp: 'probed', verdict: 'accepted' }).status).toBe( + 'valid', + ); + }); +}); + +describe('autopilot list quality', () => { + test('verifies an address before its first message and caches the verdict', async () => { + seeded = await seedDatabase('lq-verify'); + const { db } = seeded; + await makeSendable(db); + const verifier = fakeVerifier({ code: 250 }); + + const { sent, mailer } = recordingMailer(); + await runAutopilot({ db, mailer, verifier, holdLedger: new HoldLedger() }, SEED.workspaceId); + + expect(sent).toHaveLength(1); + expect(verifier.asked).toEqual(['jane@acme.com']); + expect((await readVerification(db, 'JANE@acme.com'))?.status).toBe('valid'); + }); + + test('a refused address is held and never sent', async () => { + seeded = await seedDatabase('lq-refused'); + const { db } = seeded; + await makeSendable(db); + + const { sent, mailer } = recordingMailer(); + const result = await runAutopilot( + { db, mailer, verifier: fakeVerifier({ code: 550 }), holdLedger: new HoldLedger() }, + SEED.workspaceId, + ); + + expect(sent).toHaveLength(0); + expect(result.skipped[0]?.reason).toContain('failed verification'); + }); + + test('a blocked port 25 still sends on MX alone', async () => { + seeded = await seedDatabase('lq-blocked'); + const { db } = seeded; + await makeSendable(db); + + const { sent, mailer } = recordingMailer(); + await runAutopilot( + { db, mailer, verifier: fakeVerifier({ unreachable: true }), holdLedger: new HoldLedger() }, + SEED.workspaceId, + ); + + expect(sent).toHaveLength(1); + expect((await readVerification(db, 'jane@acme.com'))?.status).toBe('unverified'); + }); + + test('a bounced address is never written to again, and its cadences stop', async () => { + seeded = await seedDatabase('lq-bounced'); + const { db } = seeded; + await makeSendable(db); + + await markAddressBounced(db, { + workspaceId: SEED.workspaceId, + address: 'jane@acme.com', + personId: SEED.personId, + detail: 'Undelivered Mail Returned to Sender', + }); + + const { sent, mailer } = recordingMailer(); + const result = await runAutopilot( + { db, mailer, holdLedger: new HoldLedger() }, + SEED.workspaceId, + ); + + expect(sent).toHaveLength(0); + // The bounced address is not offered at all, so there is nothing to write to. + expect(result.skipped[0]?.reason).toContain('no address'); + }); + + test('over 2% bounced: the campaign pauses, re-verifies, and resumes on a fresh window', async () => { + seeded = await seedDatabase('lq-gate'); + const { db } = seeded; + await makeSendable(db); + await history(db, 60, 3); + + const verifier = fakeVerifier({ code: 250 }); + const ledger = new HoldLedger(); + const { sent, mailer } = recordingMailer(); + + // A pass with no budget for checks: the campaign trips, nothing sends, and + // because the queue is not re-checked yet it stays paused. + let result = await runAutopilot( + { db, mailer, verifier, maxVerificationsPerRun: 0, holdLedger: ledger }, + SEED.workspaceId, + ); + expect(sent).toHaveLength(0); + expect(result.skipped[0]?.reason).toContain('campaign paused'); + expect((await campaignBounceState(db, SEED.campaignId)).pausedAt).not.toBeNull(); + + // A pass that re-checks the whole queue resumes it, still without sending. + result = await runAutopilot({ db, mailer, verifier, holdLedger: ledger }, SEED.workspaceId); + expect(sent).toHaveLength(0); + const state = await campaignBounceState(db, SEED.campaignId); + expect(state.pausedAt).toBeNull(); + expect(state.window.sends).toBe(0); + + // And the next one sends. + await runAutopilot({ db, mailer, verifier, holdLedger: ledger }, SEED.workspaceId); + expect(sent).toHaveLength(1); + + const event = await queryOne<{ message: string }>( + db, + `SELECT message FROM workflow_events WHERE campaign_id = ? AND message LIKE 'Resumed%'`, + [SEED.campaignId], + ); + expect(event?.message).toContain('re-verified'); + }); + + test('an accept-all address waits while the campaign bounces above 1%', async () => { + seeded = await seedDatabase('lq-catchall'); + const { db } = seeded; + await makeSendable(db); + await history(db, 100, 2); + await recordVerification(db, 'jane@acme.com', { status: 'catch_all', reason: 'accepts all' }); + + const { sent, mailer } = recordingMailer(); + const result = await runAutopilot( + { db, mailer, holdLedger: new HoldLedger() }, + SEED.workspaceId, + ); + + expect(sent).toHaveLength(0); + expect(result.skipped[0]?.reason).toContain('accept-all'); + + const report = await listHealthReport(db, SEED.campaignId); + expect(report.sends).toBe(100); + expect(report.bounces).toBe(2); + }); +}); diff --git a/packages/pipeline/src/list-quality.ts b/packages/pipeline/src/list-quality.ts new file mode 100644 index 0000000..b3a2636 --- /dev/null +++ b/packages/pipeline/src/list-quality.ts @@ -0,0 +1,332 @@ +/** + * List quality at send time: the verification cache, bounce suppression and + * the per-campaign bounce gate. The rules themselves are in + * `@outreachgraph/domain` (`list-quality.ts`); this is where they meet rows. + * + * Autopilot is the only unattended email sender, and cadence email steps + * become recommendations it sends, so gating here covers every automated + * message without a second copy of the logic. + */ + +import { + bounceGateTripped, + bounceRate, + formatRate, + statusFromEvidence, + verificationFresh, + type AddressStatus, + type AddressVerification, + type BounceWindow, +} from '@outreachgraph/domain'; +import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; +import { verifyDomainCandidates, type VerifierDeps } from '@outreachgraph/providers'; +import { emitEvent } from './events'; + +export function normalizeAddress(address: string): string { + return address.trim().toLowerCase(); +} + +export async function readVerification( + db: Client, + address: string, +): Promise { + const row = await queryOne<{ status: string; checked_at: string; reason: string | null }>( + db, + `SELECT status, checked_at, reason FROM email_verifications WHERE address = ?`, + [normalizeAddress(address)], + ); + if (!row) return undefined; + return { status: row.status as AddressStatus, checkedAt: row.checked_at, reason: row.reason }; +} + +export async function recordVerification( + db: Client, + address: string, + verdict: { readonly status: AddressStatus; readonly reason: string; readonly mx?: string }, + at: string = now(), +): Promise { + await db.execute({ + sql: `INSERT INTO email_verifications (address, status, reason, mx, checked_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT (address) DO UPDATE SET status = excluded.status, + reason = excluded.reason, mx = excluded.mx, checked_at = excluded.checked_at`, + args: [normalizeAddress(address), verdict.status, verdict.reason, verdict.mx ?? null, at], + }); +} + +export type VerifyOutcome = + | { + readonly kind: 'known'; + readonly verification: AddressVerification; + readonly checked: boolean; + } + /** The check could not be made now (DNS timeout, no budget left this run). */ + | { readonly kind: 'pending'; readonly reason: string }; + +/** + * The address's verdict, asking the domain when the cached one is missing, + * older than 90 days, or older than `notBefore` (a campaign re-verifying + * after its bounce gate tripped). + * + * `allowCheck` false means only the cache may answer — the run's verification + * budget is spent, and a stale entry is reported as pending, not trusted. + * A bounced address stays `invalid` regardless of age: the bounce is the + * strongest evidence there is, and re-probing a server that accepts at RCPT + * and bounces later would only flip it back. + */ +export async function verifyAddress( + db: Client, + address: string, + options: { + readonly verifier?: VerifierDeps | undefined; + readonly at: Date; + readonly notBefore?: string | null | undefined; + readonly allowCheck: boolean; + }, +): Promise { + const cached = await readVerification(db, address); + if (cached?.status === 'invalid' && cached.reason?.startsWith('bounced')) { + return { kind: 'known', verification: cached, checked: false }; + } + if (cached && verificationFresh(cached.checkedAt, options.at, { notBefore: options.notBefore })) { + return { kind: 'known', verification: cached, checked: false }; + } + // No verifier configured (tests, a host with DNS forbidden): the cache is + // all there is, and an unchecked address is sendable as it always was. + if (!options.verifier) { + return cached + ? { kind: 'known', verification: cached, checked: false } + : { + kind: 'known', + verification: { status: 'unverified', checkedAt: options.at.toISOString() }, + checked: false, + }; + } + if (!options.allowCheck) { + return { kind: 'pending', reason: 'waiting for its address to be verified' }; + } + + const normalized = normalizeAddress(address); + const domain = normalized.split('@')[1]; + if (!domain) { + const verdict = { status: 'invalid' as const, reason: 'not an email address' }; + await recordVerification(db, normalized, verdict, options.at.toISOString()); + return { + kind: 'known', + verification: { ...verdict, checkedAt: options.at.toISOString() }, + checked: true, + }; + } + + try { + const result = await verifyDomainCandidates(domain, [normalized], options.verifier); + const verdict = statusFromEvidence({ + mx: result.mx, + smtp: result.smtp, + catchAll: result.catchAll, + verdict: result.verdicts.get(normalized) ?? 'unknown', + }); + const stamp = options.at.toISOString(); + await recordVerification( + db, + normalized, + { ...verdict, ...(result.mx[0] ? { mx: result.mx[0] } : {}) }, + stamp, + ); + return { kind: 'known', verification: { ...verdict, checkedAt: stamp }, checked: true }; + } catch (error) { + // A resolver failure is not an answer. Nothing is cached, so the next run + // asks again. + const message = error instanceof Error ? error.message : String(error); + return { + kind: 'pending', + reason: `its address could not be checked (${message.slice(0, 80)})`, + }; + } +} + +/** + * A delivery report named this address: it never gets another message. + * + * Also ends every active cadence for the person it belonged to, with the + * reason on the enrollment. A plan whose next three steps will all bounce is + * not a plan. + */ +export async function markAddressBounced( + db: Client, + input: { + readonly workspaceId: string; + readonly address: string; + readonly personId?: string | undefined; + readonly detail?: string | undefined; + }, +): Promise<{ stoppedEnrollments: number }> { + await recordVerification(db, input.address, { + status: 'invalid', + reason: `bounced${input.detail ? `: ${input.detail.slice(0, 160)}` : ''}`, + }); + if (!input.personId) return { stoppedEnrollments: 0 }; + + const result = await db.execute({ + sql: `UPDATE cadence_enrollments + SET status = 'stopped', next_due_at = NULL, stopped_reason = ?, updated_at = ? + WHERE workspace_id = ? AND person_id = ? AND status = 'active'`, + args: ['their address bounced', now(), input.workspaceId, input.personId], + }); + return { stoppedEnrollments: result.rowsAffected ?? 0 }; +} + +export interface CampaignBounceState { + readonly campaignId: string; + readonly windowStart: string; + readonly window: BounceWindow; + readonly rate: number; + /** Set while the campaign is stopped for re-verification. */ + readonly pausedAt: string | null; +} + +/** Sends and bounces since the campaign's window opened. */ +export async function campaignBounceState( + db: Client, + campaignId: string, +): Promise { + const health = await queryOne<{ window_start: string; paused_at: string | null }>( + db, + `SELECT window_start, paused_at FROM campaign_list_health WHERE campaign_id = ?`, + [campaignId], + ); + const windowStart = health?.window_start ?? ''; + + const row = await queryOne<{ sends: number; bounces: number }>( + db, + `SELECT + (SELECT COUNT(*) FROM interactions + WHERE campaign_id = ? AND network = 'email' AND direction = 'outbound' + AND occurred_at >= ?) AS sends, + (SELECT COUNT(*) FROM interactions + WHERE campaign_id = ? AND network = 'email' AND direction = 'automated' + AND state = 'bounced' AND occurred_at >= ?) AS bounces`, + [campaignId, windowStart, campaignId, windowStart], + ); + const window = { sends: Number(row?.sends ?? 0), bounces: Number(row?.bounces ?? 0) }; + return { + campaignId, + windowStart, + window, + rate: bounceRate(window), + pausedAt: health?.paused_at ?? null, + }; +} + +/** + * Reads the campaign's gate and trips it when the rate is over the line. + * + * Returns the state after any change, so the caller treats a campaign paused + * by this call the same as one paused an hour ago. + */ +export async function applyCampaignBounceGate( + db: Client, + workspaceId: string, + campaignId: string, + at: Date, +): Promise { + const state = await campaignBounceState(db, campaignId); + if (state.pausedAt || !bounceGateTripped(state.window)) return state; + + const stamp = at.toISOString(); + await db.execute({ + sql: `INSERT INTO campaign_list_health (campaign_id, workspace_id, window_start, paused_at, + paused_rate, updated_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (campaign_id) DO UPDATE SET paused_at = excluded.paused_at, + paused_rate = excluded.paused_rate, updated_at = excluded.updated_at`, + args: [campaignId, workspaceId, state.windowStart, stamp, state.rate, stamp], + }); + + await emitEvent(db, { + workspaceId, + campaignId, + phase: 'send', + level: 'warn', + message: + `Paused sending: ${formatRate(state.rate)} of the last ${state.window.sends} messages ` + + `bounced (over 2%). Re-verifying every queued address before it starts again.`, + detail: { ...state.window, rate: state.rate }, + }); + + return { ...state, pausedAt: stamp }; +} + +/** + * Starts a paused campaign again on a fresh window. + * + * Called once a full autopilot pass found nothing in it still waiting for a + * re-check. The old bounces stay on record but stop counting, since the list + * they came from is no longer the list being sent. + */ +export async function resumeCampaign( + db: Client, + workspaceId: string, + campaignId: string, + at: Date, + removed: number, +): Promise { + const stamp = at.toISOString(); + await db.execute({ + sql: `UPDATE campaign_list_health + SET paused_at = NULL, window_start = ?, resumed_at = ?, updated_at = ? + WHERE campaign_id = ?`, + args: [stamp, stamp, stamp, campaignId], + }); + await emitEvent(db, { + workspaceId, + campaignId, + phase: 'send', + level: 'success', + message: + `Resumed sending: every queued address was re-verified` + + (removed > 0 ? ` and ${removed} that failed were dropped.` : '.'), + detail: { removed }, + }); +} + +export interface ListHealthReport { + readonly campaignId: string; + readonly sends: number; + readonly bounces: number; + readonly bounceRate: number; + readonly windowStart: string | null; + readonly pausedAt: string | null; + /** Verdicts for the addresses this campaign has sent to, by status. */ + readonly addresses: Readonly>; +} + +/** What the list-health endpoint, CLI and MCP tool report. */ +export async function listHealthReport(db: Client, campaignId: string): Promise { + const state = await campaignBounceState(db, campaignId); + const rows = await queryAll<{ status: string | null; n: number }>( + db, + `SELECT ev.status AS status, COUNT(*) AS n FROM ( + SELECT DISTINCT contact_address FROM interactions + WHERE campaign_id = ? AND network = 'email' AND direction = 'outbound' + AND contact_address IS NOT NULL + ) sent + LEFT JOIN email_verifications ev ON ev.address = sent.contact_address + GROUP BY ev.status`, + [campaignId], + ); + const addresses = { valid: 0, catch_all: 0, unverified: 0, invalid: 0, unchecked: 0 }; + for (const row of rows) { + const key = (row.status ?? 'unchecked') as keyof typeof addresses; + if (key in addresses) addresses[key] += Number(row.n); + } + return { + campaignId, + sends: state.window.sends, + bounces: state.window.bounces, + bounceRate: state.rate, + windowStart: state.windowStart || null, + pausedAt: state.pausedAt, + addresses, + }; +} diff --git a/packages/pipeline/src/receive-email.ts b/packages/pipeline/src/receive-email.ts index 3ffb54e..c9d0044 100644 --- a/packages/pipeline/src/receive-email.ts +++ b/packages/pipeline/src/receive-email.ts @@ -39,6 +39,7 @@ import type { MailReader, IncomingMessage } from '@outreachgraph/email'; import { enqueue } from './queue'; import { recordStatus } from './stages'; import { recordSenderBounce } from './sender-pool'; +import { markAddressBounced } from './list-quality'; import { emitWebhookEvent } from './webhooks'; export interface ReceiveRepliesInput { @@ -172,6 +173,24 @@ export async function receiveReplies(input: ReceiveRepliesInput): Promise Date: Wed, 7 Oct 2026 02:18:54 +0000 Subject: [PATCH 2/2] Account expansion: one contact per company at a time, opened by the budget holder Hunter's planner reaches four people per target company, one at a time, a month apart: budget holder, pain feeler (end user), blocker (legal, security, IT, finance) and champion (manager). Now automatic: - classifyPersona reads the job title deterministically (no model on the send path). - Autopilot orders each company's queued cards by persona behind its best card, without changing the order between companies. - A colleague emailed in the last 21 days, or on an active cadence, who has not answered holds the company; everyone else there waits, card intact. A reply releases it. Deliberately reverses the old "two mailboxes at one company both get written to in one run" test. - Drafts get a per-role angle added to the step's guidance. - GET /autogtm/campaigns/:id/accounts, og leads accounts, MCP get_campaign_accounts: who was reached, who is next, missing personas. Co-Authored-By: Claude Opus 5.5 --- apps/api/src/autogtm-docs.ts | 21 ++ apps/api/src/autogtm.ts | 33 +++ apps/cli/src/commands.ts | 27 ++- apps/mcp/src/tools.ts | 16 ++ packages/ai/src/draft.ts | 22 +- packages/domain/src/index.ts | 1 + packages/domain/src/personas.test.ts | 58 +++++ packages/domain/src/personas.ts | 114 ++++++++++ .../pipeline/src/account-expansion.test.ts | 177 +++++++++++++++ packages/pipeline/src/account-expansion.ts | 204 ++++++++++++++++++ packages/pipeline/src/autopilot.test.ts | 15 +- packages/pipeline/src/autopilot.ts | 60 +++++- packages/pipeline/src/index.ts | 8 + 13 files changed, 746 insertions(+), 10 deletions(-) create mode 100644 packages/domain/src/personas.test.ts create mode 100644 packages/domain/src/personas.ts create mode 100644 packages/pipeline/src/account-expansion.test.ts create mode 100644 packages/pipeline/src/account-expansion.ts diff --git a/apps/api/src/autogtm-docs.ts b/apps/api/src/autogtm-docs.ts index 7fb153a..115f481 100644 --- a/apps/api/src/autogtm-docs.ts +++ b/apps/api/src/autogtm-docs.ts @@ -646,6 +646,27 @@ export const OPERATIONS: readonly Operation[] = [ }, }, }, + { + method: 'get', + path: '/autogtm/campaigns/{campaign_id}/accounts', + id: 'getAccounts', + tag: 'Campaigns', + summary: 'Account expansion: who at each company was reached, who is next', + description: + 'Contacts are classified from their job title as budget holder, pain feeler (end user), ' + + 'blocker (legal, security, IT, finance) or champion (manager). One contact per company is ' + + 'written to at a time: a colleague emailed in the last 21 days, or on an active cadence, ' + + 'holds the company until they answer or the window closes. Queued colleagues go in persona ' + + 'order, each drafted with an angle for their role. `missing` lists personas nobody covers yet.', + params: [idParam('campaign_id', 'The campaign.')], + response: { + type: 'object', + properties: { + gap_days: { type: 'integer' }, + accounts: { type: 'array', items: { type: 'object' } }, + }, + }, + }, { method: 'get', path: '/autogtm/campaigns/{campaign_id}/list-health', diff --git a/apps/api/src/autogtm.ts b/apps/api/src/autogtm.ts index 416301a..7a64e41 100644 --- a/apps/api/src/autogtm.ts +++ b/apps/api/src/autogtm.ts @@ -30,6 +30,7 @@ import { autogtmStatus, CONTACT_PRICE_USD, isConsumerMailDomain, + EXPANSION_GAP_DAYS, mapHeaders, MAX_BOUNCE_RATE, newId, @@ -43,6 +44,7 @@ import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; import { applyProjectBudget, budgetStatus, + campaignAccounts, campaignBudgetFrom, countScreenedHeld, crawlDedupeKey, @@ -996,6 +998,37 @@ export function autogtmRoutes(deps: AutogtmDeps): Hono { }); }); + /** + * Account expansion: each company in the campaign with who has been reached, + * who is next, and which of the four personas nobody covers yet. + */ + r.get('/campaigns/:id/accounts', async (c) => { + const actor = c.get('actor'); + const db = c.get('db'); + const campaign = await ownedCampaign(db, actor.workspaceId, c.req.param('id')); + const accounts = await campaignAccounts(db, { + workspaceId: actor.workspaceId, + campaignId: campaign.id, + }); + return c.json({ + campaign_id: campaign.id, + gap_days: EXPANSION_GAP_DAYS, + accounts: accounts.map((account) => ({ + company_id: account.companyId, + company: account.company, + missing: account.missing, + contacts: account.contacts.map((contact) => ({ + person_id: contact.personId, + name: contact.name, + title: contact.title, + persona: contact.persona, + state: contact.state, + last_contacted_at: contact.lastContactedAt, + })), + })), + }); + }); + r.get('/campaigns/:id/enrichment', async (c) => { const actor = c.get('actor'); const db = c.get('db'); diff --git a/apps/cli/src/commands.ts b/apps/cli/src/commands.ts index 973d565..33c25b2 100644 --- a/apps/cli/src/commands.ts +++ b/apps/cli/src/commands.ts @@ -306,6 +306,29 @@ async function runLeads({ client, args, flags }: CommandContext): Promise'); + const result = (await client.get( + `/autogtm/campaigns/${encodeURIComponent(target)}/accounts`, + )) as Record; + const accounts = (result.accounts ?? []) as Record[]; + if (accounts.length === 0) return 'No companies in this campaign yet.'; + return accounts + .map((account) => { + const contacts = (account.contacts ?? []) as Record[]; + const missing = (account.missing ?? []) as string[]; + return [ + `${text(account, 'company')}${missing.length ? ` (missing: ${missing.join(', ')})` : ''}`, + ...contacts.map( + (contact) => + ` ${text(contact, 'state').padEnd(10)} ${text(contact, 'persona').padEnd(13)} ` + + `${text(contact, 'name')}${contact.title ? `, ${text(contact, 'title')}` : ''}`, + ), + ].join('\n'); + }) + .join('\n'); + } + if (verb === 'health') { if (!target) throw new Error('usage: og leads health '); const health = (await client.get( @@ -336,7 +359,7 @@ async function runLeads({ client, args, flags }: CommandContext): Promise --consent-source "" [--allow-flagged] [--keep-project-duplicates] [--report out.csv] | report [--csv] | screened [--all] | enrich [--max ] | enrichment | health | allow | hold ', + 'og leads add --consent-source "" [--allow-flagged] [--keep-project-duplicates] [--report out.csv] | report [--csv] | screened [--all] | enrich [--max ] | enrichment | health | accounts | allow | hold ', summary: 'Add a CSV of leads to a running campaign, with a per-row report and screening.', run: runLeads, }, diff --git a/apps/mcp/src/tools.ts b/apps/mcp/src/tools.ts index 2759cfb..12f069e 100644 --- a/apps/mcp/src/tools.ts +++ b/apps/mcp/src/tools.ts @@ -1050,6 +1050,22 @@ export const TOOLS: readonly ToolDefinition[] = [ `/autogtm/campaigns/${encodeURIComponent(require(args, 'campaignId'))}/enrichment`, ), }, + { + name: 'get_campaign_accounts', + title: 'Account expansion per company', + description: + 'Each company in a campaign with its contacts classified as budget holder, pain feeler, ' + + 'blocker or champion; who was contacted, who replied, who is next in line (one contact per ' + + 'company at a time, 21 days apart), and which personas nobody covers yet.', + readOnly: true, + inputSchema: { + type: 'object', + properties: { campaignId: { type: 'string' } }, + required: ['campaignId'], + }, + run: (client, args) => + client.get(`/autogtm/campaigns/${encodeURIComponent(require(args, 'campaignId'))}/accounts`), + }, { name: 'get_list_health', title: 'A campaign’s bounce rate and address verdicts', diff --git a/packages/ai/src/draft.ts b/packages/ai/src/draft.ts index 4697696..793adb7 100644 --- a/packages/ai/src/draft.ts +++ b/packages/ai/src/draft.ts @@ -11,7 +11,14 @@ * fabricated one. */ -import { newId, type ActionKind, type Network, type OutreachStyle } from '@outreachgraph/domain'; +import { + classifyPersona, + newId, + PERSONA_ANGLES, + type ActionKind, + type Network, + type OutreachStyle, +} from '@outreachgraph/domain'; import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; import { composeDraft, type ComposeResult, type TextModel } from '@outreachgraph/ai'; import { draftReplyForRecommendation } from './draft-reply'; @@ -66,6 +73,17 @@ export async function draftForRecommendation( return draftReplyForRecommendation(db, model, recommendation.id); } + // A fresh angle per role (account expansion): the budget holder hears the + // outcome, the end user the friction, the reviewer the objection answered, + // the manager the proof to forward. Added to whatever the step asked for. + const titled = await queryOne<{ current_title: string | null }>( + db, + 'SELECT current_title FROM people WHERE id = ?', + [recommendation.person_id], + ); + const angle = PERSONA_ANGLES[classifyPersona(titled?.current_title)]; + const guidance = [recommendation.guidance?.trim(), angle].filter(Boolean).join(' ') || undefined; + // No trigger means nothing to quote, and §14.1 forbids personalising // without evidence. if (!recommendation.trigger_signal_id) return { ok: false, reason: 'no_trigger_signal' }; @@ -189,7 +207,7 @@ export async function draftForRecommendation( : {}), minIdentityConfidence: workspace?.min_outreach_confidence ?? 0.85, priorDraftHashes: priorHashes.map((r) => r.similarity_hash), - ...(recommendation.guidance ? { guidance: recommendation.guidance } : {}), + ...(guidance ? { guidance } : {}), }); if (!result.ok) { diff --git a/packages/domain/src/index.ts b/packages/domain/src/index.ts index bb33ebb..feef77e 100644 --- a/packages/domain/src/index.ts +++ b/packages/domain/src/index.ts @@ -39,4 +39,5 @@ export * from './replies'; export * from './webhooks'; export * from './mailbox-health'; export * from './list-quality'; +export * from './personas'; export * from './warmup'; diff --git a/packages/domain/src/personas.test.ts b/packages/domain/src/personas.test.ts new file mode 100644 index 0000000..97c4ed7 --- /dev/null +++ b/packages/domain/src/personas.test.ts @@ -0,0 +1,58 @@ +import { describe, expect, test } from 'bun:test'; +import { classifyPersona, orderForExpansion, personaRank } from './personas'; + +describe('classifyPersona', () => { + test.each([ + ['CEO', 'budget_holder'], + ['Co-founder & CTO', 'budget_holder'], + ['VP of Sales', 'budget_holder'], + ['Head of Growth', 'budget_holder'], + ['Director, Marketing', 'budget_holder'], + ['General Counsel', 'blocker'], + ['VP Legal', 'blocker'], + ['Chief Information Security Officer', 'blocker'], + ['CFO', 'blocker'], + ['Procurement Manager', 'blocker'], + ['IT Manager', 'blocker'], + ['Engineering Manager', 'champion'], + ['Team Lead, Support Operations', 'champion'], + ['Senior Software Engineer', 'pain_feeler'], + ['Marketing Specialist', 'pain_feeler'], + ['Customer Success Associate', 'pain_feeler'], + ['', 'unknown'], + [null, 'unknown'], + ['Wizard', 'unknown'], + ] as const)('%s → %s', (title, persona) => { + expect(classifyPersona(title)).toBe(persona); + }); + + test('ranks in planner order, unknown last', () => { + expect(personaRank('budget_holder')).toBe(0); + expect(personaRank('pain_feeler')).toBe(1); + expect(personaRank('blocker')).toBe(2); + expect(personaRank('champion')).toBe(3); + expect(personaRank('unknown')).toBe(4); + }); +}); + +describe('orderForExpansion', () => { + test('a company opens with its budget holder, other companies keep their places', () => { + const queue = [ + { id: 'acme-engineer', company: 'acme', title: 'Software Engineer' }, + { id: 'solo', company: null, title: 'CEO' }, + { id: 'acme-ceo', company: 'acme', title: 'CEO' }, + { id: 'beta-cfo', company: 'beta', title: 'CFO' }, + ]; + const ordered = orderForExpansion( + queue, + (item) => item.company, + (item) => classifyPersona(item.title), + ); + expect(ordered.map((item) => item.id)).toEqual([ + 'acme-ceo', + 'acme-engineer', + 'solo', + 'beta-cfo', + ]); + }); +}); diff --git a/packages/domain/src/personas.ts b/packages/domain/src/personas.ts new file mode 100644 index 0000000..a539caa --- /dev/null +++ b/packages/domain/src/personas.ts @@ -0,0 +1,114 @@ +/** + * Buyer personas and account expansion (Hunter's outreach planner, run by the + * worker). + * + * Every target company has four people worth reaching, and the planner reaches + * them one at a time, a month apart, each with its own angle: + * + * 1. **Budget holder** — signs off. Reached first, while budgets are fresh. + * 2. **Pain feeler** — the end user who lives with the problem. + * 3. **Blocker** — legal, security, IT, finance, procurement: the people who + * can say no, brought in before they do. + * 4. **Champion** — the manager who carries it internally. + * + * Two rules come with it, and both are enforced at send time: + * + * - **Never two contacts at one company at once.** While anyone at a company + * is mid-sequence (`EXPANSION_GAP_DAYS` since their last message), every + * colleague waits. + * - **In order.** When several people at one company are queued, the one + * whose persona comes earliest goes first, so a company is opened by the + * person who can buy and widened from there. + * + * Classification is a deterministic reading of the job title: the policy path + * never asks a model, and a title is what we reliably have. + */ + +export const PERSONAS = ['budget_holder', 'pain_feeler', 'blocker', 'champion'] as const; +export type Persona = (typeof PERSONAS)[number]; +export type PersonaOrUnknown = Persona | 'unknown'; + +/** Days after one contact's last message before a colleague may be written to. */ +export const EXPANSION_GAP_DAYS = 21; + +export const PERSONA_LABELS: Readonly> = { + budget_holder: 'budget holder', + pain_feeler: 'pain feeler', + blocker: 'blocker', + champion: 'champion', + unknown: 'unclassified', +}; + +/** What to say to each, so the composer writes a fresh angle per role. */ +export const PERSONA_ANGLES: Readonly> = { + budget_holder: + 'Write to the person who owns the budget: lead with the business outcome (pipeline, revenue, cost, risk) and ask one question answerable in ten seconds.', + pain_feeler: + 'Write to the person who does the work day to day: name the specific friction they live with and how it goes away. No executive framing.', + blocker: + 'Write to the legal, security, IT or finance reviewer: answer their objection before they raise it (compliance, data handling, integration effort, total cost).', + champion: + 'Write to the manager who would carry this internally: give them the proof they would forward (a result, a number, a short case) and reference the colleague already in touch.', + unknown: '', +}; + +// Order matters: a "VP of Legal" is a blocker before a budget holder, and a +// "Chief Security Officer" likewise. Finance, legal and security titles are +// checked before seniority. +const BLOCKER = + /\b(legal|counsel|attorney|lawyer|compliance|privacy|security|infosec|ciso|risk|procurement|purchasing|sourcing|vendor management|finance|financial|cfo|controller|comptroller|accounting|accountant|treasury|audit|it manager|it director|head of it|it operations|sysadmin|system administrator|information technology|data protection|dpo|gdpr)\b/i; +const BUDGET = + /\b(ceo|coo|cto|cmo|cro|cpo|cio|chief|founder|co-?founder|owner|president|vp|vice president|svp|evp|head of|director|general manager|gm|managing director|partner|principal owner|proprietor|board)\b/i; +const CHAMPION = + /\b(manager|lead|team lead|principal|staff|senior manager|supervisor|coordinator lead|architect)\b/i; +const PAIN = + /\b(engineer|developer|programmer|designer|analyst|specialist|associate|coordinator|representative|rep|executive assistant|assistant|administrator|operator|technician|marketer|writer|editor|consultant|scientist|researcher|agent|recruiter|sdr|bdr|ae|account executive|support|success|intern)\b/i; + +export function classifyPersona(title: string | null | undefined): PersonaOrUnknown { + const text = (title ?? '').trim(); + if (!text) return 'unknown'; + if (BLOCKER.test(text)) return 'blocker'; + if (BUDGET.test(text)) return 'budget_holder'; + if (CHAMPION.test(text)) return 'champion'; + if (PAIN.test(text)) return 'pain_feeler'; + return 'unknown'; +} + +/** Position in the expansion order; unclassified people go last. */ +export function personaRank(persona: PersonaOrUnknown): number { + const index = (PERSONAS as readonly string[]).indexOf(persona); + return index === -1 ? PERSONAS.length : index; +} + +/** + * Orders a send queue so each company's earliest persona comes first, without + * disturbing the order between companies. + * + * A company takes the position of its best-ranked card in the original order + * (priority already decided that), and its cards follow one another in persona + * order. Cards with no company keep their own position. + */ +export function orderForExpansion( + items: readonly T[], + companyOf: (item: T) => string | null | undefined, + personaOf: (item: T) => PersonaOrUnknown, +): T[] { + const firstSeen = new Map(); + items.forEach((item, index) => { + const company = companyOf(item); + if (company && !firstSeen.has(company)) firstSeen.set(company, index); + }); + + return items + .map((item, index) => { + const company = companyOf(item); + return { + item, + index, + group: company ? (firstSeen.get(company) as number) : index, + rank: company ? personaRank(personaOf(item)) : 0, + }; + }) + .sort((a, b) => a.group - b.group || a.rank - b.rank || a.index - b.index) + .map((entry) => entry.item); +} diff --git a/packages/pipeline/src/account-expansion.test.ts b/packages/pipeline/src/account-expansion.test.ts new file mode 100644 index 0000000..49e21d5 --- /dev/null +++ b/packages/pipeline/src/account-expansion.test.ts @@ -0,0 +1,177 @@ +import { afterEach, describe, expect, test } from 'bun:test'; +import { now, 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 { HoldLedger, runAutopilot } from './autopilot'; +import { campaignAccounts, companyHeldBy } from './account-expansion'; + +let seeded: SeededDatabase | undefined; + +afterEach(() => { + seeded?.cleanup(); + seeded = undefined; +}); + +function recordingMailer(): { sent: Message[]; mailer: Mailer } { + const sent: Message[] = []; + return { + sent, + mailer: { + send: async (message): Promise => { + sent.push(message); + return { id: 'resend_1' }; + }, + }, + }; +} + +/** Jane (VP Engineering, a budget holder) on autopilot at low priority. */ +async function janeSendable(db: Client): Promise { + 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', priority = 10 + WHERE id = ?`, + args: [SEED.recommendationId], + }); + await addEmail(db, SEED.personId, 'jane@acme.com'); +} + +async function addEmail(db: Client, personId: string, address: string): Promise { + await db.execute({ + sql: `INSERT INTO social_identities (id, person_id, network, handle, platform_user_id, + confidence, source_type, verified_by, first_seen_at) + VALUES (?, ?, 'email', ?, ?, 0.95, 'public_web', '[]', ?)`, + args: [`sid_${personId}`, personId, address, address, now()], + }); +} + +/** A colleague at Acme with their own address and a higher-priority card. */ +async function colleague(db: Client, name: string, title: string): Promise { + const stamp = now(); + const personId = `per_${name.toLowerCase()}`; + await db.execute({ + sql: `INSERT INTO people (id, display_name, current_company_id, current_title, status, + believed_minor, outreach_eligible, identity_confidence, created_at, updated_at) + VALUES (?, ?, ?, ?, 'active', 0, 1, 0.95, ?, ?)`, + args: [personId, name, SEED.companyId, title, stamp, stamp], + }); + await db.execute({ + sql: `INSERT INTO campaign_people (campaign_id, person_id, workspace_id, status, + interaction_state, discovered_at, updated_at) + VALUES (?, ?, ?, 'recommended', 'never_contacted', ?, ?)`, + args: [SEED.campaignId, personId, SEED.workspaceId, stamp, stamp], + }); + const recommendationId = `rec_${name.toLowerCase()}`; + 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', 90, 'same company', + 'allow', '2026-01-01', 'pending', ?)`, + args: [recommendationId, SEED.workspaceId, SEED.campaignId, personId, 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, recommendationId, stamp, stamp], + }); + await addEmail(db, personId, `${name.toLowerCase()}@acme.com`); + return personId; +} + +describe('account expansion', () => { + test('the budget holder opens the company even behind a higher-priority card', async () => { + seeded = await seedDatabase('expand-order'); + const { db } = seeded; + await janeSendable(db); + await colleague(db, 'Sam', 'Software Engineer'); + + const { sent, mailer } = recordingMailer(); + await runAutopilot({ db, mailer, holdLedger: new HoldLedger() }, SEED.workspaceId); + + expect(sent.map((message) => message.to)).toEqual(['jane@acme.com']); + }); + + test('then pain feeler, blocker and champion, one per window', async () => { + seeded = await seedDatabase('expand-sequence'); + const { db } = seeded; + await janeSendable(db); + await colleague(db, 'Cara', 'Engineering Manager'); + await colleague(db, 'Lee', 'General Counsel'); + await colleague(db, 'Sam', 'Software Engineer'); + + const { sent, mailer } = recordingMailer(); + const ledger = new HoldLedger(); + for (let week = 0; week < 4; week += 1) { + const at = new Date(Date.now() + week * 22 * 86_400_000); + await runAutopilot({ db, mailer, now: at, holdLedger: ledger }, SEED.workspaceId); + } + + expect(sent.map((message) => message.to)).toEqual([ + 'jane@acme.com', + 'sam@acme.com', + 'lee@acme.com', + 'cara@acme.com', + ]); + }); + + test('a reply releases the company instead of holding it', async () => { + seeded = await seedDatabase('expand-reply'); + const { db } = seeded; + await janeSendable(db); + const sam = await colleague(db, 'Sam', 'Software Engineer'); + + const { mailer } = recordingMailer(); + await runAutopilot({ db, mailer, holdLedger: new HoldLedger() }, SEED.workspaceId); + + const at = new Date(); + expect( + await companyHeldBy(db, { + workspaceId: SEED.workspaceId, + companyId: SEED.companyId, + personId: sam, + at, + }), + ).toBeDefined(); + + const stamp = now(); + await db.execute({ + sql: `INSERT INTO interactions (id, workspace_id, person_id, campaign_id, network, direction, + state, occurred_at, recorded_at) + VALUES ('int_reply', ?, ?, ?, 'email', 'inbound', 'responded', ?, ?)`, + args: [SEED.workspaceId, SEED.personId, SEED.campaignId, stamp, stamp], + }); + + expect( + await companyHeldBy(db, { + workspaceId: SEED.workspaceId, + companyId: SEED.companyId, + personId: sam, + at, + }), + ).toBeUndefined(); + }); + + test('reports each company: who was reached, who is next, which personas are missing', async () => { + seeded = await seedDatabase('expand-report'); + const { db } = seeded; + await janeSendable(db); + await colleague(db, 'Sam', 'Software Engineer'); + + const { mailer } = recordingMailer(); + await runAutopilot({ db, mailer, holdLedger: new HoldLedger() }, SEED.workspaceId); + + const [account] = await campaignAccounts(db, { + workspaceId: SEED.workspaceId, + campaignId: SEED.campaignId, + }); + expect(account?.contacts.map((contact) => [contact.persona, contact.state])).toEqual([ + ['budget_holder', 'contacted'], + ['pain_feeler', 'next'], + ]); + expect(account?.missing).toEqual(['blocker', 'champion']); + }); +}); diff --git a/packages/pipeline/src/account-expansion.ts b/packages/pipeline/src/account-expansion.ts new file mode 100644 index 0000000..8bad2b1 --- /dev/null +++ b/packages/pipeline/src/account-expansion.ts @@ -0,0 +1,204 @@ +/** + * Account expansion at send time: one contact per company at a time, in + * persona order. The rules are in `@outreachgraph/domain` (`personas.ts`). + */ + +import { + classifyPersona, + EXPANSION_GAP_DAYS, + PERSONA_LABELS, + personaRank, + type PersonaOrUnknown, +} from '@outreachgraph/domain'; +import { queryAll, queryOne, type Client } from '@outreachgraph/db'; + +export interface CompanyBusy { + readonly personId: string; + readonly name: string; + readonly persona: PersonaOrUnknown; + /** When the colleague's window closes and the next person may be written to. */ + readonly until: string; +} + +/** + * The colleague currently "holding" a company, if any: someone else there who + * is on an active cadence, or was emailed within the last 21 days and has not + * answered. A reply ends the hold — the company is talking to us, and the next + * person is introduced by reference rather than kept waiting. + */ +export async function companyHeldBy( + db: Client, + input: { + readonly workspaceId: string; + readonly companyId: string; + readonly personId: string; + readonly at: Date; + }, +): Promise { + const since = new Date(input.at.getTime() - EXPANSION_GAP_DAYS * 86_400_000).toISOString(); + + const row = await queryOne<{ + person_id: string; + display_name: string; + current_title: string | null; + last_at: string | null; + enrolled: number; + }>( + db, + `SELECT p.id AS person_id, p.display_name, p.current_title, + (SELECT MAX(i.occurred_at) FROM interactions i + WHERE i.workspace_id = ? AND i.person_id = p.id + AND i.direction = 'outbound' AND i.network = 'email') AS last_at, + (SELECT COUNT(*) FROM cadence_enrollments e + WHERE e.workspace_id = ? AND e.person_id = p.id AND e.status = 'active') AS enrolled + FROM people p + WHERE p.current_company_id = ? AND p.id <> ? AND p.status = 'active' + AND NOT EXISTS (SELECT 1 FROM interactions r + WHERE r.workspace_id = ? AND r.person_id = p.id AND r.direction = 'inbound' + AND r.state IN ('replied', 'responded')) + AND ( + EXISTS (SELECT 1 FROM interactions i + WHERE i.workspace_id = ? AND i.person_id = p.id AND i.direction = 'outbound' + AND i.network = 'email' AND i.occurred_at >= ?) + OR EXISTS (SELECT 1 FROM cadence_enrollments e + WHERE e.workspace_id = ? AND e.person_id = p.id AND e.status = 'active') + ) + ORDER BY last_at DESC + LIMIT 1`, + [ + input.workspaceId, + input.workspaceId, + input.companyId, + input.personId, + input.workspaceId, + input.workspaceId, + since, + input.workspaceId, + ], + ); + if (!row) return undefined; + + const last = row.last_at ? Date.parse(row.last_at) : input.at.getTime(); + return { + personId: row.person_id, + name: row.display_name, + persona: classifyPersona(row.current_title), + until: new Date(last + EXPANSION_GAP_DAYS * 86_400_000).toISOString(), + }; +} + +export function describeCompanyHold(held: CompanyBusy, company: string | null): string { + const where = company ? ` at ${company}` : ''; + return ( + `one contact per company: ${held.name} (${PERSONA_LABELS[held.persona]})${where} ` + + `is mid-sequence until ${held.until.slice(0, 10)}` + ); +} + +export interface AccountContact { + readonly personId: string; + readonly name: string; + readonly title: string | null; + readonly persona: PersonaOrUnknown; + /** contacted, replied, next (first in line), waiting (behind them), not_queued */ + readonly state: 'replied' | 'contacted' | 'next' | 'waiting' | 'not_queued'; + readonly lastContactedAt: string | null; +} + +export interface Account { + readonly companyId: string; + readonly company: string; + readonly contacts: readonly AccountContact[]; + /** Personas nobody at the company covers yet, in planner order. */ + readonly missing: readonly PersonaOrUnknown[]; +} + +/** + * Every company in a campaign with who has been reached, who is next, and + * which personas are still missing — the account expansion view. + */ +export async function campaignAccounts( + db: Client, + input: { readonly workspaceId: string; readonly campaignId: string; readonly limit?: number }, +): Promise { + const rows = await queryAll<{ + company_id: string; + company: string; + person_id: string; + display_name: string; + current_title: string | null; + last_at: string | null; + replied: number; + pending: number; + }>( + db, + `SELECT co.id AS company_id, co.name AS company, p.id AS person_id, p.display_name, + p.current_title, + (SELECT MAX(i.occurred_at) FROM interactions i + WHERE i.workspace_id = cp.workspace_id AND i.person_id = p.id + AND i.direction = 'outbound') AS last_at, + (SELECT COUNT(*) FROM interactions r + WHERE r.workspace_id = cp.workspace_id AND r.person_id = p.id + AND r.direction = 'inbound' AND r.state IN ('replied', 'responded')) AS replied, + (SELECT COUNT(*) FROM recommendations rc + WHERE rc.workspace_id = cp.workspace_id AND rc.person_id = p.id + AND rc.campaign_id = cp.campaign_id AND rc.status = 'pending') AS pending + FROM campaign_people cp + JOIN people p ON p.id = cp.person_id + JOIN companies co ON co.id = p.current_company_id + WHERE cp.workspace_id = ? AND cp.campaign_id = ? AND p.status = 'active' + ORDER BY co.name, p.display_name`, + [input.workspaceId, input.campaignId], + ); + + const byCompany = new Map(); + for (const row of rows) { + const entry = byCompany.get(row.company_id) ?? { company: row.company, rows: [] }; + entry.rows.push(row); + byCompany.set(row.company_id, entry); + } + + const accounts: Account[] = []; + for (const [companyId, entry] of byCompany) { + const queued = entry.rows + .filter((row) => !row.last_at && Number(row.pending) > 0) + .sort( + (a, b) => + personaRank(classifyPersona(a.current_title)) - + personaRank(classifyPersona(b.current_title)), + ); + const nextId = queued[0]?.person_id; + + const contacts = entry.rows + .map((row): AccountContact => { + const persona = classifyPersona(row.current_title); + const state: AccountContact['state'] = + Number(row.replied) > 0 + ? 'replied' + : row.last_at + ? 'contacted' + : row.person_id === nextId + ? 'next' + : Number(row.pending) > 0 + ? 'waiting' + : 'not_queued'; + return { + personId: row.person_id, + name: row.display_name, + title: row.current_title, + persona, + state, + lastContactedAt: row.last_at, + }; + }) + .sort((a, b) => personaRank(a.persona) - personaRank(b.persona)); + + const covered = new Set(contacts.map((contact) => contact.persona)); + const missing = (['budget_holder', 'pain_feeler', 'blocker', 'champion'] as const).filter( + (persona) => !covered.has(persona), + ); + accounts.push({ companyId, company: entry.company, contacts, missing }); + } + + return accounts.slice(0, input.limit ?? 500); +} diff --git a/packages/pipeline/src/autopilot.test.ts b/packages/pipeline/src/autopilot.test.ts index c31bb81..2fae461 100644 --- a/packages/pipeline/src/autopilot.test.ts +++ b/packages/pipeline/src/autopilot.test.ts @@ -504,9 +504,10 @@ describe('one mailbox, several colleagues', () => { expect(result.skipped[0]?.reason).toMatch(/address|company inbox/i); }); - test('still writes to two different mailboxes', async () => { - // The guard must not become "one email per company" — two people with - // their own addresses are two conversations. + test('two colleagues with their own mailboxes are written to one at a time', async () => { + // Account expansion (Hunter's planner): never two contacts at one company + // at once. The second waits for the first one's 21-day window, and is not + // lost — its card stays pending. seeded = await seedDatabase('autopilot-distinct-inboxes'); const { db } = seeded; @@ -522,8 +523,14 @@ describe('one mailbox, several colleagues', () => { }); const { sent, mailer } = recordingMailer(); - await runAutopilot({ db, mailer }, SEED.workspaceId); + const result = await runAutopilot({ db, mailer }, SEED.workspaceId); + + expect(sent).toHaveLength(1); + expect(result.skipped.some((skip) => /one contact per company/.test(skip.reason))).toBe(true); + // Three weeks later the colleague's turn comes. + const later = new Date(Date.now() + 22 * 86_400_000); + await runAutopilot({ db, mailer, now: later }, SEED.workspaceId); expect(sent.map((message) => message.to).sort()).toEqual(['jane@acme.com', 'tom@acme.com']); }); diff --git a/packages/pipeline/src/autopilot.ts b/packages/pipeline/src/autopilot.ts index 5d28283..9977dc5 100644 --- a/packages/pipeline/src/autopilot.ts +++ b/packages/pipeline/src/autopilot.ts @@ -50,7 +50,14 @@ import { import { budgetStatus } from './metering'; import { holdReason } from './lead-screen'; import { ownDomains, refuseRecipient } from './recipient-guard'; -import { catchAllAllowed, formatRate } from '@outreachgraph/domain'; +import { + catchAllAllowed, + classifyPersona, + EXPANSION_GAP_DAYS, + formatRate, + orderForExpansion, +} from '@outreachgraph/domain'; +import { companyHeldBy, describeCompanyHold, type CompanyBusy } from './account-expansion'; import type { VerifierDeps } from '@outreachgraph/providers'; import { applyCampaignBounceGate, @@ -158,6 +165,8 @@ interface Candidate { readonly checks_json: string | null; readonly company_name: string | null; readonly company_contact_email: string | null; + readonly company_id: string | null; + readonly person_title: string | null; readonly person_email: string | null; readonly failed_attempts: number; } @@ -316,6 +325,9 @@ export function describeHold(reason: string): string { if (/talking to this person is paused/i.test(reason)) { return 'the mailbox already talking to them is paused'; } + if (/one contact per company/i.test(reason)) { + return 'a colleague at the same company is mid-sequence'; + } if (/campaign paused/i.test(reason)) return 'the campaign bounced over 2% and is re-verifying'; if (/failed verification/i.test(reason)) return 'the address failed verification'; if (/accept-all/i.test(reason)) return 'an accept-all address, held while bounces are above 1%'; @@ -392,6 +404,7 @@ export async function runAutopilot( c.approval_mode, c.budget_json, p.display_name, p.status AS person_status, p.believed_minor, p.outreach_eligible, p.identity_confidence, + p.current_company_id AS company_id, p.current_title AS person_title, (SELECT ls.findings FROM lead_screens ls WHERE ls.workspace_id = r.workspace_id AND ls.person_id = p.id AND ls.allowed_at IS NULL) AS screen_findings, @@ -466,7 +479,17 @@ export async function runAutopilot( const droppedOnRecheck = new Map(); let verificationsLeft = deps.maxVerificationsPerRun ?? DEFAULT_VERIFICATIONS_PER_RUN; - for (const row of candidates) { + // Account expansion: one contact per company at a time, opened by the + // budget holder and widened persona by persona. Each company's cards move + // up behind its best one; the order between companies is untouched. + const queue = orderForExpansion( + candidates, + (row) => row.company_id, + (row) => classifyPersona(row.person_title), + ); + const companyHolds = new Map(); + + for (const row of queue) { if (today >= cap) { completed = false; break; @@ -616,6 +639,29 @@ export async function runAutopilot( continue; } + // ---------------------------------------------------------- company gate + // + // Never two contacts at one company at once. A colleague written to in + // the last 21 days (or on an active cadence) who has not answered holds + // the company; a send in this run holds it too, from the cache below. + if (row.company_id) { + let held = companyHolds.get(row.company_id); + if (held === undefined) { + held = + (await companyHeldBy(db, { + workspaceId, + companyId: row.company_id, + personId: row.person_id, + at, + })) ?? null; + companyHolds.set(row.company_id, held); + } + if (held && held.personId !== row.person_id) { + await note(describeCompanyHold(held, row.company_name)); + continue; + } + } + // ------------------------------------------------------------- policy // // Re-evaluated from live rows, never from the stored snapshot. @@ -869,6 +915,16 @@ export async function runAutopilot( today += 1; ledger.release(workspaceId, row.recommendation_id); + // This person now holds their company for the rest of the run. + if (row.company_id) { + companyHolds.set(row.company_id, { + personId: row.person_id, + name: row.display_name, + persona: classifyPersona(row.person_title), + until: new Date(at.getTime() + EXPANSION_GAP_DAYS * 86_400_000).toISOString(), + }); + } + // 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 }); diff --git a/packages/pipeline/src/index.ts b/packages/pipeline/src/index.ts index c37746b..5da8169 100644 --- a/packages/pipeline/src/index.ts +++ b/packages/pipeline/src/index.ts @@ -564,3 +564,11 @@ export { type ListHealthReport, type VerifyOutcome, } from './list-quality'; +export { + campaignAccounts, + companyHeldBy, + describeCompanyHold, + type Account, + type AccountContact, + type CompanyBusy, +} from './account-expansion';