feat(extraction): add FieldExtractor pipeline, ProfileStore with lazy loading, and skill matching

- New multi-signal field extraction pipeline (seniority, title, company, experience_years, domain, career_history)
- FieldExtractorPipeline orchestrates cross-field extraction with cascading fallbacks
- ProfileStore with lazy loading via offset index for 100K profiles
- Smart normalizer integrates extraction pipeline; extracts skills from raw text
- Skill aliases (32 entries, 80+ aliases) with fuzzy matching via SequenceMatcher
- Cross-encoder sigmoid normalization, hybrid search cache, exact location filters
- ProfileStore wiring through main, API routes, executor, and Gradio UI

pyproject.toml

@@ -70,6 +70,9 @@ markers = [
     "integration: marks integration tests",
 ]
 
+[tool.setuptools.packages.find]
+where = ["src"]
+
 [build-system]
 requires = ["setuptools>=68.0"]
 build-backend = "setuptools.build_meta"

scripts/build_indexes.py

@@ -116,6 +116,9 @@ def build_indexes(
     bm25_search.save()
     logger.info(f"BM25 index saved: {bm25_search.size} documents")
 
+    offset_path = index_dir / "offset_index.json"
+    _save_offset_index(profiles_path, profile_ids, offset_path)
+
     elapsed = time.perf_counter() - start
     logger.info(f"All indexes built successfully in {elapsed:.1f}s")
 
@@ -140,6 +143,48 @@ def _build_document_text(profile: Profile) -> str:
     return " ".join(parts)
 
 
+def _save_offset_index(
+    profiles_path: Path, profile_ids: list[str], output_path: Path,
+) -> None:
+    import json
+    if profiles_path.suffix != ".jsonl":
+        logger.info("Skipping offset index (not a JSONL file)")
+        return
+    pid_set = set(profile_ids)
+    offsets: dict[str, int] = {}
+    with open(profiles_path, encoding="utf-8") as f:
+        while True:
+            offset = f.tell()
+            line = f.readline()
+            if not line:
+                break
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                raw = json.loads(line)
+                cand_id = (
+                    raw.get("profile_id")
+                    or raw.get("candidate_id")
+                    or raw.get("id")
+                )
+                profile_nested = raw.get("profile", {})
+                if isinstance(profile_nested, dict) and not cand_id:
+                    cand_id = (
+                        profile_nested.get("profile_id")
+                        or profile_nested.get("candidate_id")
+                        or profile_nested.get("id")
+                    )
+                if cand_id and str(cand_id) in pid_set:
+                    offsets[str(cand_id)] = offset
+            except json.JSONDecodeError:
+                continue
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    with open(output_path, "w") as f:
+        json.dump(offsets, f)
+    logger.info(f"Offset index saved: {len(offsets)} entries")
+
+
 def main():
     parser = argparse.ArgumentParser(description="Build FAISS + BM25 indexes from profiles")
     parser.add_argument("--sample", type=int, default=0,

scripts/build_offset_index.py

@@ -0,0 +1,64 @@
+"""Build offset index only — does not rebuild FAISS/BM25."""
+import json
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.core.config import DATA_DIR
+
+profiles_path = DATA_DIR / "profiles" / "candidates.jsonl"
+output_path = DATA_DIR / "indexes" / "offset_index.json"
+
+id_map_path = DATA_DIR / "indexes" / "faiss_id_map.json"
+if not id_map_path.exists():
+    print("FAISS id map not found. Run build_indexes.py first.")
+    sys.exit(1)
+
+with open(id_map_path) as f:
+    profile_ids = json.load(f)
+
+pid_set = set(profile_ids)
+print(f"Looking up offsets for {len(pid_set)} profile IDs...")
+
+start = time.perf_counter()
+offsets: dict[str, int] = {}
+with open(profiles_path, "r", encoding="utf-8") as f:
+    while True:
+        offset = f.tell()
+        line = f.readline()
+        if not line:
+            break
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            raw = json.loads(line)
+            cand_id = (
+                raw.get("profile_id")
+                or raw.get("candidate_id")
+                or raw.get("id")
+            )
+            profile_nested = raw.get("profile", {})
+            if isinstance(profile_nested, dict) and not cand_id:
+                cand_id = (
+                    profile_nested.get("profile_id")
+                    or profile_nested.get("candidate_id")
+                    or profile_nested.get("id")
+                )
+            if cand_id and str(cand_id) in pid_set:
+                offsets[str(cand_id)] = offset
+        except json.JSONDecodeError:
+            continue
+
+elapsed = time.perf_counter() - start
+print(f"Found {len(offsets)}/{len(pid_set)} profile offsets in {elapsed:.1f}s")
+missing = pid_set - set(offsets.keys())
+if missing:
+    print(f"Missing {len(missing)} profiles")
+
+output_path.parent.mkdir(parents=True, exist_ok=True)
+with open(output_path, "w") as f:
+    json.dump(offsets, f)
+print(f"Offset index saved to {output_path}")

src/agents/executor.py

@@ -1,20 +1,122 @@
 from __future__ import annotations
 
 import logging
+from difflib import SequenceMatcher
 
 from src.core.models import (
     MatchMetadata,
     MatchResult,
     ParsedQuery,
-    Profile,
     SearchFilters,
     SearchMethod,
+    Skill,
 )
+from src.core.profile_store import ProfileStore
 from src.matching.scorer import CandidateScorer
 from src.search.filters import SearchFilter
 from src.search.hybrid import HybridSearch
 from src.search.reranker import CrossEncoderReranker
 
+SKILL_ALIASES: dict[str, list[str]] = {
+    "python": ["python3", "py"],
+    "javascript": ["js", "ecmascript", "es6"],
+    "typescript": ["ts"],
+    "react": ["reactjs", "react.js"],
+    "vue": ["vuejs", "vue.js"],
+    "angular": ["angularjs", "angular.js"],
+    "node.js": ["nodejs", "node"],
+    "kubernetes": ["k8s"],
+    "machine learning": ["ml"],
+    "artificial intelligence": ["ai"],
+    "natural language processing": ["nlp"],
+    "sql": ["mysql", "postgresql", "postgres", "pl/sql"],
+    "git": ["github", "gitlab", "bitbucket"],
+    "rest api": ["rest", "restful", "restful api"],
+    "tensorflow": ["tf"],
+    "pytorch": ["torch"],
+    "fastapi": ["fast api"],
+    "deep learning": ["dl"],
+    "computer vision": ["cv"],
+    "ci/cd": ["ci", "cd", "continuous integration", "continuous deployment"],
+    "statistics": ["statistical analysis", "statistical modeling"],
+}
+
+_ALIAS_TO_CANONICAL: dict[str, str] = {}
+for _canonical, _aliases in SKILL_ALIASES.items():
+    for _a in _aliases:
+        _ALIAS_TO_CANONICAL[_a] = _canonical
+
+
+def _canonical_skill(name: str) -> str:
+    return _ALIAS_TO_CANONICAL.get(name, name)
+
+
+def _skill_match_score(
+    required_names: list[str], profile_skills: list[Skill], raw_text: str | None = None,
+) -> float:
+    if not required_names:
+        return 1.0
+    matched = 0
+    skill_names_lower = {s.name.lower() for s in profile_skills}
+    raw_lower = raw_text.lower() if raw_text else ""
+    for rn in required_names:
+        rn_lower = _canonical_skill(rn.lower())
+        if rn_lower in skill_names_lower:
+            matched += 1
+            continue
+        aliases = SKILL_ALIASES.get(rn_lower, [])
+        if any(a in skill_names_lower for a in aliases):
+            matched += 1
+            continue
+        fuzzy_match = False
+        for sn in skill_names_lower:
+            if SequenceMatcher(None, rn_lower, sn).ratio() >= 0.8:
+                fuzzy_match = True
+                break
+        if fuzzy_match:
+            matched += 1
+            continue
+        if rn_lower in raw_lower:
+            matched += 1
+            continue
+        if any(a in raw_lower for a in aliases):
+            matched += 1
+    return matched / len(required_names)
+
+
+def _match_skills_detail(
+    required_names: list[str], profile_skills: list[Skill], raw_text: str | None = None,
+) -> tuple[list[str], list[str]]:
+    matched: list[str] = []
+    missing: list[str] = []
+    skill_names_lower = {s.name.lower() for s in profile_skills}
+    raw_lower = raw_text.lower() if raw_text else ""
+    for rn in required_names:
+        rn_lower = _canonical_skill(rn.lower())
+        if rn_lower in skill_names_lower:
+            matched.append(rn)
+            continue
+        aliases = SKILL_ALIASES.get(rn_lower, [])
+        if any(a in skill_names_lower for a in aliases):
+            matched.append(rn)
+            continue
+        fuzzy_found = False
+        for sn in skill_names_lower:
+            if SequenceMatcher(None, rn_lower, sn).ratio() >= 0.8:
+                fuzzy_found = True
+                break
+        if fuzzy_found:
+            matched.append(rn)
+            continue
+        if rn_lower in raw_lower:
+            matched.append(rn)
+            continue
+        if any(a in raw_lower for a in aliases):
+            matched.append(rn)
+            continue
+        missing.append(rn)
+    return matched, missing
+
 logger = logging.getLogger(__name__)
 
 
@@ -24,12 +126,13 @@ class ExecutorAgent:
         hybrid_search: HybridSearch,
         reranker: CrossEncoderReranker,
         scorer: CandidateScorer,
-        profiles: dict[str, Profile],
+        profiles: ProfileStore,
     ) -> None:
         self.hybrid_search = hybrid_search
         self.reranker = reranker
         self.scorer = scorer
-        self.profiles = profiles
+        self.profile_store = profiles
+        self._rerank_top_k = 20
 
     async def execute(
         self,
@@ -39,15 +142,12 @@ class ExecutorAgent:
     ) -> list[MatchResult]:
         search_text = self._query_to_search_text(parsed)
 
-        # RRF-fused hybrid search for ranking order
         hybrid_results = self.hybrid_search.search(search_text, top_k=top_k * 2)
 
-        # Separate vector + BM25 searches for actual similarity scores
         query_vec = self.hybrid_search.embedder.embed_query(search_text)
         vector_raw = self.hybrid_search.vector_search.search(query_vec, top_k=top_k * 2)
         bm25_raw = self.hybrid_search.bm25_search.search(search_text, top_k=top_k * 2)
 
-        # Build score lookup: profile_id → (vec_score, bm25_score)
         vec_scores: dict[str, float] = {
             pid: self._norm_vec_score(s) for pid, s in vector_raw
         }
@@ -58,8 +158,8 @@ class ExecutorAgent:
         filtered = self._apply_filters(hybrid_results, parsed)
 
         rerank_candidates: list[tuple[str, str, float]] = []
-        for pid, score in filtered[:50]:
-            profile = self.profiles.get(pid)
+        for pid, score in filtered[: self._rerank_top_k]:
+            profile = self.profile_store.get(pid)
             if profile is not None:
                 rerank_candidates.append((pid, profile.raw_text[:2000], score))
             else:
@@ -69,15 +169,14 @@ class ExecutorAgent:
 
         results: list[MatchResult] = []
         for rank, (pid, rerank_score) in enumerate(reranked, start=1):
-            profile = self.profiles.get(pid)
+            profile = self.profile_store.get(pid)
             if profile is None:
                 continue
 
-            skills_set = set(s.name.lower() for s in profile.skills)
-            req_set = set(rs.name.lower() for rs in parsed.required_skills)
-            pref_set = set(ps.name.lower() for ps in parsed.preferred_skills)
-            all_req = req_set | pref_set
-            skill_overlap = len(all_req & skills_set) / max(len(all_req), 1)
+            req_names = [rs.name for rs in parsed.required_skills]
+            pref_names = [ps.name for ps in parsed.preferred_skills]
+            all_req = req_names + pref_names
+            skill_overlap = _skill_match_score(all_req, profile.skills, profile.raw_text)
 
             total_years = (
                 profile.professional.total_experience_years
@@ -98,9 +197,9 @@ class ExecutorAgent:
 
             match_scores = self.scorer.compute_overall(scores_dict, slider_weights)
 
-            matched_skills = [s.name for s in profile.skills]
-            req_names = [rs.name for rs in parsed.required_skills]
-            missing_skills = [n for n in req_names if n not in matched_skills]
+            matched_skills, missing_skills = _match_skills_detail(
+                req_names, profile.skills, profile.raw_text,
+            )
 
             loc = profile.personal.location
             city = loc.city if profile.personal and loc else None
@@ -173,7 +272,7 @@ class ExecutorAgent:
 
         filtered: list[tuple[str, float]] = []
         for pid, score in results:
-            profile = self.profiles.get(pid)
+            profile = self.profile_store.get(pid)
             if profile is None:
                 filtered.append((pid, score))
             elif filter_obj.passes(profile):

src/api/routes/profiles.py

@@ -5,22 +5,25 @@ import logging
 from fastapi import APIRouter, HTTPException
 
 from src.core.models import Profile
+from src.core.profile_store import ProfileStore
 
 logger = logging.getLogger(__name__)
 
 router = APIRouter()
 
-_profiles_store: dict[str, Profile] = {}
+_profile_store: ProfileStore | None = None
 
 
-def init_profiles(profiles: dict[str, Profile]) -> None:
-    global _profiles_store
-    _profiles_store = profiles
+def init_profiles(profiles: ProfileStore) -> None:
+    global _profile_store
+    _profile_store = profiles
 
 
 @router.get("/profiles/{profile_id}", response_model=Profile)
 async def get_profile(profile_id: str) -> Profile:
-    profile = _profiles_store.get(profile_id)
+    if _profile_store is None:
+        raise HTTPException(status_code=503, detail="Profile store not initialized")
+    profile = _profile_store.get(profile_id)
     if profile is None:
         raise HTTPException(status_code=404, detail=f"Profile not found: {profile_id}")
     return profile
@@ -28,5 +31,7 @@ async def get_profile(profile_id: str) -> Profile:
 
 @router.get("/profiles", response_model=list[Profile])
 async def list_profiles(skip: int = 0, limit: int = 20) -> list[Profile]:
-    all_profiles = list(_profiles_store.values())
+    if _profile_store is None:
+        raise HTTPException(status_code=503, detail="Profile store not initialized")
+    all_profiles = list(_profile_store.get_all_sample().values())
     return all_profiles[skip : skip + limit]

src/core/models.py

@@ -80,6 +80,7 @@ class ProfessionalInfo(BaseModel):
     total_experience_years: float | None = None
     industry: str | None = None
     employment_type: EmploymentType | None = None
+    seniority_level: int | None = None
 
 
 class Skill(BaseModel):

src/core/profile_store.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+import json
+import logging
+from collections import OrderedDict
+from pathlib import Path
+
+from src.core.config import DATA_DIR
+from src.core.models import Profile
+from src.ingestion.normalizer import normalize_redrob
+
+logger = logging.getLogger(__name__)
+
+
+class ProfileStore:
+    def __init__(self, jsonl_path: Path | None = None, max_cache: int = 500) -> None:
+        self.path = jsonl_path or DATA_DIR / "profiles" / "candidates.jsonl"
+        self._max_cache = max_cache
+        self._offset_index: dict[str, int] = {}
+        self._cache: OrderedDict[str, Profile] = OrderedDict()
+        self._index_built = False
+        self._sample_profiles: dict[str, Profile] = {}
+
+    def load_sample(self, sample_path: Path) -> None:
+        if not sample_path.exists():
+            return
+        with open(sample_path) as f:
+            data = json.load(f)
+            profiles_list = data if isinstance(data, list) else [data]
+            for p in profiles_list:
+                try:
+                    profile = normalize_redrob(p)
+                    pid = profile.profile_id
+                    self._sample_profiles[pid] = profile
+                    if pid not in self._offset_index:
+                        self._offset_index[pid] = -1
+                except Exception:
+                    pass
+        logger.info(f"Loaded {len(self._sample_profiles)} sample profiles")
+
+    def load_offset_index(self, index_path: Path) -> None:
+        if index_path.exists():
+            with open(index_path) as f:
+                self._offset_index = json.load(f)
+            self._index_built = True
+            logger.info(f"Loaded offset index with {len(self._offset_index)} entries")
+
+    def save_offset_index(self, path: Path) -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with open(path, "w") as f:
+            json.dump(self._offset_index, f)
+        logger.info(f"Saved offset index ({len(self._offset_index)} entries) to {path}")
+
+    def _build_offset_index(self) -> None:
+        if self._index_built:
+            return
+        if not self.path.exists():
+            logger.warning(f"Profiles file not found: {self.path}")
+            self._index_built = True
+            return
+        count = 0
+        with open(self.path, encoding="utf-8") as f:
+            while True:
+                offset = f.tell()
+                line = f.readline()
+                if not line:
+                    break
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    raw = json.loads(line)
+                    pid = self._extract_id(raw)
+                    if pid and pid not in self._offset_index:
+                        self._offset_index[pid] = offset
+                        count += 1
+                except json.JSONDecodeError:
+                    pass
+        self._index_built = True
+        logger.info(f"Built offset index: {count} profile IDs")
+
+    @staticmethod
+    def _extract_id(raw: dict) -> str | None:
+        for key in ("profile_id", "candidate_id", "id"):
+            val = raw.get(key)
+            if val:
+                return str(val)
+        profile_nested = raw.get("profile", {})
+        if isinstance(profile_nested, dict):
+            for key in ("profile_id", "candidate_id", "id"):
+                val = profile_nested.get(key)
+                if val:
+                    return str(val)
+        return None
+
+    def get(self, pid: str) -> Profile | None:
+        if pid in self._cache:
+            self._cache.move_to_end(pid)
+            return self._cache[pid]
+
+        if pid in self._sample_profiles:
+            profile = self._sample_profiles[pid]
+            if len(self._cache) >= self._max_cache:
+                self._cache.popitem(last=False)
+            self._cache[pid] = profile
+            return profile
+
+        self._build_offset_index()
+        offset = self._offset_index.get(pid)
+        if offset is None or offset < 0:
+            return None
+
+        try:
+            with open(self.path, encoding="utf-8") as f:
+                f.seek(offset)
+                line = f.readline()
+            raw = json.loads(line)
+            profile = normalize_redrob(raw)
+        except Exception:
+            return None
+
+        if len(self._cache) >= self._max_cache:
+            self._cache.popitem(last=False)
+        self._cache[pid] = profile
+        return profile
+
+    def get_all_sample(self) -> dict[str, Profile]:
+        return dict(self._sample_profiles)
+
+    def __contains__(self, pid: str) -> bool:
+        if pid in self._sample_profiles or pid in self._cache:
+            return True
+        self._build_offset_index()
+        return pid in self._offset_index
+
+    def __len__(self) -> int:
+        self._build_offset_index()
+        return len(self._offset_index)

src/extraction/__init__.py

@@ -0,0 +1,3 @@
+from src.extraction.pipeline import ExtractionBundle, ExtractionResult, FieldExtractorPipeline
+
+__all__ = ["FieldExtractorPipeline", "ExtractionBundle", "ExtractionResult"]

src/extraction/career_history_utils.py

@@ -0,0 +1,115 @@
+from __future__ import annotations
+
+from datetime import date, datetime
+from typing import Any
+
+_LOW_CONFIDENCE_THRESHOLD = 0.5
+
+
+def latest_role(history: list[dict[str, Any]]) -> dict[str, Any] | None:
+    if not history:
+        return None
+    current = [r for r in history if r.get("is_current")]
+    if current:
+        return current[0]
+    return max(
+        history,
+        key=lambda r: _parse_date_for_sort(r.get("start_date")) or date.min,
+    )
+
+
+def compute_years_from_dates(
+    history: list[dict[str, Any]],
+) -> tuple[float | None, int, str]:
+    today = date.today()
+    total_days = 0.0
+    valid_entries = 0
+    assumptions = 0
+
+    intervals: list[tuple[date, date]] = []
+    for entry in history:
+        start = _parse_date(entry.get("start_date"))
+        if start is None:
+            continue
+        end_raw = entry.get("end_date")
+        is_current = entry.get("is_current", False)
+        if end_raw:
+            end = _parse_date(end_raw)
+            if end is None:
+                continue
+            if end > today:
+                end = today
+                assumptions += 1
+            if end <= start:
+                continue
+        elif is_current:
+            end = today
+        else:
+            end = date(start.year + 1, start.month, start.day)
+            assumptions += 1
+        intervals.append((start, end))
+        valid_entries += 1
+
+    if not intervals:
+        return None, 0, "low"
+
+    merged = _merge_intervals(intervals)
+    for s, e in merged:
+        total_days += (e - s).days
+
+    total_years = round(total_days / 365.25, 1)
+
+    if valid_entries == 0:
+        return None, 0, "low"
+    ratio = assumptions / valid_entries
+    if ratio > _LOW_CONFIDENCE_THRESHOLD:
+        confidence = "low"
+    elif ratio > 0:
+        confidence = "medium"
+    else:
+        confidence = "high"
+
+    return total_years, valid_entries, confidence
+
+
+def _parse_date(raw: str | None) -> date | None:
+    if not raw:
+        return None
+    for fmt in ("%Y-%m-%d", "%Y-%m", "%Y/%m/%d", "%Y/%m", "%d-%m-%Y", "%d/%m/%Y"):
+        try:
+            return datetime.strptime(raw.strip(), fmt).date()
+        except ValueError:
+            continue
+    try:
+        return datetime.strptime(raw.strip()[:10], "%Y-%m-%d").date()
+    except (ValueError, IndexError):
+        pass
+    try:
+        year = int(raw.strip()[:4])
+        return date(year, 1, 1)
+    except (ValueError, IndexError):
+        return None
+
+
+def _parse_date_for_sort(raw: str | None) -> date | None:
+    d = _parse_date(raw)
+    if d is not None:
+        return d
+    if raw:
+        try:
+            year = int(raw.strip()[:4])
+            return date(year, 1, 1)
+        except (ValueError, IndexError):
+            pass
+    return None
+
+
+def _merge_intervals(intervals: list[tuple[date, date]]) -> list[tuple[date, date]]:
+    sorted_iv = sorted(intervals, key=lambda x: x[0])
+    merged: list[tuple[date, date]] = []
+    for start, end in sorted_iv:
+        if merged and start <= merged[-1][1]:
+            merged[-1] = (merged[-1][0], max(merged[-1][1], end))
+        else:
+            merged.append((start, end))
+    return merged

src/extraction/company.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from typing import Any
+
+from src.extraction.career_history_utils import latest_role
+
+
+def extract_company(
+    prof: dict[str, Any],
+    history: list[dict[str, Any]],
+) -> tuple[str | None, str]:
+    direct = prof.get("current_company")
+    if direct and isinstance(direct, str) and direct.strip():
+        return direct.strip(), "direct"
+
+    latest = latest_role(history)
+    if latest:
+        company = latest.get("company")
+        if company and isinstance(company, str) and company.strip():
+            return company.strip(), "history"
+
+    return None, "not_found"

src/extraction/domain.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_INDUSTRY_KEYWORDS: dict[str, list[str]] = {
+    "fintech": ["fintech", "banking", "finance", "payment", "insurance", "investment"],
+    "healthcare": ["healthcare", "health", "medical", "clinical", "pharma", "biotech"],
+    "ecommerce": ["ecommerce", "e-commerce", "retail", "marketplace", "consumer internet"],
+    "ai/ml": ["machine learning", "artificial intelligence", "deep learning", "nlp",
+              "llm", "computer vision", "mlops"],
+    "edtech": ["edtech", "education", "elearning", "learning", "online education"],
+    "saas": ["saas", "b2b", "enterprise software", "cloud software"],
+    "infrastructure": ["devops", "infrastructure", "cloud", "kubernetes", "docker",
+                        "terraform", "platform engineering"],
+    "data": ["data engineering", "data science", "data analytics", "big data", "data pipeline"],
+    "cybersecurity": ["cybersecurity", "security", "infosec", "penetration testing"],
+}
+
+_COMPANY_INDUSTRY: dict[str, str] = {
+    "mindtree": "it_services",
+    "infosys": "it_services",
+    "tcs": "it_services",
+    "wipro": "it_services",
+    "accenture": "it_services",
+    "google": "internet",
+    "amazon": "ecommerce",
+    "microsoft": "saas",
+    "flipkart": "ecommerce",
+    "swiggy": "ecommerce",
+    "zomato": "ecommerce",
+    "razorpay": "fintech",
+    "phonepe": "fintech",
+    "paytm": "fintech",
+    "byjus": "edtech",
+    "unacademy": "edtech",
+}
+
+
+def extract_industry(
+    prof: dict[str, Any],
+    skills: list[dict[str, Any]],
+    history: list[dict[str, Any]],
+) -> tuple[str | None, str]:
+    direct = prof.get("current_industry")
+    if direct and isinstance(direct, str) and direct.strip():
+        return direct.strip(), "direct"
+
+    for entry in history:
+        company = (entry.get("company") or "").lower().strip()
+        if company in _COMPANY_INDUSTRY:
+            return _COMPANY_INDUSTRY[company], "company_map"
+        for known_company, mapped_industry in _COMPANY_INDUSTRY.items():
+            if known_company in company:
+                return mapped_industry, "company_map"
+
+    skill_names = [s.get("name", "") for s in skills]
+    all_text = " ".join(skill_names).lower()
+    for industry, keywords in _INDUSTRY_KEYWORDS.items():
+        for keyword in keywords:
+            if keyword in all_text:
+                return industry, "skills"
+
+    headline = prof.get("headline", "")
+    summary = prof.get("summary", "")
+    combined = f"{headline} {summary}".lower()
+    for industry, keywords in _INDUSTRY_KEYWORDS.items():
+        for keyword in keywords:
+            if keyword in combined:
+                return industry, "headline_summary"
+
+    return None, "not_found"

src/extraction/experience_years.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from src.extraction.career_history_utils import compute_years_from_dates
+
+_YEARS_PATTERNS: list[re.Pattern[str]] = [
+    re.compile(r"(\d+)\+?\s*(?:yrs?|years?|yoe|years?\s+of\s+experience)", re.IGNORECASE),
+    re.compile(r"(\d+)\+?\s*(?:years?\s+exp)", re.IGNORECASE),
+    re.compile(r"experience[:\s]+(\d+)\+?\s*(?:years?|yrs?)", re.IGNORECASE),
+    re.compile(r"(\d+)\s*\+\s*years?", re.IGNORECASE),
+    re.compile(r"~?\s*(\d+)\s*(?:years?|yrs?)", re.IGNORECASE),
+]
+
+_AGREEMENT_RATIO = 0.2
+
+
+def extract_experience_years(
+    prof: dict[str, Any],
+    history: list[dict[str, Any]],
+) -> tuple[float | None, str]:
+    direct = _safe_float(prof.get("years_of_experience"))
+
+    dates_result, num_valid, dates_conf = compute_years_from_dates(history)
+
+    source_a = direct
+    source_b = dates_result
+
+    if source_a is not None and source_b is not None:
+        if abs(source_a - source_b) / max(source_a, source_b, 0.1) <= _AGREEMENT_RATIO:
+            return round((source_a + source_b) / 2, 1), "average"
+        if dates_conf == "high":
+            return source_b, "dates_structured"
+        return source_a, "direct_structured"
+
+    if source_a is not None:
+        return source_a, "direct_structured"
+
+    if source_b is not None:
+        return source_b, "dates_structured"
+
+    regex_val = _extract_from_text(prof.get("headline", ""), prof.get("summary", ""))
+    if regex_val is not None:
+        return regex_val, "regex_fallback"
+
+    return None, "not_found"
+
+
+def _extract_from_text(*texts: str) -> float | None:
+    for text in texts:
+        if not text:
+            continue
+        for pattern in _YEARS_PATTERNS:
+            m = pattern.search(text)
+            if m:
+                val = float(m.group(1))
+                if val >= 100:
+                    continue
+                return val
+    return None
+
+
+def _safe_float(val: Any) -> float | None:
+    if val is None:
+        return None
+    try:
+        f = float(val)
+        if f > 100:
+            return None
+        return f
+    except (ValueError, TypeError):
+        return None

src/extraction/pipeline.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from src.extraction.company import extract_company
+from src.extraction.domain import extract_industry
+from src.extraction.experience_years import extract_experience_years
+from src.extraction.seniority import extract_seniority
+from src.extraction.title import extract_title
+
+
+@dataclass
+class ExtractionResult:
+    value: Any
+    source: str
+    confidence: str = "medium"
+
+
+@dataclass
+class ExtractionBundle:
+    current_title: ExtractionResult = field(default_factory=lambda: ExtractionResult(None, "not_found"))
+    current_company: ExtractionResult = field(default_factory=lambda: ExtractionResult(None, "not_found"))
+    total_experience_years: ExtractionResult = field(default_factory=lambda: ExtractionResult(None, "not_found"))
+    industry: ExtractionResult = field(default_factory=lambda: ExtractionResult(None, "not_found"))
+    seniority_level: ExtractionResult = field(default_factory=lambda: ExtractionResult(None, "not_found"))
+
+
+class FieldExtractorPipeline:
+    def extract(self, raw: dict[str, Any]) -> ExtractionBundle:
+        prof = raw.get("profile", {})
+        history = raw.get("career_history", [])
+        skills = raw.get("skills", [])
+
+        title_val, title_src = extract_title(prof, history)
+        company_val, company_src = extract_company(prof, history)
+        years_val, years_src = extract_experience_years(prof, history)
+        industry_val, industry_src = extract_industry(prof, skills, history)
+        seniority_val, seniority_src = extract_seniority(title_val, years_val, history)
+
+        return ExtractionBundle(
+            current_title=ExtractionResult(title_val, title_src),
+            current_company=ExtractionResult(company_val, company_src),
+            total_experience_years=ExtractionResult(years_val, years_src),
+            industry=ExtractionResult(industry_val, industry_src),
+            seniority_level=ExtractionResult(seniority_val, seniority_src),
+        )

src/extraction/seniority.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+from typing import Any
+
+from src.extraction.career_history_utils import latest_role
+
+_SENIORITY_TITLES: dict[str, int] = {
+    "intern": 0,
+    "trainee": 0,
+    "junior": 1,
+    "jr": 1,
+    "mid": 2,
+    "mid-level": 2,
+    "senior": 3,
+    "sr": 3,
+    "lead": 4,
+    "staff": 4,
+    "principal": 5,
+    "architect": 5,
+    "director": 6,
+    "head": 6,
+    "vp": 6,
+    "chief": 6,
+    "cto": 6,
+}
+
+_YEARS_JUNIOR = 2
+_YEARS_MID = 5
+_YEARS_SENIOR = 8
+_YEARS_LEAD = 12
+
+_DOMAIN_OVERRIDES: dict[str, int] = {
+    "professor": 6,
+    "assistant professor": 5,
+    "associate professor": 6,
+    "lecturer": 3,
+    "principal": 5,
+    "director": 6,
+}
+
+DOMAIN_SPECIFIC_CONSTANTS_NOTE = """
+Seniority thresholds (<2 junior, <5 mid, <8 senior, <12 lead) are
+tuned for the Indian IT market — the primary data source. These
+constants should be revisited if the normalizer is used for
+academia, government, or other sectors.
+"""
+
+
+def extract_seniority(
+    title: str | None,
+    years: float | None,
+    history: list[dict[str, Any]],
+) -> tuple[int | None, str]:
+    title_lower = (title or "").lower()
+
+    for keyword, level in _DOMAIN_OVERRIDES.items():
+        if keyword in title_lower:
+            return level, "domain_override"
+
+    for keyword, level in _SENIORITY_TITLES.items():
+        if keyword in title_lower:
+            return level, "title_keyword"
+
+    latest = latest_role(history)
+    if latest:
+        latest_title = (latest.get("title") or "").lower()
+        for keyword, level in _SENIORITY_TITLES.items():
+            if keyword in latest_title:
+                return level, "history_title"
+
+    if years is not None:
+        if years < _YEARS_JUNIOR:
+            return 1, "years_fallback"
+        if years < _YEARS_MID:
+            return 2, "years_fallback"
+        if years < _YEARS_SENIOR:
+            return 3, "years_fallback"
+        if years < _YEARS_LEAD:
+            return 4, "years_fallback"
+        return 5, "years_fallback"
+
+    return None, "not_found"

src/extraction/title.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from src.extraction.career_history_utils import latest_role
+
+_SENIORITY_PREFIXES = [
+    "senior", "sr", "junior", "jr", "lead", "staff",
+    "principal", "chief", "head", "vp", "vp of", "director of",
+    "associate", "assistant", "principal",
+]
+
+_SENIORITY_PATTERN = re.compile(
+    r"^(?:" + "|".join(_SENIORITY_PREFIXES) + r")\s+",
+    re.IGNORECASE,
+)
+
+
+def extract_title(
+    prof: dict[str, Any],
+    history: list[dict[str, Any]],
+) -> tuple[str | None, str]:
+    direct = prof.get("current_title")
+    if direct and isinstance(direct, str) and direct.strip():
+        return direct.strip(), "direct"
+
+    latest = latest_role(history)
+    if latest:
+        title = latest.get("title")
+        if title and isinstance(title, str) and title.strip():
+            return title.strip(), "history"
+
+    headline = prof.get("headline", "")
+    if headline and isinstance(headline, str) and headline.strip():
+        extracted = _parse_headline(headline)
+        if extracted:
+            return extracted, "headline"
+
+    return None, "not_found"
+
+
+def _parse_headline(headline: str) -> str | None:
+    candidate = headline.strip()
+    if "|" in candidate:
+        candidate = candidate.split("|")[0].strip()
+    if " at " in candidate.lower():
+        candidate = candidate.lower().split(" at ")[0].strip().title()
+    candidate = _SENIORITY_PATTERN.sub("", candidate).strip()
+    if candidate:
+        return candidate
+    return None

src/ingestion/normalizer.py

@@ -16,6 +16,7 @@ from src.core.models import (
     Skill,
     WorkExperience,
 )
+from src.extraction.pipeline import FieldExtractorPipeline
 
 _PROFICIENCY_MAP = {
     "beginner": ProficiencyLevel.BEGINNER,
@@ -59,12 +60,15 @@ def normalize_redrob(raw: dict[str, Any], source: str = "redrob") -> Profile:
         native_language=None,
     )
 
+    extracted = FieldExtractorPipeline().extract(raw)
+
     professional = ProfessionalInfo(
-        current_title=prof.get("current_title"),
-        current_company=prof.get("current_company"),
-        total_experience_years=prof.get("years_of_experience"),
-        industry=prof.get("current_industry"),
+        current_title=extracted.current_title.value or prof.get("current_title"),
+        current_company=extracted.current_company.value or prof.get("current_company"),
+        total_experience_years=extracted.total_experience_years.value or prof.get("years_of_experience"),
+        industry=extracted.industry.value or prof.get("current_industry"),
         employment_type=None,
+        seniority_level=extracted.seniority_level.value,
     )
 
     skills = [
@@ -91,7 +95,7 @@ def normalize_redrob(raw: dict[str, Any], source: str = "redrob") -> Profile:
             end_date=entry.get("end_date"),
             is_current=entry.get("is_current", False),
             description=entry.get("description", ""),
-            location=entry.get("industry"),
+            location=entry.get("location"),
         )
         for entry in raw.get("career_history", [])
     ]
@@ -123,6 +127,21 @@ def normalize_redrob(raw: dict[str, Any], source: str = "redrob") -> Profile:
 
     raw_text = _build_raw_text(prof, experience, skills, education, certs, languages)
 
+    from src.matching.skill_matcher import SKILL_ALIASES as _KNOWN_ALIASES
+    existing_skill_names = {s.name.lower() for s in skills}
+    raw_lower = raw_text.lower()
+    for _canon, _aliases in _KNOWN_ALIASES.items():
+        if _canon not in existing_skill_names and _canon in raw_lower:
+            skills.append(
+                Skill(
+                    name=_canon.title(),
+                    category=_infer_skill_category(_canon),
+                    evidence="Extracted from profile text",
+                    confidence=0.6,
+                )
+            )
+            existing_skill_names.add(_canon)
+
     if native_langs:
         personal.native_language = native_langs[0]
     personal.languages_spoken = languages

src/main.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import json
 import logging
 from contextlib import asynccontextmanager
 
@@ -15,7 +14,7 @@ from src.api.routes.profiles import router as profiles_router
 from src.api.routes.search import init_orchestrator
 from src.api.routes.search import router as search_router
 from src.core.config import DATA_DIR
-from src.core.models import Profile
+from src.core.profile_store import ProfileStore
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -70,41 +69,18 @@ async def lifespan(app: FastAPI):
     logger.info("Cross-encoder model loaded")
     scorer = CandidateScorer()
 
-    profiles: dict[str, Profile] = {}
-    profiles_loaded = 0
-
-    from src.ingestion.normalizer import normalize_redrob
-
+    profiles = ProfileStore()
     sample_path = DATA_DIR / "samples" / "sample_candidates.json"
     if sample_path.exists():
-        with open(sample_path) as f:
-            data = json.load(f)
-            profiles_list = data if isinstance(data, list) else [data]
-            for p in profiles_list:
-                try:
-                    profile = normalize_redrob(p)
-                    profiles[profile.profile_id] = profile
-                    profiles_loaded += 1
-                except Exception:
-                    pass
-
-    cand_path = DATA_DIR / "profiles" / "candidates.jsonl"
-    if cand_path.exists():
-        from src.ingestion.parser import ProfileParser
-        parser = ProfileParser()
-        for raw in parser.parse_jsonl_file(cand_path):
-            if raw.get("candidate_id", raw.get("id", "")) in profiles:
-                continue
-            if profiles_loaded >= vector_search.size:
-                break
-            try:
-                profile = normalize_redrob(raw)
-                profiles[profile.profile_id] = profile
-                profiles_loaded += 1
-            except Exception:
-                pass
-
-    logger.info(f"Loaded {len(profiles)} profiles into memory")
+        profiles.load_sample(sample_path)
+
+    offset_index_path = DATA_DIR / "indexes" / "offset_index.json"
+    if offset_index_path.exists():
+        profiles.load_offset_index(offset_index_path)
+        logger.info(f"ProfileStore ready: {len(profiles)} profiles available")
+    else:
+        logger.info("No offset index found — profiles will be indexed on first access")
+    logger.info("ProfileStore initialized (lazy load)")
 
     planner = PlannerAgent()
     executor = ExecutorAgent(hybrid_search, reranker, scorer, profiles)
@@ -115,7 +91,7 @@ async def lifespan(app: FastAPI):
     init_health(index_size=vector_search.size)
     init_profiles(profiles)
 
-    logger.info("System initialized successfully")
+    logger.info("System initialized successfully (%.1fs)", 0.0)
 
     yield
 
@@ -134,3 +110,7 @@ app.include_router(search_router, prefix="/api/v1")
 app.include_router(profiles_router, prefix="/api/v1")
 app.include_router(ingest_router, prefix="/api/v1")
 app.include_router(health_router, prefix="/api/v1")
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run("src.main:app", host="0.0.0.0", port=8000, log_level="info")

src/ui/app.py

@@ -6,7 +6,10 @@ import time
 
 import gradio as gr
 
+from src.api.routes.search import init_orchestrator
+from src.core.config import DATA_DIR
 from src.core.models import MatchScores, SearchResultItem
+from src.core.profile_store import ProfileStore
 from src.matching.scorer import DEFAULT_SLIDER_WEIGHTS, CandidateScorer
 from src.ui.components import (
     create_analytics_dashboard,
@@ -16,6 +19,53 @@ from src.ui.components import (
 
 logger = logging.getLogger(__name__)
 
+indexes_dir = DATA_DIR / "indexes"
+faiss_path = indexes_dir / "faiss_index.bin"
+id_map_path = indexes_dir / "faiss_id_map.json"
+bm25_path = indexes_dir / "bm25_index.pkl"
+
+if faiss_path.exists():
+    from src.agents.executor import ExecutorAgent
+    from src.agents.orchestrator import Orchestrator
+    from src.agents.planner import PlannerAgent
+    from src.agents.reflector import ReflectorAgent
+    from src.language.multilingual import MultilingualEmbedder
+    from src.search.bm25_search import BM25Search
+    from src.search.hybrid import HybridSearch
+    from src.search.reranker import CrossEncoderReranker
+    from src.search.vector_search import VectorSearch
+
+    embedder = MultilingualEmbedder()
+    _ = embedder.model
+
+    vector_search = VectorSearch()
+    vector_search.load(faiss_path, id_map_path)
+
+    bm25_search = BM25Search()
+    bm25_search.load(bm25_path)
+
+    hybrid_search = HybridSearch(vector_search, bm25_search, embedder)
+    reranker = CrossEncoderReranker()
+    _ = reranker.model
+    scorer = CandidateScorer()
+
+    profiles = ProfileStore()
+    offset_index_path = DATA_DIR / "indexes" / "offset_index.json"
+    if offset_index_path.exists():
+        profiles.load_offset_index(offset_index_path)
+    sample_path = DATA_DIR / "samples" / "sample_candidates.json"
+    if sample_path.exists():
+        profiles.load_sample(sample_path)
+
+    planner = PlannerAgent()
+    executor = ExecutorAgent(hybrid_search, reranker, scorer, profiles)
+    reflector = ReflectorAgent()
+    orchestrator = Orchestrator(planner, executor, reflector)
+    init_orchestrator(orchestrator)
+    logger.info("Search system initialized")
+else:
+    logger.warning("No FAISS index found. Run 'python scripts/build_indexes.py' first.")
+
 SLIDER_DIMS = [
     ("Skill Match", "skill_match", DEFAULT_SLIDER_WEIGHTS["skill_match"]),
     ("Experience", "experience_match", DEFAULT_SLIDER_WEIGHTS["experience_match"]),

tests/test_extraction/test_career_history_utils.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+from datetime import date
+
+from src.extraction.career_history_utils import compute_years_from_dates, latest_role, _parse_date
+
+
+class TestLatestRole:
+    def test_empty_history(self):
+        assert latest_role([]) is None
+
+    def test_prefers_current(self):
+        history = [
+            {"title": "Backend Engineer", "company": "Mindtree", "is_current": True,
+             "start_date": "2024-01"},
+            {"title": "Junior Dev", "company": "OldCo", "is_current": False,
+             "start_date": "2022-01", "end_date": "2023-12"},
+        ]
+        role = latest_role(history)
+        assert role["company"] == "Mindtree"
+
+    def test_most_recent_when_no_current(self):
+        history = [
+            {"title": "Junior Dev", "company": "OldCo", "is_current": False,
+             "start_date": "2022-01", "end_date": "2023-12"},
+            {"title": "Intern", "company": "FirstCo", "is_current": False,
+             "start_date": "2021-01", "end_date": "2021-12"},
+        ]
+        role = latest_role(history)
+        assert role["company"] == "OldCo"
+
+
+class TestComputeYearsFromDates:
+    def test_simple_non_overlapping(self):
+        history = [
+            {"start_date": "2020-01", "end_date": "2022-01", "is_current": False},
+            {"start_date": "2022-02", "end_date": "2024-02", "is_current": False},
+        ]
+        years, count, conf = compute_years_from_dates(history)
+        assert count == 2
+        assert conf == "high"
+        assert years is not None and 3.9 < years < 4.1
+
+    def test_current_role_uses_today(self):
+        history = [
+            {"start_date": "2022-01", "is_current": True},
+        ]
+        years, count, conf = compute_years_from_dates(history)
+        assert count == 1
+        assert conf == "high"
+        assert years is not None and years > 0
+
+    def test_missing_end_date_assumes_one_year(self):
+        history = [
+            {"start_date": "2023-01", "is_current": False},
+        ]
+        years, count, conf = compute_years_from_dates(history)
+        assert count == 1
+        assert conf == "low"
+        assert years is not None and 0.9 < years < 1.1
+
+    def test_future_date_capped_to_today(self):
+        history = [
+            {"start_date": "2020-01", "end_date": "2030-01", "is_current": False},
+        ]
+        years, count, conf = compute_years_from_dates(history)
+        assert count == 1
+        assert conf == "low"
+
+    def test_swapped_dates_skipped(self):
+        history = [
+            {"start_date": "2024-01", "end_date": "2022-01", "is_current": False},
+        ]
+        years, count, conf = compute_years_from_dates(history)
+        assert years is None
+
+    def test_missing_start_date_skipped(self):
+        history = [
+            {"end_date": "2024-01", "is_current": False},
+        ]
+        years, count, conf = compute_years_from_dates(history)
+        assert years is None
+
+    def test_overlapping_interval_merged(self):
+        history = [
+            {"start_date": "2020-01", "end_date": "2023-06", "is_current": False},
+            {"start_date": "2023-01", "end_date": "2024-01", "is_current": False},
+        ]
+        years, count, conf = compute_years_from_dates(history)
+        assert count == 2
+        assert years is not None and 3.9 < years < 4.1
+
+    def test_empty_history_returns_none(self):
+        years, count, conf = compute_years_from_dates([])
+        assert years is None
+
+    def test_low_confidence_when_mostly_assumptions(self):
+        history = [
+            {"start_date": "2020-01", "is_current": False},
+            {"start_date": "2021-01", "end_date": "2022-01", "is_current": False},
+        ]
+        years, count, conf = compute_years_from_dates(history)
+        assert conf == "medium"
+
+
+class TestParseDate:
+    def test_iso_date(self):
+        assert _parse_date("2024-01-15") == date(2024, 1, 15)
+
+    def test_year_month(self):
+        assert _parse_date("2024-01") == date(2024, 1, 1)
+
+    def test_year_only(self):
+        assert _parse_date("2024") == date(2024, 1, 1)
+
+    def test_invalid_returns_none(self):
+        assert _parse_date("not-a-date") is None
+
+    def test_none_returns_none(self):
+        assert _parse_date(None) is None

tests/test_extraction/test_company.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from src.extraction.company import extract_company
+
+
+class TestExtractCompany:
+    def test_direct(self):
+        val, src = extract_company(
+            {"current_company": "Mindtree"}, [],
+        )
+        assert val == "Mindtree"
+        assert src == "direct"
+
+    def test_fallback_to_history(self):
+        val, src = extract_company(
+            {},
+            [{"company": "Mindtree", "start_date": "2020-01", "is_current": True}],
+        )
+        assert val == "Mindtree"
+        assert src == "history"
+
+    def test_prefers_current_role(self):
+        val, src = extract_company(
+            {"current_company": "Mindtree"},
+            [{"company": "OldCo", "start_date": "2020-01", "is_current": False}],
+        )
+        assert val == "Mindtree"
+        assert src == "direct"
+
+    def test_no_info_returns_none(self):
+        val, src = extract_company({}, [])
+        assert val is None
+        assert src == "not_found"

tests/test_extraction/test_domain.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from src.extraction.domain import extract_industry
+
+
+class TestExtractIndustry:
+    def test_direct(self):
+        val, src = extract_industry(
+            {"current_industry": "fintech"}, [], [],
+        )
+        assert val == "fintech"
+        assert src == "direct"
+
+    def test_from_company_map(self):
+        val, src = extract_industry(
+            {},
+            [],
+            [{"company": "Razorpay"}],
+        )
+        assert val == "fintech"
+        assert src == "company_map"
+
+    def test_from_skills(self):
+        val, src = extract_industry(
+            {},
+            [{"name": "NLP"}, {"name": "PyTorch"}, {"name": "Computer Vision"}],
+            [],
+        )
+        assert val == "ai/ml"
+        assert src == "skills"
+
+    def test_from_headline(self):
+        val, src = extract_industry(
+            {"headline": "Fintech Engineer at a payments startup"},
+            [],
+            [],
+        )
+        assert val == "fintech"
+        assert src == "headline_summary"
+
+    def test_no_info_returns_none(self):
+        val, src = extract_industry({}, [], [])
+        assert val is None
+        assert src == "not_found"

tests/test_extraction/test_experience_years.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+from src.extraction.experience_years import extract_experience_years
+
+
+class TestExtractExperienceYears:
+    def test_direct_field(self):
+        val, src = extract_experience_years(
+            {"years_of_experience": 5.0}, [],
+        )
+        assert val == 5.0
+        assert src == "direct_structured"
+
+    def test_dates_from_history(self):
+        val, src = extract_experience_years(
+            {},
+            [
+                {"start_date": "2020-01", "end_date": "2022-01", "is_current": False},
+                {"start_date": "2022-02", "end_date": "2024-02", "is_current": False},
+            ],
+        )
+        assert src == "dates_structured"
+        assert val is not None and 3.9 < val < 4.1
+
+    def test_average_when_close(self):
+        val, src = extract_experience_years(
+            {"years_of_experience": 4.0},
+            [
+                {"start_date": "2020-06", "end_date": "2024-06", "is_current": False},
+            ],
+        )
+        assert src == "average"
+        assert val is not None and 3.9 < val < 4.1
+
+    def test_dates_preferred_when_disagree_and_confident(self):
+        val, src = extract_experience_years(
+            {"years_of_experience": 10.0},
+            [
+                {"start_date": "2023-01", "end_date": "2024-01", "is_current": False},
+            ],
+        )
+        assert src == "dates_structured"
+        assert val is not None and 0.9 < val < 1.1
+
+    def test_regex_fallback_from_headline(self):
+        val, src = extract_experience_years(
+            {"headline": "Senior Dev with 8+ years experience"},
+            [],
+        )
+        assert val == 8.0
+        assert src == "regex_fallback"
+
+    def test_regex_fallback_from_summary(self):
+        val, src = extract_experience_years(
+            {"summary": "Experience: 12 years in software engineering"},
+            [],
+        )
+        assert val == 12.0
+        assert src == "regex_fallback"
+
+    def test_typo_100_plus_years_ignored(self):
+        val, src = extract_experience_years(
+            {"headline": "100+ years of experience"},
+            [],
+        )
+        assert val is None
+        assert src == "not_found"
+
+    def test_typo_150_years_ignored(self):
+        val, src = extract_experience_years(
+            {"years_of_experience": 150},
+            [],
+        )
+        assert val is None
+        assert src == "not_found"
+
+    def test_no_info_returns_none(self):
+        val, src = extract_experience_years({}, [])
+        assert val is None
+        assert src == "not_found"

tests/test_extraction/test_pipeline.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from src.extraction.pipeline import FieldExtractorPipeline
+
+
+class TestPipeline:
+    def test_full_profile(self):
+        raw = {
+            "candidate_id": "CAND_001",
+            "profile": {
+                "anonymized_name": "Test User",
+                "current_title": "Senior Backend Engineer",
+                "current_company": "Mindtree",
+                "years_of_experience": 6.5,
+                "current_industry": "it_services",
+                "location": "Bangalore, Karnataka",
+                "country": "India",
+                "headline": "Senior Backend Engineer | Mindtree",
+            },
+            "career_history": [
+                {"title": "Backend Engineer", "company": "Mindtree",
+                 "start_date": "2024-01", "is_current": True},
+            ],
+            "skills": [
+                {"name": "Python", "proficiency": "advanced"},
+                {"name": "AWS", "proficiency": "intermediate"},
+            ],
+        }
+        bundle = FieldExtractorPipeline().extract(raw)
+        assert bundle.current_title.value == "Senior Backend Engineer"
+        assert bundle.current_title.source == "direct"
+        assert bundle.current_company.value == "Mindtree"
+        assert bundle.current_company.source == "direct"
+        assert bundle.total_experience_years.value is not None
+        assert bundle.industry.value == "it_services"
+        assert bundle.industry.source == "direct"
+        assert bundle.seniority_level.value == 3
+
+    def test_sparse_profile_falls_back_gracefully(self):
+        raw = {
+            "candidate_id": "CAND_002",
+            "profile": {
+                "anonymized_name": "Sparse User",
+                "headline": "Python Developer | 5+ yrs experience",
+            },
+            "career_history": [],
+            "skills": [],
+        }
+        bundle = FieldExtractorPipeline().extract(raw)
+        assert bundle.current_title.value is not None
+        assert bundle.current_title.source == "headline"
+        assert bundle.current_company.value is None
+        assert bundle.total_experience_years.value == 5.0
+        assert bundle.total_experience_years.source == "regex_fallback"
+        assert bundle.industry.value is None
+        assert bundle.seniority_level.value is not None
+
+    def test_empty_raw(self):
+        bundle = FieldExtractorPipeline().extract({})
+        assert bundle.current_title.value is None
+        assert bundle.current_company.value is None
+        assert bundle.total_experience_years.value is None
+        assert bundle.industry.value is None
+        assert bundle.seniority_level.value is None

tests/test_extraction/test_regression.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from src.ingestion.normalizer import normalize_redrob as new_normalizer
+
+_PROFESSIONAL_FIELDS = [
+    "current_title",
+    "current_company",
+    "total_experience_years",
+    "industry",
+]
+
+_SAMPLE_ROWS = [
+    {
+        "candidate_id": "CAND_001",
+        "profile": {
+            "anonymized_name": "Ira Vora",
+            "current_title": "Backend Engineer",
+            "current_company": "Mindtree",
+            "years_of_experience": 6.9,
+            "current_industry": "it_services",
+            "location": "Toronto, Ontario",
+            "country": "Canada",
+            "headline": "Backend Engineer | SQL, Spark, Cloud",
+            "summary": "Software/data professional with 6.9 yrs experience",
+        },
+        "career_history": [
+            {"title": "Backend Engineer", "company": "Mindtree",
+             "start_date": "2024-03-08", "is_current": True,
+             "description": "Implemented streaming data pipelines"},
+            {"title": "Data Engineer", "company": "OldCo",
+             "start_date": "2021-01", "end_date": "2024-02",
+             "description": "Built ETL pipelines"},
+        ],
+        "education": [
+            {"institution": "LPU", "degree": "B.E.",
+             "field_of_study": "CS", "end_year": 2020, "grade": "8.24 CGPA"},
+        ],
+        "skills": [
+            {"name": "NLP", "proficiency": "advanced", "endorsements": 37, "duration_months": 26},
+            {"name": "AWS", "proficiency": "beginner", "endorsements": 5, "duration_months": 8},
+        ],
+        "languages": [{"language": "English", "proficiency": "professional"}],
+    },
+    {
+        "candidate_id": "CAND_002",
+        "profile": {
+            "anonymized_name": "Sparse Candidate",
+            "headline": "Senior Python Dev",
+        },
+        "career_history": [],
+        "skills": [],
+    },
+    {
+        "candidate_id": "CAND_003",
+        "profile": {
+            "anonymized_name": "Minimal",
+        },
+        "career_history": [],
+        "skills": [],
+    },
+]
+
+
+class TestRegressionNoDataLoss:
+    def test_all_non_none_fields_preserved(self):
+        for row in _SAMPLE_ROWS:
+            profile = new_normalizer(row)
+            for field in _PROFESSIONAL_FIELDS:
+                old_val = row.get("profile", {}).get({
+                    "current_title": "current_title",
+                    "current_company": "current_company",
+                    "total_experience_years": "years_of_experience",
+                    "industry": "current_industry",
+                }[field])
+                new_val = getattr(profile.professional, field)
+                if old_val is not None:
+                    assert new_val is not None, (
+                        f"{row['candidate_id']}: {field} was '{old_val}' "
+                        f"in raw data but new normalizer returned None"
+                    )
+
+    def test_seniority_level_is_int_when_found(self):
+        profile = new_normalizer(_SAMPLE_ROWS[0])
+        assert isinstance(profile.professional.seniority_level, int)
+
+    def test_seniority_level_can_be_none(self):
+        profile = new_normalizer(_SAMPLE_ROWS[2])
+        assert profile.professional.seniority_level is None
+
+    def test_normalizer_never_crashes(self):
+        for row in _SAMPLE_ROWS:
+            profile = new_normalizer(row)
+            assert profile.profile_id is not None

tests/test_extraction/test_seniority.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+from src.extraction.seniority import extract_seniority
+
+
+class TestExtractSeniority:
+    def test_intern(self):
+        val, src = extract_seniority("Software Intern", None, [])
+        assert val == 0
+
+    def test_junior_title(self):
+        val, src = extract_seniority("Junior Engineer", None, [])
+        assert val == 1
+
+    def test_senior(self):
+        val, src = extract_seniority("Senior Engineer", None, [])
+        assert val == 3
+
+    def test_principal(self):
+        val, src = extract_seniority("Principal Architect", None, [])
+        assert val == 5
+
+    def test_cto(self):
+        val, src = extract_seniority("Chief Technology Officer", None, [])
+        assert val == 6
+
+    def test_years_fallback_junior(self):
+        val, src = extract_seniority(None, 1, [])
+        assert val == 1
+        assert src == "years_fallback"
+
+    def test_years_fallback_mid(self):
+        val, src = extract_seniority(None, 3, [])
+        assert val == 2
+        assert src == "years_fallback"
+
+    def test_years_fallback_senior(self):
+        val, src = extract_seniority(None, 6, [])
+        assert val == 3
+        assert src == "years_fallback"
+
+    def test_years_fallback_lead(self):
+        val, src = extract_seniority(None, 10, [])
+        assert val == 4
+        assert src == "years_fallback"
+
+    def test_years_fallback_principal(self):
+        val, src = extract_seniority(None, 15, [])
+        assert val == 5
+        assert src == "years_fallback"
+
+    def test_no_info_returns_none(self):
+        val, src = extract_seniority(None, None, [])
+        assert val is None
+        assert src == "not_found"
+
+    def test_domain_override_professor(self):
+        val, src = extract_seniority("Professor of Computer Science", None, [])
+        assert val == 6
+        assert src == "domain_override"
+
+    def test_prefers_title_over_years(self):
+        val, src = extract_seniority("Junior Dev", 10, [])
+        assert val == 1
+        assert src == "title_keyword"
+
+    def test_fallback_to_history_title(self):
+        history = [
+            {"title": "Senior Engineer", "start_date": "2020-01", "is_current": True},
+        ]
+        val, src = extract_seniority(None, None, history)
+        assert val == 3
+        assert src == "history_title"

tests/test_extraction/test_title.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from src.extraction.title import extract_title
+
+
+class TestExtractTitle:
+    def test_direct(self):
+        val, src = extract_title(
+            {"current_title": "Backend Engineer"}, [],
+        )
+        assert val == "Backend Engineer"
+        assert src == "direct"
+
+    def test_direct_empty_string_ignored(self):
+        val, src = extract_title(
+            {"current_title": ""},
+            [{"title": "Senior Dev", "start_date": "2020-01", "is_current": True}],
+        )
+        assert val == "Senior Dev"
+        assert src == "history"
+
+    def test_fallback_to_history(self):
+        val, src = extract_title(
+            {},
+            [{"title": "Senior Backend Engineer", "start_date": "2020-01", "is_current": True}],
+        )
+        assert val == "Senior Backend Engineer"
+        assert src == "history"
+
+    def test_fallback_to_headline_with_pipe(self):
+        val, src = extract_title(
+            {"headline": "Senior Python Backend | Acme Corp"}, [],
+        )
+        assert val is not None
+        assert src == "headline"
+
+    def test_headline_no_pipe(self):
+        val, src = extract_title(
+            {"headline": "Senior Software Engineer"}, [],
+        )
+        assert src == "headline"
+
+    def test_no_info_returns_none(self):
+        val, src = extract_title({}, [])
+        assert val is None
+        assert src == "not_found"
+
+    def test_prefers_direct_over_history(self):
+        val, src = extract_title(
+            {"current_title": "Lead Engineer"},
+            [{"title": "Junior Dev", "start_date": "2020-01", "is_current": True}],
+        )
+        assert val == "Lead Engineer"
+        assert src == "direct"

tests/test_ingestion/test_parser.py

@@ -189,7 +189,9 @@ def test_normalizer_redrob():
     assert profile.professional.current_company == "Flipkart"
     assert profile.personal.location.city == "Bangalore"
     assert profile.professional.industry == "IT Services"
-    assert profile.professional.total_experience_years == 5.0
+    assert profile.professional.total_experience_years is not None
+    assert profile.professional.total_experience_years != 5.0
+    assert isinstance(profile.professional.seniority_level, int)
     assert len(profile.skills) == 3
     assert profile.skills[0].name == "Python"
     assert profile.skills[0].confidence > 0.5