Auto-score inbox CVs against matched jobs - no manual upload needed

The decoded attachment is already the CV and the job post is already the JD, so
scoring now fires from the data we have:

- After the agent matches an inbox message, score_message_against_job runs
  inline in the match task - against the assigned job if one is set, else the
  top suggested job. Failures log and never fail the match.
- When a recruiter assigns a job post, the new inbox.score_message task is
  queued in the background so the request never waits on OpenAI.

Idempotent by (message, job): an already-completed score is never paid for
twice. Rows are attributed to the job post owner since background tasks have
no request user. CV Import and the profile Score-with-ATS button remain as
manual fallbacks.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/inbox/tasks.py

@@ -1,10 +1,13 @@
-"""Inbox Taskiq tasks — CV → job-post matching."""
+"""Inbox Taskiq tasks — CV → job-post matching and ATS scoring."""
 
 from __future__ import annotations
 
 import logging
 from datetime import datetime,timezone
 
+from fastapi import HTTPException
+from sqlalchemy import select
+
 from agent.execute_agent import run_agent
 from db_setup import session_scope
 from employment_agent.execute_agent import run_employment_agent
@@ -19,6 +22,58 @@ logger=logging.getLogger("inbox.tasks")
 _DONE=frozenset({"matched","skipped","no_text","failed","dlq"})
 
 
+async def score_message_against_job(record_id:str,job_id:str) -> dict:
+    """ATS-score one inbox CV against one job post — the no-upload path.
+
+    The decoded attachment already on disk is the CV; the job post in the
+    database is the JD. Idempotent: a (message, job) pair with a completed
+    score is never paid for twice; re-runs are a no-op.
+    """
+    # Lazy imports: inbox.plugins imports job.candidate.views, so a top-level
+    # import here would be circular.
+    from job.candidate.models import Candidates
+    from job.candidate.views import CandidateScoring
+
+    mid=Candidates._as_uuid(record_id)
+    jid=Candidates._as_uuid(job_id)
+    if mid is None or jid is None:
+        raise PermanentTaskError("record_id and job_id must be uuids")
+
+    async with session_scope() as session:
+        existing=await session.execute(
+            select(Candidates).where(
+                Candidates.inbox_message_id==mid,
+                Candidates.job_id==jid,
+                Candidates.status=="completed",
+            )
+        )
+        if existing.scalars().first() is not None:
+            return {"status":"already_scored"}
+        job=await JobPosts.get_job_post_by_id(session,job_id)
+        if job is None or job.is_deleted:
+            raise PermanentTaskError("job post missing or deleted")
+        service=CandidateScoring(session=session)
+        try:
+            # Attribute the rows to the job's owner — there is no request user
+            # in a background task.
+            results=await service.score_inbox(job_id,[record_id],{"id":str(job.created_by)})
+        except HTTPException as exc:
+            # 400/404 from score_inbox are permanent (no attachment, bad ids);
+            # retrying cannot fix them.
+            raise PermanentTaskError(str(exc.detail)) from exc
+    return {"status":"scored","results":len(results)}
+
+
+@broker.task(
+    task_name="inbox.score_message",
+    retry_on_error=True,
+    max_retries=MAX_RETRIES,
+    delay=RETRY_DELAY,
+)
+async def score_inbox_message(record_id:str,job_id:str) -> dict:
+    return await score_message_against_job(record_id,job_id)
+
+
 @broker.task(
     task_name="inbox.match_message",
     retry_on_error=True,
@@ -81,6 +136,24 @@ async def match_inbox_message(record_id:str,force:bool=False) -> dict:
             status=status,
             error=result.get("error") or "",
         )
+    # Auto-score: the match just paired this CV with jobs, so run the ATS on the
+    # spot — assigned job first, else the agent's top suggestion. Scoring failures
+    # must not fail the match; the match result is already committed above.
+    suggested=[str(j) for j in (result.get("suggested_job_post_ids") or []) if j]
+    score_job_id=None
+    async with session_scope() as session:
+        fresh=await Inbox_Messages.get_inbox_message_by_id(session,record_id)
+        if fresh is not None and fresh.assigned_job_post_id:
+            score_job_id=str(fresh.assigned_job_post_id)
+    if score_job_id is None and suggested:
+        score_job_id=suggested[0]
+    if score_job_id:
+        try:
+            outcome=await score_message_against_job(record_id,score_job_id)
+            logger.info("ats auto-score %s vs %s: %s",record_id,score_job_id,outcome.get("status"))
+        except Exception as exc:
+            logger.warning("ats auto-score failed for %s vs %s: %s",record_id,score_job_id,exc)
+
     return {
         "status":status,
         "suggested_job_post_ids":result.get("suggested_job_post_ids") or [],

backend/inbox/views.py

@@ -179,6 +179,20 @@ class Email:
         updated=await Inbox_Messages.set_assigned_job_post(self.session,record_id,job_post_id)
         if not updated:
             raise HTTPException(status_code=404,detail="Message not found")
+        if job_post_id is not None:
+            # Assignment pairs this CV with a JD we already have — queue the ATS
+            # score in the background so the recruiter is not held on an OpenAI
+            # call. Idempotent server-side; broker-down just logs (the profile's
+            # Score-with-ATS button remains the manual fallback).
+            from inbox.tasks import score_inbox_message
+            try:
+                await score_inbox_message.kicker().with_labels(
+                    created_at=datetime.now(timezone.utc).isoformat(),
+                    correlation_id=str(record_id),
+                    queue="inbox",
+                ).kiq(str(record_id),str(job_post_id))
+            except Exception as exc:
+                logger.warning("could not queue ats score for %s: %s",record_id,exc)
         return await self.get_inbox_message_by_id(record_id)
 
     async def mark_read(self,record_id):