"""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 _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_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 "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, } 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…") sweep = await search.run_sweep(prefs, fresh=fresh) # 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, sweep["opportunities"]) await self.push("agent_thinking", message="Scout is reading your top roles…") curated = await asyncio.to_thread(_curate.curate, prefs, user_context, top) # Opus's call is final 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"))} await self.push("agent_data", action="search_complete", data=result) # Persist the feed so it survives navigation/refresh (replayed by handle_get_feed). from app.db.repo import save_feed clean_prefs = {k: v for k, v in prefs.items() if k not in ("user_context", "_fresh")} await save_feed(self.user_id, clean_prefs, 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")