From 3969d62a51791f96f8083cbf320e5c58ef7b7fdb Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Wed, 9 Sep 2026 15:52:46 +0500 Subject: [PATCH] sumkmary check udpate --- backend/README.md | 6 ++ backend/g_sheet/models.py | 5 +- backend/g_sheet/scoring.py | 17 ++-- backend/inbox/models.py | 5 +- backend/inbox/views.py | 24 ++++++ backend/job/candidate/views.py | 69 ++++++++++++++- backend/main.py | 5 ++ backend/summary_gate/agent_setup.py | 120 ++++++++++++++++++++++++++ backend/summary_gate/execute_agent.py | 77 +++++++++++++++++ backend/summary_gate/models.py | 26 ++++++ backend/summary_gate/plugins.py | 80 +++++++++++++++++ backend/summary_gate/prompt.py | 75 ++++++++++++++++ 12 files changed, 498 insertions(+), 11 deletions(-) create mode 100644 backend/summary_gate/agent_setup.py create mode 100644 backend/summary_gate/execute_agent.py create mode 100644 backend/summary_gate/models.py create mode 100644 backend/summary_gate/plugins.py create mode 100644 backend/summary_gate/prompt.py diff --git a/backend/README.md b/backend/README.md index d238b48..f8dc536 100644 --- a/backend/README.md +++ b/backend/README.md @@ -733,6 +733,12 @@ and appended to `ats_results` as a supersede-chained history — see job: OpenAI prompt caching keys on an exact prefix match, so one volatile byte (an id, a timestamp) would stop the whole batch reusing the cached JD prefix. +When the candidate already has `professional_summary`, `summary_gate` runs first: it +asks whether that stack/department could plausibly fit the job post. A no skips PDF +load and ATS (`SUMMARY_NOT_SUITABLE`). No summary always continues, so the first score +can write one. `SUMMARY_GATE_MIN_CONFIDENCE` is the gradient threshold. Disable with +`SUMMARY_GATE_ENABLED=false`. + Résumé text is run through `normalize_spaced_text` **before** scoring, so that keyword verification sees exactly the text the model saw. Designer-made CVs position every glyph individually and `pypdf` returns `S K I L L S`; the `despace_line` decorator rebuilds those. diff --git a/backend/g_sheet/models.py b/backend/g_sheet/models.py index a470a12..83dffa1 100644 --- a/backend/g_sheet/models.py +++ b/backend/g_sheet/models.py @@ -405,15 +405,16 @@ class FormData(SQLModel, table=True): @classmethod async def list_on_hold_scan_rows(cls, session: AsyncSession, sheet=None): """On-Hold Sheet Forms: id + email. Entire catalogue, optional sheet tab.""" - statement = select(cls.id, cls.candidate_email) + statement = select(cls.id, cls.candidate_email, cls.professional_summary) for clause in cls._filters(sheet=sheet, no_suggestions=True): statement = statement.where(clause) result = await session.execute(statement) rows = [] - for record_id, email in result.all(): + for record_id, email, summary in result.all(): rows.append({ "id": record_id, "email": (email or "").strip().lower() or None, + "professional_summary": (summary or "").strip() or None, }) return rows diff --git a/backend/g_sheet/scoring.py b/backend/g_sheet/scoring.py index e7d49fd..d61a416 100644 --- a/backend/g_sheet/scoring.py +++ b/backend/g_sheet/scoring.py @@ -112,19 +112,24 @@ async def score_form_against_job(form_data_id: str, job_id: str, rescan_run_id=N form_row, job = await FormData.get_with_job(session, form_data_id, job_id) if form_row is None or job is None: return {"status": "skipped", "reason": "no_join"} - text = resume_text_from_extracted(form_row.extracted_data) - if not text: - return {"status": "skipped", "reason": "no_extract"} - filename = (form_row.extracted_data or {}).get("filename") or "resume.pdf" - page_count = int((form_row.extracted_data or {}).get("page_count") or 1) - truncated = bool((form_row.extracted_data or {}).get("truncated")) settings = get_scoring_settings() jd = build_job_description(job) if len(jd) > settings.max_jd_chars: return {"status": "skipped", "reason": "jd_too_large"} + stored_summary = (form_row.professional_summary or "").strip() or None + text = resume_text_from_extracted(form_row.extracted_data) + filename = (form_row.extracted_data or {}).get("filename") or "resume.pdf" + page_count = int((form_row.extracted_data or {}).get("page_count") or 1) + truncated = bool((form_row.extracted_data or {}).get("truncated")) form_pk = form_row.id job_pk = job.id + from summary_gate.execute_agent import allow_ats + if not await allow_ats(stored_summary, jd): + return {"status": "skipped", "reason": "not_suitable"} + if not text: + return {"status": "skipped", "reason": "no_extract"} + resume = ExtractedResume( filename=str(filename), candidate_id=str(form_pk), diff --git a/backend/inbox/models.py b/backend/inbox/models.py index 5d01fe8..91101f3 100644 --- a/backend/inbox/models.py +++ b/backend/inbox/models.py @@ -1143,16 +1143,17 @@ class Inbox_Messages(SQLModel, table=True): async def list_on_hold_scan_rows(cls, session: AsyncSession): """On-Hold email applications: id + sender + CV path. No TOAST columns.""" statement = cls._apply_filters( - select(cls.id, cls.message_from, cls.file_path), + select(cls.id, cls.message_from, cls.file_path, cls.professional_summary), no_suggestions=True, ) result = await session.execute(statement) rows = [] - for record_id, email, file_path in result.all(): + for record_id, email, file_path, summary in result.all(): rows.append({ "id": record_id, "email": (email or "").strip().lower() or None, "file_path": (file_path or "").strip() or None, + "professional_summary": (summary or "").strip() or None, }) return rows diff --git a/backend/inbox/views.py b/backend/inbox/views.py index efa551a..e2aea6a 100644 --- a/backend/inbox/views.py +++ b/backend/inbox/views.py @@ -485,14 +485,32 @@ class Email: async def plan_on_hold_pairs(self,channel,sheet=None): """Build (candidate, job) pairs that have never been ATS-scored. + When professional_summary is present, the summary-vs-job gradient runs + before the already-scored pair skip: an obvious mismatch never reaches ATS. Skip a candidate who already has a score against an active job. Skip a pair that already exists on ats_results for that person/row. """ from g_sheet.models import FormData + from job.candidate.plugins import build_job_description from job.job_post.models import JobPosts + from summary_gate.execute_agent import allow_ats + from summary_gate.plugins import GATE_ENABLED job_ids=[str(jid) for jid in await JobPosts.list_ids(self.session)] active_ids={str(jid) for jid in await JobPosts.list_ids(self.session,active_only=True)} + jd_by_id={} + if GATE_ENABLED: + jobs=await JobPosts.get_by_ids(self.session,job_ids,active_only=False) + jd_by_id={str(j.id):build_job_description(j) for j in jobs} + + async def _unsuitable(summary, job_id) -> bool: + text=(summary or "").strip() + if not text or not GATE_ENABLED: + return False + jd=jd_by_id.get(str(job_id)) + if not jd: + return False + return not await allow_ats(text,jd) inbox_rows=[] form_rows=[] if channel in ("all","email"): @@ -553,6 +571,9 @@ class Email: continue known=_already(email,row_jobs) for job_id in job_ids: + if await _unsuitable(row.get("professional_summary"),job_id): + skipped_pairs += 1 + continue if job_id in known: skipped_pairs += 1 continue @@ -569,6 +590,9 @@ class Email: continue known=_already(email,row_jobs) for job_id in job_ids: + if await _unsuitable(row.get("professional_summary"),job_id): + skipped_pairs += 1 + continue if job_id in known: skipped_pairs += 1 continue diff --git a/backend/job/candidate/views.py b/backend/job/candidate/views.py index 0106187..e000ede 100644 --- a/backend/job/candidate/views.py +++ b/backend/job/candidate/views.py @@ -609,6 +609,38 @@ class CandidateScoring: def __init__(self,session:AsyncSession): self.session=session + async def _professional_summary_for_email(self,email,stored=None): + """Stored summary wins; Users.professional_summary is the identity fallback.""" + text=(stored or "").strip() or None + if text: + return text + cleaned=(email or "").strip().lower() + if not cleaned: + return None + user=await Users.get_user_by_email(self.session,cleaned) + if user is None: + return None + return (user.professional_summary or "").strip() or None + + async def _gate_source(self,source,jd): + """Suitability gradient: summary vs job post, before PDF load / ATS. + + No summary, or the gate disabled: leave the source alone so the first + score can write professional_summary. A mismatch sets precheck and + drops bytes so later extract / score_batch never run. + """ + from summary_gate.execute_agent import allow_ats + from summary_gate.plugins import SUMMARY_NOT_SUITABLE + + if source.get("precheck") is not None: + return + summary=(source.get("professional_summary") or "").strip() + if not summary: + return + if not await allow_ats(summary,jd): + source["precheck"]=(SUMMARY_NOT_SUITABLE,"Professional summary does not fit this job.") + source["data"]=None + async def score_uploads(self,job_id,files,current_user): """files: list of (filename, bytes) pairs from the route handler.""" settings=get_scoring_settings() @@ -636,6 +668,10 @@ class CandidateScoring: """Score PDF attachments of inbox messages (S3 URLs or legacy local paths).""" from inbox.plugins import load_file_bytes + job=await JobPosts.get_job_post_by_id(self.session,job_id) + if job is None or job.is_deleted: + raise HTTPException(status_code=404,detail="Job post not found") + jd=build_job_description(job) sources=[] for mid in message_ids: row=await Inbox_Messages.get_inbox_message_by_id(self.session,mid) @@ -652,8 +688,16 @@ class CandidateScoring: "file_path":path_str, "inbox_message_id":row.id, "candidate_email":(row.message_from or "").strip().lower() or None, + "professional_summary":None, "precheck":None, } + source["professional_summary"]=await self._professional_summary_for_email( + source["candidate_email"],row.professional_summary, + ) + await self._gate_source(source,jd) + if source["precheck"] is not None: + sources.append(source) + continue lower=name.lower() if lower.endswith(".doc") or lower.endswith(".docx"): source["precheck"]=(ErrorCode.UNSUPPORTED_FILE_TYPE,"DOC/DOCX extraction is not supported yet.") @@ -697,6 +741,10 @@ class CandidateScoring: status_code=413, detail=f"At most {settings.max_resumes_per_request} CVs per request", ) + job=await JobPosts.get_job_post_by_id(self.session,job_id) + if job is None or job.is_deleted: + raise HTTPException(status_code=404,detail="Job post not found") + jd=build_job_description(job) sources=[] for record_id in ids: row=await Manual_UPLOAD_CANDIDATE.get_by_id(self.session,record_id) @@ -709,8 +757,16 @@ class CandidateScoring: "file_path":(row.file_path or "").strip() or None, "candidate_email":(row.candidate_email or "").strip().lower() or None, "manual_upload_candidate_id":row.id, + "professional_summary":None, "precheck":None, } + source["professional_summary"]=await self._professional_summary_for_email( + source["candidate_email"],row.professional_summary, + ) + await self._gate_source(source,jd) + if source["precheck"] is not None: + sources.append(source) + continue file_row=await CvBankFiles.get(self.session,row.id) data=file_row.data if file_row and file_row.data else None if data is None and source["file_path"]: @@ -755,6 +811,13 @@ class CandidateScoring: jd=build_job_description(job) if len(jd)>settings.max_jd_chars: raise HTTPException(status_code=422,detail="The job post is too large to score against") + from summary_gate.plugins import SUMMARY_NOT_SUITABLE + for source in sources: + if not source.get("professional_summary"): + source["professional_summary"]=await self._professional_summary_for_email( + source.get("candidate_email"), + ) + await self._gate_source(source,jd) fields_by_slot=await self._score_sources(sources,jd,settings) # Prefer the Manual S3 URL for this email+job when scoring from a raw upload # (Add Candidate scores right after create — same link as manual_upload_candidate). @@ -776,8 +839,11 @@ class CandidateScoring: "model":settings.openai_model, } rows=[] + kept_sources=[] for slot in range(len(sources)): fields={**fields_by_slot[slot],**common} + if fields.get("error_code")==SUMMARY_NOT_SUITABLE and rescan_run_id: + continue email=(fields.get("candidate_email") or "").strip().lower() if email and not fields.get("linkedin_url"): user=await Users.get_user_by_email(self.session,email) @@ -790,7 +856,8 @@ class CandidateScoring: ): await self.session.commit() rows.append(row) - await self._sync_ats_results(source_kind,job,rows,sources,current_user,rescan_run_id=rescan_run_id) + kept_sources.append(sources[slot]) + await self._sync_ats_results(source_kind,job,rows,kept_sources,current_user,rescan_run_id=rescan_run_id) rows.sort(key=lambda r:(0,-(r.match_score or 0)) if r.status=="completed" else (1,0)) return [serialize_candidate(row) for row in rows] diff --git a/backend/main.py b/backend/main.py index d630463..56f4348 100644 --- a/backend/main.py +++ b/backend/main.py @@ -86,6 +86,11 @@ async def lifespan(app): close_classifier() except Exception as exc: logger.warning("classifier close skipped: %s",exc) + try: + from summary_gate.agent_setup import close_gate + close_gate() + except Exception as exc: + logger.warning("summary gate close skipped: %s",exc) if llm_ready and close_llm is not None: await close_llm() if sheet_broker_ready and sheet_broker is not None: diff --git a/backend/summary_gate/agent_setup.py b/backend/summary_gate/agent_setup.py new file mode 100644 index 0000000..7c9fb7a --- /dev/null +++ b/backend/summary_gate/agent_setup.py @@ -0,0 +1,120 @@ +"""Summary-gate adapter and its process-wide instance. + +Pure module: no FastAPI imports and no HTTPException. + +Mirrors inbox_classifier/agent_setup.py: responses.parse into a Pydantic +verdict, delivery-status branching, shared AsyncOpenAI client. +""" + +from __future__ import annotations + +import logging + +from openai import AsyncOpenAI +from summary_gate.models import SummarySuitabilityVerdict +from summary_gate.plugins import PROMPT_CACHE_KEY, get_gate_settings +from summary_gate.prompt import SYSTEM_PROMPT, build_input + +from app.core.config import supports_reasoning +from app.core.errors import ModelRefusedError, ModelResponseInvalidError, ModelUnavailableError + +logger=logging.getLogger("summary.gate") + +_TRUNCATED="max_output_tokens" +_FILTERED="content_filter" + + +def _first_refusal(response): + for item in getattr(response,"output",None) or []: + for part in getattr(item,"content",None) or []: + if getattr(part,"type",None)=="refusal": + refusal=getattr(part,"refusal",None) + return str(refusal) if refusal else "refused" + return None + + +class SummaryGate: + def __init__(self, client:AsyncOpenAI, model, max_output_tokens, effort, enable_cache=True): + self._client=client + self._model=model + self._max_output_tokens=max_output_tokens + self._effort=effort + self._enable_cache=enable_cache + self._supports_reasoning=supports_reasoning(model) + + async def classify(self, job_description, summary) -> SummarySuitabilityVerdict: + kwargs={ + "model":self._model, + "instructions":SYSTEM_PROMPT, + "input":build_input(job_description,summary), + "text_format":SummarySuitabilityVerdict, + "max_output_tokens":self._max_output_tokens, + } + if self._supports_reasoning: + kwargs["reasoning"]={"effort":self._effort} + if self._enable_cache: + kwargs["prompt_cache_key"]=PROMPT_CACHE_KEY + + response=await self._client.responses.parse(**kwargs) + + status=getattr(response,"status",None) + self._log_usage(response,status) + + if status=="failed": + raise ModelUnavailableError("provider reported a failed response") + + if status=="incomplete": + reason=getattr(getattr(response,"incomplete_details",None),"reason",None) + if reason==_FILTERED: + raise ModelRefusedError("content filter blocked the response") + if reason==_TRUNCATED: + raise ModelResponseInvalidError("response truncated at max_output_tokens") + raise ModelResponseInvalidError(f"incomplete response: {reason}") + + if _first_refusal(response) is not None: + raise ModelRefusedError("model declined to classify this summary") + + parsed=getattr(response,"output_parsed",None) + if not isinstance(parsed,SummarySuitabilityVerdict): + raise ModelResponseInvalidError("response did not parse into SummarySuitabilityVerdict") + return parsed + + def _log_usage(self, response, status): + usage=getattr(response,"usage",None) + input_details=getattr(usage,"input_tokens_details",None) + output_details=getattr(usage,"output_tokens_details",None) + logger.info( + "summary gate: model=%s status=%s request_id=%s in=%s out=%s cached=%s reasoning=%s", + self._model, + status, + getattr(response,"id",None), + getattr(usage,"input_tokens",None), + getattr(usage,"output_tokens",None), + getattr(input_details,"cached_tokens",None), + getattr(output_details,"reasoning_tokens",None), + ) + + +_gate=None + + +def get_gate() -> SummaryGate: + global _gate + if _gate is None: + from llm_setup import get_client + + settings=get_gate_settings() + _gate=SummaryGate( + get_client(), + model=settings.openai_model, + max_output_tokens=settings.openai_max_output_tokens, + effort=settings.openai_effort, + enable_cache=settings.openai_enable_prompt_cache, + ) + return _gate + + +def close_gate(): + global _gate + _gate=None + logger.info("summary gate closed") diff --git a/backend/summary_gate/execute_agent.py b/backend/summary_gate/execute_agent.py new file mode 100644 index 0000000..c589a39 --- /dev/null +++ b/backend/summary_gate/execute_agent.py @@ -0,0 +1,77 @@ +"""Summary-gate entrypoint — one Responses call per (summary, job) pair. + +Pure module: no FastAPI imports and no HTTPException. +Never raises: a provider outage is a policy decision (SUMMARY_GATE_FAIL_OPEN). +""" + +from __future__ import annotations + +import logging + +from summary_gate.agent_setup import get_gate +from summary_gate.plugins import ( + GATE_ENABLED, + GATE_MAX_JD_CHARS, + GATE_MAX_SUMMARY_CHARS, + clip_text, + should_score, +) + +from app.core.errors import ATSError, classify_error + +logger=logging.getLogger("summary.gate") + +EMPTY_INPUT="empty_input" +_allow_cache: dict[tuple[str,str],bool]={} + + +async def classify_summary(summary, job_description) -> tuple: + """Judge one summary against one JD. Never raises. + + (verdict, "") on success; (None, error_code) when the model could not be + consulted or returned something unusable. + """ + text=clip_text(summary,GATE_MAX_SUMMARY_CHARS).strip() + jd=clip_text(job_description,GATE_MAX_JD_CHARS).strip() + if not text or not jd: + return None,EMPTY_INPUT + + try: + gate=get_gate() + verdict=await gate.classify(jd,text) + logger.info( + "summary gate: suitable=%s confidence=%s", + verdict.is_suitable, + verdict.confidence, + ) + return verdict,"" + except ATSError as e: + logger.warning("summary gate failed: code=%s",e.error_code) + return None,e.error_code + except Exception as e: + code,_=classify_error(e) + logger.warning("summary gate failed: code=%s exc=%s",code,type(e).__name__) + return None,code + + +async def allow_ats(summary, job_description) -> bool: + """True when full ATS should run. + + No summary, or the gate disabled: pass through so the first score can + write professional_summary. A present summary is the gradient: only a + suitable verdict (above SUMMARY_GATE_MIN_CONFIDENCE) continues. + """ + if not GATE_ENABLED: + return True + text=(summary or "").strip() + if not text: + return True + jd=(job_description or "").strip() + cache_key=(text,jd) + cached=_allow_cache.get(cache_key) + if cached is not None: + return cached + verdict,error=await classify_summary(text,jd) + proceed,_,_=should_score(verdict,error) + _allow_cache[cache_key]=proceed + return proceed diff --git a/backend/summary_gate/models.py b/backend/summary_gate/models.py new file mode 100644 index 0000000..aa9caf7 --- /dev/null +++ b/backend/summary_gate/models.py @@ -0,0 +1,26 @@ +"""The suitability verdict exchanged with the summary-vs-job-post gate. + +Pure module: no FastAPI imports and no HTTPException. + +``extra="forbid"`` is load-bearing — it emits ``additionalProperties: false``, which +structured outputs requires (same reason as inbox_classifier/models.py). +""" + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field + + +class SummarySuitabilityVerdict(BaseModel): + """One coarse decision: is a full CV-vs-JD ATS score worth running? + + ``confidence`` is the gradient. The boolean is the recruiter-shaped answer; + doubt belongs here so SUMMARY_GATE_MIN_CONFIDENCE can raise the bar without + changing the prompt. + """ + + model_config = ConfigDict(extra="forbid") + + is_suitable: bool + confidence: float = Field(ge=0.0, le=1.0) + evidence: str = Field(min_length=1, max_length=200) diff --git a/backend/summary_gate/plugins.py b/backend/summary_gate/plugins.py new file mode 100644 index 0000000..03cb8e0 --- /dev/null +++ b/backend/summary_gate/plugins.py @@ -0,0 +1,80 @@ +"""Summary-gate configuration and the fail policy. + +Pure module: no FastAPI imports and no HTTPException. + +Non-DB config is module-level load_dotenv() + os.getenv (house style). Model / +token / effort / cache knobs come from the bulk-ats Settings, same as +inbox_classifier.plugins.get_triage_settings. +""" + +from __future__ import annotations + +import os + +from dotenv import load_dotenv +from summary_gate.prompt import PROMPT_VERSION + +from app.core.config import Settings, get_settings + +load_dotenv() + + +def _flag(name, default) -> bool: + raw=(os.getenv(name) or "").strip().lower() + if not raw: + return default + return raw in ("1","true","yes","on") + + +# false restores pre-gate behaviour: every pair with a CV is ATS-scored. +GATE_ENABLED=_flag("SUMMARY_GATE_ENABLED",True) + +# true: a provider outage or a missing key lets the pair through to ATS. +# Fail-closed would silently skip scoring when OPENAI_API_KEY is unset. +GATE_FAIL_OPEN=_flag("SUMMARY_GATE_FAIL_OPEN",True) + +GATE_MAX_SUMMARY_CHARS=max(int(os.getenv("SUMMARY_GATE_MAX_SUMMARY_CHARS") or 500),1) +# Same bound as the inbox intake body cap — enough for title + requirements. +GATE_MAX_JD_CHARS=max(int(os.getenv("SUMMARY_GATE_MAX_JD_CHARS") or 4000),1) +# 0 disables the uncertainty branch (0.0 < 0.0 is False). +GATE_MIN_CONFIDENCE=float(os.getenv("SUMMARY_GATE_MIN_CONFIDENCE") or 0) + +PROMPT_CACHE_KEY=f"summary-gate-{PROMPT_VERSION}" + +UNCLASSIFIED_PREFIX="unclassified:" + +CLASSIFIED="classified" +LOW_CONFIDENCE="low_confidence" +ERROR="error" + +# Backend-local skip code. Not an app.core.errors.ErrorCode — the bulk-ats +# engine never sees this pair; the CV is never opened. +SUMMARY_NOT_SUITABLE="SUMMARY_NOT_SUITABLE" + + +def get_gate_settings() -> Settings: + return get_settings() + + +def clip_text(value, limit) -> str: + text=str(value or "") + if len(text)<=limit: + return text + return text[:limit] + + +def should_score(verdict, error_code="") -> tuple[bool,str,str]: + """(proceed, status, reason) — the entire fail policy, in one place. + + No verdict means the model could not be consulted. SUMMARY_GATE_FAIL_OPEN + decides. A present verdict uses is_suitable, unless confidence is below + SUMMARY_GATE_MIN_CONFIDENCE (the gradient threshold). + """ + if verdict is None: + reason=f"{UNCLASSIFIED_PREFIX}{error_code or 'unknown'}"[:60] + return GATE_FAIL_OPEN,ERROR,reason + if verdict.confidence\n{job_description}\n" +) +_SUMMARY_TEMPLATE="\n{summary}\n" + + +def build_job_block(job_description) -> dict: + return { + "type":"input_text", + "text":_JOB_TEMPLATE.format(job_description=job_description or ""), + } + + +def build_summary_block(summary) -> dict: + return { + "type":"input_text", + "text":_SUMMARY_TEMPLATE.format(summary=summary or ""), + } + + +def build_input(job_description, summary) -> list: + """JD first (cacheable prefix), summary second (volatile).""" + return [ + { + "role":"user", + "content":[build_job_block(job_description),build_summary_block(summary)], + } + ]