perf: persist seniority_level in DB to avoid recomputation on every GET /leads

backend/agents/free_scout.py

@@ -506,6 +506,7 @@ def run(
                 learning_delta=item.get("learning_delta"),
                 learning_reason=item.get("learning_reason", ""),
                 source_meta=item.get("source_meta", {}),
+                seniority_level=(item.get("source_meta") or {}).get("seniority_level", ""),
             )
             LAST_USAGE["saved"] += 1
             leads.append(item)

backend/agents/scout.py

@@ -968,7 +968,7 @@ def run(
                 LAST_ERRORS.append(f"filtered {plat}:{u} - {quality.get('reason', 'quality gate')}")
                 continue
             source_meta = item["source_meta"]
-            save_lead(jid, t, co, u, plat, desc, source_meta=source_meta)
+            save_lead(jid, t, co, u, plat, desc, source_meta=source_meta, seniority_level=source_meta.get("seniority_level", ""))
             leads.append({
                 "job_id": jid, "title": t, "company": co, "url": u,
                 "platform": plat, "description": desc, "source_meta": source_meta,

backend/agents/x_scout.py

@@ -453,6 +453,7 @@ def run(
                 learning_delta=lead.get("learning_delta"),
                 learning_reason=lead.get("learning_reason", ""),
                 source_meta=lead["source_meta"],
+                seniority_level=lead["source_meta"].get("seniority_level", ""),
             )
             LAST_USAGE["saved"] += 1
             leads.append(lead)

backend/db/client.py

@@ -146,6 +146,7 @@ def _init_sql():
         ("learning_delta", "INTEGER DEFAULT 0"),
         ("learning_reason", "TEXT DEFAULT ''"),
         ("resume_version", "INTEGER DEFAULT 0"),
+        ("seniority_level", "TEXT DEFAULT ''"),
     ]:
         try:
             c.execute(f"ALTER TABLE leads ADD COLUMN {col} {definition}")
@@ -168,7 +169,8 @@ def _init_sql():
     "signal_reason,signal_tags,outreach_reply,outreach_dm,source_meta,feedback,"
     "feedback_note,followup_due_at,last_contacted_at,outreach_email,proposal_draft,"
     "fit_bullets,followup_sequence,proof_snippet,tech_stack,location,urgency,"
-    "base_signal_score,learning_delta,learning_reason,created_at,resume_version"
+    "base_signal_score,learning_delta,learning_reason,created_at,resume_version,"
+    "seniority_level"
 )
 
 
@@ -215,6 +217,7 @@ def save_lead(
     learning_delta: int | None = None,
     learning_reason: str = "",
     source_meta: dict | None = None,
+    seniority_level: str = "",
 ):
     lead = {
         "job_id": jid,
@@ -255,8 +258,8 @@ def save_lead(
             signal_score,signal_reason,signal_tags,outreach_reply,outreach_dm,
             outreach_email,proposal_draft,fit_bullets,followup_sequence,
             proof_snippet,tech_stack,location,urgency,base_signal_score,
-            learning_delta,learning_reason,source_meta
-        ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+            learning_delta,learning_reason,source_meta,seniority_level
+        ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
         """,
         (
             jid, t, co, u, plat, desc, lead.get("kind") or "job", lead.get("budget") or "",
@@ -273,6 +276,7 @@ def save_lead(
             int(lead.get("learning_delta") or 0),
             str(lead.get("learning_reason") or "")[:700],
             json.dumps(lead.get("source_meta") or {}, ensure_ascii=False),
+            lead.get("seniority_level") or seniority_level or "",
         ),
     )
     c.commit()
@@ -437,6 +441,7 @@ def _lead_row_dict(r) -> dict:
         "learning_reason": r[36] or "",
         "created_at": r[37] or "",
         "resume_version": r[38] or 0,
+        "seniority_level": r[39] or "",
     }
 
 
@@ -1559,3 +1564,33 @@ def _add_project_vec(pid: str, title: str, stack: str, impact: str):
                 vec.create_table("projects", data=rows)
     except Exception:
         pass
+
+
+def backfill_seniority_levels(limit: int = 5000) -> int:
+    """Compute and store seniority_level for leads that don't have one yet.
+
+    Returns the number of leads updated.
+    """
+    from agents.scout import classify_job_seniority
+
+    c = _sq.connect(sql)
+    rows = c.execute(
+        f"SELECT {_LEAD_SELECT_COLUMNS} FROM leads "
+        "WHERE COALESCE(seniority_level, '') = '' "
+        "ORDER BY created_at DESC LIMIT ?",
+        (max(1, min(int(limit or 5000), 10000)),),
+    ).fetchall()
+
+    updated = 0
+    for row in rows:
+        lead = _lead_row_dict(row)
+        level = classify_job_seniority(lead)
+        c.execute(
+            "UPDATE leads SET seniority_level=? WHERE job_id=?",
+            (level, lead["job_id"]),
+        )
+        updated += 1
+
+    c.commit()
+    c.close()
+    return updated

backend/main.py

@@ -575,7 +575,7 @@ def _annotate_job_lead(lead: dict) -> dict:
     from agents.scout import classify_job_seniority
 
     meta = dict(lead.get("source_meta") or {})
-    level = str(meta.get("seniority_level") or lead.get("seniority_level") or "").strip().lower()
+    level = str(lead.get("seniority_level") or meta.get("seniority_level") or "").strip().lower()
     if level not in {"fresher", "junior", "mid", "senior", "unknown"}:
         level = classify_job_seniority(lead)
     meta["seniority_level"] = level
@@ -750,6 +750,7 @@ async def create_manual_lead(body: ManualLeadBody):
         learning_delta=lead.get("learning_delta"),
         learning_reason=lead.get("learning_reason", ""),
         source_meta=lead["source_meta"],
+        seniority_level=lead.get("seniority_level", ""),
     )
     saved = get_lead_by_id(lead["job_id"]) or lead
     await cm.broadcast({"type": "LEAD_UPDATED", "data": saved})

backend/tests/test_regressions.py

@@ -1000,5 +1000,79 @@ def test_red_flag_lead_is_rejected(self):
         self.assertIn("red flags", quality["reason"])
 
 
+class TestSeniorityPersistence(unittest.TestCase):
+    """Verify that seniority_level is read from the stored DB column first
+    and only recomputed via classify_job_seniority when missing.
+
+    These tests replicate the _annotate_job_lead priority chain without
+    importing main.py or agents.scout (which need httpx/fastapi).
+    """
+
+    def _annotate(self, lead: dict, fallback_level: str = "unknown") -> dict:
+        """Mirror of main._annotate_job_lead for unit-testing the priority chain.
+        Uses a mock classify_job_seniority that returns fallback_level.
+        """
+        meta = dict(lead.get("source_meta") or {})
+        level = str(lead.get("seniority_level") or meta.get("seniority_level") or "").strip().lower()
+        if level not in {"fresher", "junior", "mid", "senior", "unknown"}:
+            level = fallback_level
+        meta["seniority_level"] = level
+        meta["is_beginner"] = level in {"fresher", "junior"}
+        return {**lead, "source_meta": meta, "seniority_level": level}
+
+    def test_annotate_uses_stored_seniority(self):
+        lead = {
+            "job_id": "test-seniority-001",
+            "title": "Software Engineer",
+            "company": "Acme",
+            "seniority_level": "senior",
+            "source_meta": {},
+        }
+        result = self._annotate(lead, fallback_level="junior")
+        self.assertEqual(result["seniority_level"], "senior")
+
+    def test_annotate_falls_back_to_source_meta(self):
+        lead = {
+            "job_id": "test-seniority-002",
+            "title": "Software Engineer",
+            "company": "Acme",
+            "source_meta": {"seniority_level": "junior"},
+        }
+        result = self._annotate(lead, fallback_level="senior")
+        self.assertEqual(result["seniority_level"], "junior")
+
+    def test_annotate_recomputes_when_missing(self):
+        lead = {
+            "job_id": "test-seniority-003",
+            "title": "Senior Platform Engineer",
+            "company": "Acme",
+            "source_meta": {},
+        }
+        result = self._annotate(lead, fallback_level="senior")
+        self.assertEqual(result["seniority_level"], "senior")
+
+    def test_annotate_prefers_lead_column_over_source_meta(self):
+        lead = {
+            "job_id": "test-seniority-004",
+            "title": "Software Engineer",
+            "company": "Acme",
+            "seniority_level": "mid",
+            "source_meta": {"seniority_level": "junior"},
+        }
+        result = self._annotate(lead, fallback_level="senior")
+        self.assertEqual(result["seniority_level"], "mid")
+
+    def test_empty_seniority_triggers_fallback(self):
+        lead = {
+            "job_id": "test-seniority-005",
+            "title": "Software Engineer",
+            "company": "Acme",
+            "seniority_level": "",
+            "source_meta": {},
+        }
+        result = self._annotate(lead, fallback_level="mid")
+        self.assertEqual(result["seniority_level"], "mid")
+
+
 if __name__ == "__main__":
     unittest.main()