Merge pull request 'Backend_CODEBASE' (#4) from Backend_CODEBASE into main

Reviewed-on: #4

backend/.env.example

@@ -26,3 +26,25 @@ CONFIRM_TOKEN_RESEND_SECONDS=60
 BUFFER_API=
 BUFFER_API_URL=https://api.buffer.com
 BUFFER_CHANNEL_ID=
+
+OPENAI_API_KEY=
+OPENAI_MODEL=gpt-5.4-mini
+# Blank omits the parameter, for reasoning models that reject it.
+OPENAI_TEMPERATURE=0
+OPENAI_MAX_OUTPUT_TOKENS=4096
+OPENAI_TIMEOUT=60
+OPENAI_MAX_RETRIES=3
+OPENAI_CONNECT_RETRIES=3
+# Set only for Azure OpenAI or a gateway; blank uses api.openai.com.
+OPENAI_BASE_URL=
+OPENAI_ORGANIZATION=
+OPENAI_PROJECT=
+
+REDIS_URL=redis://localhost:6379/0
+TASKIQ_QUEUE_NAME=inbox
+TASKIQ_MAX_RETRIES=3
+TASKIQ_RETRY_DELAY=5
+TASKIQ_MAX_DELAY=120
+TASKIQ_DLQ_STREAM=taskiq:dlq
+TASKIQ_IDLE_TIMEOUT_MS=600000
+APP_VERSION=dev

backend/Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+# Runs the Taskiq worker against taskiq_management.broker_setup.
+# docker-compose overrides this command if needed.
+CMD ["taskiq", "worker", "taskiq_management.broker_setup:broker", "inbox.tasks", "taskiq_management.tasks"]

backend/agent/agent_setup.py

@@ -0,0 +1,57 @@
+"""LangGraph agent framework setup for HR-ATS workflows.
+
+Pure module: no FastAPI imports and no HTTPException.
+This file only owns graph construction and lifecycle:
+
+    init_agent() -> get_graph() -> build_graph() -> graph.compile()
+
+LLM client/config lives in llm_setup. Nodes live in agent.views.
+Run entrypoint lives in agent.execute_agent.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from langgraph.graph import END, START, StateGraph
+
+from agent.models import AgentState
+from agent.views import finalize, match_jobs, prepare_context, route_after_prepare
+
+logger = logging.getLogger("agent")
+
+_graph = None
+
+
+def build_graph():
+    """Construct and compile the HR-ATS candidate matching graph."""
+    graph = StateGraph(AgentState)
+    graph.add_node("prepare", prepare_context)
+    graph.add_node("match_jobs", match_jobs)
+    graph.add_node("finalize", finalize)
+    graph.add_edge(START, "prepare")
+    graph.add_conditional_edges("prepare", route_after_prepare)
+    graph.add_edge("match_jobs", "finalize")
+    graph.add_edge("finalize", END)
+    return graph.compile()
+
+
+def get_graph():
+    """Return the cached compiled graph, building it on first use."""
+    global _graph
+    if _graph is None:
+        _graph = build_graph()
+        logger.info("langgraph compiled")
+    return _graph
+
+
+async def init_agent():
+    """Warm the compiled graph. LLM init stays on llm_setup.init_llm()."""
+    get_graph()
+
+
+async def close_agent():
+    """Drop the cached graph."""
+    global _graph
+    _graph = None
+    logger.info("agent graph closed")

backend/agent/decorators.py

@@ -0,0 +1,71 @@
+"""Agent response parsers and input normalizers.
+
+Pure module: no FastAPI imports, no HTTPException, and no module-level state.
+Mirrors job/candidate/decorators.py — helpers that clean/shape data before or
+after the graph nodes run.
+"""
+
+from __future__ import annotations
+
+import uuid
+
+
+def normalize_job_posts(job_posts) -> list[dict]:
+    """Keep only dict items with an id field; stringify ids for the LLM."""
+    if not job_posts:
+        return []
+    normalized: list[dict] = []
+    for item in job_posts:
+        if not isinstance(item, dict):
+            continue
+        job_id = item.get("id")
+        if job_id is None:
+            continue
+        normalized.append(
+            {
+                "id": str(job_id),
+                "title": item.get("title") or "",
+                "description": item.get("description") or "",
+                "post_text": item.get("post_text") or "",
+                "requirements": item.get("requirements") or [],
+                "optional_skills": item.get("optional_skills") or [],
+                "location": item.get("location") or "",
+                "employment_type": item.get("employment_type") or "",
+            }
+        )
+    return normalized
+
+
+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}")
+
+    allowed = set(allowed_ids or [])
+    raw_ids = data.get("suggested_job_post_ids") or []
+    if not isinstance(raw_ids, list):
+        raw_ids = []
+
+    suggested: list[str] = []
+    seen: set[str] = set()
+    for raw_id in raw_ids:
+        job_id = str(raw_id).strip()
+        if not job_id or job_id not in allowed or job_id in seen:
+            continue
+        try:
+            uuid.UUID(job_id)
+        except ValueError:
+            continue
+        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, summary.strip(), reasoning.strip()

backend/agent/execute_agent.py

@@ -0,0 +1,22 @@
+"""Agent entrypoint — run the compiled graph.
+
+Pure module: no FastAPI imports and no HTTPException.
+"""
+
+from __future__ import annotations
+
+from agent.agent_setup import get_graph
+from agent.serializers import serialize_agent_result
+
+
+async def run_agent(*, subject="", resume_text="", job_posts=None) -> dict:
+    """Run the default graph and return a serialized result dict."""
+    final_state = await get_graph().ainvoke(
+        {
+            "subject": subject or "",
+            "resume_text": resume_text or "",
+            "job_posts": job_posts or [],
+            "status": "pending",
+        }
+    )
+    return serialize_agent_result(final_state)

backend/agent/models.py

@@ -0,0 +1,21 @@
+"""LangGraph agent state for HR-ATS workflows.
+
+Pure module: no FastAPI imports and no HTTPException.
+"""
+
+from __future__ import annotations
+
+from typing import Literal, TypedDict
+
+
+class AgentState(TypedDict, total=False):
+    """Shared state passed between graph nodes."""
+
+    subject: str
+    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"]

backend/agent/prompt.py

@@ -0,0 +1,42 @@
+"""Agent prompt builders.
+
+Pure module: no FastAPI imports and no HTTPException.
+"""
+
+from __future__ import annotations
+
+import json
+
+
+def prompt():
+    return """You are an HR-ATS recruiting assistant.
+
+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).
+
+Identify which job posts the candidate is most likely applying for.
+
+Rules:
+- Only suggest job_post_id values that appear in the provided job_posts list.
+- A candidate may match zero, one, or multiple posts.
+- 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"
+}
+"""
+
+
+def user_prompt(state) -> str:
+    return json.dumps(
+        {
+            "subject": state.get("subject") or "",
+            "resume_text": state.get("resume_text") or "",
+            "job_posts": state.get("job_posts") or [],
+        },
+        ensure_ascii=False,
+    )

backend/agent/serializers.py

@@ -0,0 +1,17 @@
+"""Agent result serializers.
+
+Pure module: no FastAPI imports and no HTTPException.
+"""
+
+from __future__ import annotations
+
+
+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 "",
+    }

backend/agent/views.py

@@ -0,0 +1,91 @@
+"""Agent graph node logic.
+
+Pure module: no FastAPI imports and no HTTPException.
+LLM client/config lives in llm_setup — nodes call llm_call only.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Literal
+
+from langgraph.graph import END
+
+from agent.decorators import normalize_job_posts, parse_match_response
+from agent.models import AgentState
+from agent.prompt import prompt, user_prompt
+from llm_setup import llm_call
+
+logger = logging.getLogger("agent")
+
+
+async def prepare_context(state: AgentState) -> dict:
+    """Validate inputs and decide whether matching should run."""
+    subject = (state.get("subject") or "").strip()
+    resume_text = (state.get("resume_text") or "").strip()
+    job_posts = normalize_job_posts(state.get("job_posts"))
+
+    if not resume_text:
+        return {
+            "status": "skipped",
+            "error": "resume_text is empty",
+            "suggested_job_post_ids": [],
+            "summary": "",
+            "reasoning": "",
+        }
+    if not job_posts:
+        return {
+            "status": "skipped",
+            "error": "no active job posts to match against",
+            "suggested_job_post_ids": [],
+            "summary": "",
+            "reasoning": "",
+        }
+
+    return {
+        "subject": subject,
+        "resume_text": resume_text,
+        "job_posts": job_posts,
+        "status": "ready",
+        "error": "",
+    }
+
+
+def route_after_prepare(state: AgentState) -> Literal["match_jobs", "__end__"]:
+    if state.get("status") == "ready":
+        return "match_jobs"
+    return END
+
+
+async def match_jobs(state: AgentState) -> dict:
+    """Ask the LLM (via llm_setup.llm_call) to map the candidate to job posts."""
+    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, summary, reasoning = parse_match_response(data, allowed_ids)
+        return {
+            "status": "matched",
+            "suggested_job_post_ids": suggested,
+            "summary": summary,
+            "reasoning": reasoning,
+        }
+    except Exception as exc:
+        logger.exception("agent match_jobs failed")
+        return {
+            "status": "failed",
+            "error": str(exc),
+            "suggested_job_post_ids": [],
+            "summary": "",
+            "reasoning": "",
+        }
+
+
+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 "",
+    }

backend/inbox/app.py

@@ -4,13 +4,19 @@ 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(
+    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 +29,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:
+            await service.enqueue_matching(list(service.pending_match_ids),force=False)
 
         return JSONResponse(content={"data":items_lst,"status_code":200})
     
@@ -53,3 +61,20 @@ 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,
+    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)
+        task_ids=await service.enqueue_matching([queued_id], force=True)
+        return JSONResponse(content={"data":{"queued":True,"task_ids":task_ids},"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))

backend/inbox/file_decoder.py

@@ -1,5 +1,7 @@
 """Decode Graph fileAttachment contentBytes into PDF / DOC / DOCX files."""
-
+# this file is decoding the pdf and also calling in the flow of first fetch of email if i do use func from  this file rather then  touchjing the email flow and create a bg task from here that can call the llm re i add param of subject and readc the file of pdf to get
+#the location to the llm_call thne it's probable that without touching the real flow i can use background task without stopping or delaying the real result and add a column in Inbox_Messages that i can later update the file recorby using filename to pdate the answer or suggeswtions from the lmm that i can later or get from get api so user/recruiter can see and map the candidate to it's real final job_post_id that then can be linked with job_post_id
+# as job_post_id is already linked by created_by and llm_call would require to read job_post of every recruiter and user ever posted only the posts that are still active it must read all post content and then finalize that this candidate might inlcude one of or more then one job_post_id : Note use list[uuid] to map with job_post_id inside Inbox_Messages table
 from __future__ import annotations
 
 import asyncio

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")
 
@@ -74,6 +81,36 @@ class Inbox_Messages(SQLModel, table=True):
             return body
         return email_data.get("bodyPreview") or ""
 
+    @classmethod
+    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:
         return {

backend/inbox/plugins.py

@@ -1,4 +1,4 @@
-"""Inbox helpers — attachment loading and other non-routing checks."""
+"""Inbox helpers — attachment loading and resume text extraction."""
 
 from __future__ import annotations
 
@@ -6,27 +6,60 @@ import base64
 from pathlib import Path
 
 from inbox.models import Inbox_Messages
+from job.candidate.views import FileRead
+
+_ATTACHMENTS_DIR=Path(__file__).resolve().parent/"decoded_attachments"
 
 
-def load_message_files(message: Inbox_Messages) -> list[dict]:
-    """Read files from file_path when they exist on disk."""
+def resolve_attachment_path(path_str:str) -> Path:
+    """Prefer stored path; fall back to basename under decoded_attachments."""
+    path=Path(path_str.strip())
+    if path.is_file():
+        return path
+    fallback=_ATTACHMENTS_DIR/path.name
+    if fallback.is_file():
+        return fallback
+    return path
+
+
+def load_message_files(message:Inbox_Messages) -> list[dict]:
     if not message.file_path:
         return []
-
-    files: list[dict] = []
+    files=[]
     for path_str in message.file_path.split(","):
-        path = Path(path_str.strip())
+        path=resolve_attachment_path(path_str)
         if not path.is_file():
             continue
         try:
-            raw = path.read_bytes()
+            raw=path.read_bytes()
         except OSError:
             continue
-        files.append(
-            {
-                "file_name": path.name,
-                "content_base64": base64.b64encode(raw).decode("ascii"),
-                "size": len(raw),
-            }
-        )
+        files.append({
+            "file_name":path.name,
+            "content_base64":base64.b64encode(raw).decode("ascii"),
+            "size":len(raw),
+        })
     return files
+
+
+async def extract_resume_text(file_paths:list[str]) -> tuple[str,str]:
+    candidates=[resolve_attachment_path(p) for p in (file_paths or []) if p and p.strip()]
+    existing=[p for p in candidates if p.is_file() and p.suffix.lower()==".pdf"]
+    if not existing:
+        return "","no PDF attachment to extract (.doc/.docx not supported)"
+
+    texts=[]
+    errors=[]
+    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),""

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,
     }

backend/inbox/tasks.py

@@ -0,0 +1,100 @@
+"""Inbox Taskiq tasks — CV → job-post matching."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+
+from db_setup import session_scope
+from inbox.models import Inbox_Messages
+from inbox.plugins import extract_resume_text
+from job.job_post.models import JobPosts
+from job.job_post.serializers import serialize_job_post
+from taskiq_management.broker_setup import MAX_RETRIES, RETRY_DELAY, broker
+from taskiq_management.middleware import PermanentTaskError
+
+logger=logging.getLogger("inbox.tasks")
+
+_DONE_STATUSES=frozenset({"matched","skipped","no_text","failed","dlq"})
+
+
+@broker.task(
+    task_name="inbox.match_message",
+    retry_on_error=True,
+    max_retries=MAX_RETRIES,
+    delay=RETRY_DELAY,
+)
+async def match_inbox_message(record_id:str,force:bool=False) -> dict:
+    if not record_id or not str(record_id).strip():
+        raise PermanentTaskError("record_id is required")
+
+    record_id=str(record_id).strip()
+
+    async with session_scope() as session:
+        row=await Inbox_Messages.get_inbox_message_by_id(session,record_id)
+        if not row:
+            raise PermanentTaskError(f"inbox message {record_id} not found")
+
+        if not force and row.match_status in _DONE_STATUSES:
+            logger.info("skip %s — already %s",record_id,row.match_status)
+            return {"status":row.match_status,"skipped":True}
+
+        if not row.attachment or not row.file_path:
+            raise PermanentTaskError("message has no attachment to match")
+
+        row.match_status="processing"
+        row.match_error=None
+        row.matched_at=datetime.now(timezone.utc)
+        session.add(row)
+        await session.commit()
+        await session.refresh(row)
+
+        paths=[p.strip() for p in (row.file_path or "").split(",") if p.strip()]
+        subject=row.message_subject or ""
+
+    text,extract_err=await extract_resume_text(paths)
+    if not text:
+        async with session_scope() as session:
+            await Inbox_Messages.set_match_result(
+                session,
+                record_id,
+                status="no_text",
+                error=extract_err or "no text extracted",
+            )
+        return {"status":"no_text","error":extract_err}
+
+    from agent.execute_agent import run_agent
+
+    async with session_scope() as session:
+        posts=await JobPosts.get_active_job_posts(session)
+        job_posts=[serialize_job_post(p) for p in posts]
+
+    try:
+        result=await run_agent(subject=subject,resume_text=text,job_posts=job_posts)
+    except Exception as exc:
+        logger.exception("agent failed for %s",record_id)
+        raise RuntimeError(f"agent matching failed: {exc}") from exc
+
+    status=result.get("status") or "failed"
+    error=result.get("error") or ""
+
+    if status=="failed":
+        raise RuntimeError(error or "agent returned failed status")
+
+    async with session_scope() as session:
+        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=status,
+            error=error,
+        )
+
+    logger.info("matched inbox %s status=%s ids=%s",record_id,status,result.get("suggested_job_post_ids"))
+    return {
+        "status":status,
+        "suggested_job_post_ids":result.get("suggested_job_post_ids") or [],
+    }

backend/inbox/views.py

@@ -1,19 +1,24 @@
+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.file_decoder import decode_attachment
 from inbox.serializers import serialize_message
 from inbox.plugins import load_message_files
 from dotenv import load_dotenv
 load_dotenv()
 from sqlalchemy.ext.asyncio import AsyncSession
-from pydantic import BaseModel
+from datetime import datetime,timezone
+
+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]=[]
 
     async def service_email(self,top,skip):
         async with httpx.AsyncClient() as client:
@@ -39,6 +44,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 +77,26 @@ 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 enqueue_matching(self,inbox_ids,force=False):
+        from inbox.tasks import match_inbox_message
+        task_ids=[]
+        for record_id in inbox_ids or []:
+            created_at=datetime.now(timezone.utc).isoformat()
+            task=await match_inbox_message.kicker().with_labels(
+                created_at=created_at,
+                correlation_id=str(record_id),
+                queue="inbox",
+            ).kiq(str(record_id),force=force)
+            task_ids.append(task.task_id)
+        return task_ids
+
     async def count_inbox_messages(self,search=None):
         return await Inbox_Messages.count_inbox_messages(self.session,search)

backend/job/app.py

@@ -40,7 +40,7 @@ async def cv_upload(
         file_content = await file.read()
         logger.info(f"Received file: {file.filename} ({len(file_content)} bytes)")
         service=FileRead(session=session,filename=file.filename,file=file_content)
-        data=await service.read_file(file_content, file.filename)
+        data=await service.read_file()
 
         return JSONResponse(content={"data":data,"status_code":200})
     except HTTPException:

backend/job/candidate/models.py

@@ -0,0 +1,6 @@
+# from sqlmodel import SQLModel, Field
+# from uuid import UUID, uuid4
+# from datetime import datetime
+# from enum import Enum
+
+# class CV_extraction(SQLModel,table=True):

backend/job/candidate/plugins.py

@@ -27,3 +27,4 @@ def normalize_spaced_text(text) -> str:
         return ""
     lines = [re.sub(r" {2,}", " ", line).strip() for line in text.splitlines()]
     return re.sub(r"\n{3,}", "\n\n", "\n".join(lines)).strip()
+

backend/job/candidate/views.py

@@ -26,3 +26,7 @@ class FileRead:
             raise
         except Exception as e:
             raise HTTPException(400, str(e))
+    # async def get_intention(self,input):
+    #     try:
+    #         get_subject=Inbox_Messages.candidate_x_inbox(self.session,self.candidate_id)
+    #         get_file=

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)

backend/llm_setup.py

@@ -0,0 +1,142 @@
+"""OpenAI async client and a single llm_call helper.
+
+Pure module: no FastAPI imports and no HTTPException.
+
+Config is module-level `os.getenv` (house style for non-DB secrets); the client is
+lazy, created on first use like `db_setup.get_engine()`.
+
+    text = await llm_call(system, user)
+    data = await llm_call(system, user, json_mode=True)
+
+`init_llm()` confirms the key on startup and `close_llm()` disposes of the connection
+pool, so both can hang off the FastAPI lifespan beside `init_db()` / `close_db()`.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+
+from dotenv import load_dotenv
+from openai import APIError, APIStatusError, AsyncOpenAI
+
+load_dotenv()
+
+logger = logging.getLogger("llm")
+
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL") or None
+OPENAI_ORGANIZATION = os.getenv("OPENAI_ORGANIZATION") or None
+OPENAI_PROJECT = os.getenv("OPENAI_PROJECT") or None
+OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-5.4-mini")
+OPENAI_MAX_OUTPUT_TOKENS = int(os.getenv("OPENAI_MAX_OUTPUT_TOKENS") or 32768)
+OPENAI_TIMEOUT = float(os.getenv("OPENAI_TIMEOUT") or 60)
+OPENAI_MAX_RETRIES = int(os.getenv("OPENAI_MAX_RETRIES") or 3)
+OPENAI_CONNECT_RETRIES = int(os.getenv("OPENAI_CONNECT_RETRIES") or 3)
+
+# Blank OPENAI_TEMPERATURE means omit the param (some models reject it).
+_raw_temp = (os.getenv("OPENAI_TEMPERATURE") or "").strip()
+OPENAI_TEMPERATURE = float(_raw_temp) if _raw_temp else None
+
+_client: AsyncOpenAI | None = None
+
+
+def get_client() -> AsyncOpenAI:
+    """The process-wide AsyncOpenAI client, created on first use."""
+    global _client
+    if _client is None:
+        if not OPENAI_API_KEY:
+            raise RuntimeError("OPENAI_API_KEY is not configured")
+        _client = AsyncOpenAI(
+            api_key=OPENAI_API_KEY,
+            base_url=OPENAI_BASE_URL,
+            organization=OPENAI_ORGANIZATION,
+            project=OPENAI_PROJECT,
+            timeout=OPENAI_TIMEOUT,
+            max_retries=OPENAI_MAX_RETRIES,
+        )
+    return _client
+
+
+async def llm_call(system, user, *, model=None, temperature=None, json_mode=False):
+    """One system+user turn. Returns text, or a parsed dict when json_mode=True.
+
+    With json_mode the prompt must mention JSON somewhere or the API rejects the call.
+    """
+    kwargs = {
+        "model": model or OPENAI_MODEL,
+        "max_completion_tokens": OPENAI_MAX_OUTPUT_TOKENS,
+        "messages": [
+            {"role": "system", "content": system},
+            {"role": "user", "content": user},
+        ],
+    }
+    resolved = OPENAI_TEMPERATURE if temperature is None else temperature
+    if resolved is not None:
+        kwargs["temperature"] = resolved
+    if json_mode:
+        kwargs["response_format"] = {"type": "json_object"}
+
+    response = await get_client().chat.completions.create(**kwargs)
+    content = (response.choices[0].message.content or "").strip()
+    if not json_mode:
+        return content
+    try:
+        return json.loads(content)
+    except json.JSONDecodeError as exc:
+        raise RuntimeError(f"model did not return valid JSON: {content[:200]}") from exc
+
+
+async def check_connection(retries=None, delay=1.0):
+    """Confirm the key works, retrying with a capped backoff."""
+    attempts = OPENAI_CONNECT_RETRIES if retries is None else retries
+    for attempt in range(1, max(attempts, 1) + 1):
+        try:
+            await get_client().models.list()
+            logger.info("openai reachable, default model %s", OPENAI_MODEL)
+            return
+        except APIStatusError as exc:
+            if exc.status_code in (401, 403):
+                raise RuntimeError(f"OPENAI_API_KEY rejected ({exc.status_code})") from exc
+            if attempt >= attempts:
+                raise
+            logger.warning("openai not ready (%s/%s): %s", attempt, attempts, exc)
+            await asyncio.sleep(delay)
+            delay = min(delay * 2, 10.0)
+        except APIError as exc:
+            if attempt >= attempts:
+                raise
+            logger.warning("openai not ready (%s/%s): %s", attempt, attempts, exc)
+            await asyncio.sleep(delay)
+            delay = min(delay * 2, 10.0)
+
+
+async def init_llm(*, verify=True):
+    """Build the client and, unless told otherwise, confirm the key is live."""
+    get_client()
+    if verify:
+        await check_connection()
+
+
+async def close_llm():
+    """Close the underlying httpx pool and reset the cached client."""
+    global _client
+    if _client is not None:
+        await _client.close()
+        logger.info("openai client closed")
+    _client = None
+
+
+# if __name__ == "__main__":
+#     logging.basicConfig(level=logging.INFO, format="%(levelname)-8s %(name)s: %(message)s")
+
+#     async def _main():
+#         try:
+#             await init_llm()
+#             print(await llm_call("You are terse.", "Reply with the single word: ready"))
+#         finally:
+#             await close_llm()
+
+#     asyncio.run(_main())

backend/main.py

@@ -1,22 +1,59 @@
 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 fastapi import FastAPI
+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
 from forget_password.app import router as forget_password_router
 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")
 
-# lifespan connects to Postgres and brings migrations up to head on startup,
-# and disposes of the connection pool on shutdown.
-app = FastAPI(lifespan=lifespan)
+logging.basicConfig(level=logging.INFO,format="%(levelname)-8s %(name)s: %(message)s")
+logger=logging.getLogger("main")
+
+
+@asynccontextmanager
+async def lifespan(app):
+    async with db_lifespan(app):
+        broker_ready=False
+        llm_ready=False
+        agent_ready=False
+        close_llm=None
+        close_agent=None
+        broker=None
+        try:
+            from taskiq_management.broker_setup import broker as _broker
+            broker=_broker
+            await broker.startup()
+            broker_ready=True
+        except Exception as exc:
+            logger.warning("taskiq broker startup skipped: %s",exc)
+        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()
+            if broker_ready and broker is not None:
+                await broker.shutdown()
+
+
+app=FastAPI(lifespan=lifespan)
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],

backend/requirements.txt

@@ -29,3 +29,12 @@ bcrypt==5.0.0              # password hashing in users/plugins.py
 
 # --- PDF extraction --------------------------------------------------------
 pypdf==5.1.0
+
+# --- task queue ------------------------------------------------------------
+taskiq>=0.11,<0.12         # broker + worker/scheduler CLI (taskiq_management/)
+taskiq-redis>=1.0,<2.0     # RedisStreamBroker / result backend / schedule source
+redis>=5.0,<6.0            # DLQ middleware (taskiq_management/middleware.py) async client
+
+# --- LLM -------------------------------------------------------------------
+openai==2.53.0             # AsyncOpenAI client in llm_setup.py
+langgraph==1.2.10          # StateGraph agent framework in agent/agent_setup.py

backend/taskiq_management/broker_setup.py

@@ -0,0 +1,55 @@
+"""Taskiq broker — Redis Streams + smart retry + DLQ.
+
+Worker:    taskiq worker taskiq_management.broker_setup:broker inbox.tasks taskiq_management.tasks
+Scheduler: taskiq scheduler taskiq_management.broker_setup:scheduler
+"""
+
+from __future__ import annotations
+
+import os
+
+from dotenv import load_dotenv
+from taskiq import TaskiqScheduler
+from taskiq.middlewares import SmartRetryMiddleware
+from taskiq_redis import (
+    ListRedisScheduleSource,
+    RedisAsyncResultBackend,
+    RedisStreamBroker,
+)
+
+from taskiq_management.middleware import DeadLetterMiddleware
+
+load_dotenv()
+
+REDIS_URL=os.getenv("REDIS_URL","redis://localhost:6379/0")
+QUEUE_NAME=os.getenv("TASKIQ_QUEUE_NAME","inbox")
+# 2 retries after first failure → max_retries=3
+MAX_RETRIES=int(os.getenv("TASKIQ_MAX_RETRIES","3"))
+RETRY_DELAY=float(os.getenv("TASKIQ_RETRY_DELAY","5"))
+
+result_backend=RedisAsyncResultBackend(redis_url=REDIS_URL)
+schedule_source=ListRedisScheduleSource(url=REDIS_URL,prefix="taskiq:schedule")
+
+broker=(
+    RedisStreamBroker(
+        url=REDIS_URL,
+        queue_name=QUEUE_NAME,
+        consumer_group_name=os.getenv("TASKIQ_CONSUMER_GROUP","taskiq"),
+        idle_timeout=int(os.getenv("TASKIQ_IDLE_TIMEOUT_MS","600000")),
+    )
+    .with_result_backend(result_backend)
+    .with_middlewares(
+        DeadLetterMiddleware(redis_url=REDIS_URL),
+        SmartRetryMiddleware(
+            default_retry_count=MAX_RETRIES,
+            default_retry_label=True,
+            default_delay=RETRY_DELAY,
+            use_jitter=True,
+            use_delay_exponent=True,
+            max_delay_exponent=float(os.getenv("TASKIQ_MAX_DELAY","120")),
+            schedule_source=schedule_source,
+        ),
+    )
+)
+
+scheduler=TaskiqScheduler(broker=broker,sources=[schedule_source])

backend/taskiq_management/middleware.py

@@ -0,0 +1,102 @@
+"""PermanentTaskError + Redis Stream DLQ middleware for Taskiq.
+
+Middleware order: DeadLetterMiddleware before SmartRetryMiddleware so
+permanent failures can set retry_on_error=False before SmartRetry runs.
+
+Pure module: no FastAPI imports.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+import redis.asyncio as redis
+from taskiq import TaskiqMiddleware
+from taskiq.message import TaskiqMessage
+from taskiq.result import TaskiqResult
+
+from taskiq_management.models import DLQ_STREAM
+from taskiq_management.serializers import serialize_dlq_payload
+
+logger=logging.getLogger("taskiq.dlq")
+
+
+class PermanentTaskError(Exception):
+    """Validation / business failure — DLQ immediately, no retries."""
+
+
+class DeadLetterMiddleware(TaskiqMiddleware):
+    def __init__(self,redis_url:str,stream:str=DLQ_STREAM):
+        super().__init__()
+        self.redis_url=redis_url
+        self.stream=stream
+        self._redis:redis.Redis|None=None
+
+    async def startup(self) -> None:
+        self._redis=redis.from_url(self.redis_url,decode_responses=True)
+
+    async def shutdown(self) -> None:
+        if self._redis is not None:
+            await self._redis.aclose()
+            self._redis=None
+
+    def _client(self) -> redis.Redis:
+        if self._redis is None:
+            self._redis=redis.from_url(self.redis_url,decode_responses=True)
+        return self._redis
+
+    async def on_error(
+        self,
+        message:TaskiqMessage,
+        result:TaskiqResult[Any],
+        exception:BaseException,
+    ) -> None:
+        retries=int(message.labels.get("_retries",0))
+        max_retries=int(message.labels.get("max_retries",2))
+        is_permanent=isinstance(exception,PermanentTaskError)
+        retries_exhausted=(retries+1)>=max_retries
+
+        if is_permanent:
+            message.labels["retry_on_error"]=False
+
+        if not is_permanent and not retries_exhausted:
+            return
+
+        queue=getattr(self.broker,"queue_name",None)
+        payload=serialize_dlq_payload(message,exception,retries=retries+1,queue=queue)
+        try:
+            await self._client().xadd(self.stream,{"payload":json.dumps(payload,ensure_ascii=False,default=str)})
+            logger.error(
+                "task %s (%s) sent to DLQ after %s",
+                message.task_name,
+                message.task_id,
+                "permanent failure" if is_permanent else f"{retries+1} attempts",
+            )
+        except Exception:
+            logger.exception("failed to write DLQ entry for %s",message.task_id)
+
+        await self._mark_inbox_dlq(message,exception)
+
+    async def _mark_inbox_dlq(self,message:TaskiqMessage,exception:BaseException) -> None:
+        if message.task_name!="inbox.match_message":
+            return
+        record_id=(message.kwargs or {}).get("record_id")
+        if not record_id and message.args:
+            record_id=message.args[0]
+        if not record_id:
+            return
+        try:
+            from db_setup import session_scope
+            from inbox.models import Inbox_Messages
+
+            async with session_scope() as session:
+                await Inbox_Messages.set_match_result(
+                    session,
+                    record_id,
+                    status="dlq",
+                    error=f"{type(exception).__name__}: {exception}",
+                )
+        except Exception:
+            logger.exception("failed to mark inbox %s as dlq",record_id)

backend/taskiq_management/models.py

@@ -0,0 +1,15 @@
+"""Taskiq constants — DLQ stream + app version defaults.
+
+Pure module: no FastAPI imports.
+"""
+
+from __future__ import annotations
+
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+DLQ_STREAM=os.getenv("TASKIQ_DLQ_STREAM","taskiq:dlq")
+APP_VERSION=os.getenv("APP_VERSION","dev")

backend/taskiq_management/serializers.py

@@ -0,0 +1,44 @@
+"""DLQ payload serializers for Taskiq dead-letter entries.
+
+Pure module: no FastAPI imports.
+"""
+
+from __future__ import annotations
+
+import os
+import socket
+import sys
+import traceback
+from datetime import datetime, timezone
+
+from taskiq.message import TaskiqMessage
+
+from taskiq_management.models import APP_VERSION
+
+
+def serialize_dlq_payload(
+    message:TaskiqMessage,
+    exception:BaseException,
+    *,
+    retries:int,
+    queue:str|None=None,
+) -> dict:
+    now=datetime.now(timezone.utc).isoformat()
+    return {
+        "task_name":message.task_name,
+        "task_id":message.task_id,
+        "kwargs":message.kwargs or {},
+        "args":list(message.args or []),
+        "exception":type(exception).__name__,
+        "message":str(exception),
+        "traceback":"".join(traceback.format_exception(type(exception),exception,exception.__traceback__)),
+        "retry_count":retries,
+        "worker":os.getenv("TASKIQ_WORKER_NAME") or os.getenv("HOSTNAME") or socket.gethostname(),
+        "queue":message.labels.get("queue") or queue or "taskiq",
+        "created_at":message.labels.get("created_at") or now,
+        "failed_at":now,
+        "correlation_id":message.labels.get("correlation_id") or message.task_id,
+        "hostname":socket.gethostname(),
+        "python_version":sys.version.split()[0],
+        "app_version":APP_VERSION,
+    }

backend/taskiq_management/tasks.py

@@ -0,0 +1,10 @@
+"""Framework smoke tasks for Taskiq — domain tasks stay in their packages."""
+
+from __future__ import annotations
+
+from taskiq_management.broker_setup import broker
+
+
+@broker.task(task_name="ping")
+async def ping() -> str:
+    return "pong"

docker-compose.yml

@@ -1,61 +1,63 @@
 services:
-  minio:
-    image: minio/minio:RELEASE.2025-04-22T22-12-26Z
-    container_name: hrms-minio
-    command: server /data --console-address ":9001"
-    environment:
-      MINIO_ROOT_USER: ${MINIO_ROOT_USER:-minioadmin}
-      MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD:-minioadmin}
+  redis:
+    image: redis:7-alpine
+    container_name: hrms-redis
+    command: ["redis-server", "--appendonly", "yes"]
     ports:
-      - "9000:9000"   # S3 API
-      - "9001:9001"   # web console
+      - "${REDIS_PORT:-6379}:6379"
     volumes:
-      - minio-data:/data
+      - redis-data:/data
     healthcheck:
-      test: ["CMD", "mc", "ready", "local"]
+      test: ["CMD", "redis-cli", "ping"]
       interval: 10s
       timeout: 5s
       retries: 5
-      start_period: 10s
     restart: unless-stopped
 
-  # One-shot: creates the attachments bucket, then exits.
-  minio-init:
-    image: minio/mc:RELEASE.2025-04-16T18-13-26Z
-    container_name: hrms-minio-init
-    depends_on:
-      minio:
-        condition: service_healthy
+  taskiq-worker:
+    build:
+      context: ./backend
+    container_name: hrms-taskiq-worker
+    command:
+      [
+        "taskiq",
+        "worker",
+        "taskiq_management.broker_setup:broker",
+        "inbox.tasks",
+        "taskiq_management.tasks",
+        "--workers",
+        "1",
+      ]
+    env_file:
+      - ./backend/.env
     environment:
-      MINIO_ROOT_USER: ${MINIO_ROOT_USER:-minioadmin}
-      MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD:-minioadmin}
-      MINIO_BUCKET: ${MINIO_BUCKET:-hrms-attachments}
-    entrypoint: >
-      /bin/sh -c "
-      mc alias set local http://minio:9000 \"$$MINIO_ROOT_USER\" \"$$MINIO_ROOT_PASSWORD\" &&
-      mc mb --ignore-existing local/\"$$MINIO_BUCKET\" &&
-      mc version enable local/\"$$MINIO_BUCKET\" &&
-      echo 'bucket ready: '\"$$MINIO_BUCKET\"
-      "
-
-  postgres:
-    image: postgres:16-alpine
-    container_name: hrms-postgres
-    environment:
-      POSTGRES_USER: ${DB_USERNAME:-postgres}
-      POSTGRES_PASSWORD: ${DB_PASSWORD:-postgres}
-      POSTGRES_DB: ${DB_NAME:-hrms}
-    ports:
-      - "${DB_PORT:-5432}:5432"
+      REDIS_URL: redis://redis:6379/0
+      TASKIQ_QUEUE_NAME: inbox
+      TASKIQ_WORKER_NAME: worker-01
+      DB_HOST: host.docker.internal
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
     volumes:
-      - postgres-data:/var/lib/postgresql/data
-    healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U ${DB_USERNAME:-postgres} -d ${DB_NAME:-hrms}"]
-      interval: 10s
-      timeout: 5s
-      retries: 5
+      - ./backend/inbox/decoded_attachments:/app/inbox/decoded_attachments
+    depends_on:
+      redis:
+        condition: service_healthy
+    restart: unless-stopped
+
+  taskiq-scheduler:
+    build:
+      context: ./backend
+    container_name: hrms-taskiq-scheduler
+    command: ["taskiq", "scheduler", "taskiq_management.broker_setup:scheduler"]
+    env_file:
+      - ./backend/.env
+    environment:
+      REDIS_URL: redis://redis:6379/0
+      TASKIQ_QUEUE_NAME: inbox
+    depends_on:
+      redis:
+        condition: service_healthy
     restart: unless-stopped
 
 volumes:
-  minio-data:
-  postgres-data:
+  redis-data: