feat: prototype lead-scoring module (ICP gate + intent + routing)

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).

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()],

backend/app/routers/__init__.py

@@ -0,0 +1 @@
+# routers package

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,
+    )

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,
+    )

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)