Merge pull request 'Add mailbox sync functionality and related API endpoints' (#21) from Add_History into main

Reviewed-on: #21

backend/inbox/app.py

@@ -47,6 +47,7 @@ class ReadAllBody(BaseModel):
     isread: bool = True
     application_status: Candidate_application_Status = Candidate_application_Status.CLOSED
     assigned: bool | None = None
+    is_duplicate: bool | None = None
 
 
 class TriageOverrideBody(BaseModel):
@@ -73,36 +74,67 @@ async def fetch_email(
     token: str | None = Query(None),
     session: AsyncSession = Depends(get_session),
 ):
+    """Synchronous one-shot sync — kept for scripts/compat. UI uses POST /email/sync."""
     try:
         service=Email(session=session,token=token)
         if not service.token:
             raise HTTPException(status_code=401,detail="Unauthorized")
-        data=await service.service_email(top,skip)
-        value=data.get("value")
-        items_lst=[]
-        # Classify the whole page first, bounded-parallel, then replay it in upstream
-        # order: the inserts stay serial on the one request session and pending_match_ids
-        # keeps the sequence it has today.
-        decisions=await service.triage_round([item.get("id") for item in value])
-        for item in value:
-            message_id=item.get("id")
-            service_per_email=await service.get_email_by_id(message_id,test_on,decision=decisions.get(str(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)
-
-        skipped=len(service.skipped_message_ids)
-        triage={"ingested":len(items_lst)-skipped,"skipped":skipped,"errors":len(service.triage_errors)}
-
-        account_setup=[]
+        summary=await service.run_mailbox_sync_page(top=top,skip=skip,test_on=test_on)
         if test_on:
-            return JSONResponse(content={"data":items_lst,"triage":triage,"status_code":200})
-        if service.pending_confirmation_emails:
-            account_setup=await service.send_account_setup(list(service.pending_confirmation_emails))
+            return JSONResponse(content={
+                "data":summary["entries"],
+                "triage":summary["triage"],
+                "status_code":200,
+            })
+        return JSONResponse(content={
+            "data":summary["entries"],
+            "account_setup":summary["account_setup"],
+            "triage":summary["triage"],
+            "status_code":200,
+        })
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
 
-        return JSONResponse(content={"data":items_lst,"account_setup":account_setup,"triage":triage,"status_code":200})
 
+@router.post("/email/sync")
+async def start_email_sync(
+    top:int=Query(100),
+    skip:int=Query(0,ge=0),
+    test_on: bool = Query(True),
+    current_user: dict = Depends(require_permission(PermissionTag.INBOX_EDIT)),
+    session: AsyncSession = Depends(get_session),
+):
+    """Enqueue mailbox sync on the dedicated mailbox_sync Taskiq queue.
+
+    Returns immediately with a run id. Poll GET /email/sync/fetch until completed.
+    Closing the browser does not cancel the worker.
+    """
+    try:
+        service=Email(session=session)
+        if not service.token:
+            raise HTTPException(status_code=401,detail="Unauthorized")
+        data=await service.start_mailbox_sync(
+            current_user=current_user,top=top,skip=skip,test_on=test_on,
+        )
+        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("/email/sync/fetch")
+async def fetch_email_sync(
+    run_id: str | None = Query(None),
+    current_user: dict = Depends(require_permission(PermissionTag.INBOX_VIEW)),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=Email(session=session)
+        data=await service.get_mailbox_sync(run_id=run_id)
+        return JSONResponse(content={"data":data,"total":1,"status_code":200})
     except HTTPException:
         raise
     except Exception as e:
@@ -218,6 +250,7 @@ async def mark_all_inbox_read(
             isread=payload.isread,
             application_status=payload.application_status,
             assigned=payload.assigned,
+            is_duplicate=payload.is_duplicate,
         )
         return JSONResponse(content={"data":data,"total":data["updated"],"status_code":200})
     except HTTPException:
@@ -248,6 +281,7 @@ async def get_all_applications(
     application_status: Candidate_application_Status = Query(default=Candidate_application_Status.CLOSED),
     isread: bool = Query(default=True),
     assigned: bool | None = Query(default=None),
+    is_duplicate: bool | None = Query(default=None),
     search: str | None = Query(None),
     top: int | None = Query(None),
     skip: int = Query(0, ge=0),
@@ -258,19 +292,19 @@ async def get_all_applications(
         service=Email(session=session)
         
         if application_status in (Candidate_application_Status.PROCESS, Candidate_application_Status.REJECTED, Candidate_application_Status.SCREENING, Candidate_application_Status.ASSESSMENT, Candidate_application_Status.INTERVIEW, Candidate_application_Status.OFFER, Candidate_application_Status.HIRED):
-            items=await service.get_all_applications(top, skip, search, application_status=application_status, assigned=assigned)
-            total=await service.count_inbox_messages(search, application_status=application_status, assigned=assigned)
+            items=await service.get_all_applications(top, skip, search, application_status=application_status, assigned=assigned, is_duplicate=is_duplicate)
+            total=await service.count_inbox_messages(search, application_status=application_status, assigned=assigned, is_duplicate=is_duplicate)
             return JSONResponse(content={"data":items,"total":total,"status_code":200})
         if isread==False:
-            items=await service.get_all_applications(top, skip, search, isread=False, assigned=assigned)
-            total=await service.count_inbox_messages(search, isread=False, assigned=assigned)
+            items=await service.get_all_applications(top, skip, search, isread=False, assigned=assigned, is_duplicate=is_duplicate)
+            total=await service.count_inbox_messages(search, isread=False, assigned=assigned, is_duplicate=is_duplicate)
             return JSONResponse(content={"data":items,"total":total,"status_code":200})
         if record_id:
             item=await service.get_application_by_id(record_id)
             return JSONResponse(content={"data":item,"total":1,"status_code":200})
 
-        items=await service.get_all_applications(top,skip,search,assigned=assigned)
-        total=await service.count_inbox_messages(search,assigned=assigned)
+        items=await service.get_all_applications(top,skip,search,assigned=assigned,is_duplicate=is_duplicate)
+        total=await service.count_inbox_messages(search,assigned=assigned,is_duplicate=is_duplicate)
         return JSONResponse(content={"data":items,"total":total,"status_code":200})
     except HTTPException:
         raise

backend/inbox/models.py

@@ -353,6 +353,13 @@ class Inbox_Messages(SQLModel, table=True):
     is_duplicate: bool = Field(default=False, sa_column_kwargs={"server_default": "false"})
     source_channel_id: int | None = Field(default=None, foreign_key="source_channels.id")
     processing_state: str = Field(default="unread", sa_column_kwargs={"server_default": "unread"})
+    # Ingestion time — list screens order by this (newest first). server_default
+    # backfills existing rows on the ALTER so NOT NULL is safe on a populated table.
+    created_at: datetime = Field(
+        default_factory=_now,
+        sa_type=DateTime(timezone=True),
+        sa_column_kwargs={"server_default": "now()"},
+    )
     inbox: list[Inbox] = Relationship(back_populates="messages")
 
     @staticmethod
@@ -577,6 +584,7 @@ class Inbox_Messages(SQLModel, table=True):
         cls, statement, search: str | None=None, isread: bool=True,
         application_status: Candidate_application_Status=Candidate_application_Status.CLOSED,
         assigned: bool | None=None,
+        is_duplicate: bool | None=None,
     ):
         """The one WHERE chain shared by the list, the count and the bulk read UPDATE.
 
@@ -595,15 +603,21 @@ class Inbox_Messages(SQLModel, table=True):
             statement = statement.where(cls.assigned_job_post_id.is_(None))
         if isread==False:
             statement = statement.where(cls.message_read==False)
+        if is_duplicate is True:
+            statement = statement.where(cls.is_duplicate==True)  # noqa: E712
+        elif is_duplicate is False:
+            statement = statement.where(cls.is_duplicate==False)  # noqa: E712
         return statement
 
     @classmethod
     async def get_inbox_messages(
-        cls, session: AsyncSession, top: int | None, skip: int, search: str | None, isread: bool=True, application_status: Candidate_application_Status=Candidate_application_Status.CLOSED, assigned: bool | None=None
+        cls, session: AsyncSession, top: int | None, skip: int, search: str | None, isread: bool=True, application_status: Candidate_application_Status=Candidate_application_Status.CLOSED, assigned: bool | None=None, is_duplicate: bool | None=None
     ):
+        # Page size is the caller's `top` (Inbox sends 10); `skip` is (page-1)*top
+        # so page 1 -> 0..9, page 2 -> 10..19. Newest first via created_at.
         statement = cls._apply_filters(
-            select(cls).order_by(cls.message_received_time.desc()),
-            search, isread, application_status, assigned,
+            select(cls).order_by(cls.created_at.desc()),
+            search, isread, application_status, assigned, is_duplicate,
         )
         if skip:
             statement = statement.offset(skip)
@@ -654,10 +668,10 @@ class Inbox_Messages(SQLModel, table=True):
         return row
 
     @classmethod
-    async def count_inbox_messages(cls, session: AsyncSession, search: str | None, isread: bool=True, application_status: Candidate_application_Status=Candidate_application_Status.CLOSED, assigned: bool | None=None):
+    async def count_inbox_messages(cls, session: AsyncSession, search: str | None, isread: bool=True, application_status: Candidate_application_Status=Candidate_application_Status.CLOSED, assigned: bool | None=None, is_duplicate: bool | None=None):
         statement = cls._apply_filters(
             select(func.count()).select_from(cls),
-            search, isread, application_status, assigned,
+            search, isread, application_status, assigned, is_duplicate,
         )
         result = await session.execute(statement)
         return result.scalar_one()
@@ -760,6 +774,7 @@ class Inbox_Messages(SQLModel, table=True):
         cls, session: AsyncSession, read: bool, search: str | None=None, isread: bool=True,
         application_status: Candidate_application_Status=Candidate_application_Status.CLOSED,
         assigned: bool | None=None,
+        is_duplicate: bool | None=None,
     ) -> int:
         """Mark every row matching a list filter. Returns rows actually CHANGED.
 
@@ -768,7 +783,7 @@ class Inbox_Messages(SQLModel, table=True):
         was already read. It also keeps read_overridden_at off rows nobody decided
         anything about, so the Outlook sweep keeps its reach over untouched mail.
         """
-        statement=cls._apply_filters(update(cls),search,isread,application_status,assigned)
+        statement=cls._apply_filters(update(cls),search,isread,application_status,assigned,is_duplicate)
         statement=statement.where(cls.message_read!=bool(read))
         result=await session.execute(
             statement.values(message_read=bool(read),read_overridden_at=_now())
@@ -782,8 +797,8 @@ class Inbox_Messages(SQLModel, table=True):
             func.count().label("all_count"),
             func.coalesce(func.sum(case((cls.message_read == False, 1), else_=0)), 0).label("unread"),  # noqa: E712
             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.application_status == Candidate_application_Status.PROCESS, 1), else_=0)), 0).label("processed"),
+            func.coalesce(func.sum(case((cls.application_status == Candidate_application_Status.REJECTED, 1), else_=0)), 0).label("rejected"),
             func.coalesce(func.sum(case((cls.is_duplicate == True, 1), else_=0)), 0).label("duplicates"),  # noqa: E712
             func.coalesce(func.sum(case((cls.assigned_job_post_id.is_not(None), 1), else_=0)), 0).label("assigned"),
             func.coalesce(func.sum(case((cls.assigned_job_post_id.is_(None), 1), else_=0)), 0).label("unassigned"),
@@ -1148,3 +1163,76 @@ class AtsResults(SQLModel, table=True):
                 session.add(link)
         await session.commit()
         return row
+
+
+class MailboxSyncRun(SQLModel, table=True):
+    """One Outlook mailbox sync job — survives tab close because work runs in Taskiq.
+
+    The Sync button enqueues a run and returns immediately. The UI polls this row for
+    status / per-message entries / triage totals. Closing the browser does not cancel
+    the worker.
+    """
+
+    __tablename__ = "mailbox_sync_runs"
+
+    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+    status: str = Field(default="queued", index=True)  # queued|running|completed|failed
+    task_id: str | None = Field(default=None)
+    created_by: uuid.UUID | None = Field(default=None, foreign_key="users.id")
+    top: int = Field(default=100)
+    skip: int = Field(default=0)
+    test_on: bool = Field(default=True)
+    triage: dict | None = Field(default=None, sa_column=Column(JSONB))
+    entries: list | None = Field(default=None, sa_column=Column(JSONB))
+    error: str | None = Field(default=None)
+    created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
+    started_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
+    finished_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
+
+    @staticmethod
+    def _as_uuid(record_id) -> uuid.UUID | None:
+        if record_id in (None, ""):
+            return None
+        try:
+            return uuid.UUID(str(record_id))
+        except ValueError:
+            return None
+
+    @classmethod
+    async def get_by_id(cls, session: AsyncSession, record_id):
+        uid = cls._as_uuid(record_id)
+        if uid is None:
+            return None
+        result = await session.execute(select(cls).where(cls.id == uid))
+        return result.scalars().first()
+
+    @classmethod
+    async def get_active(cls, session: AsyncSession):
+        result = await session.execute(
+            select(cls)
+            .where(cls.status.in_(("queued", "running")))
+            .order_by(cls.created_at.desc())
+        )
+        return result.scalars().first()
+
+    @classmethod
+    async def insert_run(cls, session: AsyncSession, fields: dict, *, commit: bool = True):
+        row = cls(**fields)
+        session.add(row)
+        if commit:
+            await session.commit()
+            await session.refresh(row)
+        return row
+
+    @classmethod
+    async def update_run(cls, session: AsyncSession, record_id, fields: dict, *, commit: bool = True):
+        row = await cls.get_by_id(session, record_id)
+        if not row:
+            return None
+        for key, value in fields.items():
+            setattr(row, key, value)
+        session.add(row)
+        if commit:
+            await session.commit()
+            await session.refresh(row)
+        return row

backend/inbox/serializers.py

@@ -157,3 +157,21 @@ def serialize_triage(row: Inbox_Message_Triage) -> dict:
         "overridden_at": row.overridden_at.isoformat() if row.overridden_at else None,
         "classified_at": row.classified_at.isoformat() if row.classified_at else None,
     }
+
+
+def serialize_mailbox_sync_run(row) -> dict:
+    return {
+        "id": str(row.id),
+        "status": row.status,
+        "task_id": row.task_id,
+        "created_by": str(row.created_by) if row.created_by else None,
+        "top": row.top,
+        "skip": row.skip,
+        "test_on": row.test_on,
+        "triage": row.triage,
+        "entries": row.entries or [],
+        "error": row.error,
+        "created_at": row.created_at.isoformat() if row.created_at else None,
+        "started_at": row.started_at.isoformat() if row.started_at else None,
+        "finished_at": row.finished_at.isoformat() if row.finished_at else None,
+    }

backend/inbox/views.py

@@ -4,9 +4,9 @@ import uuid
 import httpx,os
 from fastapi import HTTPException
 from inbox.enums import Candidate_application_Status
-from inbox.models import Inbox_Messages,Inbox_Message_Triage
+from inbox.models import Inbox_Messages,Inbox_Message_Triage,MailboxSyncRun
 from inbox.file_decoder import decode_attachment
-from inbox.serializers import serialize_application, serialize_message, serialize_triage
+from inbox.serializers import serialize_application, serialize_message, serialize_triage, serialize_mailbox_sync_run
 from inbox.plugins import (
     EMAIL_API_TOKEN,
     fetch_message_read_status,
@@ -270,13 +270,13 @@ class Email:
             item["assigned_job_post"]=None
         return item
 
-    async def get_all_applications(self,top,skip,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,assigned=None):
+    async def get_all_applications(self,top,skip,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,assigned=None,is_duplicate=None):
         if application_status in (Candidate_application_Status.PROCESS, Candidate_application_Status.REJECTED, Candidate_application_Status.SCREENING, Candidate_application_Status.ASSESSMENT, Candidate_application_Status.INTERVIEW, Candidate_application_Status.OFFER, Candidate_application_Status.HIRED):
-            messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,application_status=application_status,assigned=assigned)
+            messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,application_status=application_status,assigned=assigned,is_duplicate=is_duplicate)
         elif isread==False:
-            messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,isread,assigned=assigned)
+            messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,isread,assigned=assigned,is_duplicate=is_duplicate)
         else:
-            messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,assigned=assigned)
+            messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,assigned=assigned,is_duplicate=is_duplicate)
         return [serialize_application(m) for m in messages]
 
     async def get_application_by_id(self,record_id):
@@ -306,6 +306,115 @@ class Email:
             task_ids.append(task.task_id)
         return task_ids
 
+    async def start_mailbox_sync(self,current_user=None,top=100,skip=0,test_on=True):
+        """Enqueue Outlook pull on the mailbox_sync queue; return the run row.
+
+        If a queued/running sync already exists, return it instead of stacking another.
+        """
+        active=await MailboxSyncRun.get_active(self.session)
+        if active:
+            return serialize_mailbox_sync_run(active)
+
+        created_by=None
+        if isinstance(current_user,dict) and current_user.get("id"):
+            created_by=MailboxSyncRun._as_uuid(current_user.get("id"))
+
+        row=await MailboxSyncRun.insert_run(self.session,{
+            "status":"queued",
+            "created_by":created_by,
+            "top":int(top or 100),
+            "skip":int(skip or 0),
+            "test_on":bool(test_on) if test_on is not None else True,
+        })
+
+        from inbox.mailbox_sync_tasks import sync_mailbox
+        task=await sync_mailbox.kicker().with_labels(
+            created_at=datetime.now(timezone.utc).isoformat(),
+            correlation_id=str(row.id),
+            queue="mailbox_sync",
+        ).kiq(str(row.id))
+        row=await MailboxSyncRun.update_run(self.session,row.id,{"task_id":task.task_id})
+        return serialize_mailbox_sync_run(row)
+
+    async def get_mailbox_sync(self,run_id=None):
+        if run_id:
+            row=await MailboxSyncRun.get_by_id(self.session,run_id)
+            if not row:
+                raise HTTPException(status_code=404,detail="Sync run not found")
+            return serialize_mailbox_sync_run(row)
+        row=await MailboxSyncRun.get_active(self.session)
+        if row:
+            return serialize_mailbox_sync_run(row)
+        # Latest finished run so the UI can still show the last result after refresh.
+        from sqlmodel import select
+        result=await self.session.execute(
+            select(MailboxSyncRun).order_by(MailboxSyncRun.created_at.desc()).limit(1)
+        )
+        row=result.scalars().first()
+        if not row:
+            raise HTTPException(status_code=404,detail="No sync runs yet")
+        return serialize_mailbox_sync_run(row)
+
+    async def run_mailbox_sync_page(self,top=100,skip=0,test_on=True):
+        """Pull one Outlook page, triage, ingest, enqueue matching. Returns summary.
+
+        Shared by the legacy synchronous /email/fetch and the background Taskiq worker.
+        """
+        data=await self.service_email(top,skip)
+        value=data.get("value") or []
+        decisions=await self.triage_round([item.get("id") for item in value])
+        entries=[]
+        for item in value:
+            message_id=item.get("id")
+            decision=decisions.get(str(message_id)) or {}
+            try:
+                result=await self.get_email_by_id(message_id,test_on,decision=decision)
+                if isinstance(result,dict) and result.get("skipped"):
+                    entries.append({
+                        "message_id":str(message_id),
+                        "status":"skipped",
+                        "reason":result.get("reason") or "",
+                        "triage_status":result.get("status") or "",
+                    })
+                else:
+                    # Prefer the decision status (known/recorded/…) when present.
+                    status=decision.get("status") or "ingested"
+                    if status in ("known","recorded","disabled"):
+                        entry_status="known" if status=="known" else "ingested"
+                    else:
+                        entry_status="ingested"
+                    entries.append({
+                        "message_id":str(message_id),
+                        "status":entry_status,
+                        "reason":decision.get("reason") or "",
+                        "triage_status":status,
+                    })
+            except Exception as e:
+                entries.append({
+                    "message_id":str(message_id),
+                    "status":"error",
+                    "reason":str(e),
+                    "triage_status":"error",
+                })
+                self.triage_errors.append(str(message_id))
+
+        if self.pending_match_ids:
+            await self.enqueue_matching(list(self.pending_match_ids),force=False)
+
+        account_setup=[]
+        if not test_on and self.pending_confirmation_emails:
+            account_setup=await self.send_account_setup(list(self.pending_confirmation_emails))
+
+        skipped=len(self.skipped_message_ids)
+        ingested=sum(1 for e in entries if e.get("status") in ("ingested","known"))
+        triage={
+            "ingested":ingested,
+            "skipped":skipped,
+            "errors":len(self.triage_errors),
+            "total":len(entries),
+        }
+        return {"entries":entries,"triage":triage,"account_setup":account_setup}
+
     async def send_account_setup(self,emails):
         results=[]
         for email in emails or []:
@@ -317,13 +426,13 @@ class Email:
                 results.append({"email":email,"sent":False})
         return results
 
-    async def count_inbox_messages(self,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,assigned=None):
+    async def count_inbox_messages(self,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,assigned=None,is_duplicate=None):
         if application_status == Candidate_application_Status.PROCESS or application_status==Candidate_application_Status.REJECTED:
-            return await Inbox_Messages.count_inbox_messages(self.session,search,application_status=application_status,assigned=assigned)
+            return await Inbox_Messages.count_inbox_messages(self.session,search,application_status=application_status,assigned=assigned,is_duplicate=is_duplicate)
         elif isread==False:
-            return await Inbox_Messages.count_inbox_messages(self.session,search,isread=False,assigned=assigned)
+            return await Inbox_Messages.count_inbox_messages(self.session,search,isread=False,assigned=assigned,is_duplicate=is_duplicate)
         else:
-            return await Inbox_Messages.count_inbox_messages(self.session,search,assigned=assigned)
+            return await Inbox_Messages.count_inbox_messages(self.session,search,assigned=assigned,is_duplicate=is_duplicate)
 
     async def assign_job_post(self,record_id,job_post_id):
         message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
@@ -381,7 +490,7 @@ class Email:
 
     async def set_read_all(self,read,search=None,isread:bool=True,
                            application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,
-                           assigned=None):
+                           assigned=None,is_duplicate=None):
         """Mark every row the SAME filter set would have listed.
 
         The filter arguments are the caller's current view, not a free-form query: the
@@ -390,10 +499,10 @@ class Email:
         """
         updated=await Inbox_Messages.set_read_scope(
             self.session,read,search=search,isread=isread,
-            application_status=application_status,assigned=assigned,
+            application_status=application_status,assigned=assigned,is_duplicate=is_duplicate,
         )
-        logger.info("scope read: updated=%s read=%s isread=%s assigned=%s status=%s",
-                    updated,bool(read),isread,assigned,getattr(application_status,"value",application_status))
+        logger.info("scope read: updated=%s read=%s isread=%s assigned=%s status=%s dup=%s",
+                    updated,bool(read),isread,assigned,getattr(application_status,"value",application_status),is_duplicate)
         return {"updated":updated,"read":bool(read)}
 
     async def refresh_read_status(self,record_id):

docker-compose.dev.yml

@@ -58,3 +58,8 @@ services:
     volumes:
       - ./backend:/app
       - ./app:/app/app
+
+  taskiq-mailbox-sync-worker:
+    volumes:
+      - ./backend:/app
+      - ./app:/app/app

docker-compose.yml

@@ -231,6 +231,25 @@ services:
       <<: *backend-env
       TASKIQ_CV_QUEUE_NAME: cv_upload
 
+  # Dedicated stream: Outlook pull/triage must not block match/ATS or CV uploads.
+  taskiq-mailbox-sync-worker:
+    <<: *backend-service
+    container_name: hrms-taskiq-mailbox-sync-worker
+    command:
+      [
+        "taskiq",
+        "worker",
+        "taskiq_management.mailbox_sync_broker_setup:mailbox_sync_broker",
+        "inbox.mailbox_sync_tasks",
+        "--workers",
+        "1",
+      ]
+    environment:
+      <<: *backend-env
+      TASKIQ_MAILBOX_SYNC_QUEUE_NAME: mailbox_sync
+      TASKIQ_WORKER_NAME: mailbox-sync-worker-01
+    volumes: *attachments
+
   # --- Postgres: DEFINED HERE, NOT STARTED BY DEFAULT -------------------------------
   # The profile is what keeps it out of `docker compose build` and `docker compose up`.
   # Nothing about the default stack changes by its presence in this file.

frontend/src/api/inbox.js

@@ -20,13 +20,14 @@ export function listMessages() {
  * `assigned` is tri-valued: omit for no filter, true for rows with an
  * assigned_job_post_id, false for the Job Matching queue.
  */
-export function listApplications({ search, top, skip, recordId, isread, applicationStatus, assigned } = {}) {
+export function listApplications({ search, top, skip, recordId, isread, applicationStatus, assigned, isDuplicate } = {}) {
   return request('/inbox/all-applications', {
     // `isread` is tri-valued on the wire: omit it for every tab (server defaults
     // to true = no filter), send false for the Unread tab only. buildUrl drops
     // undefined but keeps false, so `isread: undefined` sends no param at all.
     // Same for `application_status`: omit for every tab (server defaults to
     // CLOSED = no filter), send PROCESS / REJECTED for those tabs only.
+    // Same for `is_duplicate`: omit unless the Duplicates tab.
     params: {
       search,
       top,
@@ -35,6 +36,7 @@ export function listApplications({ search, top, skip, recordId, isread, applicat
       isread,
       application_status: applicationStatus,
       assigned,
+      is_duplicate: isDuplicate,
     },
   })
 }
@@ -51,7 +53,23 @@ export function getMessage(recordId) {
   return request('/inbox/fetch', { params: { record_id: recordId } })
 }
 
-/** Triggers the Graph proxy to pull new mail and persist it. */
+/** Enqueue Outlook pull on the mailbox_sync worker. Returns a run immediately. */
+export function startMailboxSync({ top, skip, testOn } = {}) {
+  return request('/email/sync', {
+    method: 'POST',
+    params: { top, skip, test_on: testOn },
+  })
+}
+
+/** Poll one sync run (or the active/latest run when runId is omitted). */
+export function getMailboxSync(runId) {
+  return request('/email/sync/fetch', { params: { run_id: runId } })
+}
+
+/**
+ * Legacy synchronous sync — blocks until the page is ingested. Prefer
+ * startMailboxSync + getMailboxSync so closing the tab cannot kill the job.
+ */
 export function syncMailbox({ token, top, skip } = {}) {
   return request('/email/fetch', { params: { token, top, skip } })
 }
@@ -96,7 +114,7 @@ export function bulkSetRead(recordIds, read) {
  * Resolves to `{updated, read}`, where `updated` counts rows that actually
  * CHANGED state, so it is safe to show in a toast.
  */
-export function setReadAll({ read, search, isread, applicationStatus, assigned } = {}) {
+export function setReadAll({ read, search, isread, applicationStatus, assigned, isDuplicate } = {}) {
   return request('/inbox/read-all', {
     method: 'PATCH',
     body: {
@@ -105,6 +123,7 @@ export function setReadAll({ read, search, isread, applicationStatus, assigned }
       isread,
       application_status: applicationStatus,
       assigned,
+      is_duplicate: isDuplicate,
     },
   })
 }

frontend/src/lib/queryKeys.js

@@ -27,6 +27,7 @@ export const qk = {
     // qk.mailbox.all() already refreshes the intake gate's ledger, so an
     // override or a sync needs no extra invalidation.
     triage: (p = {}) => ['mailbox', 'triage', p],
+    sync: (id) => ['mailbox', 'sync', id],
   },
   assessments: {
     all: () => ['assessments'],

frontend/src/screens/Inbox.jsx

@@ -1,12 +1,6 @@
 /* ============================================================
-   Recruitment Inbox — application tabs plus the Email tab, which is the
-   app's oldest real network call (GET /inbox/fetch, previously the only fetch
-   in the entire prototype).
-
-   The email body used to be interpolated raw into markup at js/inbox.js:292 —
-   the single widest XSS sink in the repository, and the one that mattered most
-   because inbound mail is attacker-supplied by definition. It renders as text
-   now, which is the structural fix.
+   Recruitment Inbox — application tabs over GET /inbox/all-applications.
+   Page size is 10, newest first (order by created_at on the server).
    ============================================================ */
 
 import { useCallback, useEffect, useMemo, useRef, useState } from 'react'
@@ -16,6 +10,7 @@ import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
 import Modal from '../ui/Modal'
 import EmailBody, { looksLikeHtml } from '../ui/EmailBody'
 import { Tabs } from '../ui/Tabs'
+import { Pagination, pageWindow } from '../ui/DataTable'
 import { Avatar, Badge, EmptyState, Icon, ScoreChip } from '../ui/primitives'
 import { JobCard, PickRoleModal } from '../ui/SuggestedRoles'
 import { useToast } from '../ui/Toast'
@@ -29,33 +24,22 @@ import {
   inboxSources, sourceMeta,
 } from '../data/seed'
 
-const TABS = ['All Applications', 'Unread', 'Processed', 'Rejected', 'Duplicates', 'Email']
+const TABS = ['All Applications', 'Unread', 'Processed', 'Rejected', 'Duplicates']
+const PAGE_SIZE = 10
 
 /**
- * Tabs that do not read /inbox/all-applications at all. Email reads
- * /inbox/fetch; every other tab is a view over the applications list.
- */
-const SPECIAL_TABS = new Set(['Email'])
-
-/**
- * Server-side filters for tabs /inbox/all-applications can narrow.
- * Processed / Rejected / Duplicates filter client-side on
- * processing_state + is_duplicate (see GET /inbox/counts).
+ * Server-side filters for each tab. Processed / Rejected use
+ * Candidate_application_Status (PROCESS / REJECTED), not processing_state.
  */
 const TAB_FILTERS = {
   Unread: { isread: false },
+  Processed: { applicationStatus: 'PROCESS' },
+  Rejected: { applicationStatus: 'REJECTED' },
+  Duplicates: { isDuplicate: true },
 }
 
-/**
- * Tabs whose visible set is EXACTLY what the server returns for TAB_FILTERS[tab].
- *
- * Only these may use the scope endpoint for "mark all". Processed / Rejected /
- * Duplicates narrow client-side over rows the server already handed back in full,
- * so a scope call from one of those carries no such predicate and would mark the
- * entire mailbox — rows the user never saw, with no undo. Those tabs send the
- * visible ids instead.
- */
-const SERVER_SCOPED_TABS = new Set(['All Applications', 'Unread'])
+/** Every tab above is a true server-side scope — safe for "mark all". */
+const SERVER_SCOPED_TABS = new Set(TABS)
 
 /**
  * message_received_time / message_sent_time are plain string columns
@@ -205,11 +189,14 @@ async function fetchMessageDetail(recordId) {
  *
  * `processing` prefers processing_state (imported/processed/rejected); otherwise
  * Read/Unread from message_read. Duplicate comes from is_duplicate.
+ * Returns `{ rows, total }` so the pager can show page 1 / 2 / 3… without
+ * loading the whole mailbox.
  */
 async function fetchApplications(params) {
   const res = await inboxApi.listApplications(params)
   const rows = Array.isArray(res?.data) ? res.data : []
-  return rows.map((row) => {
+  return {
+    rows: rows.map((row) => {
       const name = row.name || row.email || 'Unknown'
       return {
         id: String(row.id),
@@ -236,7 +223,9 @@ async function fetchApplications(params) {
         suggestedIds: (row.suggested_job_post_ids || []).map(String),
         assignedId: row.assigned_job_post_id ? String(row.assigned_job_post_id) : null,
       }
-  })
+    }),
+    total: Number(res?.total ?? rows.length) || 0,
+  }
 }
 
 // Shared with the sidebar badge through qk.mailbox.counts — see inboxApi.fetchCounts.
@@ -258,9 +247,10 @@ async function setReadChunked(ids, read) {
 }
 
 /**
- * Rewrites ONE cache entry for a read-state flip: a row list, or the
- * single-message detail object. Anything else (the counts object) passes through
- * untouched — it has no `id`, and its delta is applied separately.
+ * Rewrites ONE cache entry for a read-state flip: a row list, a paginated
+ * `{rows, total}` page, or the single-message detail object. Anything else
+ * (the counts object) passes through untouched — it has no `id`, and its
+ * delta is applied separately.
  */
 function patchReadState(data, idSet, read) {
   const patchRow = (r) => (idSet.has(r.id)
@@ -275,6 +265,7 @@ function patchReadState(data, idSet, read) {
     }
     : r)
   if (Array.isArray(data)) return data.map(patchRow)
+  if (data && Array.isArray(data.rows)) return { ...data, rows: data.rows.map(patchRow) }
   if (data && typeof data === 'object' && data.id && idSet.has(data.id)) return patchRow(data)
   return data
 }
@@ -304,8 +295,9 @@ async function optimisticRead(qc, ids, read) {
   // A Set, because the same id appears in several cached lists at once.
   const flipping = new Set()
   for (const [, data] of previous) {
-    if (!Array.isArray(data)) continue
-    for (const r of data) if (idSet.has(r.id) && Boolean(r.unread) === read) flipping.add(r.id)
+    const rows = Array.isArray(data) ? data : data?.rows
+    if (!Array.isArray(rows)) continue
+    for (const r of rows) if (idSet.has(r.id) && Boolean(r.unread) === read) flipping.add(r.id)
   }
 
   qc.setQueriesData({ queryKey: qk.mailbox.all() }, (data) => patchReadState(data, idSet, read))
@@ -559,60 +551,37 @@ export default function Inbox() {
   const updateInbox = useSeedMutation('inbox')
 
   const [tab, setTab] = useState('All Applications')
+  const [page, setPage] = useState(1)
   const [selectedId, setSelectedId] = useState(null)
   const [q, setQ] = useState('')
   const [previewing, setPreviewing] = useState(null)
   const [assigning, setAssigning] = useState(null)
   const [noting, setNoting] = useState(null)
 
-  // Tabs with a server-side filter pass their params; everything else (and
-  // countsQuery) passes `{}` so the backend defaults mean "no filter".
   const tabFilter = TAB_FILTERS[tab] ?? {}
+  const listParams = useMemo(() => ({
+    ...tabFilter,
+    top: PAGE_SIZE,
+    skip: (page - 1) * PAGE_SIZE,
+    ...(q.trim() ? { search: q.trim() } : {}),
+  }), [tabFilter, page, q])
 
   const applicationsQuery = useQuery({
-    queryKey: qk.mailbox.applications(tabFilter),
-    queryFn: () => fetchApplications(tabFilter),
-    enabled: !SPECIAL_TABS.has(tab),
+    queryKey: qk.mailbox.applications(listParams),
+    queryFn: () => fetchApplications(listParams),
   })
 
   const countsQuery = useQuery({
     queryKey: qk.mailbox.counts(),
     queryFn: fetchInboxCounts,
-    enabled: tab !== 'Email',
   })
 
-  const inbox = applicationsQuery.data ?? []
+  const inbox = applicationsQuery.data?.rows ?? []
+  const total = applicationsQuery.data?.total ?? 0
+  const pages = Math.max(1, Math.ceil(total / PAGE_SIZE))
+  const currentPage = Math.min(page, pages)
   const serverCounts = countsQuery.data ?? {}
 
-  const emailsQuery = useQuery({
-    queryKey: qk.mailbox.messages(),
-    queryFn: async () => {
-      const res = await inboxApi.listMessages()
-      const rows = Array.isArray(res?.data) ? res.data : []
-      return rows.map((row) => ({
-        id: String(row.id),
-        from: row.sender_name || row.fromEmail || 'Unknown',
-        fromEmail: row.fromEmail || '',
-        subject: row.subject || '',
-        body: row.body || '',
-        when: parseDate(row.when) ?? parseDate(row.message_sent_time),
-        unread: Boolean(row.unread),
-        attachment: row.attachment_name || 'Resume.pdf',
-        attachmentSize: '—',
-        // The agent's verdict, straight off backend/inbox/serializers.py:44-48.
-        // suggested_job_post_ids is deliberately NOT carried: job posts stay
-        // dark to the inbox.
-        matchStatus: row.match_status || null,
-        matchSummary: row.match_summary || '',
-        matchReasoning: row.match_reasoning || '',
-        matchError: row.match_error || '',
-        matchedAt: parseDate(row.matched_at),
-        imported: false,
-      }))
-    },
-    enabled: tab === 'Email',
-  })
-
   const counts = useMemo(
     () => ({
       'All Applications': serverCounts.all ?? 0,
@@ -620,28 +589,16 @@ export default function Inbox() {
       Processed: serverCounts.processed ?? 0,
       Rejected: serverCounts.rejected ?? 0,
       Duplicates: serverCounts.duplicates ?? 0,
-      Email: (emailsQuery.data ?? []).filter((e) => e.unread).length,
     }),
-    [serverCounts, emailsQuery.data],
+    [serverCounts],
   )
 
-  const list = useMemo(() => {
-    let l = inbox
-    if (tab === 'Unread') l = l.filter((i) => i.processing === 'Unread')
-    else if (tab === 'Processed') l = l.filter((i) => i.processing === 'Processed')
-    else if (tab === 'Rejected') l = l.filter((i) => i.processing === 'Rejected')
-    else if (tab === 'Duplicates') l = l.filter((i) => i.duplicate)
-    if (q) l = l.filter((i) => (i.name + i.position + i.source).toLowerCase().includes(q.toLowerCase()))
-    return l
-  }, [inbox, tab, q])
+  const list = inbox
 
-  // Clicking a row fetches that one record from /inbox/fetch. The list row is
-  // kept as the base and the detail is overlaid, so the pane paints instantly
-  // from cached list data and fills in body/attachments when the fetch lands.
   const detailQuery = useQuery({
     queryKey: qk.mailbox.message(selectedId),
     queryFn: () => fetchMessageDetail(selectedId),
-    enabled: !SPECIAL_TABS.has(tab) && Boolean(selectedId),
+    enabled: Boolean(selectedId),
   })
 
   const selectedRow = inbox.find((i) => i.id === selectedId)
@@ -677,15 +634,13 @@ export default function Inbox() {
   }
 
   function setReadEverything(read) {
-    // The scope endpoint may only be used where the server-side filter IS the
-    // view. The search box narrows client-side on name/position/source while the
-    // server's `search` matches subject/from/body, and three of the tabs narrow
-    // client-side entirely — handing either to a WHERE clause would mark rows
-    // that were never on screen. Those cases send the visible ids instead, which
-    // is exact and, since the list endpoint is unpaginated, complete.
-    if (SERVER_SCOPED_TABS.has(tab) && !q.trim()) {
-      // `ids` drives the optimistic patch only; the write itself is the filter.
-      setReadAll.mutate({ read, filter: tabFilter, ids: list.map((i) => i.id) })
+    // Search goes on the wire now, so every tab is a true server scope.
+    if (SERVER_SCOPED_TABS.has(tab)) {
+      setReadAll.mutate({
+        read,
+        filter: { ...tabFilter, ...(q.trim() ? { search: q.trim() } : {}) },
+        ids: list.map((i) => i.id),
+      })
       return
     }
     const ids = list.map((i) => i.id)
@@ -791,14 +746,16 @@ export default function Inbox() {
         <div style={{ margin: '0 16px', paddingTop: 8 }}>
           <Tabs
             value={tab}
-            onChange={(t) => { setTab(t); setSelectedId(null); selection.clear() }}
+            onChange={(t) => {
+              setTab(t)
+              setPage(1)
+              setSelectedId(null)
+              selection.clear()
+            }}
             tabs={TABS.map((t) => ({ key: t, label: t, count: counts[t] }))}
           />
         </div>
 
-        {tab === 'Email' ? (
-          <EmailTab query={emailsQuery} toast={toast} />
-        ) : (
         <div className="split inbox-split">
             <div className="split-list inbox-queue">
               {applicationsQuery.isSuccess && (
@@ -815,7 +772,11 @@ export default function Inbox() {
               <div style={{ padding: '12px 16px', borderBottom: '1px solid var(--border)' }}>
                 <div className="toolbar-search" style={{ maxWidth: 'none' }}>
                   <Icon name="search" />
-                  <input value={q} onChange={(e) => setQ(e.target.value)} placeholder="Search applications…" />
+                  <input
+                    value={q}
+                    onChange={(e) => { setQ(e.target.value); setPage(1) }}
+                    placeholder="Search applications…"
+                  />
                 </div>
               </div>
               <div>
@@ -823,7 +784,7 @@ export default function Inbox() {
                   <EmptyState icon="inbox" title="Loading…">Fetching applications from the server.</EmptyState>
                 )}
                 {applicationsQuery.isError && (
-                  <EmptyState icon="inbox" title="Couldn’t load applications">
+                  <EmptyState icon="inbox" title="Couldn't load applications">
                     {friendlyAuthError(applicationsQuery.error, 'Request failed')}
                   </EmptyState>
                 )}
@@ -861,9 +822,6 @@ export default function Inbox() {
                         <div className="ii-time">
                           {outlookListTime(i.received)}
                         </div>
-                        {/* No ATS score exists server-side — the agent returns a
-                            verdict, not a number. The chip stays off rather than
-                            rendering a placeholder that reads as a real score. */}
                         {i.atsScore != null && (
                           <div style={{ marginTop: 6 }}><ScoreChip score={i.atsScore} /></div>
                         )}
@@ -872,6 +830,17 @@ export default function Inbox() {
                   ))
                 )}
               </div>
+              {applicationsQuery.isSuccess && total > 0 && (
+                <Pagination
+                  from={total ? (currentPage - 1) * PAGE_SIZE + 1 : 0}
+                  to={Math.min(currentPage * PAGE_SIZE, total)}
+                  total={total}
+                  page={currentPage}
+                  pages={pages}
+                  setPage={(p) => { setPage(p); setSelectedId(null); selection.clear() }}
+                  pageButtons={pageWindow(currentPage, pages)}
+                />
+              )}
             </div>
 
             <div className="split-detail">
@@ -883,7 +852,7 @@ export default function Inbox() {
                 </div>
               ) : detailQuery.isError ? (
                 <div style={{ padding: '100px 20px' }}>
-                  <EmptyState icon="inbox" title="Couldn’t load this application">
+                  <EmptyState icon="inbox" title="Couldn't load this application">
                     {friendlyAuthError(detailQuery.error, 'Request failed')}
                   </EmptyState>
                 </div>
@@ -904,7 +873,6 @@ export default function Inbox() {
               )}
             </div>
           </div>
-        )}
       </div>
 
       {previewing && (
@@ -1368,232 +1336,3 @@ function AssignRecruiter({ item, recruiters, onClose, onSave }) {
     </Modal>
   )
 }
-
-/** The live tab: real fetch, real loading state, real error state. */
-function EmailTab({ query, toast }) {
-  const qc = useQueryClient()
-  const { can } = useAuth()
-  const canEdit = can('inbox.edit')
-  const [selectedId, setSelectedId] = useState(null)
-  const [replying, setReplying] = useState(null)
-  const [replyBody, setReplyBody] = useState('')
-  const [importedIds, setImportedIds] = useState(() => new Set())
-
-  const emails = query.data ?? []
-  const selected = emails.find((e) => e.id === selectedId)
-  const unread = emails.filter((e) => e.unread).length
-
-  const selection = useRowSelection(emails)
-  const setRead = useSetRead(toast)
-  const setReadAll = useSetReadAll(toast)
-
-  function setReadSelected(read) {
-    const ids = [...selection.selectedIds]
-    if (!ids.length) return
-    setRead.mutate({ ids, read }, {
-      onSuccess: () => {
-        toast(`${ids.length} ${ids.length === 1 ? 'message' : 'messages'} marked ${read ? 'read' : 'unread'}`, 'success')
-        selection.clear()
-      },
-    })
-  }
-
-  /**
-   * This list is /inbox/fetch with no params — every persisted message, no
-   * filter, no pagination — so the empty scope really is the whole mailbox and
-   * the WHERE clause matches what is on screen exactly.
-   */
-  function setReadEverything(read) {
-    setReadAll.mutate({ read, filter: {}, ids: emails.map((e) => e.id) })
-  }
-
-  const importMsg = useMutation({
-    mutationFn: (id) => inboxApi.setProcessingState(id, 'imported'),
-    onSuccess: (_d, id) => {
-      setImportedIds((s) => new Set(s).add(id))
-      toast('Marked as imported', 'success')
-    },
-    onError: (err) => toast(friendlyAuthError(err, 'Could not import.'), 'error'),
-    onSettled: () => {
-      qc.invalidateQueries({ queryKey: qk.mailbox.all() })
-      qc.invalidateQueries({ queryKey: qk.mailbox.counts() })
-    },
-  })
-
-  const reply = useMutation({
-    mutationFn: ({ recordId, body }) => inboxApi.replyEmail({ recordId, body }),
-    onSuccess: () => {
-      toast('Reply sent', 'success')
-      setReplying(null)
-      setReplyBody('')
-    },
-    onError: (err) => toast(friendlyAuthError(err, 'Could not send reply.'), 'error'),
-  })
-
-  function selectEmail(e) {
-    setSelectedId(e.id)
-    if (e.unread) setRead.mutate({ ids: [e.id], read: true })
-  }
-
-  const isImported = (e) => importedIds.has(e.id)
-
-  return (
-    <>
-      <div style={{ padding: '12px 18px', borderBottom: '1px solid var(--border)', display: 'flex', alignItems: 'center', gap: 12 }}>
-        <span className="integration-status"><span className="pulse" />Outlook · Microsoft Graph API</span>
-        <span className="text-muted text-sm">
-          {query.isPending ? 'Loading…' : query.isError ? 'Failed to load' : `${emails.length} messages · ${unread} unread`}
-        </span>
-      </div>
-
-      <div className="split inbox-split">
-        <div className="split-list inbox-queue">
-          {query.isSuccess && emails.length > 0 && (
-            <BulkReadBar
-              rows={emails}
-              selection={selection}
-              onSetRead={setReadSelected}
-              onSetAllRead={setReadEverything}
-              busy={setRead.isPending || setReadAll.isPending}
-              canEdit={canEdit}
-              scopeLabel="every message in the mailbox"
-            />
-          )}
-          {query.isPending && <EmptyState icon="mail" title="Loading…">Fetching mailbox from the server.</EmptyState>}
-          {query.isError && (
-            <EmptyState icon="mail" title="Couldn’t load mailbox">
-              {friendlyAuthError(query.error, 'Request failed')}
-            </EmptyState>
-          )}
-          {query.isSuccess && emails.length === 0 && (
-            <EmptyState icon="mail" title="Nothing here">No emails in the mailbox.</EmptyState>
-          )}
-          {query.isSuccess && emails.map((e) => (
-            <div
-              key={e.id}
-              className={`inbox-item${e.unread ? ' unread' : ''}${selectedId === e.id ? ' active' : ''}`}
-              onClick={() => selectEmail(e)}
-            >
-              <RowCheck
-                checked={selection.selectedIds.has(e.id)}
-                onToggle={() => selection.toggle(e.id)}
-                label={`Select mail from ${e.from}`}
-              />
-              <Avatar name={e.from} />
-              <div className="ii-main">
-                <div className="ii-name">{e.from}</div>
-                <div className="ii-pos">{e.subject}</div>
-                <div className="ii-meta">
-                  <span className="source-chip" style={{ '--chip': '#0078d4' }}>
-                    <Icon name="mail" />Outlook
-                  </span>
-                  {isImported(e) && <Badge className="b-green">Imported</Badge>}
-                </div>
-              </div>
-              <div className="ii-time">{outlookListTime(e.when)}</div>
-            </div>
-          ))}
-        </div>
-
-        <div className="split-detail">
-          {!selected ? (
-            <div style={{ padding: '100px 20px' }}>
-              <EmptyState icon="mail" title="Select an email">
-                Preview email body and resume attachments here.
-              </EmptyState>
-            </div>
-          ) : (
-            <div style={{ padding: 24 }}>
-              <div className="flex items-center gap-12" style={{ marginBottom: 20 }}>
-                <Avatar name={selected.from} />
-                <div>
-                  <div className="fw-600">{selected.from}</div>
-                  <div className="cell-sub">
-                    {selected.fromEmail} · {selected.when ? fmtDate(selected.when) : 'Date unavailable'}
-                  </div>
-                </div>
-              </div>
-
-              {/* Same Subject-strip + framed-body template as /matching. It replaces
-                  the old <h2> subject rather than sitting under it — two subject
-                  lines on one panel is worse than none. The Imported/New badge
-                  moves into the strip, which is where a mail client puts status. */}
-              <div style={{ marginBottom: 18 }}>
-                <div className="email-head flex items-center gap-12" style={{ justifyContent: 'space-between' }}>
-                  <span>Subject: {selected.subject || '(no subject)'}</span>
-                  {isImported(selected) ? <Badge className="b-green">Imported</Badge> : <Badge className="b-blue">New</Badge>}
-                </div>
-                {looksLikeHtml(selected.body) ? (
-                  <EmailBody html={selected.body} />
-                ) : (
-                  <pre className="resume-thumb is-full email-plain">
-                    {htmlToText(selected.body) || 'This email has no message body.'}
-                  </pre>
-                )}
-              </div>
-
-              <div className="attach-card" style={{ marginBottom: 18 }}>
-                <span className="attach-icn"><Icon name="file" /></span>
-                <div style={{ flex: 1 }}>
-                  <div className="fw-600">{selected.attachment}</div>
-                  <div className="cell-sub">{selected.attachmentSize} · PDF</div>
-                </div>
-              </div>
-
-              <div className="flex gap-8">
-                {isImported(selected) ? (
-                  <button className="btn btn-secondary" disabled><Icon name="check" /> Already Imported</button>
-                ) : (
-                  <button
-                    className="btn btn-primary"
-                    disabled={importMsg.isPending}
-                    onClick={() => importMsg.mutate(selected.id)}
-                  >
-                    <Icon name="user-plus" /> Import Candidate
-                  </button>
-                )}
-                <button className="btn btn-secondary" onClick={() => { setReplying(selected); setReplyBody('') }}>
-                  <Icon name="mail" /> Reply
-                </button>
-              </div>
-            </div>
-          )}
-        </div>
-      </div>
-
-      {replying && (
-        <Modal
-          title="Reply"
-          subtitle={`Re: ${replying.subject || '(no subject)'}`}
-          onClose={() => setReplying(null)}
-          footer={
-            <>
-              <button className="btn btn-secondary" onClick={() => setReplying(null)} disabled={reply.isPending}>Cancel</button>
-              <button
-                className="btn btn-primary"
-                disabled={reply.isPending || !replyBody.trim()}
-                onClick={() => reply.mutate({ recordId: replying.id, body: replyBody.trim() })}
-              >
-                <Icon name="send" /> {reply.isPending ? 'Sending…' : 'Send Reply'}
-              </button>
-            </>
-          }
-        >
-          <div className="form-field">
-            <label>To</label>
-            <input value={replying.fromEmail || ''} disabled />
-          </div>
-          <div className="form-field">
-            <label>Message</label>
-            <textarea
-              rows={6}
-              value={replyBody}
-              onChange={(e) => setReplyBody(e.target.value)}
-              placeholder="Write your reply…"
-            />
-          </div>
-        </Modal>
-      )}
-    </>
-  )
-}

frontend/src/ui/DataTable.jsx

@@ -62,7 +62,7 @@ export function useDataTable({ columns, rows, pageSize = 10 }) {
 }
 
 /** 1 … cur-1 cur cur+1 … n — the prototype's windowing, unchanged. */
-function pageWindow(cur, pages) {
+export function pageWindow(cur, pages) {
   const list = []
   for (let i = 1; i <= pages; i++) {
     if (i === 1 || i === pages || Math.abs(i - cur) <= 1) list.push(i)