crm agentic-first CRM

28export async function claimDue(
29	limit: number,
30	kinds: { only: readonly string[] } | { except: readonly string[] },
31	leaseMs = LEASE_MS,
32): Promise<LeasedTask[]> {
33	const now = new Date();
34	const until = new Date(now.getTime() + leaseMs);
35
36	const list = "only" in kinds ? kinds.only : kinds.except;
37	if ("only" in kinds && list.length === 0) return [];
38
39	const match = Prisma.sql`t2.kind ${"only" in kinds ? Prisma.sql`IN` : Prisma.sql`NOT IN`} (${Prisma.join(list)})`;
40
41	const claimed = await db.$queryRaw<LeasedTask[]>`
42		UPDATE "agentTask" AS t
43		SET "leasedUntil" = ${until},
44			"startedAt" = COALESCE(t."startedAt", ${now}),
45			"attempts" = t."attempts" + 1
46		FROM (
47			SELECT t2.id FROM "agentTask" AS t2
48			WHERE t2."finishedAt" IS NULL
49				AND t2."dueAt" <= ${now}
50				AND (t2."leasedUntil" IS NULL OR t2."leasedUntil" < ${now})
51				AND t2."attempts" < ${MAX_ATTEMPTS}
52				AND ${match}
53			ORDER BY t2."priority" DESC, t2."dueAt" ASC
54			LIMIT ${limit}
55			FOR UPDATE SKIP LOCKED
56		) AS due
57		WHERE t.id = due.id
58		RETURNING t.id, t."contactId", t."companyId", t.kind, t.reason,
59			t.budget, t.attempts, t.priority, t."dueAt";
60	`;
61
62	return claimed.sort(
63		(a, b) => b.priority - a.priority || a.dueAt.getTime() - b.dueAt.getTime(),
64	);
65}

67export async function retireExhausted(): Promise<TaskSubject[]> {
68	const now = new Date();
69
70	return db.$queryRaw<TaskSubject[]>`
71		UPDATE "agentTask" AS t
72		SET "finishedAt" = ${now},
73			"outcome" = ${`Gave up after ${MAX_ATTEMPTS} attempts: the session never reported back.`}
74		WHERE t."finishedAt" IS NULL
75			AND t."attempts" >= ${MAX_ATTEMPTS}
76			AND (t."leasedUntil" IS NULL OR t."leasedUntil" < ${now})
77		RETURNING t.id, t."contactId", t."companyId", t.kind;
78	`;
79}
80
81export async function completeTask(
82	taskId: string,
83	outcome: string,
84	sessionId?: string,
85): Promise<TaskSubject | null> {
86	const { count } = await db.agentTask.updateMany({
87		where: { id: taskId, finishedAt: null },
88		data: {
89			finishedAt: new Date(),
90			outcome: outcome.slice(0, 500),
91			...(sessionId ? { sessionId } : {}),
92		},
93	});
94
95	if (count === 0) return null;
96
97	return db.agentTask.findUnique({
98		where: { id: taskId },
99		select: { id: true, contactId: true, companyId: true, kind: true },
100	});

42export async function runVisibleLane(): Promise<number> {
43	let handled = 0;
44
45	while (handled < VISIBLE_BATCH) {
46		const tasks = await claimDue(
47			Math.min(VISIBLE_CONCURRENCY, VISIBLE_BATCH - handled),
48			{ only: DIRECT_KINDS },
49			VISIBLE_LEASE_MS,
50		);
51
52		if (tasks.length === 0) break;
53
54		await runLimited(VISIBLE_CONCURRENCY, tasks, runDirect);
55		handled += tasks.length;
56	}
57
58	return handled;
59}
60
61async function runDirect(task: LeasedTask): Promise<void> {
62	try {
63		if (task.kind === "brand" && task.companyId) {
64			const result = await runBrand({ companyId: task.companyId });
65			if (result.retryable) return;
66
67			await completeTask(task.id, brandOutcome(result));
68			return;
69		}
70
71		if (task.kind === "portrait" && task.contactId) {
72			const portrait = await runPortrait({
73				contactId: task.contactId,
74				spend: () => ({ ok: true }),
75			});
76
77			await completeTask(
78				task.id,
79				portrait.stored
80					? `Picture stored from ${portrait.source}.`
81					: (portrait.reason ?? "No picture found."),
82			);
83			return;
84		}
85
86		await completeTask(task.id, "The record this names is gone.");
87	} catch (error) {
88		const reason = error instanceof Error ? error.message : String(error);
89		await settle(task, EnrichmentStatus.FAILED, reason).catch(() => {});
90	}

5export default defineSchedule({
6	cron: "* * * * *",
7	async run({ receive, waitUntil, appAuth }) {
8		waitUntil(
9			drainAll((task) =>
10				receive(crm, {
11					message: brief(task),
12					target: { taskId: task.id },
13					auth: taskAuth(task, appAuth),
14				}),
15			),
16		);
17	},

1import { defaultBackend, defineSandbox } from "eve/sandbox";
2
3export default defineSandbox({
4	backend: defaultBackend({
5		vercel: { networkPolicy: "deny-all" },
6		docker: { networkPolicy: "deny-all" },
7		microsandbox: { networkPolicy: "deny-all" },
8	}),
9});

8export default defineTool({
9	description:
10		"Record one claim about a contact — title, employer, a profile URL, seniority — together with the evidence for it. The evidence decides whether it is written to the record or offered to a rep as a suggestion. Never invent evidence you did not observe.",
11	inputSchema: z.object({
12		contactId: z.string(),
13		field: z
14			.enum(FACT_FIELDS as [FactField, ...FactField[]])
15			.describe("Which fact about them this is."),
16		value: z
17			.string()
18			.describe("The claim itself, exactly as the source states it."),
19		evidence: z
20			.array(
21				z.object({
22					kind: z
23						.enum(Object.keys(WEIGHTS) as [EvidenceKind, ...EvidenceKind[]])
24						.describe(
25							"What kind of thing you saw. Use `contradiction` when two sources disagree.",
26						),
27					detail: z
28						.string()
29						.describe(
30							"What it actually said, in one line a rep would understand.",
31						),
32					sourceUrl: z.string().optional(),
33				}),
34			)
35			.min(1)
36			.describe("Everything you observed. One entry per independent source."),
37		method: z
38			.string()
39			.describe(
40				'Where it came from: "linkedin.profile", "github.api", "crm.thread", "web".',
41			),
42		sourceUrl: z
43			.string()
44			.optional()
45			.describe("The page a rep should open to check."),
46	}),
47	async execute(input) {
48		focusOn({ contactId: input.contactId });
49
50		const result = await recordFact({
51			contactId: input.contactId,
52			field: input.field,
53			value: input.value,
54			evidence: input.evidence as Evidence[],
55			method: input.method,
56			sourceUrl: input.sourceUrl,
57		});
58
59		return {
60			stored: result.stored,
61			applied: result.applied,
62			band: result.band,
63			score: Number(result.score.toFixed(2)),
64			rationale: result.rationale,
65			...(result.reason ? { reason: result.reason } : {}),
66		};
67	},

22export const WEIGHTS: Record<EvidenceKind, Weighting> = {
23	"profile.email-match": {
24		weight: 0.95,
25		primary: true,
26		label: "their email address is on the profile",
27	},
28	"linkedin.employer-and-name": {
29		weight: 0.85,
30		primary: true,
31		label: "LinkedIn: employer and name both match",
32	},
33	"crm.thread-reply": {
34		weight: 0.85,
35		primary: true,
36		label: "they replied on a thread we have",
37	},
38	"crm.signature-block": {
39		weight: 0.8,
40		primary: true,
41		label: "their own email signature says so",
42	},
43	"github.account-identity": {
44		weight: 0.8,
45		primary: true,
46		label: "the GitHub account names them or their employer",
47	},
48	"crm.meeting-attendance": {
49		weight: 0.7,
50		primary: true,
51		label: "they attended a meeting on our calendar",
52	},
53	"web.cited-claim": {
54		weight: 0.4,
55		primary: false,
56		label: "a cited web source states it",
57	},
58	"handle.name-form": {
59		weight: 0.35,
60		primary: false,
61		label: "the handle is a form of their name",
62	},
63	"search.cites-profile": {
64		weight: 0.35,
65		primary: false,
66		label: "a search for them cites this profile",
67	},
68	"employer-only": {
69		weight: 0.2,
70		primary: false,
71		label: "the employer matches, the name does not",
72	},
73	contradiction: {
74		weight: 0,
75		primary: false,
76		label: "another source disagrees",
77	},
78};
79
80export type Evidence = {
81	kind: EvidenceKind;
82	detail: string;
83	sourceUrl?: string;
84};
85
86export type Scored = {
87	score: number;
88	band: FactBand | null;
89	hasPrimary: boolean;
90	rationale: string;
91};
92
93const CEILING = 0.99;
94
95const CONTRADICTED = 0.45;
96
97export const BAND_FLOOR = { VERIFIED: 0.85, PROBABLE: 0.55, POSSIBLE: 0.3 };
98
99export function scoreEvidence(evidence: Evidence[]): Scored {
100	if (evidence.length === 0) {
101		return {
102			score: 0,
103			band: null,
104			hasPrimary: false,
105			rationale: "No evidence.",
106		};
107	}
108
109	const contradicted = evidence.some((item) => item.kind === "contradiction");
110	const hasPrimary = evidence.some((item) => WEIGHTS[item.kind].primary);
111
112	const combined = evidence.reduce(
113		(remaining, item) => remaining * (1 - WEIGHTS[item.kind].weight),
114		1,
115	);
116
117	let score = Math.min(CEILING, 1 - combined);
118	if (contradicted) score = Math.min(score, CONTRADICTED);
119
120	return {
121		score,
122		band: bandFor(score, hasPrimary),
123		hasPrimary,
124		rationale: rationaleFor(evidence, contradicted, hasPrimary),
125	};
126}
127
128export function bandFor(score: number, hasPrimary: boolean): FactBand | null {
129	if (score >= BAND_FLOOR.VERIFIED && hasPrimary) return FactBand.VERIFIED;
130	if (score >= BAND_FLOOR.PROBABLE) return FactBand.PROBABLE;
131	if (score >= BAND_FLOOR.POSSIBLE) return FactBand.POSSIBLE;
132	return null;

12export async function mintBridgeToken(
13	user: {
14		id: string;
15		email: string;
16		name: string;
17	},
18	record: { contactId?: string; companyId?: string; dealId?: string } = {},
19): Promise<string> {
20	const secret = process.env.AGENT_BRIDGE_SECRET;
21	if (!secret) throw new Error("AGENT_BRIDGE_SECRET is not set.");
22
23	const now = Math.floor(Date.now() / 1000);
24
25	const header = { alg: "HS256", typ: "JWT" };
26	const payload = {
27		iss: ISSUER,
28		aud: AUDIENCE,
29		sub: user.id,
30		email: user.email,
31		name: user.name,
32		...(record.contactId ? { contactId: record.contactId } : {}),
33		...(record.companyId ? { companyId: record.companyId } : {}),
34		...(record.dealId ? { dealId: record.dealId } : {}),
35		iat: now,
36		nbf: now - 5,
37		exp: now + TTL_SECONDS,
38	};
39
40	const signingInput = `${base64url(JSON.stringify(header))}.${base64url(
41		JSON.stringify(payload),
42	)}`;
43
44	const key = await crypto.subtle.importKey(
45		"raw",
46		new TextEncoder().encode(secret),
47		{ name: "HMAC", hash: "SHA-256" },
48		false,
49		["sign"],
50	);
51
52	const signature = await crypto.subtle.sign(
53		"HMAC",
54		key,
55		new TextEncoder().encode(signingInput),
56	);
57
58	return `${signingInput}.${base64url(new Uint8Array(signature))}`;
59}

471function useSavedConversation({
472	record,
473	conversation,
474	opening,
475	session,
476	messages,
477}: {
478	record: { contactId?: string; companyId?: string; dealId?: string };
479	conversation: Conversation | null;
480	opening: React.RefObject<string | null>;
481	session: {
482		sessionId?: string;
483		continuationToken?: string;
484		streamIndex: number;
485	} | null;
486	messages: number;
487}) {
488	const trpc = useTRPC();
489	const queryClient = useQueryClient();
490	const save = useMutation(trpc.conversations.save.mutationOptions({}));
491
492	const sessionId = session?.sessionId ?? null;
493	const token = session?.continuationToken ?? null;
494	const streamIndex = session?.streamIndex ?? 0;
495	const { contactId, companyId, dealId } = record;
496
497	const isNew = conversation === null || conversation.sessionId !== sessionId;
498
499	const latest = useRef({ save, queryClient, trpc, opening });
500	latest.current = { save, queryClient, trpc, opening };
501
502	const written = useRef<string | null>(null);
503
504	useEffect(() => {
505		if (!sessionId) return;
506
507		const cursor = `${sessionId}:${token ?? ""}:${messages}`;
508		if (written.current === cursor) return;
509		written.current = cursor;
510
511		const {
512			save: mutation,
513			queryClient: cache,
514			trpc: api,
515			opening: title,
516		} = latest.current;
517
518		mutation.mutate(
519			{
520				...(contactId ? { contactId } : {}),
521				...(companyId ? { companyId } : {}),
522				...(dealId ? { dealId } : {}),
523				sessionId,
524				continuationToken: token,
525				streamIndex,
526				messageCount: messages,
527				...(isNew ? { title: title.current ?? undefined } : {}),
528			},
529			{
530				onSuccess: () => {
531					if (!isNew) return;
532					void cache.invalidateQueries({
533						queryKey: api.conversations.list.pathKey(),
534					});
535				},
536			},
537		);
538	}, [
539		sessionId,
540		token,
541		streamIndex,
542		messages,
543		contactId,
544		companyId,
545		dealId,
546		isNew,
547	]);
548}

57describe("claimDue", () => {
58	it("claims due work and leases it", async () => {
59		const task = await queue();
60
61		const claimed = await claimDue(10, RESEARCH);
62		expect(claimed.map((t) => t.id)).toContain(task.id);
63
64		const row = await db.agentTask.findUnique({ where: { id: task.id } });
65		expect(row?.leasedUntil).not.toBeNull();
66		expect(row?.startedAt).not.toBeNull();
67	});
68
69	it("does not hand the same row to two dispatchers", async () => {
70		await Promise.all([queue(), queue(), queue()]);
71
72		const [first, second] = await Promise.all([
73			claimDue(3, RESEARCH),
74			claimDue(3, RESEARCH),
75		]);
76		const ids = [...first, ...second].map((t) => t.id);
77
78		expect(new Set(ids).size).toBe(ids.length);
79		expect(ids).toHaveLength(3);
80	});
81
82	it("leaves work that is not due yet", async () => {
83		await queue({ dueAt: new Date(Date.now() + 60_000) });
84		const claimed = await claimDue(10, RESEARCH);
85		expect(claimed).toHaveLength(0);
86	});
87
88	it("takes the most urgent first", async () => {
89		const low = await queue({ priority: 0 });
90		const high = await queue({ priority: 100 });
91
92		const claimed = await claimDue(1, RESEARCH);
93		expect(claimed[0]?.id).toBe(high.id);
94		expect(claimed[0]?.id).not.toBe(low.id);
95	});
96
97	it("does not re-claim a leased row, and does re-claim an expired one", async () => {
98		const task = await queue();
99		await claimDue(10, RESEARCH);
100
101		expect(await claimDue(10, RESEARCH)).toHaveLength(0);
102
103		await db.agentTask.update({
104			where: { id: task.id },
105			data: { leasedUntil: new Date(Date.now() - 1000) },
106		});
107
108		expect((await claimDue(10, RESEARCH)).map((t) => t.id)).toContain(task.id);
109	});
110
111	it("stops handing out a row that has spent its attempts", async () => {
112		const task = await queue();
113
114		for (let attempt = 0; attempt < MAX_ATTEMPTS; attempt++) {
115			expect((await claimDue(10, RESEARCH)).map((t) => t.id)).toContain(
116				task.id,
117			);
118			await expire(task.id);
119		}
120
121		expect(await claimDue(10, RESEARCH)).toHaveLength(0);
122	});
123
124	it("counts the attempts it has handed out", async () => {
125		const task = await queue();
126
127		expect((await claimDue(10, RESEARCH))[0]?.attempts).toBe(1);
128		await expire(task.id);
129		expect((await claimDue(10, RESEARCH))[0]?.attempts).toBe(2);
130	});
131
132	it("stops claiming once the work is finished", async () => {
133		const task = await queue();
134		await claimDue(10, RESEARCH);
135		await completeTask(task.id, "ran");
136
137		await db.agentTask.update({
138			where: { id: task.id },
139			data: { leasedUntil: null },
140		});
141
142		expect(await claimDue(10, RESEARCH)).toHaveLength(0);
143	});

12describe("scoreEvidence", () => {
13	it("writes a fact when a primary source names them", () => {
14		const scored = scoreEvidence(of("profile.email-match"));
15		expect(scored.band).toBe("VERIFIED");
16		expect(scored.hasPrimary).toBe(true);
17	});
18
19	it("treats our own mailbox as primary evidence", () => {
20		expect(scoreEvidence(of("crm.thread-reply")).band).toBe("VERIFIED");
21	});
22
23	it("writes a title a signature block states on a thread they replied to", () => {
24		expect(scoreEvidence(of("crm.signature-block")).band).toBe("PROBABLE");
25		expect(
26			scoreEvidence(of("crm.thread-reply", "crm.signature-block")).band,
27		).toBe("VERIFIED");
28	});
29
30	it("refuses to write anything without a primary source, however much of it there is", () => {
31		const scored = scoreEvidence(
32			of(
33				"handle.name-form",
34				"search.cites-profile",
35				"web.cited-claim",
36				"employer-only",
37			),
38		);
39
40		expect(scored.score).toBeGreaterThan(BAND_FLOOR.PROBABLE);
41		expect(scored.hasPrimary).toBe(false);
42		expect(scored.band).toBe("PROBABLE");
43	});
44
45	it("holds the primary-source rule even if the weights are re-calibrated upwards", () => {
46		expect(bandFor(0.99, false)).toBe("PROBABLE");
47		expect(bandFor(0.85, true)).toBe("VERIFIED");
48	});
49
50	it("makes an unreadable X handle a suggestion, not a value", () => {
51		const scored = scoreEvidence(
52			of("handle.name-form", "search.cites-profile"),
53		);
54		expect(scored.band).toBe("PROBABLE");
55	});
56
57	it("holds a fact when sources disagree, rather than averaging them", () => {
58		const scored = scoreEvidence([
59			...of("linkedin.employer-and-name"),
60			{
61				kind: "contradiction",
62				detail: "their signature says a different employer",
63			},
64		]);
65
66		expect(scored.band).toBe("POSSIBLE");
67		expect(scored.rationale).toContain("Held:");
68	});
69
70	it("drops evidence too weak to keep at all", () => {
71		expect(scoreEvidence(of("employer-only")).band).toBeNull();
72		expect(scoreEvidence([]).band).toBeNull();
73	});
74
75	it("never claims certainty", () => {
76		const everything = scoreEvidence(
77			of(
78				"profile.email-match",
79				"linkedin.employer-and-name",
80				"crm.thread-reply",
81				"github.account-identity",
82			),
83		);
84		expect(everything.score).toBeLessThan(1);
85	});
86
87	it("explains itself in words a rep could read", () => {
88		const scored = scoreEvidence(
89			of("github.account-identity", "handle.name-form"),
90		);
91		expect(scored.rationale).toContain("GitHub account");
92		expect(scored.rationale).toContain("handle");
93	});