diff --git a/backend/agent/decorators.py b/backend/agent/decorators.py index ed36b2e..a5c0079 100644 --- a/backend/agent/decorators.py +++ b/backend/agent/decorators.py @@ -36,7 +36,7 @@ def normalize_job_posts(job_posts) -> list[dict]: return normalized -def parse_match_response(data, allowed_ids) -> tuple[list[str], str]: +def parse_match_response(data, allowed_ids) -> tuple[list[str], str, str]: """Filter model JSON ids to the allowed job-post set.""" if not isinstance(data, dict): raise RuntimeError(f"model did not return a JSON object: {data!r}") @@ -59,7 +59,13 @@ def parse_match_response(data, allowed_ids) -> tuple[list[str], str]: seen.add(job_id) suggested.append(job_id) + summary = data.get("summary") + if not isinstance(summary, str): + summary = "" + reasoning = data.get("reasoning") + if isinstance(reasoning, list): + reasoning = "\n".join(str(item) for item in reasoning) if not isinstance(reasoning, str): reasoning = "" - return suggested, reasoning.strip() + return suggested, summary.strip(), reasoning.strip() diff --git a/backend/agent/models.py b/backend/agent/models.py index 9be9175..fa4e5b6 100644 --- a/backend/agent/models.py +++ b/backend/agent/models.py @@ -15,6 +15,7 @@ class AgentState(TypedDict, total=False): resume_text: str job_posts: list[dict] suggested_job_post_ids: list[str] + summary: str reasoning: str error: str status: Literal["pending", "ready", "matched", "skipped", "failed"] diff --git a/backend/agent/prompt.py b/backend/agent/prompt.py index 486fa68..25e979a 100644 --- a/backend/agent/prompt.py +++ b/backend/agent/prompt.py @@ -11,20 +11,23 @@ import json def prompt(): return """You are an HR-ATS recruiting assistant. -Given a candidate email subject, resume text, and a list of active job posts, -identify which job posts the candidate is most likely applying for. +You are given a candidate email subject, CV/resume text extracted from an +attachment, and a list of active job posts (id, title, description, requirements). -Return JSON only with this shape: -{ - "suggested_job_post_ids": ["uuid-string", ...], - "reasoning": "brief explanation" -} +Identify which job posts the candidate is most likely applying for. Rules: -- suggested_job_post_ids must only contain ids from the provided job_posts list. -- Return an empty list when nothing matches confidently. +- Only suggest job_post_id values that appear in the provided job_posts list. - A candidate may match zero, one, or multiple posts. -- Prefer title/subject alignment, then skills and experience in the resume. +- Base matches on skills, role title, experience, and the subject line — not guesses. +- If confidence is low, return an empty list rather than forcing a match. + +Respond with JSON only: +{ + "suggested_job_post_ids": ["uuid", "..."], + "summary": "one short sentence for the recruiter", + "reasoning": "brief bullet-style explanation per suggested match" +} """ diff --git a/backend/agent/serializers.py b/backend/agent/serializers.py index 0aa7fe8..163027e 100644 --- a/backend/agent/serializers.py +++ b/backend/agent/serializers.py @@ -10,6 +10,7 @@ def serialize_agent_result(state: dict) -> dict: """Plain dict for services/serializers — no ORM objects.""" return { "suggested_job_post_ids": state.get("suggested_job_post_ids") or [], + "summary": state.get("summary") or "", "reasoning": state.get("reasoning") or "", "status": state.get("status") or "failed", "error": state.get("error") or "", diff --git a/backend/agent/views.py b/backend/agent/views.py index 049e33d..0a6a867 100644 --- a/backend/agent/views.py +++ b/backend/agent/views.py @@ -30,6 +30,7 @@ async def prepare_context(state: AgentState) -> dict: "status": "skipped", "error": "resume_text is empty", "suggested_job_post_ids": [], + "summary": "", "reasoning": "", } if not job_posts: @@ -37,6 +38,7 @@ async def prepare_context(state: AgentState) -> dict: "status": "skipped", "error": "no active job posts to match against", "suggested_job_post_ids": [], + "summary": "", "reasoning": "", } @@ -60,10 +62,11 @@ async def match_jobs(state: AgentState) -> dict: try: data = await llm_call(prompt(), user_prompt(state), json_mode=True) allowed_ids = {item["id"] for item in state.get("job_posts") or []} - suggested, reasoning = parse_match_response(data, allowed_ids) + suggested, summary, reasoning = parse_match_response(data, allowed_ids) return { "status": "matched", "suggested_job_post_ids": suggested, + "summary": summary, "reasoning": reasoning, } except Exception as exc: @@ -72,6 +75,7 @@ async def match_jobs(state: AgentState) -> dict: "status": "failed", "error": str(exc), "suggested_job_post_ids": [], + "summary": "", "reasoning": "", } @@ -80,6 +84,7 @@ async def finalize(state: AgentState) -> dict: """Normalize terminal state for callers.""" return { "suggested_job_post_ids": state.get("suggested_job_post_ids") or [], + "summary": state.get("summary") or "", "reasoning": state.get("reasoning") or "", "status": state.get("status") or "failed", "error": state.get("error") or "", diff --git a/backend/inbox/app.py b/backend/inbox/app.py index c817262..2670808 100644 --- a/backend/inbox/app.py +++ b/backend/inbox/app.py @@ -1,16 +1,23 @@ -from fastapi import APIRouter,Depends, Query +from fastapi import APIRouter,BackgroundTasks,Depends, Query from fastapi.responses import JSONResponse from fastapi import HTTPException from db_setup import get_session from sqlalchemy.ext.asyncio import AsyncSession from inbox.views import Email +from users.permissions import PermissionTag, require_permission from dotenv import load_dotenv load_dotenv() router = APIRouter() @router.get("/email/fetch") -async def fetch_email(top:int=Query(100),skip:int=Query(0,ge=0),token=Query(...),session: AsyncSession = Depends(get_session)): +async def fetch_email( + background_tasks: BackgroundTasks, + top:int=Query(100), + skip:int=Query(0,ge=0), + token=Query(...), + session: AsyncSession = Depends(get_session), +): try: if not token: raise HTTPException(status_code=401,detail="Unauthorized") @@ -23,6 +30,8 @@ async def fetch_email(top:int=Query(100),skip:int=Query(0,ge=0),token=Query(...) service_per_email=await service.get_email_by_id(message_id) items_lst.append({"message_id":message_id,"email_contents":service_per_email}) + if service.pending_match_ids: + background_tasks.add_task(Email.run_inbox_matching, list(service.pending_match_ids)) return JSONResponse(content={"data":items_lst,"status_code":200}) @@ -53,3 +62,21 @@ async def fetch_inbox( raise except Exception as e: raise HTTPException(status_code=500,detail=str(e)) + + +@router.post("/inbox/{record_id}/match") +async def rematch_inbox( + record_id: str, + background_tasks: BackgroundTasks, + current_user: dict = Depends(require_permission(PermissionTag.INBOX_EDIT)), + session: AsyncSession = Depends(get_session), +): + try: + service=Email(session=session) + queued_id=await service.queue_rematch(record_id) + background_tasks.add_task(Email.run_inbox_matching, [queued_id], force=True) + return JSONResponse(content={"data":{"queued":True},"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) diff --git a/backend/inbox/models.py b/backend/inbox/models.py index 1af3171..a2e4a64 100644 --- a/backend/inbox/models.py +++ b/backend/inbox/models.py @@ -1,8 +1,8 @@ import uuid -from datetime import datetime +from datetime import datetime, timezone from typing import Any, Optional -from sqlalchemy import Column, func, or_ +from sqlalchemy import Column, DateTime, func, or_ from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.ext.asyncio import AsyncSession from sqlmodel import Field, Relationship, SQLModel, select @@ -62,6 +62,13 @@ class Inbox_Messages(SQLModel, table=True): message_reply: str | None = Field(default=None) file_name: str | None = Field(default=None) file_path: str | None = Field(default=None) + resume_text: str | None = Field(default=None) + suggested_job_post_ids: list[str] | None = Field(default=None, sa_column=Column(JSONB)) + match_summary: str | None = Field(default=None) + match_reasoning: str | None = Field(default=None) + match_status: str | None = Field(default=None) + match_error: str | None = Field(default=None) + matched_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True)) inbox: list[Inbox] = Relationship(back_populates="messages") @@ -73,12 +80,36 @@ class Inbox_Messages(SQLModel, table=True): if isinstance(body, str): return body return email_data.get("bodyPreview") or "" + @classmethod - async def candidate_x_inbox(cls,session:AsyncSession,candidate_id:Any): - try: - pass - except Exception as e: - raise HTTPException(status_code=500,detail=str(e)) + async def set_match_result( + cls, + session: AsyncSession, + record_id, + *, + resume_text=None, + suggested_job_post_ids=None, + summary="", + reasoning="", + status="", + error="", + ): + """Persist agent output onto one inbox row; returns the row or None.""" + row = await cls.get_inbox_message_by_id(session, record_id) + if not row: + return None + if resume_text is not None: + row.resume_text = resume_text + row.suggested_job_post_ids = suggested_job_post_ids + row.match_summary = summary or None + row.match_reasoning = reasoning or None + row.match_status = status or None + row.match_error = error or None + row.matched_at = datetime.now(timezone.utc) + session.add(row) + await session.commit() + await session.refresh(row) + return row @classmethod def _fields_from_email(cls, email_data: dict, file_path: list[str] | None = None) -> dict: diff --git a/backend/inbox/plugins.py b/backend/inbox/plugins.py index b0795ed..aaac986 100644 --- a/backend/inbox/plugins.py +++ b/backend/inbox/plugins.py @@ -6,6 +6,7 @@ import base64 from pathlib import Path from inbox.models import Inbox_Messages +from job.candidate.views import FileRead def load_message_files(message: Inbox_Messages) -> list[dict]: @@ -30,3 +31,32 @@ def load_message_files(message: Inbox_Messages) -> list[dict]: } ) return files + + +async def extract_resume_text(file_paths: list[str]) -> tuple[str, str]: + """Extract text from the PDFs among file_paths. Returns (combined_text, error).""" + pdf_paths = [ + Path(p.strip()) + for p in (file_paths or []) + if p and p.strip() and Path(p.strip()).suffix.lower() == ".pdf" + ] + existing = [p for p in pdf_paths if p.is_file()] + if not existing: + return "", "no PDF attachment to extract (.doc/.docx not supported)" + + texts: list[str] = [] + errors: list[str] = [] + for path in existing: + try: + raw = path.read_bytes() + result = await FileRead(session=None, filename=path.name, file=raw).read_file() + text = (result.get("text") or "").strip() + if text: + texts.append(text) + except Exception as exc: + errors.append(f"{path.name}: {exc}") + + if not texts: + return "", "; ".join(errors) if errors else "no text extracted from PDF" + + return "\n\n---\n\n".join(texts), "" diff --git a/backend/inbox/serializers.py b/backend/inbox/serializers.py index 712db4b..1a1ec02 100644 --- a/backend/inbox/serializers.py +++ b/backend/inbox/serializers.py @@ -40,4 +40,10 @@ def serialize_message(message: Inbox_Messages) -> dict: "message_sent_time": message.message_sent_time, "message_reply": message.message_reply, "file_path": message.file_path, + "suggested_job_post_ids": list(message.suggested_job_post_ids or []), + "match_summary": message.match_summary, + "match_reasoning": message.match_reasoning, + "match_status": message.match_status, + "match_error": message.match_error, + "matched_at": message.matched_at.isoformat() if message.matched_at else None, } diff --git a/backend/inbox/views.py b/backend/inbox/views.py index e561a44..82742e7 100644 --- a/backend/inbox/views.py +++ b/backend/inbox/views.py @@ -1,19 +1,96 @@ +import logging import httpx,os from fastapi import HTTPException from inbox.models import Inbox_Messages from inbox.file_decoder import decode_attachment, AttachmentDecodeError from inbox.serializers import serialize_message -from inbox.plugins import load_message_files +from inbox.plugins import load_message_files, extract_resume_text from dotenv import load_dotenv load_dotenv() from sqlalchemy.ext.asyncio import AsyncSession from pydantic import BaseModel +from db_setup import session_scope +from job.job_post.models import JobPosts +from job.job_post.serializers import serialize_job_post + +logger = logging.getLogger("inbox.match") + + class Email: def __init__(self,session:AsyncSession,token=None): self.session=session self.get_url=os.getenv("EMAIL_URL") self.token=token + self.pending_match_ids: list[str] = [] + + @staticmethod + async def run_inbox_matching(inbox_ids: list[str], *, force: bool = False) -> None: + """Background worker: extract CV text, run the matching agent, persist results. + + Uses its own session_scope per message - never the request session. + """ + if not inbox_ids: + return + + # Local import so inbox routes still load when langgraph/openai are missing. + from agent.execute_agent import run_agent + + cached_posts = None + for record_id in inbox_ids: + try: + async with session_scope() as session: + row = await Inbox_Messages.get_inbox_message_by_id(session, record_id) + if not row: + continue + if not force and row.match_status is not None: + continue + + paths = [p.strip() for p in (row.file_path or "").split(",") if p.strip()] + text, extract_err = await extract_resume_text(paths) + if not text: + await Inbox_Messages.set_match_result( + session, + record_id, + status="no_text", + error=extract_err or "no text extracted", + ) + continue + + if cached_posts is None: + posts = await JobPosts.get_active_job_posts(session) + cached_posts = [serialize_job_post(p) for p in posts] + + result = await run_agent( + subject=row.message_subject or "", + resume_text=text, + job_posts=cached_posts, + ) + await Inbox_Messages.set_match_result( + session, + record_id, + resume_text=text, + suggested_job_post_ids=result.get("suggested_job_post_ids") or [], + summary=result.get("summary") or "", + reasoning=result.get("reasoning") or "", + status=result.get("status") or "failed", + error=result.get("error") or "", + ) + logger.info( + "matched inbox %s status=%s ids=%s", + record_id, + result.get("status"), + result.get("suggested_job_post_ids"), + ) + except Exception as exc: + logger.exception("inbox matching failed for %s", record_id) + try: + async with session_scope() as session: + await Inbox_Messages.set_match_result( + session, record_id, status="failed", error=str(exc) + ) + except Exception: + logger.exception("failed to persist match failure for %s", record_id) async def service_email(self,top,skip): async with httpx.AsyncClient() as client: @@ -39,6 +116,12 @@ class Email: data=response.json() re_create_file=await decode_attachment(data.get("attachments")) insert_func=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file) + if ( + insert_func.attachment + and insert_func.file_path + and insert_func.match_status is None + ): + self.pending_match_ids.append(str(insert_func.id)) return response.json() else: raise HTTPException(status_code=response.status_code,detail=response.text) @@ -66,5 +149,13 @@ class Email: item["files"]=files return item + async def queue_rematch(self,record_id): + message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id) + if not message: + raise HTTPException(status_code=404,detail="Message not found") + if not message.attachment or not message.file_path: + raise HTTPException(status_code=400,detail="Message has no attachment to match") + return str(message.id) + async def count_inbox_messages(self,search=None): return await Inbox_Messages.count_inbox_messages(self.session,search) diff --git a/backend/job/job_post/models.py b/backend/job/job_post/models.py index 0c00b49..d87c564 100644 --- a/backend/job/job_post/models.py +++ b/backend/job/job_post/models.py @@ -62,6 +62,13 @@ class JobPosts(SQLModel, table=True): result = await session.execute(select(cls).where(cls.id == uid)) return result.scalars().first() + @classmethod + async def get_active_job_posts(cls, session: AsyncSession): + result = await session.execute( + select(cls).where(cls.is_active == True, cls.is_deleted == False) # noqa: E712 + ) + return result.scalars().all() + @classmethod async def insert_job_post(cls, session: AsyncSession, fields: dict): row = cls(**fields) diff --git a/backend/llm_setup.py b/backend/llm_setup.py index acc8550..4b94063 100644 --- a/backend/llm_setup.py +++ b/backend/llm_setup.py @@ -66,7 +66,7 @@ async def llm_call(system, user, *, model=None, temperature=None, json_mode=Fals With json_mode the prompt must mention JSON somewhere or the API rejects the call. """ kwargs = { - "model": OPENAI_MODEL, + "model": model or OPENAI_MODEL, "max_completion_tokens": OPENAI_MAX_OUTPUT_TOKENS, "messages": [ {"role": "system", "content": system}, diff --git a/backend/main.py b/backend/main.py index 2c07775..4ffbe83 100644 --- a/backend/main.py +++ b/backend/main.py @@ -1,10 +1,11 @@ import logging +from contextlib import asynccontextmanager import fastapi from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from fastapi import FastAPI,APIRouter -from db_setup import lifespan +from db_setup import lifespan as db_lifespan from inbox.app import router as inbox_router from users.app import router as users_router from role.app import router as role_router @@ -13,6 +14,37 @@ from job.app import router as candidate_router from notifications.app import router as confirmation_router # Without this the db/migration logs have no handler and are swallowed under uvicorn. logging.basicConfig(level=logging.INFO, format="%(levelname)-8s %(name)s: %(message)s") +logger = logging.getLogger("main") + + +@asynccontextmanager +async def lifespan(app): + """Compose DB lifespan with optional LLM/agent warm-up (degrades on failure).""" + async with db_lifespan(app): + llm_ready = False + agent_ready = False + close_llm = None + close_agent = None + try: + from llm_setup import init_llm, close_llm as _close_llm + from agent.agent_setup import init_agent, close_agent as _close_agent + + close_llm = _close_llm + close_agent = _close_agent + await init_llm() + llm_ready = True + await init_agent() + agent_ready = True + except Exception as exc: + logger.warning("llm/agent startup skipped: %s", exc) + try: + yield + finally: + if agent_ready and close_agent is not None: + await close_agent() + if llm_ready and close_llm is not None: + await close_llm() + # lifespan connects to Postgres and brings migrations up to head on startup, # and disposes of the connection pool on shutdown. @@ -30,4 +62,4 @@ app.include_router(users_router) app.include_router(role_router) app.include_router(forget_password_router) app.include_router(confirmation_router) -app.include_router(candidate_router) \ No newline at end of file +app.include_router(candidate_router)