src/agents/executor.py

from __future__ import annotations

import asyncio
import logging
from concurrent.futures import ThreadPoolExecutor

from src.core.models import (
    MatchMetadata,
    MatchResult,
    ParsedQuery,
    Profile,
    SearchFilters,
    SearchMethod,
    Skill,
)
from src.core.profile_store import ProfileStore
from src.matching.behavioral_scorer import (
    compute_behavioral_score,
    compute_career_trajectory,
    compute_skill_proficiency,
    detect_honeypot,
)
from src.matching.scorer import CandidateScorer
from src.matching.skill_matcher import SkillMatcher
from src.search.filters import SearchFilter
from src.search.hybrid import HybridSearch
from src.search.reranker import CrossEncoderReranker


def _match_skills_detail(
    required_names: list[str], profile_skills: list[Skill],
    raw_text: str | None = None, subskills: dict[str, list[str]] | None = None,
) -> tuple[list[str], list[str]]:
    """Match skills using ONLY structured skills array.

    Returns (matched_names, missing_names) based on explicit skill entries.
    """
    matched: list[str] = []
    missing: list[str] = []
    _matcher = SkillMatcher(similarity_threshold=0.85)
    for rn in required_names:
        rn_subskills = subskills.get(rn) if subskills else None
        result = _matcher.find_best_match(rn, profile_skills, rn_subskills)
        if result is not None:
            matched.append(rn)
        else:
            missing.append(rn)
    return matched, missing

logger = logging.getLogger(__name__)


class ExecutorAgent:
    def __init__(
        self,
        hybrid_search: HybridSearch,
        reranker: CrossEncoderReranker,
        scorer: CandidateScorer,
        profiles: ProfileStore,
    ) -> None:
        self.hybrid_search = hybrid_search
        self.reranker = reranker
        self.scorer = scorer if scorer is not None else CandidateScorer()
        self.profile_store = profiles
        self._rerank_top_k = 20

    async def execute(
        self,
        parsed: ParsedQuery,
        top_k: int = 50,
        slider_weights: dict[str, float] | None = None,
        skip_reranker: bool = False,
    ) -> list[MatchResult]:
        search_text = self._query_to_search_text(parsed)

        # Check if there are active filters to decide retrieval size
        has_filters = False
        if parsed.location:
            if parsed.location.city and parsed.location.city.strip():
                has_filters = True
            if parsed.location.remote_ok:
                has_filters = True
        if parsed.experience:
            if parsed.experience.min_years is not None or parsed.experience.max_years is not None:
                has_filters = True
        if parsed.filters:
            if parsed.filters.exclude_companies or parsed.filters.include_companies:
                has_filters = True

        retrieval_k = max(1000, top_k * 10) if has_filters else top_k * 2

        query_vec = self.hybrid_search.embedder.embed_query(search_text)
        vector_raw = self.hybrid_search.vector_search.search(query_vec, top_k=retrieval_k)
        bm25_raw = self.hybrid_search.bm25_search.search(search_text, top_k=retrieval_k)

        hybrid_results = self.hybrid_search.reciprocal_rank_fusion(
            [vector_raw, bm25_raw], k=self.hybrid_search.rrf_k
        )

        vec_scores: dict[str, float] = {
            pid: self._norm_vec_score(s) for pid, s in vector_raw
        }
        bm25_scores: dict[str, float] = {
            pid: self._norm_bm25_score(s, bm25_raw) for pid, s in bm25_raw
        }

        fetch_top_k = max(self._rerank_top_k, top_k) if not skip_reranker else top_k * 2
        filtered = self._apply_filters(hybrid_results, parsed, limit=fetch_top_k)

        local_profile_cache: dict[str, Profile] = {}
        def _get_profile(pid: str) -> Profile | None:
            if pid in local_profile_cache:
                return local_profile_cache[pid]
            p = self.profile_store.get(pid)
            if p is not None:
                local_profile_cache[pid] = p
            return p

        pids_to_fetch = [pid for pid, _ in filtered[:fetch_top_k]]
        loop = asyncio.get_running_loop()
        num_fetch_workers = min(16, len(pids_to_fetch) or 1)
        with ThreadPoolExecutor(max_workers=num_fetch_workers) as fetch_pool:
            fetch_tasks = [
                loop.run_in_executor(fetch_pool, self.profile_store.get, pid)
                for pid in pids_to_fetch
            ]
            fetched_profiles = await asyncio.gather(*fetch_tasks)

        for pid, p in zip(pids_to_fetch, fetched_profiles):
            if p is not None:
                local_profile_cache[pid] = p

        if skip_reranker:
            candidate_scores = filtered[:top_k * 2]
            # Normalize RRF scores to [0, 1] so they work as cross_encoder_score dimension
            if candidate_scores:
                max_score = max(s for _, s in candidate_scores)
                if max_score > 0:
                    candidate_scores = [(pid, s / max_score) for pid, s in candidate_scores]
        else:
            rerank_candidates: list[tuple[str, str, float]] = []
            for pid, score in filtered[:fetch_top_k]:
                profile = local_profile_cache.get(pid)
                if profile is not None:
                    rerank_candidates.append((pid, profile.raw_text[:2000], score))
                else:
                    rerank_candidates.append((pid, "", score))
            candidate_scores = self.reranker.rerank(
                parsed.original_query or search_text, rerank_candidates, top_k=top_k
            )

        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

        scoring_args = []
        for pid, rerank_score in candidate_scores:
            profile = _get_profile(pid)
            if profile is None:
                continue
            scoring_args.append(
                (pid, rerank_score, profile, parsed, vec_scores, bm25_scores,
                 slider_weights, req_names, all_req, skip_reranker)
            )

        loop = asyncio.get_running_loop()
        num_workers = min(8, len(scoring_args) or 1)
        with ThreadPoolExecutor(max_workers=num_workers) as pool:
            tasks = [
                loop.run_in_executor(pool, self._score_single_candidate, *args)
                for args in scoring_args
            ]
            scored = await asyncio.gather(*tasks)
            results = [r for r in scored if r is not None]

        results.sort(key=lambda r: (-r.scores.overall, r.profile_id))
        for rank, r in enumerate(results, start=1):
            r.rank = rank

        return results

    @staticmethod
    def _norm_vec_score(score: float) -> float:
        return max(0.0, min(1.0, (score + 1.0) / 2.0))

    @staticmethod
    def _norm_bm25_score(score: float, all_results: list[tuple[str, float]]) -> float:
        if not all_results:
            return 0.0
        max_score = max(s for _, s in all_results)
        if max_score <= 0:
            return 0.0
        return max(0.0, min(1.0, score / max_score))

    @staticmethod
    def _prepare_scores_dict(
        pid: str,
        profile: Profile,
        vec_scores: dict[str, float],
        bm25_scores: dict[str, float],
        rerank_score: float | None,
        skill_overlap: float,
        exp_match: float,
        all_req: list[str],
    ) -> dict[str, float | None]:
        """Build the raw scores dict for a single candidate, before weighting."""
        honeypot_reason = detect_honeypot(profile)
        honeypot_penalty = 0.15 if honeypot_reason else 1.0

        return {
            "semantic_similarity": vec_scores.get(pid),
            "keyword_match": bm25_scores.get(pid),
            "skill_match": skill_overlap * honeypot_penalty,
            "experience_match": exp_match * honeypot_penalty,
            "location_match": None,
            "education_match": None,
            "cross_encoder_score": rerank_score * honeypot_penalty if rerank_score else None,
            "behavioral_score": compute_behavioral_score(profile.signals) * honeypot_penalty,
            "career_trajectory_score": compute_career_trajectory(profile) * honeypot_penalty,
            "skill_proficiency_score": (
                compute_skill_proficiency(profile, all_req) * honeypot_penalty
            ),
        }

    @staticmethod
    def _extract_candidate_info(profile: Profile, pid: str) -> tuple[str, str | None, str | None, str | None, float | None]:  # noqa: E501
        """Extract basic candidate display info from a Profile."""
        loc = profile.personal.location
        return (
            profile.personal.name if profile.personal else "",
            profile.professional.current_title if profile.professional else None,
            profile.professional.current_company if profile.professional else None,
            loc.city if profile.personal and loc else None,
            profile.professional.total_experience_years if profile.professional else None,
        )

    def _score_single_candidate(
        self,
        pid: str,
        rerank_score: float | None,
        profile: Profile,
        parsed: ParsedQuery,
        vec_scores: dict[str, float],
        bm25_scores: dict[str, float],
        slider_weights: dict[str, float] | None,
        req_names: list[str],
        all_req: list[str],
        skip_reranker: bool = False,
    ) -> MatchResult | None:
        subskills = parsed.subskills if hasattr(parsed, "subskills") else None
        matched_skills_list, missing_skills_list = _match_skills_detail(
            all_req, profile.skills, profile.raw_text, subskills,
        )
        skill_overlap = len(matched_skills_list) / len(all_req) if all_req else 1.0

        total_years = (
            profile.professional.total_experience_years
            if profile.professional and profile.professional.total_experience_years
            else 0
        )

        from src.matching.experience_matcher import ExperienceMatcher
        exp_matcher = ExperienceMatcher()
        years_match = exp_matcher.match(
            required_min_years=parsed.experience.min_years,
            required_max_years=parsed.experience.max_years,
            candidate_years=total_years,
        )
        title_match = exp_matcher.match_title(parsed.original_query or "", profile)

        exp_match = min(1.0, total_years / 10.0) * years_match

        scores_dict = ExecutorAgent._prepare_scores_dict(
            pid, profile, vec_scores, bm25_scores, rerank_score,
            skill_overlap, exp_match, all_req,
        )

        match_scores = self.scorer.compute_overall(scores_dict, slider_weights)
        match_scores.overall = max(0.0, min(1.0, match_scores.overall * title_match))

        req_only_matched, req_only_missing = _match_skills_detail(
            req_names, profile.skills, profile.raw_text, subskills,
        )

        name, title, company, city, exp_years = ExecutorAgent._extract_candidate_info(profile, pid)

        return MatchResult(
            query_id="",
            profile_id=pid,
            rank=0,
            name=name,
            current_title=title,
            current_company=company,
            location=city,
            experience_years=exp_years,
            scores=match_scores,
            matched_skills=list(set(req_only_matched)),
            missing_skills=list(set(req_only_missing)),
            metadata=MatchMetadata(search_method=SearchMethod.HYBRID, reranked=True),
        )

    def _query_to_search_text(self, parsed: ParsedQuery) -> str:
        parts: list[str] = []
        for rs in parsed.required_skills:
            parts.append(rs.name)
        for ps in parsed.preferred_skills:
            parts.append(ps.name)
        if parsed.location and parsed.location.city:
            parts.append(parsed.location.city)
        return " ".join(parts) if parts else "software engineer"

    def _apply_filters(
        self,
        results: list[tuple[str, float]],
        parsed: ParsedQuery,
        limit: int | None = None,
    ) -> list[tuple[str, float]]:
        filters = SearchFilters(
            location=parsed.location.city,
            min_experience_years=parsed.experience.min_years,
            max_experience_years=parsed.experience.max_years,
            remote_ok=parsed.location.remote_ok,
            exclude_companies=parsed.filters.exclude_companies,
            include_companies=parsed.filters.include_companies,
        )
        filter_obj = SearchFilter(filters)

        filtered: list[tuple[str, float]] = []
        for pid, score in results:
            profile = self.profile_store.get(pid)
            if profile is None:
                filtered.append((pid, score))
            elif filter_obj.passes(profile):
                filtered.append((pid, score))

            if limit is not None and len(filtered) >= limit:
                break

        return filtered