matchmaking-v2/app/agent/session.py

"""Matchmaking-v2 agent session.

Mirrors the mesh contract: the orchestrator calls /a2a/tasks, which drives
on_session_start / on_user_action; handlers `push` messages back (agent_thinking,
agent_data{action,data}, agent_error) that the orchestrator forwards to the frontend.

Step-0 scaffold: the dispatch + contract are wired; handlers are honest stubs that
say "not implemented yet" so we can fill them in one full-stack slice at a time.
"""
from __future__ import annotations

import asyncio
import logging

logger = logging.getLogger(__name__)


class MatchmakingAgentSession:
    def __init__(self, websocket, user_id: str):
        self.ws = websocket
        self.user_id = user_id
        self.context: dict = {}

    async def push(self, msg_type: str, **kwargs):
        await self.ws.send_json({"type": msg_type, **kwargs})

    async def _with_heartbeat(self, awaitable, messages: list[str], interval: float = 5.0):
        """Await `awaitable` while emitting a rotating progress message every `interval`s.

        A live search runs ~30-40s (Apify scrape + Opus curation). Without a steady trickle of
        messages, the Redis pub/sub channel that relays the result back to the browser idles out and
        the final `search_complete` is dropped — which is why users were seeing 0 matches. The
        heartbeat keeps that channel warm AND gives the user real progress during the wait.
        """
        task = asyncio.ensure_future(awaitable)
        i = 0
        while True:
            done, _ = await asyncio.wait({task}, timeout=interval)
            if task in done:
                return task.result()
            await self.push("agent_thinking", message=messages[i % len(messages)])
            i += 1

    async def _stub(self, action: str, note: str):
        await self.push("agent_data", action=action, data={"status": "not_implemented", "note": note})

    async def on_session_start(self, params: dict):
        self.context["user_context"] = params.get("user_context")
        await self.push("agent_thinking", message="Scout is warming up…")
        # Real flow lands in a later slice: resolve prefs → run_search → emit feed.
        await self.push("agent_data", action="session_ready", data={"service": "matchmaking-v2", "version": "2.0.0"})

    async def on_user_action(self, action: str, params: dict):
        handlers = {
            # existing contract
            "sync_preferences": self.handle_sync_preferences,
            "get_feed": self.handle_get_feed,
            "get_scout_feed": self.handle_get_feed,   # unique name (avoids the course-service "get_feed" collision)
            "record_feedback": self.handle_record_feedback,
            "get_opportunity_detail": self.handle_get_opportunity_detail,
            # new in v2
            "generate_matches": self.handle_run_search,
            "run_search": self.handle_run_search,
            "suggest_bubbles": self.handle_suggest_bubbles,   # the Fine-tune "pathway finder" engine
            "tailor_resume": self.handle_tailor_resume,
            "submit_application": self.handle_submit_application,
            "get_apply_proof": self.handle_get_apply_proof,
            # per-opportunity state — persists across deck refreshes (applied / dismissed / saved docs)
            "dismiss_opportunity": self.handle_dismiss_opportunity,
            "restore_opportunity": self.handle_restore_opportunity,
            "mark_applied": self.handle_mark_applied,
            "save_apply_docs": self.handle_save_apply_docs,
            # activity tracking → the dashboard funnel + engagement rank
            "mark_viewed": self.handle_mark_viewed,
            "mark_saved": self.handle_mark_saved,
            "get_scout_stats": self.handle_get_scout_stats,
        }
        handler = handlers.get(action)
        if not handler:
            await self.push("agent_error", message=f"Unknown action: {action}")
            return
        try:
            await handler(params)
        except Exception as e:  # noqa: BLE001 — surface to the client, log the trace
            logger.error("action '%s' failed: %s", action, e, exc_info=True)
            await self.push("agent_error", message=f"Failed to {action}: {e}")

    # ── handlers (stubs — replaced one full-stack slice at a time) ──
    async def handle_sync_preferences(self, params: dict):
        await self._stub("preferences_synced", "persist ScoutPrefs → label store")

    async def handle_get_feed(self, params: dict):
        """Replay the user's last persisted feed so matches survive navigation (no re-search)."""
        from app.db.repo import get_feed
        feed = await get_feed(self.user_id)
        await self.push("agent_data", action="feed_loaded", data=feed or {"opportunities": [], "cached": False})

    async def handle_record_feedback(self, params: dict):
        await self._stub("feedback_recorded", "store SAVE/DISMISS/APPLY label")

    async def handle_get_opportunity_detail(self, params: dict):
        await self._stub("opportunity_detail", "return match-score breakdown")

    async def handle_run_search(self, params: dict):
        """Slice-1a: ScoutPrefs → live multi-board fetch → cards. No engine, no user-arm."""
        from app.engine import search

        prefs = params  # the frontend's ScoutPrefs arrive as params (user_context, if present, is ignored)
        fresh = bool(prefs.get("_fresh"))  # dev toggle: force a live sweep instead of replaying cache
        if not search.has_mvq(prefs):
            await self.push("agent_data", action="search_complete",
                            data={"opportunities": [], "needs": ["title", "location"]})
            return
        await self.push("agent_thinking",
                        message="Fetching fresh roles across job boards…" if fresh
                        else "Scanning live roles across job boards…")
        # Per-board search cursors: re-running the SAME query advances EACH board's own page so every
        # run fetches GENUINELY NEW jobs. Resets when the query changes OR >24h (boards refresh ~daily).
        # Dedup is the per-user seen-net below — never cross-user, no opaque actor-side state.
        from app.db import repo as _repo
        sig = _repo.search_signature(prefs)
        cursors = await _repo.get_search_cursors(self.user_id, sig)
        prefs["_cursors"] = cursors
        # Heartbeat through the long Apify scrape so the relay channel stays alive (see _with_heartbeat).
        sweep = await self._with_heartbeat(
            search.run_sweep(prefs, fresh=fresh),
            ["Scanning live roles across the boards…", "Pulling the latest Naukri + LinkedIn + Foundit roles…",
             "Gathering fresh postings for you…"])
        # Safety net: drop anything already shown / applied / dismissed (covers boards that can't
        # paginate, so the same posting never reappears even if a board re-returns it).
        seen = await _repo.get_seen_ids(self.user_id)
        candidates = [o for o in sweep["opportunities"] if o.get("id") not in seen]
        # Engine: assemble FULL context (profile + prefs). Cheap rankers SIFT 90 → top ~18 (white-box ‖
        # embedding vibe); then Opus reads only those and curates the honest shortlist + report cards.
        from app.engine import curate as _curate
        from app.engine import rank as _rank
        from app.engine import sift as _sift
        user_context = params.get("user_context")   # resume skills / experience / education / QScore
        await self.push("agent_thinking", message="Scoring roles against your profile…")
        # Run the blocking LLM work (embeddings + Opus) OFF the event loop, or the long sync OpenAI
        # call freezes the loop and the Redis response can't publish (→ the loader hangs forever).
        top, dbg = await asyncio.to_thread(_sift.sift, prefs, user_context, candidates)
        # Heartbeat through the long Opus curation (the other 30-40s gap that was dropping responses).
        curated = await self._with_heartbeat(
            asyncio.to_thread(_curate.curate, prefs, user_context, top),   # Opus's call is final
            ["Scout is reading your top roles…", "Scoring fit + writing your report cards…",
             "Ranking your strongest matches…", "Almost there…"])
        if curated is None:                                       # safety net: sift + templated cards
            curated = _rank.select(top, prefs.get("stretch", "balanced"))
            engine = f"fallback:{dbg['mode']}"
        else:
            engine = "opus"
        result = {"opportunities": curated, "sources": sweep["sources"],
                  "shortlisted": len(curated), "scanned": dbg["scored"], "engine": engine,
                  "has_profile": bool(user_context and user_context.get("skills"))}
        # Persist the RAW deck + ADVANCE every board that ran (its page +1) so the next run of this
        # query fetches new jobs. Per-opportunity state (applied/dismissed/docs) lives separately.
        clean_prefs = {k: v for k, v in prefs.items()
                       if k not in ("user_context", "_fresh", "_cursors")}
        # Store the LAST page fetched per paginating board (last + 1 = the page we just pulled). So a
        # NEW query's first search stores 1 (page 1), not 2. Naukri/Foundit have no page param → no
        # cursor (they dedup via the per-user seen-net).
        next_cursors = {b: int(cursors.get(b, 0)) + 1
                        for b in (sweep.get("sources") or {}) if b in search.PAGINATING_BOARDS}
        await _repo.save_feed(self.user_id, clean_prefs, result, query_sig=sig, cursors=next_cursors)
        # Accumulate the salary band — each deck contributes its peak salary, averaged over decks
        # (survives decks that disclose nothing, like Naukri "Not disclosed").
        await _repo.update_salary_band(self.user_id, [o.get("salary_lpa") for o in (result["opportunities"] or [])])
        # Mark the shown deck as SEEN so it never reappears in a future run.
        await _repo.mark_seen(self.user_id, [o.get("id") for o in (result["opportunities"] or []) if o.get("id")])
        # Hand the UI a state-aware view: dismissed dropped, applied + saved-doc ids annotated.
        states = await _repo.get_opportunity_states(self.user_id)
        result["opportunities"] = _repo.annotate_and_filter(result["opportunities"], states)
        result["shortlisted"] = len(result["opportunities"])
        await self.push("agent_data", action="search_complete", data=result)

    async def handle_suggest_bubbles(self, params: dict):
        """Fine-tune bubbling — stage-aware (broad→narrow→role) bubbles from the picks so far."""
        from app.engine import suggest
        stage = params.get("stage", "role")
        items = await asyncio.to_thread(
            suggest.suggest, stage, params.get("picks") or {}, params.get("profile") or {}
        )
        await self.push("agent_data", action="bubbles_suggested",
                        data={"stage": stage, "items": items or []})

    async def handle_tailor_resume(self, params: dict):
        await self._stub("resume_tailored", "resume-builder: tailor resume to the chosen role")

    async def handle_submit_application(self, params: dict):
        await self._stub("application_submitted", "master-key auto-apply (tier-aware)")

    async def handle_get_apply_proof(self, params: dict):
        await self._stub("apply_proof", "return captured confirmation screenshots")

    # ── per-opportunity state (persists across deck refreshes) ──
    async def handle_dismiss_opportunity(self, params: dict):
        """Mark a posting stale/dismissed → it never resurfaces in the deck."""
        from app.db.repo import set_opportunity_state
        await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), status="dismissed")
        await self.push("agent_data", action="opportunity_state_saved",
                        data={"opportunity_id": params.get("opportunity_id"), "status": "dismissed"})

    async def handle_restore_opportunity(self, params: dict):
        """Undo a dismiss/pass → clear the status so the role returns to the deck."""
        from app.db.repo import set_opportunity_state
        await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), status=None)
        await self.push("agent_data", action="opportunity_state_saved",
                        data={"opportunity_id": params.get("opportunity_id"), "status": None})

    async def handle_mark_applied(self, params: dict):
        """Record that the user applied to a role (sticks across reloads)."""
        from app.db.repo import set_opportunity_state
        await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), status="applied")
        await self.push("agent_data", action="opportunity_state_saved",
                        data={"opportunity_id": params.get("opportunity_id"), "status": "applied"})

    async def handle_save_apply_docs(self, params: dict):
        """Remember the resume-builder document ids generated for a role so we never re-pay to
        re-tailor / re-generate them. Only the provided ids are written."""
        from app.db.repo import set_opportunity_state
        fields = {k: params[k] for k in ("tailored_resume_id", "tailored_version_id", "cover_letter_id")
                  if params.get(k)}
        await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), **fields)
        await self.push("agent_data", action="opportunity_state_saved",
                        data={"opportunity_id": params.get("opportunity_id"), **fields})

    # ── activity tracking (dashboard funnel + engagement rank) ──
    async def handle_mark_viewed(self, params: dict):
        """The user opened a job → funnel 'Viewed'. Fire-and-forget (no response needed)."""
        from app.db.repo import set_opportunity_state
        await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), viewed=True)

    async def handle_mark_saved(self, params: dict):
        """The user saved/swiped-right a job → funnel 'Shortlisted'."""
        from app.db.repo import set_opportunity_state
        await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), saved=True)

    async def handle_get_scout_stats(self, params: dict):
        """Assemble the REAL Summary dashboard metrics (Phase 2 fills the cross-service bits)."""
        from app.engine import stats
        data = await stats.build_scout_stats(
            self.user_id,
            user_uuid=params.get("user_uuid"),
        )
        await self.push("agent_data", action="scout_stats", data=data)