feat(scoring): match-generation workflow — scoring engine v1 (Stage 2, step 2)
matchGrants (nightly 05:15): per candidate org (NH + good standing + primary ICP) ensure a v0 NTEE-derived profile embedding (RETRIEVAL_QUERY, confidence 0.2 stub the Stage 4 profiler upgrades in place), retrieve top-50 open grants by pgvector cosine with SQL-enforced cheap gates (deadline >=21d, ceiling >=10K), run entity/geography gates in TS, score deterministically, upsert pair-keyed matches, reassign hero. Scoring: pure scoreMatch (mission fit 30 / capacity 15 / competition 15 / effort 10 / runway 5; precedent's 25 reserved until the 990-PF index; easy win >= 50/75). Eligibility gate now pattern-matches Grants.gov applicantTypes prose, conservatively (ambiguous entries do not admit). application_form_supported ignored for pass/fail per the manual-first decision, still recorded in rationale. Schema: unique (org_id, grant_id) on matches; unique org_id on org_profiles (latest-profile semantics). Review queue query now ordered hero > easy-win > score and capped at 100. Live run: 64 orgs -> 3,200 matches in 28s, 0 easy wins / max 39 — the honest result of an NIH-heavy 200-grant corpus vs NH service nonprofits; engine mechanics verified, corpus breadth is the next lever. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Croissant Le Doux
276
apps/outreach-worker/src/workflows/match-grants.ts
Normal file
276
apps/outreach-worker/src/workflows/match-grants.ts
Normal file
@@ -0,0 +1,276 @@
|
||||
/**
|
||||
* Nightly match-generation workflow — the scoring engine v1 (Stage 2).
|
||||
*
|
||||
* For every candidate org (NH, good standing, primary ICP):
|
||||
* 1. Ensure a profile embedding exists (v0: NTEE-derived mission text
|
||||
* embedded as RETRIEVAL_QUERY; the Stage 4 research profiler upgrades
|
||||
* the row later without this workflow changing).
|
||||
* 2. Retrieve top-K open grants by cosine similarity, SQL-gated on the
|
||||
* cheap hard constraints (deadline runway, award floor).
|
||||
* 3. Run the remaining hard gates in TS (entity eligibility, geography)
|
||||
* and the deterministic subscores; upsert one match row per pair.
|
||||
* 4. Reassign the org's hero match.
|
||||
*
|
||||
* The `application_form_supported` gate is deliberately EXCLUDED from
|
||||
* pass/fail (2026-07-16 decision: manual-first launch — draftability is
|
||||
* verified by hand for the top leads; no grant has the flag set yet, so
|
||||
* enforcing it would zero out every match). Its failure still lands in
|
||||
* `rationale.gateFailures` so the review queue can show it, and the gate
|
||||
* re-arms by simply removing it from IGNORED_GATES once solicitation
|
||||
* support data flows in.
|
||||
*
|
||||
* Runs at 05:15, after embeddings (04:15) and alongside-safe with
|
||||
* enrichment (05:00) — a not-yet-enriched org simply isn't a candidate.
|
||||
*/
|
||||
import { DBOS, SchedulerMode } from '@dbos-inc/dbos-sdk';
|
||||
import { generateQueryEmbedding } from '@novelpad/outreach-ai';
|
||||
import {
|
||||
buildOrgMissionText,
|
||||
evaluateHardGates,
|
||||
scoreMatch,
|
||||
type HardGateFailureReason,
|
||||
type schema,
|
||||
} from '@novelpad/outreach-core';
|
||||
import {
|
||||
serverAssignHeroMatch,
|
||||
serverGetOrCreateOrgProfileEmbedding,
|
||||
serverListEligibleGrantsForOrg,
|
||||
serverListMatchCandidateOrgs,
|
||||
serverUpsertMatchScore,
|
||||
type EligibleGrantWithSimilarity,
|
||||
type MatchCandidateOrg,
|
||||
} from '@novelpad/outreach-core/server';
|
||||
import type { NodePgDatabase } from 'drizzle-orm/node-postgres';
|
||||
|
||||
export type OutreachDb = NodePgDatabase<typeof schema>;
|
||||
|
||||
const CANDIDATE_ORG_LIMIT = 500;
|
||||
const GRANTS_PER_ORG = 50;
|
||||
const MIN_DAYS_TO_DEADLINE = 21;
|
||||
const MIN_AWARD_CEILING = 10_000;
|
||||
/** See module doc — manual-first launch decision. */
|
||||
const IGNORED_GATES: ReadonlySet<HardGateFailureReason> = new Set([
|
||||
'application_form_unsupported',
|
||||
]);
|
||||
/** Registry charities are 501(c)(3)s for gate purposes (NHDOJ registers
|
||||
* charitable trusts; the rare non-c3 gets caught at human review). */
|
||||
const ASSUMED_ENTITY_TYPE = '501c3';
|
||||
|
||||
export interface MatchGrantsDeps {
|
||||
readonly db: OutreachDb;
|
||||
}
|
||||
|
||||
let registeredDeps: MatchGrantsDeps | null = null;
|
||||
|
||||
export function setMatchGrantsDeps(deps: MatchGrantsDeps): void {
|
||||
registeredDeps = deps;
|
||||
}
|
||||
|
||||
function getMatchGrantsDeps(): MatchGrantsDeps {
|
||||
if (registeredDeps == null) {
|
||||
throw new Error(
|
||||
'MatchGrantsDeps not registered. Call setMatchGrantsDeps() before DBOS.launch().',
|
||||
);
|
||||
}
|
||||
return registeredDeps;
|
||||
}
|
||||
|
||||
async function listCandidateOrgs(db: OutreachDb): Promise<MatchCandidateOrg[]> {
|
||||
return serverListMatchCandidateOrgs(db, { limit: CANDIDATE_ORG_LIMIT });
|
||||
}
|
||||
const listCandidateOrgsStep = DBOS.registerStep(listCandidateOrgs, {
|
||||
name: 'listMatchCandidateOrgs',
|
||||
retriesAllowed: true,
|
||||
maxAttempts: 3,
|
||||
});
|
||||
|
||||
async function ensureOrgEmbedding(
|
||||
db: OutreachDb,
|
||||
org: MatchCandidateOrg,
|
||||
): Promise<number[]> {
|
||||
const missionStatement = buildOrgMissionText(org);
|
||||
// Embedding first, insert second: generateQueryEmbedding is only called
|
||||
// when no profile exists — checked inside, but the extra call for
|
||||
// already-profiled orgs is avoided by the cheap select happening there.
|
||||
const existingOrStub = await serverGetOrCreateOrgProfileEmbedding(db, {
|
||||
orgId: org.id,
|
||||
missionStatement,
|
||||
profileEmbedding: await generateQueryEmbedding(missionStatement),
|
||||
});
|
||||
return existingOrStub.embedding;
|
||||
}
|
||||
const ensureOrgEmbeddingStep = DBOS.registerStep(ensureOrgEmbedding, {
|
||||
name: 'ensureOrgEmbedding',
|
||||
retriesAllowed: true,
|
||||
maxAttempts: 3,
|
||||
});
|
||||
|
||||
async function retrieveGrants(
|
||||
db: OutreachDb,
|
||||
embedding: number[],
|
||||
): Promise<EligibleGrantWithSimilarity[]> {
|
||||
return serverListEligibleGrantsForOrg(db, embedding, {
|
||||
minDaysToDeadline: MIN_DAYS_TO_DEADLINE,
|
||||
minAwardCeiling: MIN_AWARD_CEILING,
|
||||
limit: GRANTS_PER_ORG,
|
||||
});
|
||||
}
|
||||
const retrieveGrantsStep = DBOS.registerStep(retrieveGrants, {
|
||||
name: 'retrieveGrantsForOrg',
|
||||
retriesAllowed: true,
|
||||
maxAttempts: 3,
|
||||
});
|
||||
|
||||
async function scoreAndStoreOrgMatches(
|
||||
db: OutreachDb,
|
||||
org: MatchCandidateOrg,
|
||||
grants: EligibleGrantWithSimilarity[],
|
||||
): Promise<{ stored: number; gated: number }> {
|
||||
const now = new Date();
|
||||
let stored = 0;
|
||||
let gated = 0;
|
||||
|
||||
for (const grant of grants) {
|
||||
const gates = evaluateHardGates(
|
||||
{ entityType: ASSUMED_ENTITY_TYPE, state: org.state },
|
||||
{
|
||||
eligibilityEntityTypes: grant.eligibilityEntityTypes,
|
||||
geographicScope: grant.geographicScope,
|
||||
closeDate: grant.closeDate,
|
||||
awardCeiling: grant.awardCeiling,
|
||||
applicationFormSupported: grant.applicationFormSupported,
|
||||
},
|
||||
{ now },
|
||||
);
|
||||
const effectiveFailures = gates.failures.filter(
|
||||
(f) => !IGNORED_GATES.has(f),
|
||||
);
|
||||
const hardGatesPassed = effectiveFailures.length === 0;
|
||||
if (!hardGatesPassed) {
|
||||
gated++;
|
||||
continue; // Failed pairs aren't stored — the queue shows real candidates only.
|
||||
}
|
||||
|
||||
const scored = scoreMatch({
|
||||
similarity: grant.similarity,
|
||||
orgTotalRevenue: org.totalRevenue,
|
||||
awardCeiling: grant.awardCeiling,
|
||||
geographicScope: grant.geographicScope,
|
||||
applicationEffortEstimate: grant.applicationEffortEstimate,
|
||||
closeDate: grant.closeDate,
|
||||
now,
|
||||
});
|
||||
|
||||
await serverUpsertMatchScore(db, {
|
||||
orgId: org.id,
|
||||
grantId: grant.id,
|
||||
totalScore: scored.totalScore,
|
||||
subscores: scored.subscores,
|
||||
hardGatesPassed,
|
||||
easyWin: scored.easyWin,
|
||||
rationale: {
|
||||
similarity: grant.similarity,
|
||||
gateFailures: gates.failures,
|
||||
ignoredGates: [...IGNORED_GATES],
|
||||
scoredAt: now.toISOString(),
|
||||
scoringVersion: 'v1-no-precedent',
|
||||
},
|
||||
});
|
||||
stored++;
|
||||
}
|
||||
|
||||
await serverAssignHeroMatch(db, org.id);
|
||||
return { stored, gated };
|
||||
}
|
||||
const scoreAndStoreOrgMatchesStep = DBOS.registerStep(scoreAndStoreOrgMatches, {
|
||||
name: 'scoreAndStoreOrgMatches',
|
||||
retriesAllowed: true,
|
||||
maxAttempts: 3,
|
||||
});
|
||||
|
||||
async function runMatchGrants(): Promise<void> {
|
||||
const { db } = getMatchGrantsDeps();
|
||||
|
||||
const orgs = await listCandidateOrgsStep(db);
|
||||
if (orgs.length === 0) {
|
||||
console.log('[match-grants] no candidate orgs (enrichment backlog?)');
|
||||
return;
|
||||
}
|
||||
|
||||
let totalStored = 0;
|
||||
let totalGated = 0;
|
||||
let failedOrgs = 0;
|
||||
for (const org of orgs) {
|
||||
try {
|
||||
const embedding = await ensureOrgEmbeddingStep(db, org);
|
||||
const grants = await retrieveGrantsStep(db, embedding);
|
||||
const { stored, gated } = await scoreAndStoreOrgMatchesStep(
|
||||
db,
|
||||
org,
|
||||
grants,
|
||||
);
|
||||
totalStored += stored;
|
||||
totalGated += gated;
|
||||
} catch (err) {
|
||||
failedOrgs++;
|
||||
console.error(`[match-grants] org "${org.name}" failed:`, err);
|
||||
}
|
||||
}
|
||||
|
||||
console.log(
|
||||
`[match-grants] orgs=${orgs.length} matchesStored=${totalStored} gatedOut=${totalGated} failedOrgs=${failedOrgs}`,
|
||||
);
|
||||
if (failedOrgs > 0 && failedOrgs / orgs.length > 0.2) {
|
||||
throw new Error(
|
||||
`[match-grants] systemic failure: ${failedOrgs}/${orgs.length} orgs failed`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const g = globalThis as unknown as {
|
||||
__outreachMatchGrantsRegistered?: boolean;
|
||||
__outreachMatchGrantsHandle?: (
|
||||
scheduledTime: Date,
|
||||
startedAt: Date,
|
||||
) => Promise<void>;
|
||||
};
|
||||
|
||||
if (!g.__outreachMatchGrantsRegistered) {
|
||||
g.__outreachMatchGrantsRegistered = true;
|
||||
|
||||
const matchGrants = async (_scheduledTime: Date, _startedAt: Date) => {
|
||||
try {
|
||||
await runMatchGrants();
|
||||
} catch (err) {
|
||||
console.error('[match-grants] pass failed:', err);
|
||||
throw err;
|
||||
}
|
||||
};
|
||||
|
||||
// Must be registered as BOTH a workflow and a scheduled function,
|
||||
// referencing the same function object — see ingest-grants.ts.
|
||||
g.__outreachMatchGrantsHandle = DBOS.registerWorkflow(matchGrants, {
|
||||
name: 'matchGrants',
|
||||
});
|
||||
DBOS.registerScheduled(matchGrants, {
|
||||
crontab: '15 5 * * *',
|
||||
name: 'matchGrants',
|
||||
mode: SchedulerMode.ExactlyOncePerInterval,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Starts one durable run of this workflow immediately through DBOS —
|
||||
* the exact production path (workflow + checkpointed steps), used by
|
||||
* `run-once.ts` for supervised/manual passes. Requires deps injected and
|
||||
* `DBOS.launch()` completed.
|
||||
*/
|
||||
export function runMatchGrantsNow(): Promise<void> {
|
||||
const handle = g.__outreachMatchGrantsHandle;
|
||||
if (handle == null) {
|
||||
throw new Error(
|
||||
'matchGrants is not registered; was this module imported before DBOS.launch()?',
|
||||
);
|
||||
}
|
||||
return handle(new Date(), new Date());
|
||||
}
|
||||
Reference in New Issue
Block a user