.

backend/g_sheet/decorators.py

@@ -268,7 +268,7 @@ async def ingest_form_resume_links(session,sheet,credentials,temp_root=None):
                 credentials,resume_link,job_dir,max_chars,max_bytes,
             )
             try:
-                await FormData.set_extracted_data(session,record_id,payload)
+                saved=await FormData.set_extracted_data(session,record_id,payload)
             except Exception:
                 logger.exception("could not persist extracted_data for %s",record_id)
                 failed+=1
@@ -280,6 +280,9 @@ async def ingest_form_resume_links(session,sheet,credentials,temp_root=None):
             status=payload.get("status")
             if status=="completed":
                 completed+=1
+                if saved is not None:
+                    from g_sheet.scoring import enqueue_form_row_scores
+                    await enqueue_form_row_scores(saved)
             elif status=="failed":
                 failed+=1
         logger.info(

backend/g_sheet/enums.py

@@ -171,6 +171,8 @@ class FormDataColumn(str, Enum):
     ID = "id"
     SHEET = "sheet"
     JOB_POST_ID = "job_post_id"
+    ASSIGNED_JOB_POST_ID = "assigned_job_post_id"
+    SUGGESTED_JOB_POST_IDS = "suggested_job_post_ids"
     MANUAL_UPLOAD_CANDIDATE_ID = "manual_upload_candidate_id"
     ROW_NUMBER = "row_number"
     SERIAL_NO = "serial_no"

backend/g_sheet/models.py

@@ -28,9 +28,15 @@ class FormData(SQLModel, table=True):
 
     id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
     sheet: str = Field(nullable=False, index=True)
-    # Optional link to a job post. DB FK only — no ORM Relationship (avoids
+    # Recruiter-assigned job. DB FK only — no ORM Relationship (avoids
     # pulling job_posts into the sheet worker metadata graph).
     job_post_id: uuid.UUID | None = Field(default=None, index=True)
+    assigned_job_post_id: uuid.UUID | None = Field(default=None, index=True)
+    # ILIKE title matches from Position Applied For. One form row → many jobs.
+    # Suggested, not assigned. ATS scores each id separately.
+    suggested_job_post_ids: list[str] | None = Field(
+        default=None, sa_column=Column(JSONB),
+    )
     # Set when this form row is promoted into the hiring pipeline (Users +
     # manual_upload_candidate). Idempotency key for assign / shortlist.
     manual_upload_candidate_id: uuid.UUID | None = Field(default=None, index=True)
@@ -133,19 +139,65 @@ class FormData(SQLModel, table=True):
         result = await session.execute(select(cls).where(cls.id == rid))
         return result.scalars().first()
 
+    @classmethod
+    async def get_with_job(cls, session: AsyncSession, record_id, job_post_id):
+        """Form row + one job it may be scored against (suggested or assigned)."""
+        from job.job_post.models import JobPosts
+
+        form = await cls.get_form_data_by_id(session, record_id)
+        if form is None:
+            return None, None
+        try:
+            jid = uuid.UUID(str(job_post_id))
+        except (TypeError, ValueError):
+            return None, None
+        if jid not in set(cls.score_job_ids(form)):
+            return None, None
+        job = await JobPosts.get_job_post_by_id(session, jid)
+        if job is None or job.is_deleted:
+            return None, None
+        return form, job
+
+    @staticmethod
+    def score_job_ids(row) -> list[uuid.UUID]:
+        """Jobs ATS may score: assigned only, else every suggested id."""
+        if row is None:
+            return []
+        getter = row.get if isinstance(row, dict) else lambda key, default=None: getattr(row, key, default)
+        assigned = getter("assigned_job_post_id") or getter("job_post_id")
+        if assigned not in (None, ""):
+            try:
+                return [uuid.UUID(str(assigned))]
+            except (TypeError, ValueError):
+                return []
+        out: list[uuid.UUID] = []
+        seen: set[uuid.UUID] = set()
+        for raw in getter("suggested_job_post_ids") or []:
+            try:
+                uid = uuid.UUID(str(raw))
+            except (TypeError, ValueError):
+                continue
+            if uid not in seen:
+                seen.add(uid)
+                out.append(uid)
+        return out
+
     @classmethod
     async def set_job_post(cls, session: AsyncSession, record_id, job_post_id):
-        """Set or clear job_post_id; returns the row or None if missing."""
+        """Set or clear the recruiter assignment; returns the row or None if missing."""
         row = await cls.get_form_data_by_id(session, record_id)
         if not row:
             return None
         if job_post_id is None:
             row.job_post_id = None
+            row.assigned_job_post_id = None
         else:
             try:
-                row.job_post_id = uuid.UUID(str(job_post_id))
+                uid = uuid.UUID(str(job_post_id))
             except (TypeError, ValueError):
                 return None
+            row.job_post_id = uid
+            row.assigned_job_post_id = uid
         row.updated_at = _now()
         session.add(row)
         await session.commit()
@@ -330,6 +382,29 @@ class FormData(SQLModel, table=True):
         await session.commit()
         return {"deleted": deleted, "inserted": inserted}
 
+    @classmethod
+    async def stamp_suggested_job_posts(
+        cls, session: AsyncSession, records: list[dict],
+    ) -> list[dict]:
+        """Set suggested_job_post_ids from ILIKE title match on position_applied_for.
+
+        One applied-for title can match many job_posts. Blank or no match → [].
+        Recruiter assignment (job_post_id / assigned_job_post_id) stays unset.
+        """
+        from job.job_post.models import JobPosts
+
+        found = await JobPosts.ids_for_titles_ilike(
+            session,
+            [r.get("position_applied_for") for r in records],
+        )
+        for record in records:
+            applied = (record.get("position_applied_for") or "").strip()
+            hits = found.get(applied) or [] if applied else []
+            record["suggested_job_post_ids"] = [str(uid) for uid in hits]
+            record["job_post_id"] = None
+            record["assigned_job_post_id"] = None
+        return records
+
     @staticmethod
     def _cell(data: dict, key: str):
         """Sheet cell → stripped str, or None if missing/blank."""

backend/g_sheet/plugins.py

@@ -457,6 +457,7 @@ _CURRENCY_STRIP_RE=re.compile(r"(?:rs\.?|pkr|inr|usd|\$|€|£)",re.I)
 # Every typed column key the mapper must emit (uniform dicts for bulk insert).
 _FORM_DATA_COLUMN_KEYS=tuple(field.value for field in FormDataField)+(
     "age_raw","current_salary_value","expected_salary_value","job_post_id",
+    "assigned_job_post_id","suggested_job_post_ids",
 )
 
 

backend/g_sheet/scoring.py

@@ -0,0 +1,169 @@
+"""ATS-score a form_data CV against every linked job post.
+
+A form row can match many jobs (suggested_job_post_ids). Recruiter assignment
+is assigned_job_post_id. If assigned is set, ATS scores only that job; else
+every suggested id. Resume text comes from extracted_data. Each
+(form_data_id, job_post_id) pair lands as its own ats_results row.
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime, timezone
+
+from app.models.scoring import CompletedCandidate
+from app.services.pdf import ExtractedResume
+from app.services.scoring import score_batch
+from db_setup import session_scope
+from g_sheet.models import FormData
+from inbox.models import AtsResults
+from job.candidate.plugins import build_job_description, get_scorer, get_scoring_settings
+from job.job_post.models import JobPosts
+
+logger = logging.getLogger("g_sheet.scoring")
+
+
+def resume_text_from_extracted(payload) -> str | None:
+    """Usable CV text from form_data.extracted_data, or None."""
+    if not isinstance(payload, dict):
+        return None
+    if payload.get("status") != "completed":
+        return None
+    text = (payload.get("text") or "").strip()
+    return text or None
+
+
+def _band(score) -> str:
+    if score is None:
+        return ""
+    return "Strong Match" if score >= 82 else "Potential Match" if score >= 65 else "Weak Match"
+
+
+def serialize_form_ats(row) -> dict:
+    """One current ats_results row for a Sheet Forms applicant."""
+    return {
+        "job_post_id": str(row.job_post_id) if row.job_post_id else None,
+        "overall_score": row.overall_score,
+        "band": row.band or None,
+        "computed_at": row.computed_at.isoformat() if row.computed_at else None,
+    }
+
+
+async def enqueue_form_score(form_data_id, job_post_id) -> None:
+    """Queue ATS for one form row against one job. Broker-down only logs."""
+    if not form_data_id or not job_post_id:
+        return
+    await enqueue_form_scores(form_data_id, [job_post_id])
+
+
+async def enqueue_form_scores(form_data_id, job_post_ids) -> None:
+    """Queue ATS for one form row against each job. Broker-down only logs."""
+    if not form_data_id:
+        return
+    from inbox.tasks import score_form_data
+
+    seen: set[str] = set()
+    for raw in job_post_ids or []:
+        job_id = str(raw or "").strip()
+        if not job_id or job_id in seen:
+            continue
+        seen.add(job_id)
+        try:
+            await score_form_data.kicker().with_labels(
+                created_at=datetime.now(timezone.utc).isoformat(),
+                correlation_id=str(form_data_id),
+                queue="inbox",
+            ).kiq(str(form_data_id), job_id)
+        except Exception as exc:
+            logger.warning(
+                "could not queue form ats score for %s vs %s: %s",
+                form_data_id, job_id, exc,
+            )
+
+
+async def enqueue_form_row_scores(form_row) -> None:
+    """Queue ATS: assigned job only, else every suggested job."""
+    if form_row is None:
+        return
+    assigned=getattr(form_row,"assigned_job_post_id",None) or getattr(form_row,"job_post_id",None)
+    if assigned:
+        await enqueue_form_score(form_row.id,assigned)
+        return
+    job_ids=FormData.score_job_ids(form_row)
+    if not job_ids:
+        return
+    await enqueue_form_scores(form_row.id,job_ids)
+
+
+async def score_form_against_job(form_data_id: str, job_id: str) -> dict:
+    """Score one Sheet Forms CV against one job. Idempotent per (form, job)."""
+    try:
+        uuid.UUID(str(form_data_id))
+        uuid.UUID(str(job_id))
+    except (TypeError, ValueError):
+        return {"status": "skipped", "reason": "invalid_ids"}
+
+    async with session_scope() as session:
+        existing = await AtsResults.get_for_form_job(session, form_data_id, job_id)
+        if existing is not None:
+            return {"status": "already_scored"}
+        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"}
+        form_pk = form_row.id
+        job_pk = job.id
+
+    resume = ExtractedResume(
+        filename=str(filename),
+        candidate_id=str(form_pk),
+        text=text,
+        page_count=page_count,
+        truncated=truncated,
+    )
+    scored = await score_batch(
+        [resume],
+        job_description=jd,
+        scorer=get_scorer(),
+        concurrency=1,
+    )
+    result = scored[0] if scored else None
+    if not isinstance(result, CompletedCandidate):
+        error = getattr(result, "error_code", None) if result is not None else "MODEL_UNAVAILABLE"
+        logger.warning("form ats failed form_data=%s job=%s code=%s", form_data_id, job_id, error)
+        return {"status": "failed", "error_code": error}
+
+    async with session_scope() as session:
+        existing = await AtsResults.get_for_form_job(session, form_pk, job_pk)
+        if existing is not None:
+            return {"status": "already_scored"}
+        job = await JobPosts.get_job_post_by_id(session, job_pk)
+        if job is None or job.is_deleted:
+            return {"status": "skipped", "reason": "job_gone"}
+        await AtsResults.insert_result(session, {
+            "inbox_id": None,
+            "user_id": None,
+            "candidate_id": None,
+            "form_data_id": form_pk,
+            "job_post_id": job_pk,
+            "overall_score": float(result.match_score),
+            "band": _band(result.match_score),
+            "model_name": settings.openai_model,
+            "is_current": True,
+        })
+    return {
+        "status": "scored",
+        "overall_score": result.match_score,
+        "band": _band(result.match_score),
+        "job_post_id": str(job_pk),
+    }

backend/g_sheet/serializers.py

@@ -111,6 +111,8 @@ def serialize_form_data(row) -> dict:
             out[key] = _iso(value)
         elif isinstance(value, uuid.UUID):
             out[key] = str(value)
+        elif key == "suggested_job_post_ids":
+            out[key] = [str(v) for v in (value or []) if v not in (None, "")]
         else:
             out[key] = value
     return out

backend/g_sheet/views.py

@@ -17,6 +17,7 @@ Hierarchy:
 import asyncio
 import logging
 import threading
+import uuid
 from datetime import datetime,timezone
 
 from fastapi import HTTPException
@@ -249,6 +250,7 @@ class SheetImport(SheetRead):
             FormData.from_sheet_row(tab,row_number,record)
             for row_number,record in indexed
         ]
+        mapped=await FormData.stamp_suggested_job_posts(session,mapped)
         result=await FormData.replace_sheet(session,tab,mapped)
         return serialize_import({
             "tab":tab,
@@ -334,39 +336,97 @@ class SheetFormData(Sheet):
     """FormData DB mirror — query / delete only (no Google client)."""
 
     async def _hydrate_job_posts(self,items):
-        """Attach matching job_posts (title == position_applied_for) + assigned_job_post.
+        """Attach suggested job_posts, assigned_job_post, and per-job ATS scores.
 
-        No AI suggestions — form applicants already name the role. One query for
-        titles on the page, one for any assigned ids.
+        Preferred source is suggested_job_post_ids (ILIKE matches stored on
+        import). Legacy rows without that list still title-match. ATS is one
+        current score per (form, job).
         """
         if not items:
             return items
+        from g_sheet.scoring import serialize_form_ats
+        from inbox.models import AtsResults
         from job.job_post.models import JobPosts
         from job.job_post.serializers import serialize_job_post
 
         session=self._require_session()
+
+        def _job_payload(post):
+            payload=serialize_job_post(post)
+            if post.is_deleted or not post.is_active:
+                payload={**payload,"unavailable":True}
+            return payload
+
+        suggested_ids=[]
+        for item in items:
+            for raw in item.get("suggested_job_post_ids") or []:
+                if raw:
+                    suggested_ids.append(raw)
+        assigned_ids=[]
+        for item in items:
+            aid=item.get("assigned_job_post_id") or item.get("job_post_id")
+            if aid:
+                assigned_ids.append(aid)
+        wanted=list(dict.fromkeys([*suggested_ids,*assigned_ids]))
+        by_id={}
+        if wanted:
+            for post in await JobPosts.get_by_ids(session,wanted,active_only=False):
+                by_id[str(post.id)]=_job_payload(post)
+
         titles=[(item.get("position_applied_for") or "").strip() for item in items]
         titles=[t for t in titles if t]
         by_title={}
-        if titles:
+        needs_title=any(not (item.get("suggested_job_post_ids") or []) for item in items)
+        if titles and needs_title:
             for post in await JobPosts.get_by_titles(session,titles):
                 key=(post.title or "").strip().lower()
-                payload=serialize_job_post(post)
-                if post.is_deleted or not post.is_active:
-                    payload={**payload,"unavailable":True}
-                by_title.setdefault(key,[]).append(payload)
+                by_title.setdefault(key,[]).append(_job_payload(post))
 
-        assigned_ids=[item.get("job_post_id") for item in items if item.get("job_post_id")]
-        assigned_map={}
-        if assigned_ids:
-            for post in await JobPosts.get_by_ids(session,assigned_ids,active_only=False):
-                assigned_map[str(post.id)]=serialize_job_post(post)
+        ats_by_form=await AtsResults.get_current_for_forms(
+            session,[item.get("id") for item in items],
+        )
 
         for item in items:
-            key=(item.get("position_applied_for") or "").strip().lower()
-            item["job_posts"]=list(by_title.get(key) or [])
-            aid=item.get("job_post_id")
-            item["assigned_job_post"]=assigned_map.get(str(aid)) if aid else None
+            suggested=[str(raw) for raw in (item.get("suggested_job_post_ids") or []) if raw]
+            item["suggested_job_post_ids"]=suggested
+            if suggested:
+                posts=[]
+                for sid in suggested:
+                    payload=by_id.get(sid)
+                    if payload is None:
+                        posts.append({"id":sid,"unavailable":True})
+                    else:
+                        posts.append(dict(payload))
+                item["job_posts"]=posts
+            else:
+                key=(item.get("position_applied_for") or "").strip().lower()
+                item["job_posts"]=[dict(p) for p in (by_title.get(key) or [])]
+
+            aid=item.get("assigned_job_post_id") or item.get("job_post_id")
+            item["assigned_job_post_id"]=str(aid) if aid else None
+            item["assigned_job_post"]=by_id.get(str(aid)) if aid else None
+
+            fid=item.get("id")
+            try:
+                form_uid=uuid.UUID(str(fid)) if fid else None
+            except (TypeError,ValueError):
+                form_uid=None
+            scores=[serialize_form_ats(row) for row in (ats_by_form.get(form_uid) or [])]
+            item["ats_results"]=scores
+            score_by_job={
+                str(s["job_post_id"]):s for s in scores if s.get("job_post_id")
+            }
+            for post in item["job_posts"]:
+                hit=score_by_job.get(str(post.get("id")))
+                if hit:
+                    post["overall_score"]=hit.get("overall_score")
+                    post["band"]=hit.get("band")
+            assigned_score=score_by_job.get(str(aid)) if aid else None
+            if assigned_score and assigned_score.get("overall_score") is not None:
+                item["ats_score"]=assigned_score.get("overall_score")
+            else:
+                nums=[s.get("overall_score") for s in scores if s.get("overall_score") is not None]
+                item["ats_score"]=max(nums) if nums else None
         return items
 
     async def get_form_data(
@@ -394,7 +454,7 @@ class SheetFormData(Sheet):
         return items[0]
 
     async def assign_job_post(self,record_id,job_post_id):
-        """Set or clear form_data.job_post_id (same contract as inbox assign).
+        """Set or clear form_data.assigned_job_post_id (same contract as inbox assign).
 
         Setting a job promotes the row into Users + manual_upload_candidate so
         Candidates / Talent Pool / Pipeline can see it (platform tag: Form).
@@ -410,6 +470,8 @@ class SheetFormData(Sheet):
             raise HTTPException(status_code=404,detail="Form data not found")
         if job_post_id is not None:
             await self._promote_to_application(updated)
+            from g_sheet.scoring import enqueue_form_score
+            await enqueue_form_score(updated.id,job_post_id)
         return await self.get_form_data_by_id(record_id)
 
     async def set_processing_state(self,record_id,processing_state,current_user=None):

backend/inbox/models.py

@@ -1472,6 +1472,8 @@ class AtsResults(SQLModel, table=True):
     candidate_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="candidates.id")
     user_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="users.id")
     job_post_id: uuid.UUID | None = Field(default=None, foreign_key="job_posts.id")
+    # Sheet Forms scores: no inbox/user. Identity is (form_data_id, job_post_id).
+    form_data_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="form_data.id")
     overall_score: float = Field(default=0.0)
     band: str = Field(default="")
     is_current: bool = Field(default=True)
@@ -1530,6 +1532,53 @@ class AtsResults(SQLModel, table=True):
         )
         return result.scalars().first()
 
+    @classmethod
+    async def get_for_form_job(cls, session: AsyncSession, form_data_id, job_post_id):
+        """Any score for this form_data row against this job — current or superseded."""
+        fid = cls._as_uuid(form_data_id)
+        jid = cls._as_uuid(job_post_id)
+        if fid is None or jid is None:
+            return None
+        result = await session.execute(
+            select(cls)
+            .where(cls.form_data_id == fid, cls.job_post_id == jid)
+            .order_by(cls.computed_at.desc())
+        )
+        return result.scalars().first()
+
+    @classmethod
+    async def get_current_for_form_job(cls, session: AsyncSession, form_data_id, job_post_id=None):
+        """Current Sheet Forms score, optionally pinned to one job post."""
+        fid = cls._as_uuid(form_data_id)
+        if fid is None:
+            return None
+        qry = select(cls).where(cls.form_data_id == fid, cls.is_current == True)  # noqa: E712
+        jid = cls._as_uuid(job_post_id) if job_post_id is not None else None
+        if jid is not None:
+            qry = qry.where(cls.job_post_id == jid)
+        result = await session.execute(qry.order_by(cls.computed_at.desc()))
+        return result.scalars().first()
+
+    @classmethod
+    async def get_current_for_forms(cls, session: AsyncSession, form_data_ids) -> dict:
+        """Current Sheet Forms scores grouped by form_data_id (newest first)."""
+        keys = []
+        for raw in form_data_ids or []:
+            uid = cls._as_uuid(raw)
+            if uid is not None:
+                keys.append(uid)
+        if not keys:
+            return {}
+        result = await session.execute(
+            select(cls)
+            .where(cls.form_data_id.in_(keys), cls.is_current == True)  # noqa: E712
+            .order_by(cls.computed_at.desc())
+        )
+        grouped: dict = {}
+        for row in result.scalars():
+            grouped.setdefault(row.form_data_id, []).append(row)
+        return grouped
+
     @classmethod
     async def resolve_identity(cls, session: AsyncSession, email, candidate_id):
         """XOR identity for a score row from the scored candidate's email.
@@ -1579,15 +1628,20 @@ class AtsResults(SQLModel, table=True):
 
         Inbox scores chain on inbox_id and repoint inbox.ats_id (candidate_id
         may be NULL). Upload scores chain on candidate_id, or on (user_id,
-        job_post_id) when identity resolved to a user. Flush the INSERT first:
-        with no relationship() edge the unit of work emits the UPDATEs first,
-        and the FKs reject a pointer to a row not yet inserted."""
+        job_post_id) when identity resolved to a user. Sheet Forms scores chain
+        on (form_data_id, job_post_id) with inbox_id and user_id left NULL.
+        Flush the INSERT first: with no relationship() edge the unit of work
+        emits the UPDATEs first, and the FKs reject a pointer to a row not yet
+        inserted."""
         inbox_id = fields.get("inbox_id")
         candidate_id = fields.get("candidate_id")
         user_id = fields.get("user_id")
         job_post_id = fields.get("job_post_id")
+        form_data_id = fields.get("form_data_id")
         if inbox_id is not None:
             prev = await cls.get_current_for_inbox(session, inbox_id)
+        elif form_data_id is not None:
+            prev = await cls.get_current_for_form_job(session, form_data_id, job_post_id)
         elif candidate_id is not None:
             prev = await cls.get_current_for_candidate(session, candidate_id)
         elif user_id is not None:

backend/inbox/tasks.py

@@ -62,6 +62,27 @@ async def score_message_against_job(record_id:str,job_id:str) -> dict:
     return {"status":"scored","results":len(results)}
 
 
+@broker.task(
+    task_name="g_sheet.score_form",
+    retry_on_error=True,
+    max_retries=MAX_RETRIES,
+    delay=RETRY_DELAY,
+)
+async def score_form_data(form_data_id:str,job_id:str) -> dict:
+    """ATS-score one Sheet Forms CV (extracted_data) against one job post."""
+    from g_sheet.scoring import score_form_against_job
+
+    try:
+        uuid.UUID(str(form_data_id))
+        uuid.UUID(str(job_id))
+    except (TypeError,ValueError):
+        raise PermanentTaskError("form_data_id and job_id must be uuids")
+    result=await score_form_against_job(form_data_id,job_id)
+    if result.get("status")=="failed":
+        raise RuntimeError(result.get("error_code") or "form ats failed")
+    return result
+
+
 @broker.task(
     task_name="inbox.score_message",
     retry_on_error=True,

backend/job/job_post/models.py

@@ -148,6 +148,52 @@ class JobPosts(SQLModel, table=True):
         result = await session.execute(statement)
         return list(result.scalars().all())
 
+    @staticmethod
+    def title_ilike_pattern(applied_for: str) -> str | None:
+        """ILIKE pattern so job_posts.title contains the form's Position Applied For."""
+        needle = (applied_for or "").strip()
+        if not needle:
+            return None
+        escaped = needle.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+        return f"%{escaped}%"
+
+    @classmethod
+    async def ids_for_title_ilike(cls, session: AsyncSession, applied_for: str) -> list[uuid.UUID]:
+        """All non-deleted job_posts.id whose title ILIKE-contains applied_for.
+
+        One form title can match many posts. Order is created_at DESC, id DESC
+        so the UI/ATS list is stable. Empty if blank or no row. Suggested,
+        not recruiter-assigned.
+        """
+        pattern = cls.title_ilike_pattern(applied_for)
+        if not pattern:
+            return []
+        statement = (
+            select(cls.id)
+            .where(cls.is_deleted == False)  # noqa: E712
+            .where(cls.title.ilike(pattern, escape="\\"))
+            .order_by(cls.created_at.desc(), cls.id.desc())
+        )
+        result = await session.execute(statement)
+        return list(result.scalars().all())
+
+    @classmethod
+    async def ids_for_titles_ilike(
+        cls, session: AsyncSession, titles: list[str],
+    ) -> dict[str, list[uuid.UUID]]:
+        """Map stripped Position Applied For → every matching job_posts.id."""
+        out: dict[str, list[uuid.UUID]] = {}
+        seen: set[str] = set()
+        for raw in titles or []:
+            key = (raw or "").strip()
+            if not key or key in seen:
+                continue
+            seen.add(key)
+            found = await cls.ids_for_title_ilike(session, key)
+            if found:
+                out[key] = found
+        return out
+
     @classmethod
     async def fetch_job_posts(
         cls,

backend/tests/test_g_sheet_plugins.py

@@ -180,3 +180,65 @@ def test_cv_dest_path_stays_inside_dir(tmp_path):
     assert dest.parent == tmp_path.resolve()
     assert dest.name.endswith(".pdf")
     assert dest.name.startswith("1l5HOY5R6KiL6270A_sV56FkdX6EIGfI4")
+
+
+def test_title_ilike_pattern_contains_and_escapes():
+    from job.job_post.models import JobPosts
+
+    assert JobPosts.title_ilike_pattern("") is None
+    assert JobPosts.title_ilike_pattern("  ") is None
+    assert JobPosts.title_ilike_pattern("Brand Manager") == "%Brand Manager%"
+    assert JobPosts.title_ilike_pattern("C++ _dev%") == r"%C++ \_dev\%%"
+
+
+async def test_stamp_suggested_job_posts_sets_ids_or_empty(monkeypatch):
+    import uuid
+
+    job_a = uuid.uuid4()
+    job_b = uuid.uuid4()
+
+    async def fake_ids(session, titles):
+        return {"Executive Secretary": [job_a, job_b]}
+
+    monkeypatch.setattr(
+        "job.job_post.models.JobPosts.ids_for_titles_ilike",
+        fake_ids,
+    )
+    records = [
+        {"position_applied_for": "Executive Secretary"},
+        {"position_applied_for": "Unknown Role"},
+        {"position_applied_for": None},
+        {"position_applied_for": "  "},
+    ]
+    out = await FormData.stamp_suggested_job_posts(object(), records)
+    assert out[0]["suggested_job_post_ids"] == [str(job_a), str(job_b)]
+    assert out[0]["job_post_id"] is None
+    assert out[0]["assigned_job_post_id"] is None
+    assert out[1]["suggested_job_post_ids"] == []
+    assert out[1]["job_post_id"] is None
+    assert out[1]["assigned_job_post_id"] is None
+    assert out[2]["suggested_job_post_ids"] == []
+    assert out[3]["suggested_job_post_ids"] == []
+
+
+def test_score_job_ids_assigned_wins_else_suggested():
+    import uuid
+
+    a = uuid.uuid4()
+    b = uuid.uuid4()
+    assigned = uuid.uuid4()
+    assert FormData.score_job_ids({
+        "suggested_job_post_ids": [str(a), str(b)],
+        "assigned_job_post_id": assigned,
+    }) == [assigned]
+    assert FormData.score_job_ids({
+        "suggested_job_post_ids": [str(a), str(b)],
+        "job_post_id": assigned,
+    }) == [assigned]
+    assert FormData.score_job_ids({
+        "suggested_job_post_ids": [str(a), str(b), str(a)],
+        "assigned_job_post_id": None,
+        "job_post_id": None,
+    }) == [a, b]
+    assert FormData.score_job_ids({"suggested_job_post_ids": None}) == []
+    assert FormData.score_job_ids(None) == []

frontend/src/screens/Inbox.jsx

@@ -169,11 +169,22 @@ function formReceivedAt(entryDate, entryTime) {
 /**
  * GET /sheet/form-data/fetch row → the same list/detail shape the email channel
  * uses for name / avatar / position / source / time, plus form-only profile fields.
- * job_posts are title-matched (position_applied_for ↔ job_posts.title), not AI.
+ * job_posts come from suggested_job_post_ids (ILIKE title matches), not AI.
  */
 function mapFormRow(row) {
   const name = (row.name || row.candidate_email || 'Unknown').trim()
   const jobPosts = Array.isArray(row.job_posts) ? row.job_posts : []
+  const atsResults = Array.isArray(row.ats_results) ? row.ats_results : []
+  const assignedId = row.assigned_job_post_id
+    ? String(row.assigned_job_post_id)
+    : (row.job_post_id ? String(row.job_post_id) : null)
+  const assignedScore = atsResults.find((s) => String(s.job_post_id) === String(assignedId))
+  const numericScores = atsResults
+    .map((s) => Number(s.overall_score))
+    .filter((n) => Number.isFinite(n))
+  const atsScore = row.ats_score
+    ?? assignedScore?.overall_score
+    ?? (numericScores.length ? Math.max(...numericScores) : null)
   const state = row.processing_state || 'unread'
   return {
     kind: 'form',
@@ -212,7 +223,12 @@ function mapFormRow(row) {
     processingState: state,
     duplicate: Boolean(row.is_duplicate),
     jobPosts,
-    assignedId: row.job_post_id ? String(row.job_post_id) : null,
+    suggestedIds: Array.isArray(row.suggested_job_post_ids)
+      ? row.suggested_job_post_ids.map(String)
+      : [],
+    atsResults,
+    atsScore,
+    assignedId,
     assignedPost: row.assigned_job_post || null,
   }
 }
@@ -1324,7 +1340,7 @@ function firstResumeKey(item) {
 
 /**
  * Sheet form applicant detail — profile grids + resume/LinkedIn links +
- * title-matched job selection (position_applied_for ↔ job_posts.title) +
+ * ILIKE-matched job selection (one applied-for title → many job posts) +
  * the same Import / Shortlist / Duplicate / Reject actions as email.
  */
 function FormApplicantDetail({
@@ -1384,6 +1400,23 @@ function FormApplicantDetail({
   const alreadyProcessed = i.processing === 'Processed'
   const shortlistLocked = !alreadyProcessed && !jobChosen
   const panelBusy = busy || assignMutation.isPending
+  const selectedAts = useMemo(() => {
+    const jobId = selection || i.assignedId
+    if (!jobId) return null
+    const fromPost = (i.jobPosts || []).find((p) => String(p.id) === String(jobId))
+    const fromResults = (i.atsResults || []).find((s) => String(s.job_post_id) === String(jobId))
+    const score = fromPost?.overall_score ?? fromResults?.overall_score
+    if (score == null || !Number.isFinite(Number(score))) return null
+    const n = Number(score)
+    return {
+      score: n,
+      band: fromPost?.band || fromResults?.band
+        || (n >= 82 ? 'Strong Match' : n >= 65 ? 'Potential Match' : 'Weak Match'),
+    }
+  }, [selection, i.assignedId, i.jobPosts, i.atsResults])
+  const selectedRingColor = selectedAts == null
+    ? undefined
+    : selectedAts.score >= 82 ? 'var(--success)' : selectedAts.score >= 65 ? 'var(--warning)' : 'var(--danger)'
 
   async function handleMove() {
     if (busy || alreadyProcessed) return
@@ -1426,6 +1459,14 @@ function FormApplicantDetail({
             <div className="fw-600">{i.rowNumber}</div>
           </div>
         )}
+        {selectedAts != null && (
+          <div style={{ textAlign: 'center' }}>
+            <div className="ats-ring" style={{ width: 84, height: 84, '--pct': selectedAts.score, '--c': selectedRingColor }}>
+              <div className="ats-val"><div className="ats-num" style={{ fontSize: 22 }}>{Math.round(selectedAts.score)}</div></div>
+            </div>
+            <div className="cell-sub" style={{ marginTop: 4 }}>{selectedAts.band || 'ATS Score'}</div>
+          </div>
+        )}
       </div>
 
       {(resumeHref || profileHref) && (
@@ -1563,6 +1604,7 @@ function FormApplicantDetail({
           <div className="fw-600" style={{ marginBottom: 8 }}>Matching roles</div>
           <div className="cell-sub" style={{ marginBottom: 10 }}>
             Matched by position applied for: {orDash(i.position)}
+            {i.jobPosts?.length > 1 ? ` · ${i.jobPosts.length} roles` : ''}
           </div>
           {matchCards.length === 0 && !manualPost ? (
             <EmptyState icon="alert" title="No matching roles">

frontend/src/ui/SuggestedRoles.jsx

@@ -10,7 +10,7 @@ import { useState } from 'react'
 import { useQuery } from '@tanstack/react-query'
 
 import Modal from './Modal'
-import { Badge, EmptyState, Icon } from './primitives'
+import { Badge, EmptyState, Icon, ScoreChip } from './primitives'
 import { friendlyAuthError } from '../lib/errors'
 import { qk } from '../lib/queryKeys'
 import * as jobPostsApi from '../api/jobPosts'
@@ -74,6 +74,7 @@ export function JobCard({ post, rank, selected, onSelect, resumeText, manual, ba
           ) : (
             <Badge>{post.status || 'draft'}</Badge>
           )}
+          {post?.overall_score != null && <ScoreChip score={post.overall_score} />}
           {selected && <Icon name="check-circle" />}
         </div>
         {meta && <div className="cell-sub">{meta}</div>}