394 lines
22 KiB
Python
394 lines
22 KiB
Python
"""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
|
|
|
|
from app.grow_events import emit_grow_event
|
|
|
|
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 _grow_activity_fields(self) -> dict:
|
|
"""Cheap cumulative activity context shared across Grow payloads — the funnel counts the
|
|
backend extractor maps to qscore signals (applications_submitted, opportunities_viewed).
|
|
Degrades to empty dict if the DB is off."""
|
|
from app.db.repo import get_activity_stats
|
|
try:
|
|
act = await get_activity_stats(self.user_id)
|
|
return {
|
|
"applications_submitted": act.get("applied", 0), # ← projector reads applications_submitted
|
|
"opportunities_viewed": act.get("viewed", 0),
|
|
"jobs_viewed": act.get("viewed", 0), # ← projector reads jobs_viewed
|
|
"opportunities_saved": act.get("saved", 0),
|
|
"searches": act.get("searches", 0),
|
|
"matches_generated": act.get("matches", 0),
|
|
"matchCount": act.get("matches", 0), # ← projector reads matchCount
|
|
}
|
|
except Exception: # noqa: BLE001 — never fatal
|
|
return {}
|
|
|
|
@staticmethod
|
|
def _match_scores(opps: list[dict]) -> list[int]:
|
|
"""Extract integer match scores from opportunities for source score fields."""
|
|
out = []
|
|
for o in opps:
|
|
s = (o.get("match") or {}).get("score") or o.get("matchScore")
|
|
if isinstance(s, (int, float)) and s > 0:
|
|
out.append(int(s))
|
|
return out
|
|
|
|
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})
|
|
feed_opps = (feed or {}).get("opportunities") or []
|
|
feed_scores = self._match_scores(feed_opps)
|
|
await emit_grow_event(
|
|
"matchmaking.feed.viewed",
|
|
self.user_id,
|
|
payload={
|
|
"action": "get_feed",
|
|
"match_count": len(feed_opps),
|
|
"cached": bool((feed or {}).get("cached", True)),
|
|
"source_scores": feed_scores,
|
|
"matchMin": min(feed_scores) if feed_scores else None,
|
|
"matchMax": max(feed_scores) if feed_scores else None,
|
|
"match_min": min(feed_scores) if feed_scores else None,
|
|
"match_max": max(feed_scores) if feed_scores else None,
|
|
**await self._grow_activity_fields(),
|
|
},
|
|
subject={"serviceId": "matchmaking-service", "externalId": "feed"},
|
|
dedupe_key=f"matchmaking:feed:{self.user_id}:{(feed or {}).get('updated_at') or 'none'}",
|
|
)
|
|
|
|
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):
|
|
"""ScoutPrefs → warm-pool gate (serve banked jobs; sweep Apify only when the pool dips) →
|
|
sift → Opus curate → cards."""
|
|
from app.engine import search
|
|
from app.config import get_settings
|
|
|
|
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 serving the pool
|
|
if not search.has_mvq(prefs):
|
|
await self.push("agent_data", action="search_complete",
|
|
data={"opportunities": [], "needs": ["title", "location"]})
|
|
return
|
|
from app.db import repo as _repo
|
|
s = get_settings()
|
|
sig = _repo.search_signature(prefs)
|
|
seen = await _repo.get_seen_ids(self.user_id)
|
|
|
|
# ── Warm-pool gate ──────────────────────────────────────────────────────────────────────────
|
|
# Bank surplus fetched jobs per (user, query); serve from the pool and re-pay Apify only when the
|
|
# fresh-unseen count dips below the floor. Reorder EARLY (background top-up) so it never runs thin.
|
|
pooled = await _repo.pool_load_fresh_unseen(self.user_id, sig, seen) if s.POOL_ENABLED else []
|
|
exhausted = await _repo.is_pool_exhausted(self.user_id, sig) if s.POOL_ENABLED else False
|
|
decision = search.pool_decision(len(pooled), fresh, s.POOL_SAFETY_FLOOR, s.POOL_REORDER_AT, exhausted)
|
|
|
|
if decision == "sweep":
|
|
# Pool below the hard floor (or forced): blocking sweep → bank ALL of it. No cursors — every
|
|
# fetch pulls the freshest page; the seen-net dedups + daily board refresh brings new jobs.
|
|
await self.push("agent_thinking",
|
|
message="Fetching fresh roles across job boards…" if fresh
|
|
else "Scanning live roles across job boards…")
|
|
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…"])
|
|
if sweep.get("all_boards_failed"):
|
|
await self.push(
|
|
"agent_error",
|
|
action="search_failed",
|
|
message="All enabled job boards are temporarily unavailable.",
|
|
data={
|
|
"code": "job_boards_unavailable",
|
|
"message": "All enabled job boards are temporarily unavailable.",
|
|
"failed_boards": sweep.get("failed_boards", 0),
|
|
},
|
|
)
|
|
return
|
|
new_count = await _repo.pool_save(self.user_id, sig, sweep["opportunities"])
|
|
# tapped out? a sweep that added almost nothing new → flag it so we stop block-fetching dupes.
|
|
if new_count < s.POOL_MIN_NEW_PER_REFILL:
|
|
await _repo.mark_pool_exhausted(self.user_id, sig)
|
|
pooled = await _repo.pool_load_fresh_unseen(self.user_id, sig, seen)
|
|
sources = sweep["sources"]
|
|
else:
|
|
# Pool healthy → serve from the shelf, zero Apify. (Cursors unchanged; refill, if any, is bg.)
|
|
await self.push("agent_thinking", message="Pulling your latest matches…")
|
|
sources = {"pool": len(pooled)}
|
|
|
|
candidates = pooled
|
|
# Engine: cheap rankers SIFT the pool → top ~22 (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": sources,
|
|
"shortlisted": len(curated), "scanned": dbg["scored"], "engine": engine,
|
|
"has_profile": bool(user_context and user_context.get("skills"))}
|
|
# Persist the deck (no cursors — pagination removed; freshness is date-sort + the seen-net).
|
|
clean_prefs = {k: v for k, v in prefs.items()
|
|
if k not in ("user_context", "_fresh", "_cursors")}
|
|
await _repo.save_feed(self.user_id, clean_prefs, result, query_sig=sig)
|
|
# 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)
|
|
# Grow event: matches were generated for this user → backend extracts qscore signals
|
|
gen_opps = result["opportunities"] or []
|
|
gen_scores = self._match_scores(gen_opps)
|
|
await emit_grow_event(
|
|
"matchmaking.matches.generated",
|
|
self.user_id,
|
|
payload={
|
|
"action": "run_search",
|
|
"matchCount": len(gen_opps), # ← projector reads matchCount/matches/shortlisted
|
|
"match_count": len(gen_opps), # snake_case alias
|
|
"scanned": result.get("scanned"),
|
|
"total_candidates": result.get("scanned"), # ← projector reads total_candidates for match_rate
|
|
"shortlisted": result.get("shortlisted"),
|
|
"sources": result.get("sources"),
|
|
"engine": result.get("engine"),
|
|
"source_scores": gen_scores,
|
|
"matchMin": min(gen_scores) if gen_scores else None, # ← scout_stats camelCase
|
|
"matchMax": max(gen_scores) if gen_scores else None,
|
|
"match_min": min(gen_scores) if gen_scores else None, # snake_case alias
|
|
"match_max": max(gen_scores) if gen_scores else None,
|
|
"top_match_score": max(gen_scores) if gen_scores else None,
|
|
**await self._grow_activity_fields(),
|
|
},
|
|
subject={"serviceId": "matchmaking-service", "externalId": sig},
|
|
correlation={"taskId": sig},
|
|
dedupe_key=f"matchmaking:generated:{self.user_id}:{sig}",
|
|
)
|
|
|
|
# The persisted feed intentionally drops resume context, but a refill must use the same
|
|
# retrieval inputs as the blocking sweep whose signature owns the pool. Preserve only
|
|
# the adapter's top-skill input; never retain the rest of the resume context.
|
|
if decision == "pool+refill":
|
|
refill_prefs = {**clean_prefs, "_skills": (user_context or {}).get("skills") or prefs.get("_skills") or []}
|
|
asyncio.create_task(self._background_refill(refill_prefs, sig))
|
|
|
|
async def _background_refill(self, prefs: dict, sig: str):
|
|
"""Proactive pool top-up: re-fetch the freshest jobs in the background + bank the new ones, so the
|
|
fresh-unseen count climbs back above the reorder mark before the user's next search. No cursors —
|
|
the seen-net dedups and daily board refresh supplies new jobs. Best-effort, non-fatal."""
|
|
from app.engine import search
|
|
from app.db import repo as _repo
|
|
try:
|
|
from app.config import get_settings
|
|
sweep = await search.run_sweep(prefs, fresh=True)
|
|
if sweep.get("all_boards_failed"):
|
|
logger.warning("pool refill skipped: all configured job boards failed, sig=%s", sig[:8])
|
|
return
|
|
new_count = await _repo.pool_save(self.user_id, sig, sweep["opportunities"])
|
|
if new_count < get_settings().POOL_MIN_NEW_PER_REFILL: # tapped out → relax the floor
|
|
await _repo.mark_pool_exhausted(self.user_id, sig)
|
|
logger.info("pool refill (bg): +%d NEW jobs, sig=%s", new_count, sig[:8])
|
|
except Exception as e: # noqa: BLE001
|
|
logger.warning("background refill failed (non-fatal): %s", e)
|
|
|
|
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"})
|
|
await emit_grow_event(
|
|
"matchmaking.match.applied",
|
|
self.user_id,
|
|
payload={
|
|
"opportunity_id": params.get("opportunity_id"),
|
|
"status": "applied",
|
|
**await self._grow_activity_fields(),
|
|
},
|
|
subject={"serviceId": "matchmaking-service", "externalId": params.get("opportunity_id", "")},
|
|
dedupe_key=f"matchmaking:applied:{self.user_id}:{params.get('opportunity_id', '')}",
|
|
)
|
|
|
|
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)
|
|
await emit_grow_event(
|
|
"matchmaking.match.viewed",
|
|
self.user_id,
|
|
payload={
|
|
"opportunity_id": params.get("opportunity_id"),
|
|
"action": "viewed",
|
|
**await self._grow_activity_fields(),
|
|
},
|
|
subject={"serviceId": "matchmaking-service", "externalId": params.get("opportunity_id", "")},
|
|
dedupe_key=f"matchmaking:viewed:{self.user_id}:{params.get('opportunity_id', '')}",
|
|
)
|
|
|
|
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)
|