diff --git a/backend/app/config.py b/backend/app/config.py index 4516583..806541b 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -32,6 +32,10 @@ class Settings(BaseSettings): # When True, hide dollar amounts in the UI (cost visibility toggle). vecna_hide_costs: bool = Field(default=False, validation_alias="VECNA_HIDE_COSTS") + # Lead-scoring flag (default off); LEAD_SCORING_KILL_SWITCH=1 overrides. + lead_scoring_enabled: bool = Field(default=False, validation_alias="LEAD_SCORING_ENABLED") + lead_scoring_kill_switch: bool = Field(default=False, validation_alias="LEAD_SCORING_KILL_SWITCH") + # Platform pricing on top of LLM token cost (USD). Zero by default. vecna_scan_fee_usd: float = Field(default=0.0, validation_alias="VECNA_SCAN_FEE_USD") # Legacy: single $/call when no tiered JSON is set (VECNA_TOOL_FAMILY_FEES_JSON / OVERRIDES / BY_TOOL). diff --git a/backend/app/main.py b/backend/app/main.py index 23f7cfe..a4b53c9 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -26,6 +26,7 @@ from app.services.runner import cancel_task, execute_operation, register_task from app.schema_migrate import _sqlite_add_missing_columns from app.url_guard import is_safe_public_target +from app.routers import lead_scoring as lead_scoring_router logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @@ -67,6 +68,7 @@ async def lifespan(app: FastAPI): app = FastAPI(title="Vecna Operations Console", lifespan=lifespan) +app.include_router(lead_scoring_router.router) app.add_middleware( CORSMiddleware, allow_origins=[o.strip() for o in settings.cors_origins.split(",") if o.strip()], diff --git a/backend/app/routers/__init__.py b/backend/app/routers/__init__.py new file mode 100644 index 0000000..9c8ddfa --- /dev/null +++ b/backend/app/routers/__init__.py @@ -0,0 +1 @@ +# routers package diff --git a/backend/app/routers/lead_scoring.py b/backend/app/routers/lead_scoring.py new file mode 100644 index 0000000..83fb38e --- /dev/null +++ b/backend/app/routers/lead_scoring.py @@ -0,0 +1,95 @@ +"""Lead-scoring API router. + +GET /api/lead-scoring/status -- feature-flag probe (always responds) +GET /api/lead-scoring/rubric -- encoded ICP rubric + intent table +POST /api/lead-scoring/score -- score a single lead + +Returns 503 when flag is off or kill-switch is set. +""" +from __future__ import annotations + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, Field + +from app.services.lead_scoring import ( + ATTRIBUTE_POINTS, + INTENT_HIGH_THRESHOLD, + INTENT_POINTS, + NEGATIVE_POINTS, + ROUTING_MATRIX, + FirmographicAttribute, + IntentSignal, + NegativeSignal, + lead_scoring_enabled, + score_lead, +) + +router = APIRouter(prefix="/api/lead-scoring", tags=["lead-scoring"]) + + +def _require_enabled() -> None: + if not lead_scoring_enabled(): + raise HTTPException( + status_code=503, + detail="Lead-scoring is disabled. Set LEAD_SCORING_ENABLED=1 to enable.", + ) + + +class ScoreRequest(BaseModel): + attributes: list[FirmographicAttribute] = Field(default_factory=list) + negatives: list[NegativeSignal] = Field(default_factory=list) + intent_signals: list[IntentSignal] = Field(default_factory=list) + + +class RoutingOut(BaseModel): + action: str + label: str + owner: str + sla_hours: int | None + + +class ScoreResponse(BaseModel): + fit_score: int + fit_band: str + fit_attributes: list[str] + negative_signals: list[str] + intent_score: int + intent_level: str + intent_signals: list[str] + routing: RoutingOut + meddpicc_note: str + + +@router.get("/status", summary="Feature-flag probe") +def lead_scoring_status() -> dict: + return {"enabled": lead_scoring_enabled()} + + +@router.get("/rubric", summary="Encoded ICP rubric and intent table") +def get_rubric() -> dict: + _require_enabled() + return { + "firmographic_attributes": {k.value: v for k, v in ATTRIBUTE_POINTS.items()}, + "negative_signals": {k.value: v for k, v in NEGATIVE_POINTS.items()}, + "intent_signals": {k.value: v for k, v in INTENT_POINTS.items()}, + "intent_high_threshold": INTENT_HIGH_THRESHOLD, + "fit_bands": {"A": ">=70", "B": "40-69", "C": "<40"}, + "routing_matrix": { + f"{band}_{level}": route + for (band, level), route in ROUTING_MATRIX.items() + }, + } + + +@router.post("/score", response_model=ScoreResponse, summary="Score a lead") +def score_lead_endpoint(body: ScoreRequest) -> ScoreResponse: + _require_enabled() + r = score_lead(body.attributes, body.negatives, body.intent_signals) + return ScoreResponse( + fit_score=r.fit_score, fit_band=r.fit_band, + fit_attributes=r.fit_attributes, negative_signals=r.negative_signals, + intent_score=r.intent_score, intent_level=r.intent_level, + intent_signals=r.intent_signals, + routing=RoutingOut(**r.routing), + meddpicc_note=r.meddpicc_note, + ) diff --git a/backend/app/services/lead_scoring.py b/backend/app/services/lead_scoring.py new file mode 100644 index 0000000..9fd5c80 --- /dev/null +++ b/backend/app/services/lead_scoring.py @@ -0,0 +1,145 @@ +"""Lead-scoring: ICP gate + intent signals + routing matrix. + +Source rubrics: + icp-scoring-rubric.md — firmographic/technographic attrs + negative signals + lead-scoring-model.md — intent behaviors (caller handles time-decay) + routing + meddpicc-qualification.md — deal-health context surfaced in output + +Feature flag: LEAD_SCORING_ENABLED=1; kill-switch: LEAD_SCORING_KILL_SWITCH=1. +""" +from __future__ import annotations + +import os +from dataclasses import dataclass +from enum import Enum +from typing import Literal + + +def lead_scoring_enabled() -> bool: + if os.environ.get("LEAD_SCORING_KILL_SWITCH", "").strip() in ("1", "true", "yes"): + return False + return os.environ.get("LEAD_SCORING_ENABLED", "").strip() in ("1", "true", "yes") + + +class FirmographicAttribute(str, Enum): + COMPANY_SIZE_OK = "company_size_200_5000" + INDUSTRY_IDEAL = "industry_ideal" + REVENUE_OK = "annual_revenue_20m_1b" + TECH_STACK_OK = "tech_stack_cloud_native" + REGION_OK = "region_us_eu" + TOOLING_GAP = "no_existing_agent_platform" + +ATTRIBUTE_POINTS: dict[FirmographicAttribute, int] = { + FirmographicAttribute.COMPANY_SIZE_OK: 25, # 200–5 000 employees + FirmographicAttribute.INDUSTRY_IDEAL: 20, # FinTech/SaaS/e-com/regulated + FirmographicAttribute.REVENUE_OK: 15, # $20M–$1B ARR + FirmographicAttribute.TECH_STACK_OK: 15, # cloud-native + SRE team + FirmographicAttribute.REGION_OK: 10, # US/EU supported data residency + FirmographicAttribute.TOOLING_GAP: 15, # no current agent platform +} + + +class NegativeSignal(str, Enum): + TINY_COMPANY = "lt_50_employees_preseed" + PERSONAL_DOMAIN = "student_personal_free_email" + BAD_INDUSTRY = "industry_we_cant_serve" + COMPETITOR_PATTERN = "competitor_tirekicker" + NO_BUDGET_AUTHORITY = "no_budget_authority" + +NEGATIVE_POINTS: dict[NegativeSignal, int] = { + NegativeSignal.TINY_COMPANY: -30, # <50 emp / pre-seed + NegativeSignal.PERSONAL_DOMAIN: -40, # student/personal/free-email + NegativeSignal.BAD_INDUSTRY: -25, # regulated we can't support + NegativeSignal.COMPETITOR_PATTERN: -20, # competitor employee / tire-kicker + NegativeSignal.NO_BUDGET_AUTHORITY: -20, # no budget authority anywhere +} + + +class IntentSignal(str, Enum): + PRICING_PAGE_VISIT = "pricing_page_visit" # +20 / 14-day decay + DEMO_BOOKED_ATTENDED = "demo_booked_attended" # +30 / 30-day decay + DOCS_API_DEEP_READ = "docs_api_deep_read_3plus" # +15 / 14-day decay + COMPETITOR_COMPARE = "competitor_compare_page" # +15 / 14-day decay + FREE_TRIAL_SIGNUP = "free_trial_signup" # +25 / 30-day decay + MULTI_PERSON_VISIT = "repeat_visit_2plus_people" # +20 / 14-day decay + EMAIL_OPEN_COLD = "email_open_cold_list" # +2 / 7-day decay + UNSUBSCRIBE = "unsubscribe_not_now" # -15 / no decay + +INTENT_POINTS: dict[IntentSignal, int] = { + IntentSignal.PRICING_PAGE_VISIT: 20, + IntentSignal.DEMO_BOOKED_ATTENDED: 30, + IntentSignal.DOCS_API_DEEP_READ: 15, + IntentSignal.COMPETITOR_COMPARE: 15, + IntentSignal.FREE_TRIAL_SIGNUP: 25, + IntentSignal.MULTI_PERSON_VISIT: 20, + IntentSignal.EMAIL_OPEN_COLD: 2, + IntentSignal.UNSUBSCRIBE: -15, +} + +INTENT_HIGH_THRESHOLD = 30 + +FitBand = Literal["A", "B", "C"] +IntentLevel = Literal["high", "low"] + +ROUTING_MATRIX: dict[tuple[FitBand, IntentLevel], dict] = { + ("A", "high"): {"action": "mql_to_sql", "owner": "ae", "sla_hours": 8, + "label": "MQL -> SQL now. Route to AE same day."}, + ("A", "low"): {"action": "nurture_targeted", "owner": "marketing", "sla_hours": None, + "label": "Nurture with targeted content; sales-assist on next signal."}, + ("B", "high"): {"action": "ae_qualify_hard", "owner": "ae", "sla_hours": 48, + "label": "AE qualifies hard (champion + pain required)."}, + ("B", "low"): {"action": "marketing_nurture", "owner": "marketing", "sla_hours": None, + "label": "Marketing nurture; re-score monthly."}, + ("C", "high"): {"action": "self_serve_plg", "owner": "plg", "sla_hours": None, + "label": "Self-serve / PLG; do not staff an AE."}, + ("C", "low"): {"action": "suppress", "owner": "none", "sla_hours": None, + "label": "Suppress."}, +} + + +@dataclass +class LeadScoreResult: + fit_score: int + fit_band: FitBand + fit_attributes: list[str] + negative_signals: list[str] + intent_score: int + intent_level: IntentLevel + intent_signals: list[str] + routing: dict + meddpicc_note: str + + +def score_lead( + attributes: list[FirmographicAttribute], + negatives: list[NegativeSignal], + intent: list[IntentSignal], +) -> LeadScoreResult: + fit_score = ( + sum(ATTRIBUTE_POINTS.get(a, 0) for a in attributes) + + sum(NEGATIVE_POINTS.get(n, 0) for n in negatives) + ) + fit_band: FitBand = "A" if fit_score >= 70 else ("B" if fit_score >= 40 else "C") + + intent_score = sum(INTENT_POINTS.get(s, 0) for s in intent) + intent_level: IntentLevel = "high" if intent_score >= INTENT_HIGH_THRESHOLD else "low" + + routing = ROUTING_MATRIX[(fit_band, intent_level)] + + if fit_band == "A" and intent_level == "high": + note = "Begin MEDDPICC discovery immediately. Confirm EB and champion before first forecast commit." + elif fit_band == "B" and intent_level == "high": + note = "Gate on champion test and pain tied to a metric before advancing." + elif fit_band in ("A", "B"): + note = "Do not open MEDDPICC until intent score crosses threshold." + else: + note = "ICP gate failed -- do not enter MEDDPICC discovery." + + return LeadScoreResult( + fit_score=fit_score, fit_band=fit_band, + fit_attributes=[a.value for a in attributes], + negative_signals=[n.value for n in negatives], + intent_score=intent_score, intent_level=intent_level, + intent_signals=[s.value for s in intent], + routing=routing, meddpicc_note=note, + ) diff --git a/backend/tests/test_lead_scoring.py b/backend/tests/test_lead_scoring.py new file mode 100644 index 0000000..c8cd9aa --- /dev/null +++ b/backend/tests/test_lead_scoring.py @@ -0,0 +1,124 @@ +"""Tests: lead-scoring service (fit, intent, routing) + API feature-flag behavior.""" +from __future__ import annotations + +import os + +from fastapi.testclient import TestClient + +from app.services.lead_scoring import ( + FirmographicAttribute as FA, + IntentSignal as IS, + NegativeSignal as NS, + score_lead, +) + + +class TestFitScoring: + def test_all_positives_band_a(self): + r = score_lead(list(FA), [], []) + assert r.fit_score == 100 and r.fit_band == "A" + + def test_band_b_boundary(self): + r = score_lead([FA.COMPANY_SIZE_OK, FA.REVENUE_OK], [], []) # 25+15=40 + assert r.fit_score == 40 and r.fit_band == "B" + + def test_negative_signals_subtract(self): + r = score_lead([FA.REGION_OK], [NS.PERSONAL_DOMAIN], []) # 10-40=-30 + assert r.fit_score == -30 and r.fit_band == "C" + + def test_combined_stays_a(self): + r = score_lead(list(FA), [NS.TINY_COMPANY], []) # 100-30=70 + assert r.fit_score == 70 and r.fit_band == "A" + + +class TestIntentClassification: + def test_demo_alone_is_high(self): + r = score_lead([], [], [IS.DEMO_BOOKED_ATTENDED]) + assert r.intent_score == 30 and r.intent_level == "high" + + def test_cold_email_is_low(self): + r = score_lead([], [], [IS.EMAIL_OPEN_COLD]) + assert r.intent_score == 2 and r.intent_level == "low" + + def test_unsubscribe_drags_down(self): + r = score_lead([], [], [IS.PRICING_PAGE_VISIT, IS.UNSUBSCRIBE]) # 20-15=5 + assert r.intent_score == 5 and r.intent_level == "low" + + def test_two_signals_cross_threshold(self): + r = score_lead([], [], [IS.PRICING_PAGE_VISIT, IS.DOCS_API_DEEP_READ]) # 35 + assert r.intent_level == "high" + + +class TestRoutingMatrix: + def test_a_high_mql_to_sql(self): + r = score_lead(list(FA), [], [IS.DEMO_BOOKED_ATTENDED]) + assert r.routing["action"] == "mql_to_sql" and r.routing["sla_hours"] == 8 + + def test_a_low_nurture(self): + r = score_lead(list(FA), [], []) + assert r.routing["action"] == "nurture_targeted" + + def test_b_high_qualify_hard(self): + r = score_lead([FA.COMPANY_SIZE_OK, FA.REVENUE_OK], [], [IS.DEMO_BOOKED_ATTENDED]) + assert r.fit_band == "B" and r.routing["action"] == "ae_qualify_hard" + + def test_c_high_self_serve(self): + r = score_lead([], [], [IS.DEMO_BOOKED_ATTENDED]) + assert r.fit_band == "C" and r.routing["action"] == "self_serve_plg" + + def test_c_low_suppress_and_no_meddpicc(self): + r = score_lead([], [], []) + assert r.routing["action"] == "suppress" + assert "ICP gate failed" in r.meddpicc_note + + def test_meddpicc_note_a_high_start_immediately(self): + r = score_lead(list(FA), [], [IS.DEMO_BOOKED_ATTENDED]) + assert "immediately" in r.meddpicc_note + + +from app.main import app # noqa: E402 +client = TestClient(app) + + +class TestFeatureFlag: + def test_status_always_ok(self): + r = client.get("/api/lead-scoring/status") + assert r.status_code == 200 and "enabled" in r.json() + + def test_disabled_by_default(self): + os.environ.pop("LEAD_SCORING_ENABLED", None) + os.environ.pop("LEAD_SCORING_KILL_SWITCH", None) + assert client.post("/api/lead-scoring/score", json={}).status_code == 503 + + def test_enabled_scores_a_high(self): + os.environ["LEAD_SCORING_ENABLED"] = "1" + os.environ.pop("LEAD_SCORING_KILL_SWITCH", None) + try: + r = client.post("/api/lead-scoring/score", json={ + "attributes": [a.value for a in FA], + "negatives": [], + "intent_signals": [IS.DEMO_BOOKED_ATTENDED.value], + }) + assert r.status_code == 200 + assert r.json()["routing"]["action"] == "mql_to_sql" + finally: + os.environ.pop("LEAD_SCORING_ENABLED", None) + + def test_kill_switch_overrides_enabled(self): + os.environ["LEAD_SCORING_ENABLED"] = "1" + os.environ["LEAD_SCORING_KILL_SWITCH"] = "1" + try: + assert client.post("/api/lead-scoring/score", json={}).status_code == 503 + finally: + os.environ.pop("LEAD_SCORING_ENABLED", None) + os.environ.pop("LEAD_SCORING_KILL_SWITCH", None) + + def test_rubric_returns_matrix(self): + os.environ["LEAD_SCORING_ENABLED"] = "1" + try: + r = client.get("/api/lead-scoring/rubric") + assert r.status_code == 200 + assert r.json()["fit_bands"]["A"] == ">=70" + assert "A_high" in r.json()["routing_matrix"] + finally: + os.environ.pop("LEAD_SCORING_ENABLED", None)