feat: implement session-based candidate matching architecture

backend/alembic/versions/002_add_sessions.py

@@ -0,0 +1,42 @@
+"""add sessions table and session_id columns
+
+Revision ID: 002
+Revises: 001
+Create Date: 2026-04-12 10:00:00.000000
+"""
+from typing import Sequence, Union
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import UUID
+
+revision: str = "002"
+down_revision: Union[str, None] = "001"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "sessions",
+        sa.Column("id", UUID(as_uuid=True), primary_key=True),
+        sa.Column("name", sa.String(255), nullable=False),
+        sa.Column("description", sa.Text, nullable=True),
+        sa.Column("candidate_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("status", sa.String(32), nullable=False, server_default="active"),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
+    )
+
+    op.add_column("candidates", sa.Column("session_id", UUID(as_uuid=True), sa.ForeignKey("sessions.id", ondelete="SET NULL"), nullable=True))
+    op.create_index("ix_candidates_session_id", "candidates", ["session_id"])
+
+    op.add_column("match_results", sa.Column("session_id", UUID(as_uuid=True), sa.ForeignKey("sessions.id", ondelete="CASCADE"), nullable=True))
+    op.create_index("ix_match_results_session_id", "match_results", ["session_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_match_results_session_id", "match_results")
+    op.drop_column("match_results", "session_id")
+    op.drop_index("ix_candidates_session_id", "candidates")
+    op.drop_column("candidates", "session_id")
+    op.drop_table("sessions")

backend/requirements.txt

@@ -8,7 +8,7 @@ qdrant-client==1.11.0
 celery[redis]==5.4.0
 redis==5.1.0
 sentence-transformers==3.3.0
-FlagEmbedding==1.2.15
+FlagEmbedding==1.3.5
 groq==0.12.0
 python-multipart==0.0.12
 pydantic==2.9.2

backend/src/models/__init__.py

@@ -1,5 +1,6 @@
 from .jd import JobDescription
 from .candidate import Candidate
 from .match_result import MatchResult
+from .session import Session
 
-__all__ = ["JobDescription", "Candidate", "MatchResult"]
+__all__ = ["JobDescription", "Candidate", "MatchResult", "Session"]

backend/src/models/candidate.py

@@ -1,6 +1,6 @@
 import uuid
 from datetime import datetime
-from sqlalchemy import String, Text, DateTime, JSON, Float, Integer, Boolean, func
+from sqlalchemy import String, Text, DateTime, JSON, Float, Integer, Boolean, func, ForeignKey
 from sqlalchemy.orm import Mapped, mapped_column
 from sqlalchemy.dialects.postgresql import UUID
 from ..database import Base
@@ -45,6 +45,7 @@ class Candidate(Base):
     time_in_current_company: Mapped[float | None] = mapped_column(Float, nullable=True)
     is_actively_or_passively_looking: Mapped[str | None] = mapped_column(String(100), nullable=True)
 
+    session_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("sessions.id", ondelete="SET NULL"), nullable=True, index=True)
     growth_velocity: Mapped[float] = mapped_column(Float, default=0.5)
     embedding_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
     qdrant_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)

backend/src/models/match_result.py

@@ -1,6 +1,6 @@
 import uuid
 from datetime import datetime
-from sqlalchemy import String, DateTime, JSON, Float, ForeignKey, func
+from sqlalchemy import String, DateTime, JSON, Float, ForeignKey, func, Integer
 from sqlalchemy.orm import Mapped, mapped_column
 from sqlalchemy.dialects.postgresql import UUID
 from ..database import Base
@@ -12,6 +12,7 @@ class MatchResult(Base):
     id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
     jd_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("job_descriptions.id", ondelete="CASCADE"), index=True)
     candidate_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("candidates.id", ondelete="CASCADE"), index=True)
+    session_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("sessions.id", ondelete="CASCADE"), nullable=True, index=True)
 
     rank: Mapped[int | None] = mapped_column(nullable=True)
     stage1_score: Mapped[float] = mapped_column(Float, default=0.0)

backend/src/models/session.py

@@ -0,0 +1,18 @@
+import uuid
+from datetime import datetime
+from sqlalchemy import String, Text, DateTime, Integer, func
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.dialects.postgresql import UUID
+from ..database import Base
+
+
+class Session(Base):
+    __tablename__ = "sessions"
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    name: Mapped[str] = mapped_column(String(255))
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    candidate_count: Mapped[int] = mapped_column(Integer, default=0)
+    status: Mapped[str] = mapped_column(String(32), default="active")
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

backend/src/routers/__init__.py

@@ -1,5 +1,6 @@
 from .jds import router as jds_router
 from .candidates import router as candidates_router
 from .matching import router as matching_router
+from .sessions import router as sessions_router
 
-__all__ = ["jds_router", "candidates_router", "matching_router"]
+__all__ = ["jds_router", "candidates_router", "matching_router", "sessions_router"]

backend/src/routers/candidates.py

@@ -1,12 +1,14 @@
 import io
 import json
+import uuid
 import pandas as pd
-from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
+from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Query
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select, func
 
 from ..database import get_db
 from ..models.candidate import Candidate
+from ..models.session import Session
 from ..schemas.candidate import CandidateResponse, UploadResponse, TaskStatusResponse
 from ..workers.celery_app import celery_app
 from ..workers.ingest import ingest_candidates_batch
@@ -17,7 +19,17 @@ BATCH_SIZE = 100
 
 
 @router.post("/upload", response_model=UploadResponse)
-async def upload_candidates(file: UploadFile = File(...)):
+async def upload_candidates(
+    file: UploadFile = File(...),
+    session_id: uuid.UUID | None = Query(None, description="Session to attach candidates to"),
+    db: AsyncSession = Depends(get_db),
+):
+    if session_id:
+        result = await db.execute(select(Session).where(Session.id == session_id))
+        sess = result.scalar_one_or_none()
+        if not sess:
+            raise HTTPException(status_code=404, detail="Session not found")
+
     content = await file.read()
     filename = file.filename or ""
 
@@ -48,7 +60,7 @@ async def upload_candidates(file: UploadFile = File(...)):
                 else:
                     clean[k] = v
             serializable.append(clean)
-        task = ingest_candidates_batch.delay(serializable)
+        task = ingest_candidates_batch.delay(serializable, str(session_id) if session_id else None)
         task_ids.append(task.id)
 
     return UploadResponse(
@@ -69,17 +81,21 @@ async def task_status(task_id: str):
 
 
 @router.get("/count")
-async def candidate_count(db: AsyncSession = Depends(get_db)):
-    result = await db.execute(select(func.count()).select_from(Candidate))
-    count = result.scalar()
-    return {"count": count}
+async def candidate_count(
+    session_id: uuid.UUID | None = Query(None),
+    db: AsyncSession = Depends(get_db),
+):
+    q = select(func.count()).select_from(Candidate)
+    if session_id:
+        q = q.where(Candidate.session_id == session_id)
+    result = await db.execute(q)
+    return {"count": result.scalar()}
 
 
 @router.get("/{candidate_id}", response_model=CandidateResponse)
 async def get_candidate(candidate_id: str, db: AsyncSession = Depends(get_db)):
-    import uuid as _uuid
     try:
-        cid = _uuid.UUID(candidate_id)
+        cid = uuid.UUID(candidate_id)
     except ValueError:
         raise HTTPException(status_code=400, detail="Invalid candidate ID")
     result = await db.execute(select(Candidate).where(Candidate.id == cid))

backend/src/routers/sessions.py

@@ -0,0 +1,45 @@
+import uuid
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, func
+
+from ..database import get_db
+from ..models.session import Session
+from ..models.candidate import Candidate
+from ..schemas.session import SessionCreate, SessionResponse
+
+router = APIRouter()
+
+
+@router.post("", response_model=SessionResponse, status_code=201)
+async def create_session(payload: SessionCreate, db: AsyncSession = Depends(get_db)):
+    session = Session(id=uuid.uuid4(), name=payload.name, description=payload.description)
+    db.add(session)
+    await db.commit()
+    await db.refresh(session)
+    return session
+
+
+@router.get("", response_model=list[SessionResponse])
+async def list_sessions(db: AsyncSession = Depends(get_db)):
+    result = await db.execute(select(Session).order_by(Session.created_at.desc()).limit(50))
+    return result.scalars().all()
+
+
+@router.get("/{session_id}", response_model=SessionResponse)
+async def get_session(session_id: uuid.UUID, db: AsyncSession = Depends(get_db)):
+    result = await db.execute(select(Session).where(Session.id == session_id))
+    sess = result.scalar_one_or_none()
+    if not sess:
+        raise HTTPException(status_code=404, detail="Session not found")
+    return sess
+
+
+@router.delete("/{session_id}", status_code=204)
+async def delete_session(session_id: uuid.UUID, db: AsyncSession = Depends(get_db)):
+    result = await db.execute(select(Session).where(Session.id == session_id))
+    sess = result.scalar_one_or_none()
+    if not sess:
+        raise HTTPException(status_code=404, detail="Session not found")
+    await db.delete(sess)
+    await db.commit()

backend/src/schemas/match.py

@@ -44,6 +44,7 @@ class MatchResponse(BaseModel):
     total_matched: int
     results: list[MatchedCandidate]
     weights_used: dict[str, float] = {}
+    session_id: UUID | None = None
 
 
 class CandidateDetailResponse(BaseModel):

backend/src/schemas/session.py

@@ -0,0 +1,19 @@
+import uuid
+from datetime import datetime
+from pydantic import BaseModel
+
+
+class SessionCreate(BaseModel):
+    name: str
+    description: str | None = None
+
+
+class SessionResponse(BaseModel):
+    id: uuid.UUID
+    name: str
+    description: str | None = None
+    candidate_count: int
+    status: str
+    created_at: datetime
+
+    model_config = {"from_attributes": True}

backend/src/workers/ingest.py

@@ -1,9 +1,9 @@
 import uuid
 import asyncio
 import re
+import json
 from typing import Any
-from sqlalchemy import text, select
-from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy import select, func
 from qdrant_client import QdrantClient
 from qdrant_client.models import PointStruct
 
@@ -11,8 +11,9 @@ from .celery_app import celery_app
 from ..config import get_settings
 from ..database import AsyncSessionLocal
 from ..models.candidate import Candidate
+from ..models.session import Session
 from ..models.jd import JobDescription
-from ..ml.embedder import embed_texts, embed_query, compute_text_hash
+from ..ml.embedder import embed_texts, compute_text_hash
 from ..ml.feature_builder import (
     build_candidate_text,
     compute_growth_velocity,
@@ -52,40 +53,36 @@ def _parse_list(val: Any) -> list:
 
 
 @celery_app.task(bind=True, name="ingest_candidates_batch", max_retries=3)
-def ingest_candidates_batch(self, rows: list[dict]) -> dict:
+def ingest_candidates_batch(self, rows: list[dict], session_id: str | None = None) -> dict:
     try:
-        return asyncio.run(_ingest_candidates_async(rows))
+        return asyncio.run(_ingest_candidates_async(rows, session_id))
     except Exception as exc:
         raise self.retry(exc=exc, countdown=30)
 
 
-async def _ingest_candidates_async(rows: list[dict]) -> dict:
+async def _ingest_candidates_async(rows: list[dict], session_id: str | None) -> dict:
     settings = get_settings()
     qdrant = _get_qdrant()
     texts = []
     processed = []
+    sess_uuid = uuid.UUID(session_id) if session_id else None
 
     for row in rows:
         work_exp = row.get("parsed_work_experience") or []
         if isinstance(work_exp, str):
             try:
-                import json
                 work_exp = json.loads(work_exp)
             except Exception:
                 work_exp = []
 
         is_funded = _normalize_bool(row.get("most_recent_company_is_funded"))
         velocity = compute_growth_velocity(work_exp, is_funded=bool(is_funded))
-
         candidate_text = build_candidate_text({**row, "parsed_work_experience": work_exp})
         text_hash = compute_text_hash(candidate_text)
 
         processed.append({
-            "row": row,
-            "work_exp": work_exp,
-            "velocity": velocity,
-            "text": candidate_text,
-            "hash": text_hash,
+            "row": row, "work_exp": work_exp, "velocity": velocity,
+            "text": candidate_text, "hash": text_hash,
         })
         texts.append(candidate_text)
 
@@ -100,8 +97,17 @@ async def _ingest_candidates_async(rows: list[dict]) -> dict:
             candidate_id = uuid.uuid4()
             qdrant_id = str(candidate_id)
 
+            all_skills = (
+                _parse_list(row.get("programming_languages"))
+                + _parse_list(row.get("backend_frameworks"))
+                + _parse_list(row.get("frontend_technologies"))
+            )
+            if row.get("parsed_skills"):
+                all_skills.extend(_parse_list(row["parsed_skills"]))
+
             cand = Candidate(
                 id=candidate_id,
+                session_id=sess_uuid,
                 external_id=str(row.get("id") or row.get("candidate_id") or ""),
                 name=row.get("name"),
                 email=row.get("email"),
@@ -136,20 +142,13 @@ async def _ingest_candidates_async(rows: list[dict]) -> dict:
             )
             session.add(cand)
 
-            all_skills = (
-                _parse_list(row.get("programming_languages"))
-                + _parse_list(row.get("backend_frameworks"))
-                + _parse_list(row.get("frontend_technologies"))
-            )
-            if row.get("parsed_skills"):
-                all_skills.extend(_parse_list(row["parsed_skills"]))
-
             qdrant_points.append(
                 PointStruct(
                     id=qdrant_id,
                     vector=embeddings[i].tolist(),
                     payload={
                         "candidate_id": str(candidate_id),
+                        "session_id": session_id or "",
                         "role_type": row.get("role_type"),
                         "engineer_type": row.get("engineer_type"),
                         "years_of_experience": _normalize_float(row.get("years_of_experience")),
@@ -170,7 +169,18 @@ async def _ingest_candidates_async(rows: list[dict]) -> dict:
         if qdrant_points:
             qdrant.upsert(collection_name=settings.collection_name, points=qdrant_points)
 
-    return {"inserted": inserted, "total": len(rows)}
+        if sess_uuid:
+            result = await session.execute(
+                select(func.count()).select_from(Candidate).where(Candidate.session_id == sess_uuid)
+            )
+            count = result.scalar() or 0
+            sess_result = await session.execute(select(Session).where(Session.id == sess_uuid))
+            sess_obj = sess_result.scalar_one_or_none()
+            if sess_obj:
+                sess_obj.candidate_count = count
+                await session.commit()
+
+    return {"inserted": inserted, "total": len(rows), "session_id": session_id}
 
 
 @celery_app.task(bind=True, name="ingest_jd", max_retries=3)
@@ -186,9 +196,7 @@ async def _ingest_jd_async(jd_id: str, raw_text: str, title: str) -> dict:
     jd_quality = compute_jd_quality(raw_text, parsed)
 
     async with AsyncSessionLocal() as session:
-        result = await session.execute(
-            select(JobDescription).where(JobDescription.id == uuid.UUID(jd_id))
-        )
+        result = await session.execute(select(JobDescription).where(JobDescription.id == uuid.UUID(jd_id)))
         jd = result.scalar_one_or_none()
         if jd:
             jd.parsed_requirements = parsed