App Growth Agent Orchestrator source

Imported TypeScript file from src/lib/agent/orchestrator.ts.

Source file

src/lib/agent/orchestrator.ts

TypeScript371 lines

import { prisma } from '@/lib/prisma';
import { discoverLeads, createOutreachDraft } from '@/lib/agent/jobs';
import { agentQueue } from '@/queue/client';
import { TASKS } from '@/queue/tasks';
import { loadAppContext, loadCampaignContext } from '@/lib/agent/context';
import { runStructured } from '@/lib/llm/client';
import { PlanSchema, QualificationSchema, DraftSchema, CritiqueSchema } from '@/lib/llm/schemas';
import { buildPlanPrompt, buildQualificationPrompt, buildDraftPrompt, buildCritiquePrompt } from '@/lib/llm/prompts';
import type { CampaignPlan, QualificationResult, DraftResult, CritiqueResult } from '@/lib/agent/types';
import { scoreLead } from '@/lib/scoring';
import { Prisma } from '@prisma/client';
export async function runCampaignTick(campaignId: string) {
const run = await prisma.agentRun.create({ data: { campaignId } });
try {
const campaign = await prisma.campaign.findUnique({
where: { id: campaignId },
include: { app: true, policies: true }
});
if (!campaign) throw new Error('campaign_not_found');
const appContext = await loadAppContext(campaign.appId);
const campaignContext = await loadCampaignContext(campaignId);
const planRaw = await runStepWithFallback<CampaignPlan>(
run.id,
'plan',
{ campaignId },
async (stepId) => {
const prompt = buildPlanPrompt({ campaign, appContext, campaignContext });
return runStructured({
system: prompt.system,
user: prompt.user,
schema: PlanSchema,
runId: run.id,
stepId
});
},
buildDefaultPlan(campaign)
);
const plan = normalizePlan(planRaw);
const discovery = await runStepWithFallback(
run.id,
'discover',
{ queries: plan.queryVariants, limit: plan.maxCandidates },
async () =>
discoverLeads(campaignId, {
queries: plan.queryVariants,
limit: plan.maxCandidates,
minScore: plan.minHeuristicScore
}),
{ created: 0, error: 'discovery_failed' }
);
const leads = await prisma.lead.findMany({
where: { campaignId, status: 'new' },
orderBy: [{ qualityScore: 'desc' }, { createdAt: 'asc' }],
take: plan.maxLeadsToContact
});
let qualifiedCount = 0;
let draftedCount = 0;
let queuedCount = 0;
let skippedExisting = 0;
for (const lead of leads) {
const hasRecent = await hasRecentOutbound(lead.id, 14);
if (hasRecent) {
skippedExisting += 1;
continue;
}
const source = await prisma.leadSource.findFirst({
where: { leadId: lead.id },
orderBy: { ingestedAt: 'desc' }
});
const excerpt = source?.sourceExcerpt ?? lead.painStatement ?? '';
const qualification = await runStepWithFallback<QualificationResult>(
run.id,
'qualify_lead',
{ leadId: lead.id, sourceUrl: source?.sourceUrl },
async (stepId) => {
const prompt = buildQualificationPrompt({
lead,
sourceExcerpt: excerpt,
sourceUrl: source?.sourceUrl,
appContext,
campaignContext
});
return runStructured({
system: prompt.system,
user: prompt.user,
schema: QualificationSchema,
runId: run.id,
stepId
});
},
buildFallbackQualification(lead, excerpt)
);
if (!qualification.shouldContact || qualification.fitScore < plan.minHeuristicScore) {
continue;
}
qualifiedCount += 1;
await prisma.lead.update({
where: { id: lead.id },
data: {
qualityScore: qualification.fitScore,
painStatement: qualification.painSummary
}
});
const channel = lead.platform === 'email' ? 'email' : 'reddit_dm';
const draft = await runStepWithFallback<DraftResult>(
run.id,
'draft_outreach',
{ leadId: lead.id, channel },
async (stepId) => {
const prompt = buildDraftPrompt({
lead,
qualification,
appContext,
campaignContext,
channel
});
return runStructured({
system: prompt.system,
user: prompt.user,
schema: DraftSchema,
runId: run.id,
stepId
});
},
buildFallbackDraft(lead, channel)
);
const critique = await runStepWithFallback<CritiqueResult>(
run.id,
'critic_outreach',
{ leadId: lead.id, channel },
async (stepId) => {
const prompt = buildCritiquePrompt({
channel,
draft,
appContext,
campaignContext
});
return runStructured({
system: prompt.system,
user: prompt.user,
schema: CritiqueSchema,
runId: run.id,
stepId
});
},
{
riskLevel: 'medium',
issues: ['llm_unavailable'],
revisedBody: undefined,
requiresApproval: false,
confidence: 0.4
}
);
const finalBody = critique.revisedBody ?? draft.body;
// Only ever *raise* the approval requirement; never lower it.
const requiresApprovalOverride =
critique.requiresApproval || critique.riskLevel === 'high' ? true : undefined;
const { messageId, lint, requiresApproval } = await createOutreachDraft(lead.id, {
channel,
subject: draft.subject,
body: finalBody,
metadata: {
plan,
qualification,
draft,
critique,
evidence: qualification.evidence,
sourceUrl: source?.sourceUrl
},
requiresApprovalOverride
});
draftedCount += 1;
const outboundDisabled = isOutboundDisabled();
if (lint.ok && !requiresApproval && !outboundDisabled) {
await agentQueue.add(TASKS.MESSAGE_SEND, { messageId });
queuedCount += 1;
} else if (outboundDisabled) {
await prisma.agentAction.create({
data: {
campaignId,
actionType: 'send',
status: 'blocked',
reason: 'outbound_disabled',
input: { messageId }
}
});
}
}
await prisma.agentRun.update({
where: { id: run.id },
data: {
status: 'success',
endedAt: new Date(),
summary: {
ok: true,
plan,
discovery,
totals: {
leadsConsidered: leads.length,
qualified: qualifiedCount,
drafted: draftedCount,
queued: queuedCount,
skippedExisting
}
}
}
});
return { ok: true };
} catch (e: any) {
await prisma.agentRun.update({
where: { id: run.id },
data: {
status: 'failed',
endedAt: new Date(),
summary: { ok: false, error: e?.message ?? String(e) }
}
});
throw e;
}
}
async function runStepWithFallback<T>(
runId: string,
stepName: string,
input: Record<string, unknown>,
fn: (stepId: string) => Promise<T>,
fallback: T
): Promise<T> {
const toJson = (value: unknown) => {
try {
const str = JSON.stringify(value);
if (str === undefined) return Prisma.JsonNull;
return JSON.parse(str) as Prisma.InputJsonValue;
} catch (err) {
return {
error: 'non_serializable_value',
message: (err as Error).message
} as unknown as Prisma.InputJsonValue;
}
};
const step = await prisma.agentStep.create({
data: {
runId,
stepName,
status: 'running',
input: toJson(input)
}
});
try {
const output = await fn(step.id);
await prisma.agentStep.update({
where: { id: step.id },
data: {
status: 'success',
output: toJson(output),
endedAt: new Date()
}
});
return output;
} catch (err: any) {
await prisma.agentStep.update({
where: { id: step.id },
data: {
status: 'failed',
output: toJson(fallback),
error: err?.message ?? String(err),
endedAt: new Date()
}
});
return fallback;
}
}
function buildDefaultPlan(campaign: { keywords: unknown; dailyDmCap: number }): CampaignPlan {
const keywords = Array.isArray(campaign.keywords) ? campaign.keywords.map(String) : [];
const queryVariants = keywords.length
? keywords.map(k => `${k} shopify`)
: ['shopify bulk edit', 'shopify bulk update'];
return {
goal: 'Find Shopify merchants with bulk editing pain',
queryVariants,
maxCandidates: 25,
maxLeadsToContact: Math.max(1, Math.min(5, campaign.dailyDmCap || 5)),
minHeuristicScore: 35,
notes: 'fallback_plan'
};
}
function buildFallbackQualification(lead: { painStatement: string | null }, excerpt: string): QualificationResult {
const painText = excerpt || lead.painStatement || '';
const isMerchantLikely = /my store|our store|we sell|my products/i.test(painText);
const recencyDays = 14;
const fitScore = scoreLead({ painText, isMerchantLikely, recencyDays });
return {
fitScore,
isMerchantLikely,
painSummary: painText.slice(0, 200),
evidence: painText ? [painText.slice(0, 200)] : [],
reasons: ['fallback_heuristic_scoring'],
shouldContact: fitScore >= 35
};
}
function buildFallbackDraft(lead: { painStatement: string | null }, channel: 'reddit_dm' | 'email'): DraftResult {
const pain = (lead.painStatement ?? '').slice(0, 220);
const subject = channel === 'email' ? 'Quick question about your Shopify store' : undefined;
const body = `Hey — saw your post about struggling with bulk edits in Shopify.\n\n` +
`I’m building a lightweight bulk editor that’s meant to help with the kind of issue you described ("${pain}").\n` +
`I’m looking for a small number of early users to stress-test it and give honest feedback.\n\n` +
`If you want, I can give you free access while we iterate. No obligation — just trying to make sure it actually solves this problem.\n\n` +
`If you're interested, reply here and I’ll send setup details.`;
return { subject, body, personalization: [], rationale: 'fallback_template' };
}
async function hasRecentOutbound(leadId: string, days: number) {
const since = new Date();
since.setDate(since.getDate() - days);
const existing = await prisma.message.findFirst({
where: {
leadId,
direction: 'outbound',
createdAt: { gte: since },
status: { in: ['draft', 'sent'] }
},
select: { id: true }
});
return !!existing;
}
function isOutboundDisabled() {
return ['true', '1', 'yes'].includes((process.env.STOP_OUTBOUND_SEND ?? '').toLowerCase());
}
function normalizePlan(plan: CampaignPlan): CampaignPlan {
const clamp = (value: number, min: number, max: number) => Math.min(max, Math.max(min, value));
const queryVariants = (plan.queryVariants || []).map(q => q.trim()).filter(Boolean).slice(0, 12);
return {
...plan,
queryVariants: queryVariants.length ? queryVariants : ['shopify bulk edit'],
maxCandidates: clamp(plan.maxCandidates, 5, 100),
maxLeadsToContact: clamp(plan.maxLeadsToContact, 1, 20),
minHeuristicScore: clamp(plan.minHeuristicScore, 0, 100)
};
}