.

backend/g_sheet/app.py

@@ -167,6 +167,14 @@ class AssignFormJobPostBody(BaseModel):
     job_post_id: str | None = None
 
 
+class FormProcessingStateBody(BaseModel):
+    processing_state: str
+
+
+class FormDuplicateBody(BaseModel):
+    is_duplicate: bool
+
+
 @router.get("/sheet/form-data/sheets")
 async def fetch_form_data_sheets(
     current_user: dict = Depends(_FORM_DATA_READ),
@@ -186,6 +194,8 @@ async def fetch_form_data_sheets(
 async def fetch_form_data(
     sheet: str | None = Query(None),
     search: str | None = Query(None),
+    processing_state: str | None = Query(None),
+    is_duplicate: bool | None = Query(None),
     offset: int = Query(0,ge=0),
     limit: int | None = Query(None,ge=1),
     current_user: dict = Depends(_FORM_DATA_READ),
@@ -193,7 +203,10 @@ async def fetch_form_data(
 ):
     try:
         service=SheetFormData(session=session)
-        items,total=await service.get_form_data(sheet=sheet,search=search,offset=offset,limit=limit)
+        items,total=await service.get_form_data(
+            sheet=sheet,search=search,offset=offset,limit=limit,
+            processing_state=processing_state,is_duplicate=is_duplicate,
+        )
         return JSONResponse(content={"data":items,"total":total,"status_code":200})
     except HTTPException:
         raise
@@ -201,6 +214,22 @@ async def fetch_form_data(
         raise HTTPException(status_code=500,detail=str(e))
 
 
+@router.get("/sheet/form-data/counts")
+async def fetch_form_data_counts(
+    sheet: str | None = Query(None),
+    current_user: dict = Depends(_FORM_DATA_READ),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=SheetFormData(session=session)
+        data=await service.get_counts(sheet=sheet)
+        return JSONResponse(content={"data":data,"total":1,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
 @router.get("/sheet/form-data/{record_id}")
 async def fetch_form_data_by_id(
     record_id: str,
@@ -234,6 +263,40 @@ async def assign_form_job_post(
         raise HTTPException(status_code=500,detail=str(e))
 
 
+@router.patch("/sheet/form-data/{record_id}/processing-state")
+async def set_form_processing_state(
+    record_id: str,
+    payload: FormProcessingStateBody,
+    current_user: dict = Depends(_FORM_DATA_EDIT),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=SheetFormData(session=session)
+        data=await service.set_processing_state(record_id,payload.processing_state)
+        return JSONResponse(content={"data":data,"total":1,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
+@router.patch("/sheet/form-data/{record_id}/duplicate")
+async def set_form_duplicate(
+    record_id: str,
+    payload: FormDuplicateBody,
+    current_user: dict = Depends(_FORM_DATA_EDIT),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=SheetFormData(session=session)
+        data=await service.set_duplicate(record_id,payload.is_duplicate)
+        return JSONResponse(content={"data":data,"total":1,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
 @router.delete("/sheet/form-data/{tab}/delete")
 async def delete_form_data_sheet(
     tab: str,

backend/g_sheet/enums.py

@@ -171,6 +171,7 @@ class FormDataColumn(str, Enum):
     ID = "id"
     SHEET = "sheet"
     JOB_POST_ID = "job_post_id"
+    MANUAL_UPLOAD_CANDIDATE_ID = "manual_upload_candidate_id"
     ROW_NUMBER = "row_number"
     SERIAL_NO = "serial_no"
     ENTRY_YEAR = "entry_year"
@@ -216,6 +217,9 @@ class FormDataColumn(str, Enum):
     DIRECTOR_POC_CATEGORY = "director_poc_category"
     PROS = "pros"
     CONS = "cons"
+    # Same vocabulary as inbox_messages — Import / Shortlist / Reject / Duplicate.
+    PROCESSING_STATE = "processing_state"
+    IS_DUPLICATE = "is_duplicate"
     RAW_RECORD = "raw_record"
     IMPORTED_AT = "imported_at"
     CREATED_AT = "created_at"

backend/g_sheet/models.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import uuid
 from datetime import datetime, timezone
 
-from sqlalchemy import Column, DateTime, Index, delete, func, insert, or_
+from sqlalchemy import Column, DateTime, Index, case, delete, func, insert, or_
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlmodel import Field, SQLModel, select
@@ -31,6 +31,9 @@ class FormData(SQLModel, table=True):
     # Optional link to a job post. 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)
+    # 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)
     row_number: int | None = Field(default=None)
     serial_no: str | None = Field(default=None)
     entry_year: str | None = Field(default=None)
@@ -77,16 +80,25 @@ class FormData(SQLModel, table=True):
     pros: str | None = Field(default=None)
     cons: str | None = Field(default=None)
 
+    # Same allowlist as inbox_messages.processing_state: unread|imported|processed|rejected.
+    # server_default is load-bearing — ALTER on a populated form_data table.
+    processing_state: str = Field(default="unread", sa_column_kwargs={"server_default": "unread"})
+    is_duplicate: bool = Field(default=False, sa_column_kwargs={"server_default": "false"})
+
     raw_record: dict | None = Field(default=None, sa_column=Column(JSONB))
     imported_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
     created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
     updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
 
     @classmethod
-    def _filters(cls, *, sheet=None, search=None):
+    def _filters(cls, *, sheet=None, search=None, processing_state=None, is_duplicate=None):
         filters = []
         if sheet:
             filters.append(cls.sheet == sheet)
+        if processing_state:
+            filters.append(cls.processing_state == processing_state)
+        if is_duplicate is not None:
+            filters.append(cls.is_duplicate == bool(is_duplicate))
         if search:
             # Twelve unanchored ILIKEs over ~26k rows is a sequential scan of a few
             # tens of ms — acceptable at this size; a pg_trgm GIN index is the
@@ -139,26 +151,101 @@ class FormData(SQLModel, table=True):
         return row
 
     @classmethod
-    async def fetch_form_data(cls, session: AsyncSession, *, sheet=None, search=None, offset=0, limit=None):
+    async def set_processing_state(cls, session: AsyncSession, record_id, processing_state: str):
+        row = await cls.get_form_data_by_id(session, record_id)
+        if not row:
+            return None
+        row.processing_state = processing_state
+        row.updated_at = _now()
+        session.add(row)
+        await session.commit()
+        await session.refresh(row)
+        return row
+
+    @classmethod
+    async def set_duplicate(cls, session: AsyncSession, record_id, is_duplicate: bool):
+        row = await cls.get_form_data_by_id(session, record_id)
+        if not row:
+            return None
+        row.is_duplicate = bool(is_duplicate)
+        row.updated_at = _now()
+        session.add(row)
+        await session.commit()
+        await session.refresh(row)
+        return row
+
+    @classmethod
+    async def link_manual_upload(cls, session: AsyncSession, record_id, manual_upload_candidate_id, *, commit: bool = True):
+        row = await cls.get_form_data_by_id(session, record_id)
+        if not row:
+            return None
+        try:
+            row.manual_upload_candidate_id = uuid.UUID(str(manual_upload_candidate_id))
+        except (TypeError, ValueError):
+            return None
+        row.updated_at = _now()
+        session.add(row)
+        if commit:
+            await session.commit()
+            await session.refresh(row)
+        return row
+
+    @classmethod
+    async def fetch_form_data(
+        cls, session: AsyncSession, *, sheet=None, search=None,
+        processing_state=None, is_duplicate=None, offset=0, limit=None,
+    ):
         statement = select(cls).order_by(cls.sheet, cls.row_number)
-        for clause in cls._filters(sheet=sheet, search=search):
+        for clause in cls._filters(
+            sheet=sheet, search=search,
+            processing_state=processing_state, is_duplicate=is_duplicate,
+        ):
             statement = statement.where(clause)
         if offset:
             statement = statement.offset(offset)
         if limit is not None:
             statement = statement.limit(limit)
-        statement=statement.order_by(cls.row_number)
+        statement = statement.order_by(cls.row_number)
         result = await session.execute(statement)
         return result.scalars().all()
 
     @classmethod
-    async def count_form_data(cls, session: AsyncSession, *, sheet=None, search=None):
+    async def count_form_data(
+        cls, session: AsyncSession, *, sheet=None, search=None,
+        processing_state=None, is_duplicate=None,
+    ):
         statement = select(func.count()).select_from(cls)
-        for clause in cls._filters(sheet=sheet, search=search):
+        for clause in cls._filters(
+            sheet=sheet, search=search,
+            processing_state=processing_state, is_duplicate=is_duplicate,
+        ):
             statement = statement.where(clause)
         result = await session.execute(statement)
         return result.scalar_one()
 
+    @classmethod
+    async def count_processing(cls, session: AsyncSession, *, sheet=None):
+        """Tab badge counts for the Sheet Forms channel."""
+        statement = select(
+            func.count().label("all"),
+            func.coalesce(func.sum(case((cls.processing_state == "unread", 1), else_=0)), 0).label("unread"),
+            func.coalesce(func.sum(case((cls.processing_state == "imported", 1), else_=0)), 0).label("imported"),
+            func.coalesce(func.sum(case((cls.processing_state == "processed", 1), else_=0)), 0).label("processed"),
+            func.coalesce(func.sum(case((cls.processing_state == "rejected", 1), else_=0)), 0).label("rejected"),
+            func.coalesce(func.sum(case((cls.is_duplicate == True, 1), else_=0)), 0).label("duplicates"),  # noqa: E712
+        ).select_from(cls)
+        if sheet:
+            statement = statement.where(cls.sheet == sheet)
+        row = (await session.execute(statement)).one()
+        return {
+            "all": int(row.all or 0),
+            "unread": int(row.unread or 0),
+            "imported": int(row.imported or 0),
+            "processed": int(row.processed or 0),
+            "rejected": int(row.rejected or 0),
+            "duplicates": int(row.duplicates or 0),
+        }
+
     @classmethod
     async def get_sheet_names(cls, session: AsyncSession):
         result = await session.execute(

backend/g_sheet/views.py

@@ -368,12 +368,19 @@ class SheetFormData(Sheet):
             item["assigned_job_post"]=assigned_map.get(str(aid)) if aid else None
         return items
 
-    async def get_form_data(self,sheet=None,search=None,offset=0,limit=None):
+    async def get_form_data(
+        self,sheet=None,search=None,offset=0,limit=None,
+        processing_state=None,is_duplicate=None,
+    ):
         session=self._require_session()
         rows=await FormData.fetch_form_data(
             session,sheet=sheet,search=search,offset=offset,limit=limit,
+            processing_state=processing_state,is_duplicate=is_duplicate,
+        )
+        total=await FormData.count_form_data(
+            session,sheet=sheet,search=search,
+            processing_state=processing_state,is_duplicate=is_duplicate,
         )
-        total=await FormData.count_form_data(session,sheet=sheet,search=search)
         items=await self._hydrate_job_posts([serialize_form_data(row) for row in rows])
         return items,total
 
@@ -386,7 +393,11 @@ 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.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).
+        """
         session=self._require_session()
         if job_post_id is not None:
             from job.job_post.models import JobPosts
@@ -396,8 +407,106 @@ class SheetFormData(Sheet):
         updated=await FormData.set_job_post(session,record_id,job_post_id)
         if not updated:
             raise HTTPException(status_code=404,detail="Form data not found")
+        if job_post_id is not None:
+            await self._promote_to_application(updated)
         return await self.get_form_data_by_id(record_id)
 
+    async def set_processing_state(self,record_id,processing_state):
+        allowed=("unread","imported","processed","rejected")
+        if processing_state not in allowed:
+            raise HTTPException(status_code=422,detail=f"processing_state must be one of {', '.join(allowed)}")
+        session=self._require_session()
+        row=await FormData.get_form_data_by_id(session,record_id)
+        if not row:
+            raise HTTPException(status_code=404,detail="Form data not found")
+        # Shortlist requires a job — promote (idempotent) then flip the queue label.
+        if processing_state=="processed":
+            if not row.job_post_id:
+                raise HTTPException(status_code=422,detail="Assign a job post before moving to shortlist")
+            await self._promote_to_application(row)
+        updated=await FormData.set_processing_state(session,record_id,processing_state)
+        if not updated:
+            raise HTTPException(status_code=404,detail="Form data not found")
+        return await self.get_form_data_by_id(record_id)
+
+    async def _promote_to_application(self,form_row):
+        """Create Users + manual_upload_candidate from a form_data row (idempotent).
+
+        Pipeline / Candidates / Talent Pool all read manual_upload_candidate (or
+        the CANDIDATE user it creates). platform='Form' is the source badge.
+        """
+        session=self._require_session()
+        from job.candidate.models import Manual_UPLOAD_CANDIDATE
+        from job.history.views import HistoryRecorder
+        from job.history.enums import HistoryEvent
+
+        if getattr(form_row,"manual_upload_candidate_id",None):
+            existing=await Manual_UPLOAD_CANDIDATE.get_by_id(session,form_row.manual_upload_candidate_id)
+            if existing:
+                if form_row.job_post_id and existing.job_post_id!=form_row.job_post_id:
+                    existing.job_post_id=form_row.job_post_id
+                    session.add(existing)
+                    await session.commit()
+                return existing
+
+        email=(form_row.candidate_email or "").strip().lower()
+        if not email:
+            raise HTTPException(status_code=422,detail="candidate_email is required to promote this form applicant")
+        if not form_row.job_post_id:
+            raise HTTPException(status_code=422,detail="job_post_id is required to promote this form applicant")
+
+        existing=await Manual_UPLOAD_CANDIDATE.get_by_email_and_job(
+            session,email,form_row.job_post_id,
+        )
+        if existing:
+            await FormData.link_manual_upload(session,form_row.id,existing.id)
+            return existing
+
+        resume=(form_row.resume_link or "").strip()
+        file_name=""
+        if resume:
+            file_name=resume.rsplit("/",1)[-1][:180] or "resume"
+
+        row=await Manual_UPLOAD_CANDIDATE.create_manual_upload_candidate(session,{
+            "candidate_email":email,
+            "candidate_name":(form_row.name or "").strip() or email,
+            "candidate_phone":(form_row.candidate_number or "").strip(),
+            "job_post_id":str(form_row.job_post_id),
+            "current_company":(form_row.current_company or "").strip(),
+            "current_position":(form_row.position_applied_for or "").strip(),
+            "platform":"Form",
+            "apply_via":"form",
+            "experience":(form_row.experience or "").strip(),
+            "status":"PENDING",
+            "file_name":file_name,
+            "file_path":resume,
+            "full_text":"",
+        })
+        await FormData.link_manual_upload(session,form_row.id,row.id)
+        try:
+            await HistoryRecorder(session).record(
+                HistoryEvent.CANDIDATE_CREATED.value,
+                actor_id=None,user_id=row.user_id,
+                manual_upload_candidate_id=row.id,
+                entity_type="manual_upload_candidate",entity_id=row.id,
+                to_value=row.candidate_email,
+                description="Form",commit=True,
+            )
+        except Exception:
+            logger.exception("form promote history record failed for %s",form_row.id)
+        return row
+
+    async def set_duplicate(self,record_id,is_duplicate):
+        if not isinstance(is_duplicate,bool):
+            raise HTTPException(status_code=422,detail="is_duplicate must be a boolean")
+        updated=await FormData.set_duplicate(self._require_session(),record_id,is_duplicate)
+        if not updated:
+            raise HTTPException(status_code=404,detail="Form data not found")
+        return await self.get_form_data_by_id(record_id)
+
+    async def get_counts(self,sheet=None):
+        return await FormData.count_processing(self._require_session(),sheet=sheet)
+
     async def get_imported_sheets(self):
         session=self._require_session()
         sheets=await FormData.get_sheet_names(session)

backend/inbox/models.py

@@ -27,7 +27,7 @@ logger = logging.getLogger("inbox.models")
 # Placeholder only. The account lands inactive and the candidate is mailed a
 # confirmation link; the real password comes from the reset flow afterwards.
 DEFAULT_CANDIDATE_PASSWORD = os.getenv("DEFAULT_CANDIDATE_PASSWORD", "Utopia!@#")
-CANDIDATE_ROLE_ID_FALLBACK = 8          # mirrors users/views.py:signup_user
+CANDIDATE_ROLE_ID_FALLBACK = 4          # mirrors users/views.py:signup_user
 SKIP_SENDER_PREFIXES = ("noreply", "no-reply", "donotreply", "do-not-reply",
                         "mailer-daemon", "postmaster", "bounce")
 

backend/job/app.py

@@ -218,7 +218,7 @@ async def create_manual_candidate(
 
 @router.get("/candidate/fetch/users")
 async def fetch_users(
-    role_id:int=Query(8),
+    role_id:int=Query(4),
     top:int=Query(10),
     skip:int=Query(0),
     search:str=Query(None),

backend/job/candidate/models.py

@@ -3,7 +3,7 @@ from datetime import datetime, timezone
 from typing import TYPE_CHECKING, List, Optional
 
 from fastapi import HTTPException
-from sqlalchemy import JSON, DateTime, Index, func, UniqueConstraint
+from sqlalchemy import JSON, DateTime, Index, UniqueConstraint, func, or_
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
@@ -78,6 +78,8 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
                     cls.current_company,
                     cls.current_position,
                     cls.experience,
+                    cls.platform,
+                    cls.apply_via,
                     cls.created_at,
                     cls.updated_at,
                     AtsResults.id.label("ats_result_id"),
@@ -135,6 +137,8 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
                     "current_company":row["current_company"] or None,
                     "current_position":row["current_position"] or None,
                     "experience":row["experience"] or None,
+                    "platform":row["platform"] or None,
+                    "apply_via":row["apply_via"] or None,
                     "created_at":row["created_at"].isoformat() if row["created_at"] else None,
                     "updated_at":row["updated_at"].isoformat() if row["updated_at"] else None,
                     "ats_result":ats,
@@ -192,7 +196,7 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
             user=await Users.insert_user(session,{
                 "name":name,
                 "email":email,
-                "role_id":role.id if role else 8,
+                "role_id":role.id if role else 4,
                 "password":hash_password(default_pw),
                 "is_active":True,
                 "is_deleted":False,
@@ -207,7 +211,7 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
             linkedin_slug=primary_slug_from_text(fields.get("full_text") or ""),
             current_company=(fields.get("current_company") or "").strip(),
             current_position=(fields.get("current_position") or "").strip(),
-            apply_via="manual_upload",
+            apply_via=(fields.get("apply_via") or "manual_upload").strip() or "manual_upload",
             user_id=user.id,
             platform=(fields.get("platform") or "").strip(),
             created_by=cls._as_uuid(fields.get("created_by")),
@@ -240,6 +244,70 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
         result = await session.execute(select(cls).where(cls.id == uid))
         return result.scalars().first()
 
+    @classmethod
+    async def get_by_email_and_job(cls, session: AsyncSession, email: str, job_post_id):
+        """Idempotency for form / re-import promotes against the same role."""
+        cleaned = (email or "").strip().lower()
+        jid = cls._as_uuid(job_post_id)
+        if not cleaned or jid is None:
+            return None
+        result = await session.execute(
+            select(cls)
+            .where(cls.candidate_email == cleaned, cls.job_post_id == jid)
+            .order_by(cls.created_at.desc())
+        )
+        return result.scalars().first()
+
+    @classmethod
+    async def list_for_talent_pool(cls, session: AsyncSession, *, limit=100, offset=0, search=None):
+        """Newest applications with a user + job for Talent Pool (manual / form)."""
+        from users.models import Users
+
+        statement = (
+            select(cls)
+            .join(Users, cls.user_id == Users.id)
+            .where(cls.user_id.is_not(None), cls.job_post_id.is_not(None))
+            .order_by(cls.created_at.desc())
+        )
+        if search:
+            like = f"%{search.strip()}%"
+            statement = statement.where(
+                or_(
+                    cls.candidate_name.ilike(like),
+                    cls.candidate_email.ilike(like),
+                    Users.name.ilike(like),
+                    Users.email.ilike(like),
+                )
+            )
+        statement = statement.limit(limit).offset(offset)
+        result = await session.execute(statement)
+        return list(result.scalars().all())
+
+    @classmethod
+    async def sources_by_user_ids(cls, session: AsyncSession, user_ids) -> dict[str, str]:
+        """Newest platform/apply_via label per user — Candidates Form badges."""
+        parsed = []
+        for raw in (user_ids or []):
+            uid = cls._as_uuid(raw)
+            if uid is not None:
+                parsed.append(uid)
+        if not parsed:
+            return {}
+        result = await session.execute(
+            select(cls.user_id, cls.platform, cls.apply_via, cls.created_at)
+            .where(cls.user_id.in_(parsed))
+            .order_by(cls.created_at.desc())
+        )
+        out: dict[str, str] = {}
+        for user_id, platform, apply_via, _created in result.all():
+            key = str(user_id)
+            if key in out:
+                continue
+            label = (platform or "").strip() or (apply_via or "").strip()
+            if label:
+                out[key] = label
+        return out
+
 
 class Candidates(SQLModel, table=True):
 

backend/job/candidate/views.py

@@ -555,6 +555,7 @@ class CandidateView:
                 "current_company":(current_company or "").strip(),
                 "current_position":(current_position or "").strip(),
                 "platform":(platform or "").strip(),
+                "apply_via":"manual_upload",
                 "experience":(experience or "").strip(),
                 "status":(status or "").strip(),
                 "referral_by":(referral_by or "").strip(),
@@ -619,7 +620,51 @@ class CandidateView:
                     if score.get("job_post_id"):
                         payload["scored_job_post_id"]=score["job_post_id"]
                 return payload
-            return await self.attach_job_posts(rows)
+            # List mode: inbox applications + manual/form applications (dedupe by user).
+            inbox_payloads=await self.attach_job_posts(rows)
+            if not isinstance(inbox_payloads,list):
+                inbox_payloads=[inbox_payloads] if inbox_payloads else []
+            manual_rows=await Manual_UPLOAD_CANDIDATE.list_for_talent_pool(
+                self.session,limit=fetch_limit,offset=0,search=search,
+            )
+            seen={p.get("user_id") for p in inbox_payloads if p.get("user_id")}
+            manual_payloads=[]
+            for manual in manual_rows:
+                uid=str(manual.user_id) if manual.user_id else None
+                if uid and uid in seen:
+                    continue
+                user=await Users.get_user_by_id(self.session,manual.user_id) if manual.user_id else None
+                job_post=None
+                if manual.job_post_id:
+                    job_post=await JobPosts.get_job_post_by_id(self.session,str(manual.job_post_id))
+                payload=serialize_manual_candidate_profile(manual,user,job_post)
+                # List shape matches attach_job_posts: keep job_posts, drop heavy detail.
+                manual_payloads.append({
+                    "inbox_id":None,
+                    "manual_upload_candidate_id":payload["manual_upload_candidate_id"],
+                    "user_id":payload["user_id"],
+                    "candidate_id":None,
+                    "name":payload["name"],
+                    "email":payload["email"],
+                    "is_active":payload.get("is_active"),
+                    "message_id":None,
+                    "created_at":payload.get("created_at"),
+                    "application_status":payload.get("application_status"),
+                    "experience":payload.get("experience"),
+                    "current_employment":payload.get("current_employment"),
+                    "current_title":payload.get("current_title"),
+                    "resume_text":None,
+                    "suggested_job_post_ids":[],
+                    "assigned_job_post_id":payload.get("assigned_job_post_id"),
+                    "job_posts":payload.get("job_posts") or [],
+                    "assigned_job_post":payload.get("assigned_job_post"),
+                    "source":payload.get("source"),
+                    "ai_score":None,
+                    "recommendation":None,
+                })
+                if uid:
+                    seen.add(uid)
+            return inbox_payloads+manual_payloads
         except HTTPException:
             raise
         except Exception as e:

backend/users/views.py

@@ -54,7 +54,7 @@ class User:
         if not fields.get("password"):
             raise HTTPException(status_code=400,detail="Password is required")
         role=await Roles.get_role_by_name(self.session,EnumRoles.CANDIDATE.value)
-        fields["role_id"]=role.id if role else 8
+        fields["role_id"]=role.id if role else 4
         user=await Users.insert_user(self.session,fields)
         # Signup lands inactive; the mailed link is what flips is_active.
         service=Confirmation(session=self.session)

frontend/src/api/candidates.js

@@ -97,7 +97,7 @@ export function toCandidateView(row) {
  * Candidate USER accounts — `users` rows filtered by role, not the scored
  * `candidates` table. Needs candidates.view.
  *
- * role_id 8 is the seeded `candidate` role (backend/role/models.py::EnumRoles);
+ * role_id 4 is the seeded `candidate` role (backend/role/models.py::EnumRoles);
  * the route defaults to it, and we send it explicitly so a re-seed that renumbers
  * the roles fails loudly here rather than silently listing the wrong people.
  *
@@ -110,7 +110,7 @@ export function toCandidateView(row) {
  *    layer (`get_users(role_id=, top=, skip=)`), so searching is a no-op
  *    server-side. Filtering stays client-side until that is fixed.
  */
-export function listCandidateUsers({ roleId = 8, top = 500, skip = 0 } = {}) {
+export function listCandidateUsers({ roleId = 4, top = 500, skip = 0 } = {}) {
   return request('/candidate/fetch/users', {
     params: { role_id: roleId, top, skip },
   })

frontend/src/api/pipeline.js

@@ -153,6 +153,7 @@ function sourceFields(row, kind) {
       jobTitle: row.title ?? null,
       currentTitle: row.current_position || null,
       currentCompany: row.current_company || null,
+      source: row.platform || row.apply_via || 'Manual',
     }
   }
   return {
@@ -163,6 +164,7 @@ function sourceFields(row, kind) {
     jobTitle: row.title ?? null,
     currentTitle: row.current_title || null,
     currentCompany: row.current_employment || null,
+    source: null,
   }
 }
 

frontend/src/api/sheet.js

@@ -16,13 +16,21 @@ export function listFormDataSheets() {
  * Paginated form_data rows.
  *
  * `offset` / `limit` map 1:1 to the backend Query params (not skip/top).
+ * Optional `processing_state` / `is_duplicate` power the Sheet Forms tabs.
  */
-export function listFormData({ sheet, search, offset = 0, limit } = {}) {
+export function listFormData({
+  sheet, search, offset = 0, limit, processing_state, is_duplicate,
+} = {}) {
   return request('/sheet/form-data/fetch', {
-    params: { sheet, search, offset, limit },
+    params: { sheet, search, offset, limit, processing_state, is_duplicate },
   })
 }
 
+/** Tab badge counts for one sheet (or all sheets when sheet omitted). */
+export function fetchFormCounts({ sheet } = {}) {
+  return request('/sheet/form-data/counts', { params: { sheet } })
+}
+
 /** One form_data row by UUID. */
 export function getFormData(recordId) {
   return request(`/sheet/form-data/${recordId}`)
@@ -35,3 +43,18 @@ export function assignJobPost(recordId, jobPostId) {
     body: { job_post_id: jobPostId },
   })
 }
+
+/** unread | imported | processed | rejected — same allowlist as inbox. */
+export function setProcessingState(recordId, processingState) {
+  return request(`/sheet/form-data/${recordId}/processing-state`, {
+    method: 'PATCH',
+    body: { processing_state: processingState },
+  })
+}
+
+export function setDuplicate(recordId, isDuplicate) {
+  return request(`/sheet/form-data/${recordId}/duplicate`, {
+    method: 'PATCH',
+    body: { is_duplicate: isDuplicate },
+  })
+}

frontend/src/lib/queryKeys.js

@@ -33,6 +33,7 @@ export const qk = {
     formSheets: () => ['mailbox', 'form-sheets'],
     formData: (p = {}) => ['mailbox', 'form-data', p],
     formRow: (id) => ['mailbox', 'form-row', id],
+    formCounts: (p = {}) => ['mailbox', 'form-counts', p],
   },
   assessments: {
     all: () => ['assessments'],

frontend/src/screens/Candidates.jsx

@@ -36,7 +36,7 @@ const STAGE_ORDER = ['Shortlist', 'Screening', 'Assessment', 'Interview', 'Offer
 /** The seeded `candidate` role (backend/role/models.py::EnumRoles). */
 const CANDIDATE_ROLE_ID = 8
 
-/* Rows are candidate USER accounts (GET /candidate/fetch/users?role_id=8), not
+/* Rows are candidate USER accounts (GET /candidate/fetch/users?role_id=4), not
    rows of the scored `candidates` table.
 
    Why: /candidate/scored/fetch only ever returns CVs that have been through the
@@ -48,9 +48,22 @@ const CANDIDATE_ROLE_ID = 8
    toCandidateUserView. Open a candidate to get their score, which the shared
    Talent Pool profile modal (CandidateProfile.jsx) reads live by userId. */
 async function fetchCandidates() {
-  const res = await candidatesApi.listCandidateUsers({ roleId: CANDIDATE_ROLE_ID })
-  const rows = Array.isArray(res?.data) ? res.data : []
-  return rows.map(candidatesApi.toCandidateUserView)
+  const [usersRes, appsRes] = await Promise.all([
+    candidatesApi.listCandidateUsers({ roleId: CANDIDATE_ROLE_ID }),
+    candidatesApi.list({ limit: 500 }).catch(() => null),
+  ])
+  const rows = Array.isArray(usersRes?.data) ? usersRes.data : []
+  const sourceByUser = new Map()
+  for (const app of Array.isArray(appsRes?.data) ? appsRes.data : []) {
+    const uid = app.user_id
+    if (!uid || sourceByUser.has(uid)) continue
+    if (app.source) sourceByUser.set(String(uid), app.source)
+  }
+  return rows.map((row) => {
+    const view = candidatesApi.toCandidateUserView(row)
+    const source = sourceByUser.get(String(view.userId))
+    return source ? { ...view, source } : view
+  })
 }
 
 async function fetchJobs() {
@@ -386,7 +399,12 @@ export default function Candidates() {
                           <div className="user-cell">
                             <Avatar name={c.name} initials={initialsOf(c.name)} color={avatarColor(c.name)} />
                             <div>
-                              <div className="cell-primary">{c.name}</div>
+                              <div className="cell-primary">
+                                {c.name}
+                                {c.source === 'Form' && (
+                                  <Badge className="b-gray" style={{ marginLeft: 6, fontSize: 10 }}>Form</Badge>
+                                )}
+                              </div>
                               <div className="cell-sub">{c.roleName ?? '—'}</div>
                             </div>
                           </div>

frontend/src/screens/Inbox.jsx

@@ -26,6 +26,8 @@ import {
 } from '../data/seed'
 
 const TABS = ['All Applications', 'Unread', 'Processed', 'Rejected', 'Duplicates']
+/** Sheet Forms have no mailbox read state — no Unread tab on that channel. */
+const FORM_TABS = ['All Applications', 'Processed', 'Rejected', 'Duplicates']
 const PAGE_SIZE = 10
 
 /** Inbox channel: Outlook email queue vs imported Google Form rows. */
@@ -38,8 +40,9 @@ const DEFAULT_FORM_SHEET = 'Form Responses - Candidate Database Sheet 2026'
 const SHEET_SOURCE_META = { icon: 'layers', color: 'var(--c4)', channel: 'Sheet' }
 
 /**
- * Server-side filters for each tab. Processed / Rejected use
- * Candidate_application_Status (PROCESS / REJECTED), not processing_state.
+ * Server-side filters for each tab. Email Processed / Rejected use
+ * Candidate_application_Status (PROCESS / REJECTED). Sheet Forms use
+ * form_data.processing_state (same vocabulary as inbox Import/Reject).
  */
 const TAB_FILTERS = {
   Unread: { isread: false },
@@ -48,6 +51,19 @@ const TAB_FILTERS = {
   Duplicates: { isDuplicate: true },
 }
 
+const FORM_TAB_FILTERS = {
+  Processed: { processing_state: 'processed' },
+  Rejected: { processing_state: 'rejected' },
+  Duplicates: { is_duplicate: true },
+}
+
+const FORM_PROCESSING_LABEL = {
+  unread: 'New',
+  imported: 'Imported',
+  processed: 'Processed',
+  rejected: 'Rejected',
+}
+
 /** Every tab above is a true server-side scope — safe for "mark all". */
 const SERVER_SCOPED_TABS = new Set(TABS)
 
@@ -137,6 +153,7 @@ function formReceivedAt(entryDate, entryTime) {
 function mapFormRow(row) {
   const name = (row.name || row.candidate_email || 'Unknown').trim()
   const jobPosts = Array.isArray(row.job_posts) ? row.job_posts : []
+  const state = row.processing_state || 'unread'
   return {
     kind: 'form',
     id: String(row.id),
@@ -168,8 +185,11 @@ function mapFormRow(row) {
     resumeLink: row.resume_link || '',
     sheet: row.sheet || '',
     rowNumber: row.row_number ?? null,
-    unread: false,
-    processing: row.screened_by ? 'Screened' : 'New',
+    unread: state === 'unread',
+    processing: FORM_PROCESSING_LABEL[state]
+      || (row.screened_by ? 'Screened' : 'New'),
+    processingState: state,
+    duplicate: Boolean(row.is_duplicate),
     jobPosts,
     assignedId: row.job_post_id ? String(row.job_post_id) : null,
     assignedPost: row.assigned_job_post || null,
@@ -230,7 +250,7 @@ const RESUME_STATUS = {
   failed: 'Failed', dlq: 'Failed', skipped: 'Pending',
 }
 
-const SHORTLIST_JOB_WARNING = 'Choose one of the suggested jobs above to add this candidate to the shortlist.'
+const SHORTLIST_JOB_WARNING = 'Choose a matching job above to add this candidate to the shortlist.'
 
 /**
  * GET /inbox/fetch?record_id=<pk> -> the detail behind one application row.
@@ -656,8 +676,10 @@ export default function Inbox() {
   const [noting, setNoting] = useState(null)
 
   const isForms = channel === 'forms'
+  const channelTabs = isForms ? FORM_TABS : TABS
 
   const tabFilter = TAB_FILTERS[tab] ?? {}
+  const formTabFilter = FORM_TAB_FILTERS[tab] ?? {}
   const listParams = useMemo(() => ({
     ...tabFilter,
     top: PAGE_SIZE,
@@ -669,8 +691,9 @@ export default function Inbox() {
     sheet: formSheet || undefined,
     offset: (page - 1) * PAGE_SIZE,
     limit: PAGE_SIZE,
+    ...formTabFilter,
     ...(q.trim() ? { search: q.trim() } : {}),
-  }), [formSheet, page, q])
+  }), [formSheet, page, q, formTabFilter])
 
   const applicationsQuery = useQuery({
     queryKey: qk.mailbox.applications(listParams),
@@ -700,6 +723,15 @@ export default function Inbox() {
     enabled: !isForms,
   })
 
+  const formCountsQuery = useQuery({
+    queryKey: qk.mailbox.formCounts({ sheet: formSheet || undefined }),
+    queryFn: async () => {
+      const res = await sheetApi.fetchFormCounts({ sheet: formSheet || undefined })
+      return res?.data ?? {}
+    },
+    enabled: isForms,
+  })
+
   // Prefer the imported sheet list; keep the known 2026 tab even when the
   // sheets endpoint is still loading so the first paint is not blank.
   const formSheetOptions = useMemo(() => {
@@ -720,7 +752,7 @@ export default function Inbox() {
   const total = activeQuery.data?.total ?? 0
   const pages = Math.max(1, Math.ceil(total / PAGE_SIZE))
   const currentPage = Math.min(page, pages)
-  const serverCounts = countsQuery.data ?? {}
+  const serverCounts = isForms ? (formCountsQuery.data ?? {}) : (countsQuery.data ?? {})
 
   const counts = useMemo(
     () => ({
@@ -768,6 +800,8 @@ export default function Inbox() {
     setPage(1)
     setSelectedId(null)
     setQ('')
+    // Unread is email-only; leave it behind when opening Sheet Forms.
+    if (next === 'forms' && tab === 'Unread') setTab('All Applications')
     selection.clear()
   }
 
@@ -803,7 +837,11 @@ export default function Inbox() {
   }
 
   const setState = useMutation({
-    mutationFn: ({ id, state }) => inboxApi.setProcessingState(id, state),
+    mutationFn: ({ id, state, kind }) => (
+      kind === 'form'
+        ? sheetApi.setProcessingState(id, state)
+        : inboxApi.setProcessingState(id, state)
+    ),
     onSuccess: (_data, vars) => {
       const labels = { imported: 'Imported', processed: 'Processed', rejected: 'Rejected', unread: 'Unread' }
       toast(`${vars.name || 'Application'} marked ${labels[vars.state] || vars.state}`, vars.state === 'rejected' ? 'warning' : 'success')
@@ -813,11 +851,17 @@ export default function Inbox() {
     onSettled: () => {
       qc.invalidateQueries({ queryKey: qk.mailbox.all() })
       qc.invalidateQueries({ queryKey: qk.mailbox.counts() })
+      qc.invalidateQueries({ queryKey: qk.pipeline.all() })
+      qc.invalidateQueries({ queryKey: qk.candidates.all() })
     },
   })
 
   const markDuplicate = useMutation({
-    mutationFn: ({ id, isDuplicate }) => inboxApi.setDuplicate(id, isDuplicate),
+    mutationFn: ({ id, isDuplicate, kind }) => (
+      kind === 'form'
+        ? sheetApi.setDuplicate(id, isDuplicate)
+        : inboxApi.setDuplicate(id, isDuplicate)
+    ),
     onSuccess: (_d, vars) => toast(vars.isDuplicate ? 'Marked as duplicate' : 'Duplicate cleared', 'success'),
     onError: (err) => toast(friendlyAuthError(err, 'Could not update duplicate flag.'), 'error'),
     onSettled: () => {
@@ -834,19 +878,19 @@ export default function Inbox() {
   }
 
   function importItem(item) {
-    setState.mutate({ id: item.id, state: 'imported', name: item.name })
+    setState.mutate({ id: item.id, state: 'imported', name: item.name, kind: item.kind })
   }
 
   function moveToPipeline(item) {
-    setState.mutate({ id: item.id, state: 'processed', name: item.name })
+    setState.mutate({ id: item.id, state: 'processed', name: item.name, kind: item.kind })
   }
 
   function reject(item) {
-    setState.mutate({ id: item.id, state: 'rejected', name: item.name })
+    setState.mutate({ id: item.id, state: 'rejected', name: item.name, kind: item.kind })
   }
 
   function toggleDuplicate(item) {
-    markDuplicate.mutate({ id: item.id, isDuplicate: !item.duplicate })
+    markDuplicate.mutate({ id: item.id, isDuplicate: !item.duplicate, kind: item.kind })
   }
 
   const sync = useMutation({
@@ -918,7 +962,6 @@ export default function Inbox() {
       </div>
 
       <div className="card">
-        {!isForms && (
         <div style={{ margin: '0 16px', paddingTop: 8 }}>
           <Tabs
             value={tab}
@@ -928,10 +971,9 @@ export default function Inbox() {
               setSelectedId(null)
               selection.clear()
             }}
-              tabs={TABS.map((t) => ({ key: t, label: t, count: counts[t] }))}
+            tabs={channelTabs.map((t) => ({ key: t, label: t, count: counts[t] }))}
           />
         </div>
-        )}
 
         <div className="split inbox-split">
             <div className="split-list inbox-queue">
@@ -1072,8 +1114,14 @@ export default function Inbox() {
                 <FormApplicantDetail
                   item={selected}
                   loading={detailQuery.isPending}
+                  busy={setState.isPending || markDuplicate.isPending}
                   canEdit={canEdit}
                   toast={toast}
+                  onImport={() => importItem(selected)}
+                  onMove={() => moveToPipeline(selected)}
+                  onNote={() => setNoting(selected)}
+                  onReject={() => reject(selected)}
+                  onToggleDuplicate={() => toggleDuplicate(selected)}
                 />
               ) : (
                 <ApplicationDetail
@@ -1169,9 +1217,13 @@ function externalHref(url) {
 
 /**
  * Sheet form applicant detail — profile grids + resume/LinkedIn links +
- * title-matched job selection (position_applied_for ↔ job_posts.title).
+ * title-matched job selection (position_applied_for ↔ job_posts.title) +
+ * the same Import / Shortlist / Duplicate / Reject actions as email.
  */
-function FormApplicantDetail({ item: i, loading, canEdit, toast }) {
+function FormApplicantDetail({
+  item: i, loading, busy, canEdit, toast,
+  onImport, onMove, onNote, onReject, onToggleDuplicate,
+}) {
   const qc = useQueryClient()
   const resumeHref = externalHref(i.resumeLink)
   const profileHref = externalHref(i.profileLink)
@@ -1213,11 +1265,34 @@ function FormApplicantDetail({ item: i, loading, canEdit, toast }) {
     onSettled: (_res, _err, vars) => {
       qc.invalidateQueries({ queryKey: qk.mailbox.all() })
       qc.invalidateQueries({ queryKey: qk.mailbox.formRow(vars.recordId) })
+      qc.invalidateQueries({ queryKey: qk.pipeline.all() })
+      qc.invalidateQueries({ queryKey: qk.candidates.all() })
     },
   })
 
   const assigned = i.assignedPost
   const canAssign = canEdit && selection && selection !== i.assignedId && !assignMutation.isPending
+  const jobChosen = Boolean(i.assignedId || selection)
+  const alreadyProcessed = i.processing === 'Processed'
+  const shortlistLocked = !alreadyProcessed && !jobChosen
+  const panelBusy = busy || assignMutation.isPending
+
+  async function handleMove() {
+    if (busy || alreadyProcessed) return
+    if (!jobChosen) {
+      toast(SHORTLIST_JOB_WARNING, 'warning')
+      return
+    }
+    const jobId = selection || i.assignedId
+    if (jobId && String(jobId) !== String(i.assignedId || '')) {
+      try {
+        await assignMutation.mutateAsync({ recordId: i.id, jobPostId: jobId })
+      } catch {
+        return
+      }
+    }
+    onMove()
+  }
 
   return (
     <div style={{ padding: 24 }}>
@@ -1228,6 +1303,7 @@ function FormApplicantDetail({ item: i, loading, canEdit, toast }) {
           <div className="ph-role">{i.position}</div>
           <div className="ph-tags" style={{ marginTop: 8 }}>
             <SourceChip item={i} /> <Badge>{i.processing}</Badge>{' '}
+            {i.duplicate && <><Badge className="b-red">Duplicate</Badge>{' '}</>}
             {i.hoAvailability && (
               <Badge className={String(i.hoAvailability).toLowerCase() === 'yes' ? 'b-green' : 'b-amber'}>
                 Relocate: {i.hoAvailability}
@@ -1435,6 +1511,36 @@ function FormApplicantDetail({ item: i, loading, canEdit, toast }) {
         </div>
       </div>
 
+      <div className="flex gap-8" style={{ flexWrap: 'wrap' }}>
+        <button className="btn btn-primary" onClick={onImport} disabled={panelBusy || i.processing === 'Imported'}>
+          <Icon name="user-plus" /> {i.processing === 'Imported' ? 'Imported' : 'Import Candidate'}
+        </button>
+        <button
+          className="btn btn-secondary"
+          aria-disabled={shortlistLocked || panelBusy || alreadyProcessed}
+          disabled={panelBusy || alreadyProcessed}
+          style={shortlistLocked ? { opacity: 0.55, cursor: 'not-allowed' } : undefined}
+          title={shortlistLocked ? SHORTLIST_JOB_WARNING : undefined}
+          onClick={handleMove}
+        >
+          <Icon name="layers" /> Move to Shortlist
+        </button>
+        <button className="btn btn-secondary" onClick={onNote} disabled title="Notes attach to a candidate profile — open the candidate first">
+          <Icon name="edit" /> Add Note
+        </button>
+        <button className="btn btn-secondary" onClick={onToggleDuplicate} disabled={panelBusy}>
+          <Icon name="alert" /> {i.duplicate ? 'Clear Duplicate' : 'Mark Duplicate'}
+        </button>
+        <button
+          className="btn btn-ghost"
+          style={{ color: 'var(--danger)' }}
+          onClick={onReject}
+          disabled={panelBusy || i.processing === 'Rejected'}
+        >
+          <Icon name="trash" /> Reject
+        </button>
+      </div>
+
       {showPicker && (
         <PickRoleModal
           onClose={() => setShowPicker(false)}

frontend/src/screens/Pipeline.jsx

@@ -18,7 +18,7 @@ import { useMemo, useState } from 'react'
 import { useNavigate } from 'react-router-dom'
 import { keepPreviousData, useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
 
-import { Avatar, EmptyState, Icon, ScoreChip } from '../ui/primitives'
+import { Avatar, Badge, EmptyState, Icon, ScoreChip } from '../ui/primitives'
 import { useToast } from '../ui/Toast'
 import { useAuth } from '../auth/AuthContext'
 import { qk } from '../lib/queryKeys'
@@ -250,7 +250,12 @@ export default function Pipeline() {
                       <div className="k-card-top">
                         <Avatar name={c.name} />
                         <div>
-                          <div className="kc-name">{c.name}</div>
+                          <div className="kc-name" style={{ display: 'flex', alignItems: 'center', gap: 10, flexWrap: 'wrap' }}>
+                            <span>{c.name}</span>
+                            {c.source === 'Form' && (
+                              <Badge className="b-gray" style={{ fontSize: 10 }}>Form</Badge>
+                            )}
+                          </div>
                           <div className="kc-role">{c.currentTitle}</div>
                         </div>
                       </div>

frontend/src/screens/TalentPool.jsx

@@ -77,6 +77,8 @@ function years(value) {
 function merge(row, template) {
   const name = row.name || template.name
   const title = (row.job_posts || []).map((j) => j.title).find(Boolean)
+    || row.job_title
+    || row.current_title
   const stage = STAGE_FROM_STATUS[row.application_status] || template.stage
   const experience = years(row.experience)
 
@@ -92,6 +94,8 @@ function merge(row, template) {
     status: stage,
     currentTitle: title || template.currentTitle,
     jobTitle: title || template.jobTitle,
+    // Prefer real Form / platform tags from manual_upload; seed only as fallback.
+    source: row.source || template.source,
     // NO seed fallback. `ai_score` is the candidate's current ats_results row,
     // resolved server-side; null means the scoring engine never scored this
     // person, and the card renders nothing rather than a plausible fake number