s3 configured pending testing

backend/.env.example

@@ -123,21 +123,24 @@ UVICORN_WORKERS=2
 # VITE_API_BASE=http://127.0.0.1:8000 (backend is published on BACKEND_PORT).
 VITE_API_BASE=
 
-# --- AWS S3 (s3/) — permanent public object URLs (not presigned) ------------
+# --- AWS S3 (s3/) — private CVs (no Principal "*" public policy) ------------
 # Bucket from your console, e.g. hr-ats-416818527652-us-east-2-an
 AWS_ACCESS_KEY_ID=
 AWS_SECRET_ACCESS_KEY=
 AWS_REGION=us-east-2
 S3_BUCKET=
-# Optional CDN / custom domain. Blank → https://{bucket}.s3.{region}.amazonaws.com/{key}
+# Optional CDN / custom domain for stable DB identity URLs only (objects stay private).
 S3_PUBLIC_BASE_URL=
-# Leave blank when ACLs are disabled (Object Ownership = Bucket owner enforced).
+# Leave blank. Do NOT set public-read — CVs are confidential.
-# Use public-read only if the bucket still allows ACLs.
 S3_OBJECT_ACL=
+# Short-lived browser open links via GET /s3/open (seconds; max 604800).
+S3_PRESIGN_EXPIRES_SECONDS=900
 # CV object keys (after DB row exists):
 #   Email/{inbox_messages.id}/{user_id}/{file}.pdf
 #   Manual/{manual_upload_candidate.id}/{user_id}/{file}.pdf
 #   Form/{form_data.id}/{recruiter_id}/{file}.pdf
+# Open a CV: GET /s3/open?key=<file_path or key> (auth) → temporary URL
+# Or stream: GET /s3/download?key=... (auth)
 
 LOG_FORMAT=json
 LOG_LEVEL=INFO

backend/employment_agent/decorators.py

@@ -5,14 +5,15 @@ Mirrors job/candidate/decorators.py — stacked wrappers that clean LLM output
 before the task persists it:
 
     raw JSON -> require_json_object -> clamp_company_to_resume
-              -> clamp_education_to_resume -> parse_employment_response
+              -> clamp_education_to_resume -> clamp_linkedin_url
+              -> parse_employment_response
 """
 
 from __future__ import annotations
 
 from functools import wraps
 
-from employment_agent.prompt import EDUCATION,NO_COMPANY
+from employment_agent.prompt import EDUCATION,NO_COMPANY,NO_LINKEDIN
 
 
 def require_json_object(func):
@@ -32,14 +33,14 @@ def clamp_company_to_resume(func):
 
     @wraps(func)
     def wrapper(data,resume_text="",*args,**kwargs):
-        company,education,current_title=func(data,resume_text,*args,**kwargs)
+        company,education,current_title,linkedin_url=func(data,resume_text,*args,**kwargs)
         company=(company or "").strip()
         if not company or company.lower()==NO_COMPANY.lower():
-            return NO_COMPANY,education,current_title
+            return NO_COMPANY,education,current_title,linkedin_url
         haystack=(resume_text or "").lower()
         if company.lower() not in haystack:
-            return NO_COMPANY,education,current_title
+            return NO_COMPANY,education,current_title,linkedin_url
-        return company,education,current_title
+        return company,education,current_title,linkedin_url
 
     return wrapper
 
@@ -49,14 +50,39 @@ def clamp_education_to_resume(func):
 
     @wraps(func)
     def wrapper(data,resume_text="",*args,**kwargs):
-        company,education,current_title=func(data,resume_text,*args,**kwargs)
+        company,education,current_title,linkedin_url=func(data,resume_text,*args,**kwargs)
         education=(education or "").strip()
         if not education or education.lower()==EDUCATION.lower():
-            return company,EDUCATION,current_title
+            return company,EDUCATION,current_title,linkedin_url
         haystack=(resume_text or "").lower()
         if education.lower() not in haystack:
-            return company,EDUCATION,current_title
+            return company,EDUCATION,current_title,linkedin_url
-        return company,education,current_title
+        return company,education,current_title,linkedin_url
+
+    return wrapper
+
+
+def clamp_linkedin_url(func):
+    """Keep linkedin_url only when the model returned a LinkedIn profile URL.
+
+    This is output validation, not CV scanning: the URL is the agent's own
+    `linkedin_url` key. Company pages and non-LinkedIn URLs are dropped.
+    """
+
+    @wraps(func)
+    def wrapper(data,resume_text="",*args,**kwargs):
+        company,education,current_title,linkedin_url=func(data,resume_text,*args,**kwargs)
+        url=(linkedin_url or "").strip()
+        if not url or url.lower() in (NO_LINKEDIN.lower(),"none","null","n/a","-"):
+            return company,education,current_title,None
+        lowered=url.lower()
+        if "linkedin.com/company/" in lowered:
+            return company,education,current_title,None
+        if "linkedin.com" not in lowered and "lnkd.in" not in lowered:
+            return company,education,current_title,None
+        if not lowered.startswith("http://") and not lowered.startswith("https://"):
+            url="https://"+url.lstrip("/")
+        return company,education,current_title,url
 
     return wrapper
 
@@ -64,15 +90,19 @@ def clamp_education_to_resume(func):
 @require_json_object
 @clamp_company_to_resume
 @clamp_education_to_resume
-def parse_employment_response(data,resume_text:str="") -> tuple[str,str]:
+@clamp_linkedin_url
-    """Pull company + education from LLM JSON; decorators clamp to the resume."""
+def parse_employment_response(data,resume_text:str="") -> tuple[str,str,str,str|None]:
+    """Pull company, education, title, and linkedin_url from the agent JSON."""
     current=data.get("current_employment")
     education=data.get("education")
     current_title=data.get("current_title")
+    linkedin_url=data.get("linkedin_url")
     if not isinstance(current,str):
         current=""
     if not isinstance(education,str):
         education=""
     if not isinstance(current_title,str):
         current_title=""
-    return current.strip(),education.strip(),current_title.strip()
+    if not isinstance(linkedin_url,str):
+        linkedin_url=""
+    return current.strip(),education.strip(),current_title.strip(),linkedin_url.strip()

backend/employment_agent/execute_agent.py

@@ -15,10 +15,10 @@ from llm_setup import llm_call
 logger=logging.getLogger("employment_agent")
 
 
-async def run_employment_agent(*,resume_text="") -> tuple[str,str]:
+async def run_employment_agent(*,resume_text="") -> tuple[str,str,str,str|None]:
     text=(resume_text or "").strip()
     if not text:
-        return NO_COMPANY,EDUCATION,CURRENT_TITLE
+        return NO_COMPANY,EDUCATION,CURRENT_TITLE,None
     try:
         data=await llm_call(prompt(),user_prompt(text),json_mode=True)
         return parse_employment_response(data,text)

backend/employment_agent/prompt.py

@@ -10,12 +10,15 @@ import json
 NO_COMPANY="no company was mentioned"
 EDUCATION="No Education Mentioned"
 CURRENT_TITLE="No JOB POSITION MENTIONED"
+NO_LINKEDIN="no linkedin url mentioned"
+
 
 def prompt():
     return f"""You are an HR-ATS recruiting assistant.
 
 You are given CV/resume text. Identify the candidate's CURRENT employer company
-name and their education (degree / school) when present.
+name, their education (degree / school), their current job title, and their
+LinkedIn profile URL when present.
 
 Rules:
 - Return only the company name that appears in the resume text for the ongoing / most recent role.
@@ -28,11 +31,19 @@ Rules:
 - Do not invent education. If none is mentioned, return exactly: {EDUCATION}
 - Do not invent job title. If none is mentioned, return exactly: {CURRENT_TITLE}
 
+linkedin_url (its own key — extract this separately from the other fields):
+- Return the candidate's own public LinkedIn profile URL (linkedin.com/in/..., /pub/..., /mwlite/in/..., or lnkd.in/...).
+- Reconstruct the URL if PDF extraction wrapped or spaced it (e.g. "linkedin.com/in/\\njane-doe" or "linkedin . com / in / jane-doe").
+- Clickable icon links may appear as bare URLs on their own lines at the end of the text; use those.
+- Do not return a company page (linkedin.com/company/...), a search URL, or a URL that is not LinkedIn.
+- Do not invent a profile. If none is mentioned, return exactly: {NO_LINKEDIN}
+
 Respond with JSON only:
 {{
   "current_employment": "Company Name",
   "education": "Degree / School",
-  "current_title": "Job Title"
+  "current_title": "Job Title",
+  "linkedin_url": "https://www.linkedin.com/in/slug"
 }}
 """
 

backend/g_sheet/views.py

@@ -467,6 +467,12 @@ class SheetFormData(Sheet):
         if resume:
             file_name=resume.rsplit("/",1)[-1][:180] or "resume"
 
+        # Sheet already stores LinkedIn on profile_link — copy it through, do not parse the CV.
+        profile=(form_row.profile_link or "").strip()
+        linkedin_url=None
+        if profile:
+            linkedin_url=profile if profile.lower().startswith("http") else f"https://{profile.lstrip('/')}"
+
         row=await Manual_UPLOAD_CANDIDATE.create_manual_upload_candidate(session,{
             "candidate_email":email,
             "candidate_name":(form_row.name or "").strip() or email,
@@ -481,6 +487,7 @@ class SheetFormData(Sheet):
             "file_name":file_name,
             "file_path":resume,
             "full_text":"",
+            "linkedin_url":linkedin_url,
         })
         await FormData.link_manual_upload(session,form_row.id,row.id)
         try:

backend/inbox/file_decoder.py

@@ -1,150 +1,62 @@
-"""Decode Graph fileAttachment contentBytes into PDF / DOC / DOCX files."""
+"""Decode Graph fileAttachment contentBytes — PDF only, in memory (no disk).
-# this file is decoding the pdf and also calling in the flow of first fetch of email if i do use func from  this file rather then  touchjing the email flow and create a bg task from here that can call the llm re i add param of subject and readc the file of pdf to get
+
-#the location to the llm_call thne it's probable that without touching the real flow i can use background task without stopping or delaying the real result and add a column in Inbox_Messages that i can later update the file recorby using filename to pdate the answer or suggeswtions from the lmm that i can later or get from get api so user/recruiter can see and map the candidate to it's real final job_post_id that then can be linked with job_post_id
+Email / Manual CV flows upload bytes to S3 after the DB row exists. Nothing
-# as job_post_id is already linked by created_by and llm_call would require to read job_post of every recruiter and user ever posted only the posts that are still active it must read all post content and then finalize that this candidate might inlcude one of or more then one job_post_id : Note use list[uuid] to map with job_post_id inside Inbox_Messages table
+writes under inbox/decoded_attachments anymore.
+"""
+
 from __future__ import annotations
 
-import asyncio
 import base64
 import binascii
-import io
-import zipfile
 from pathlib import Path
 from typing import Any
 
 
 class AttachmentDecodeError(ValueError):
-    """Raised when contentBytes is malformed or is not the expected format."""
+    """Raised when contentBytes is malformed or is not a PDF."""
-
-
-_DEFAULT_OUT_DIR = Path(__file__).resolve().parent / "decoded_attachments"
 
 
 def _decode_bytes(attachment: dict) -> bytes:
-    """base64 -> raw bytes.
+    """base64 -> raw bytes."""
-
+    b64=attachment.get("contentBytes")
-    Graph's ``size`` often includes MIME/encoding overhead and may not equal
-    ``len(contentBytes)`` after decode, so it is not treated as a hard check.
-    """
-    b64 = attachment.get("contentBytes")
     if not b64:
         raise AttachmentDecodeError(f"{attachment.get('name')!r}: no contentBytes")
-
     try:
-        return base64.b64decode(b64, validate=True)
+        return base64.b64decode(b64,validate=True)
     except binascii.Error as exc:
         raise AttachmentDecodeError(
             f"{attachment.get('name')!r}: bad base64: {exc}"
         ) from exc
 
 
-def _write(out_dir: Path, name: str, raw: bytes) -> Path:
-    out_dir = Path(out_dir)
-    out_dir.mkdir(parents=True, exist_ok=True)
-    dest = out_dir / Path(name).name  # basename only — strip path traversal
-    dest.write_bytes(raw)
-    return dest
-
-
-def decode_pdf(attachment: dict, out_dir: str | Path) -> Path:
-    """Decode a PDF attachment and write it under out_dir."""
-    raw = _decode_bytes(attachment)
-    name = attachment.get("name")
-    if not raw.startswith(b"%PDF-"):
-        raise AttachmentDecodeError(f"{name!r}: not a PDF (missing %PDF- header)")
-    if b"%%EOF" not in raw[-2048:]:
-        raise AttachmentDecodeError(f"{name!r}: not a PDF (missing %%EOF trailer)")
-    return _write(Path(out_dir), name or "attachment.pdf", raw)
-
-
-def decode_docx(attachment: dict, out_dir: str | Path) -> Path:
-    """Decode a DOCX attachment and write it under out_dir."""
-    raw = _decode_bytes(attachment)
-    name = attachment.get("name")
-    if not raw.startswith(b"PK\x03\x04"):
-        raise AttachmentDecodeError(f"{name!r}: not a DOCX (missing ZIP signature)")
-
-    bio = io.BytesIO(raw)
-    if not zipfile.is_zipfile(bio):
-        raise AttachmentDecodeError(f"{name!r}: not a DOCX (invalid ZIP)")
-    bio.seek(0)
-    with zipfile.ZipFile(bio) as zf:
-        if not any(member.startswith("word/") for member in zf.namelist()):
-            raise AttachmentDecodeError(f"{name!r}: not a DOCX (no word/ entry)")
-
-    return _write(Path(out_dir), name or "attachment.docx", raw)
-
-
-def decode_doc(attachment: dict, out_dir: str | Path) -> Path:
-    """Decode a legacy DOC (OLE2) attachment and write it under out_dir."""
-    raw = _decode_bytes(attachment)
-    name = attachment.get("name")
-    if raw.startswith(b"PK\x03\x04"):
-        raise AttachmentDecodeError(
-            f"{name!r}: named .doc but content is DOCX — use decode_docx"
-        )
-    ole2 = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1"
-    if not raw.startswith(ole2):
-        raise AttachmentDecodeError(f"{name!r}: not a DOC (missing OLE2 signature)")
-    return _write(Path(out_dir), name or "attachment.doc", raw)
-
-
-_DECODERS = {
-    ".pdf": decode_pdf,
-    ".docx": decode_docx,
-    ".doc": decode_doc,
-}
-
-
-def _decode_one(attachment: dict, out_dir: str | Path) -> Path:
-    """Route on the file extension to the right decoder."""
-    ext = Path(attachment.get("name", "")).suffix.lower()
-    if ext not in _DECODERS:
-        raise AttachmentDecodeError(f"unsupported extension {ext!r}")
-    return _DECODERS[ext](attachment, out_dir)
-
-
 def _normalize_attachments(attachments: Any) -> list[dict]:
-    """Accept None, a single dict, or a list; return only dict items."""
     if attachments is None:
         return []
-    if isinstance(attachments, dict):
+    if isinstance(attachments,dict):
         return [attachments]
-    if isinstance(attachments, list):
+    if isinstance(attachments,list):
-        return [a for a in attachments if isinstance(a, dict)]
+        return [a for a in attachments if isinstance(a,dict)]
     return []
 
 
-def _decode_attachments_sync(
+def extract_pdf_attachments(attachments: Any) -> list[dict]:
-    attachments: Any,
+    """Return ``[{name, body}]`` for PDF Graph attachments — no disk writes.
-    out_dir: str | Path | None = None,
-) -> list[str]:
-    """Decode supported file attachments; skip empty / non-file / unsupported."""
-    dest_dir = Path(out_dir) if out_dir is not None else _DEFAULT_OUT_DIR
-    paths: list[str] = []
 
+    Non-PDF / empty / reference attachments are skipped. PDF gate is extension
+    + ``%PDF-`` header (same bar as assert_pdf / Manual create).
+    """
+    out: list[dict]=[]
     for attachment in _normalize_attachments(attachments):
-        # Graph itemAttachment / referenceAttachment have no contentBytes
         if not attachment.get("contentBytes"):
             continue
-        ext = Path(attachment.get("name") or "").suffix.lower()
+        name=Path(attachment.get("name") or "resume.pdf").name or "resume.pdf"
-        if ext not in _DECODERS:
+        if not name.lower().endswith(".pdf"):
             continue
-        path = _decode_one(attachment, dest_dir).resolve()
+        try:
-        paths.append(str(path))
+            raw=_decode_bytes(attachment)
-
+        except AttachmentDecodeError:
-    return paths
+            continue
-
+        if not raw.startswith(b"%PDF-"):
-
+            continue
-async def decode_attachment(
+        out.append({"name":name,"body":raw})
-    attachments: Any,
+    return out
-    out_dir: str | Path | None = None,
-) -> list[str]:
-    """
-    Decode Graph attachments into files under out_dir.
-
-    Designed for views: ``await decode_attachment(data.get("attachments"))``.
-    Accepts None, a single attachment dict, or a list of attachment dicts.
-    Returns absolute file_path strings for successfully converted files.
-    """
-    return await asyncio.to_thread(_decode_attachments_sync, attachments, out_dir)

backend/inbox/models.py

@@ -17,7 +17,7 @@ from sqlalchemy.orm import selectinload
 from sqlmodel import Field, Relationship, SQLModel, select, true
 
 from job.candidate.models import Activity, Feedback, Interviews
-from linkedin_utils import primary_slug_from_text
+from linkedin_utils import slug_from_url, NO_SLUG
 from users.models import Users
 from users.plugins import hash_password
 
@@ -83,6 +83,7 @@ class Inbox(SQLModel, table=True):
                     cls.user_id,
                     Users.name,
                     Users.email,
+                    Users.linkedin_url,
                     Inbox_Messages.candidate_phone_number.label("phone"),
                     Inbox_Messages.assigned_job_post_id,
                     Inbox_Messages.application_status,
@@ -138,6 +139,7 @@ class Inbox(SQLModel, table=True):
                     "user_id":str(row["user_id"]) if row["user_id"] else None,
                     "name":row["name"],
                     "email":row["email"],
+                    "linkedin_url":row["linkedin_url"] or None,
                     "application_status":status.value if status else None,
                     "phone":row["phone"],
                     "assigned_job_post_id":str(row["assigned_job_post_id"]) if row["assigned_job_post_id"] else None,
@@ -152,6 +154,25 @@ class Inbox(SQLModel, table=True):
         except Exception as e:
             raise HTTPException(status_code=500,detail=str(e))
 
+    @classmethod
+    async def linkedin_urls_by_message_ids(cls, session: AsyncSession, message_ids) -> dict:
+        """users.linkedin_url keyed by inbox_messages.id for one list page."""
+        ids = [mid for mid in (message_ids or []) if mid is not None]
+        if not ids:
+            return {}
+        result = await session.execute(
+            select(cls.message_id, Users.linkedin_url)
+            .join(Users, Users.id == cls.user_id)
+            .where(cls.message_id.in_(ids))
+            .where(Users.linkedin_url.is_not(None))
+            .where(Users.linkedin_url != "")
+        )
+        out = {}
+        for mid, url in result.all():
+            if mid not in out and url:
+                out[mid] = url
+        return out
+
     @classmethod
     async def count_by_status(cls,session:AsyncSession,job_post_id=None):
         try:
@@ -285,6 +306,32 @@ class Inbox(SQLModel, table=True):
         )
         return result.scalars().first()
 
+    @classmethod
+    async def newest_cv_by_user_ids(cls,session:AsyncSession,user_ids):
+        """Newest inbox.id + first file_path per user — search Open resume."""
+        ids=[]
+        for raw in (user_ids or []):
+            try:
+                ids.append(uuid.UUID(str(raw)))
+            except (TypeError,ValueError):
+                continue
+        if not ids:
+            return {}
+        result=await session.execute(
+            select(cls.user_id,cls.id,Inbox_Messages.file_path)
+            .join(Inbox_Messages,cls.message_id==Inbox_Messages.id)
+            .where(cls.user_id.in_(ids))
+            .order_by(cls.created_at.desc())
+        )
+        out={}
+        for user_id,inbox_id,file_path in result.all():
+            key=str(user_id)
+            if key in out:
+                continue
+            first=(file_path or "").split(",")[0].strip() or None
+            out[key]={"inbox_id":inbox_id,"file_path":first}
+        return out
+
     @classmethod
     async def update_inbox(cls,session:AsyncSession,record_id,fields:dict):
         row=await cls.get_inbox_by_id(session,record_id)
@@ -406,6 +453,7 @@ class Inbox_Messages(SQLModel, table=True):
         candidate_phone_number=None,
         current_employment=None,
         current_title=None,
+        linkedin_url=None,
         suggested_job_post_ids=None,
         summary="",
         reasoning="",
@@ -418,7 +466,18 @@ class Inbox_Messages(SQLModel, table=True):
             return None
         if resume_text is not None:
             row.resume_text = resume_text
-            row.linkedin_slug = primary_slug_from_text(resume_text)
+        url = (linkedin_url or "").strip() or None
+        if url:
+            row.linkedin_slug = slug_from_url(url) or NO_SLUG
+            user_id = await cls.get_linked_user_id(session, row.id)
+            if user_id:
+                await Users.set_linkedin_url_if_empty(
+                    session, user_id=user_id, url=url,
+                )
+        elif resume_text is not None:
+            # Agent ran and found no profile — mark scanned so talent backfill
+            # does not regex-scan this CV again.
+            row.linkedin_slug = NO_SLUG
         if candidate_phone_number is not None:
             row.candidate_phone_number = candidate_phone_number
         if candidate_education is not None:
@@ -561,12 +620,15 @@ class Inbox_Messages(SQLModel, table=True):
             ).scalars().first()
             if existing:
                 for key, value in fields.items():
+                    # Keep prior S3 URLs until attach_email_pdfs_to_s3 replaces them.
+                    if key in ("file_path","file_name") and not value:
+                        continue
                     setattr(existing, key, value)
                 session.add(existing)
                 await session.commit()
                 await session.refresh(existing)
 
-                if fields.get("attachment"):
+                if fields.get("attachment") or existing.attachment:
                     link_user=await cls._link_sender(session, email_data, existing)
                     # _link_sender may rollback (IntegrityError); that expires this row
                     await session.refresh(existing)
@@ -582,6 +644,46 @@ class Inbox_Messages(SQLModel, table=True):
             await session.refresh(email)
         return email, link_user
 
+    @classmethod
+    async def get_linked_user_id(cls,session:AsyncSession,message_id):
+        try:
+            mid=uuid.UUID(str(message_id))
+        except (ValueError,TypeError):
+            return None
+        return (
+            await session.execute(select(Inbox.user_id).where(Inbox.message_id==mid))
+        ).scalar_one_or_none()
+
+    @classmethod
+    async def set_file_paths(cls,session:AsyncSession,record_id,file_paths,file_names=None):
+        row=await cls.get_inbox_message_by_id(session,record_id)
+        if not row:
+            return None
+        paths=file_paths if isinstance(file_paths,list) else ([file_paths] if file_paths else [])
+        cleaned=[str(p).strip() for p in paths if p and str(p).strip()]
+        row.file_path=",".join(cleaned) if cleaned else None
+        row.attachment=bool(cleaned)
+        if file_names is not None:
+            names=file_names if isinstance(file_names,list) else [file_names]
+            row.file_name=",".join(str(n).strip() for n in names if n and str(n).strip()) or row.file_name
+        session.add(row)
+        await session.commit()
+        await session.refresh(row)
+        return row
+
+    @classmethod
+    async def delete_by_id(cls,session:AsyncSession,record_id):
+        """Hard-delete message + inbox links — roll back when S3 upload fails after insert."""
+        row=await cls.get_inbox_message_by_id(session,record_id)
+        if not row:
+            return False
+        links=(await session.execute(select(Inbox).where(Inbox.message_id==row.id))).scalars().all()
+        for link in links:
+            session.delete(link)
+        session.delete(row)
+        await session.commit()
+        return True
+
     @classmethod
     def _search_filter(cls, search: str):
         pattern = f"%{search}%"
@@ -679,6 +781,23 @@ class Inbox_Messages(SQLModel, table=True):
         await session.refresh(row)
         return row
 
+    @classmethod
+    async def counts_by_job_post_ids(cls, session: AsyncSession, job_post_ids) -> dict[str, int]:
+        """Resolve {job_post_id: applicant_count} for a page of rows in a single query.
+
+        Counts Inbox_Messages rows, not Inbox rows: one message fans out to several
+        Inbox rows (one per recipient), so counting Inbox would over-count.
+        """
+        uids = {u for u in (job_post_ids or []) if u}
+        if not uids:
+            return {}
+        result = await session.execute(
+            select(cls.assigned_job_post_id, func.count().label("applicants"))
+            .where(cls.assigned_job_post_id.in_(uids))
+            .group_by(cls.assigned_job_post_id)
+        )
+        return {str(job_id): int(n) for job_id, n in result.all()}
+
     @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, is_duplicate: bool | None=None):
         statement = cls._apply_filters(

backend/inbox/plugins.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import asyncio
 import base64
+import logging
 import os
 import re
 import uuid
@@ -20,6 +22,8 @@ from job.candidate.views import FileRead
 
 load_dotenv()
 
+logger=logging.getLogger("inbox.plugins")
+
 EMAIL_URL=os.getenv("EMAIL_URL")
 EMAIL_API_TOKEN=os.getenv("EMAIL_API_TOKEN")
 BACKEND_URL=os.getenv("BACKEND_URL","http://localhost:8000")
@@ -102,12 +106,10 @@ async def fetch_message_read_status(message_id, token=None):
 
 
 def resolve_attachment_path(path_str:str) -> Path:
-    """Prefer stored path; fall back to basename under decoded_attachments.
+    """Legacy local-path resolver — kept for any old rows still on disk.
 
-    Stored paths may be Windows absolutes written by the host API. The Taskiq
+    New Email/Manual rows store HTTPS S3 URLs in file_path; callers should use
-    worker runs in Linux, where ``Path(r"D:\\...\\file.pdf").name`` is the
+    ``load_file_bytes`` / ``extract_resume_text`` which handle URLs first.
-    whole string (backslash is not a separator), so normalize separators before
-    taking the basename for the mounted attachments dir.
     """
     raw=path_str.strip()
     path=Path(raw)
@@ -120,11 +122,43 @@ def resolve_attachment_path(path_str:str) -> Path:
     return path
 
 
+def load_file_bytes(path_or_url: str) -> bytes | None:
+    """Load CV bytes from an S3 URL (preferred) or a leftover local path."""
+    raw=(path_or_url or "").strip()
+    if not raw:
+        return None
+    if raw.lower().startswith("http://") or raw.lower().startswith("https://"):
+        from s3.plugins import S3,S3ServiceError
+        try:
+            return S3().download_bytes(raw)
+        except S3ServiceError:
+            logger.exception("s3 download failed for %s",raw[:120])
+            return None
+    path=resolve_attachment_path(raw)
+    if not path.is_file():
+        return None
+    try:
+        return path.read_bytes()
+    except OSError:
+        return None
+
+
 def load_message_files(message:Inbox_Messages) -> list[dict]:
     if not message.file_path:
         return []
+    names=[n.strip() for n in (message.file_name or "").split(",") if n.strip()]
     files=[]
-    for path_str in message.file_path.split(","):
+    for idx,path_str in enumerate(p.strip() for p in message.file_path.split(",") if p.strip()):
+        name=names[idx] if idx<len(names) else Path(path_str.replace("\\","/")).name
+        entry={"file_name":name or "resume.pdf","url":None,"content_base64":None,"size":0}
+        if path_str.lower().startswith("http://") or path_str.lower().startswith("https://"):
+            entry["url"]=path_str
+            raw=load_file_bytes(path_str)
+            if raw is not None:
+                entry["content_base64"]=base64.b64encode(raw).decode("ascii")
+                entry["size"]=len(raw)
+            files.append(entry)
+            continue
         path=resolve_attachment_path(path_str)
         if not path.is_file():
             continue
@@ -132,14 +166,56 @@ def load_message_files(message:Inbox_Messages) -> list[dict]:
             raw=path.read_bytes()
         except OSError:
             continue
-        files.append({
+        entry["file_name"]=path.name
-            "file_name":path.name,
+        entry["content_base64"]=base64.b64encode(raw).decode("ascii")
-            "content_base64":base64.b64encode(raw).decode("ascii"),
+        entry["size"]=len(raw)
-            "size":len(raw),
+        files.append(entry)
-        })
     return files
 
 
+async def attach_email_pdfs_to_s3(session,row,pdfs,*,created_new:bool):
+    """Upload PDFs under Email/{row.id}/{user_id}/ and set file_path to permanent URLs.
+
+    Atomicity: if upload fails and ``created_new`` is True, delete the inbox_messages
+    row (and inbox links). Re-sync of an existing row does not delete on failure.
+    Returns the refreshed row.
+    """
+    from s3.plugins import S3,S3Source
+
+    if not pdfs:
+        return row
+    owner_id=await Inbox_Messages.get_linked_user_id(session,row.id)
+    if owner_id is None:
+        owner_id="unlinked"
+    s3=S3()
+    urls=[]
+    names=[]
+    uploaded_keys=[]
+    try:
+        for pdf in pdfs:
+            result=s3.upload_for_record(
+                pdf["body"],
+                pdf.get("name") or "resume.pdf",
+                source=S3Source.EMAIL,
+                record_id=row.id,
+                owner_id=owner_id,
+                content_type="application/pdf",
+            )
+            urls.append(result["url"])
+            names.append(result.get("filename") or pdf.get("name") or "resume.pdf")
+            uploaded_keys.append(result["key"])
+        return await Inbox_Messages.set_file_paths(session,row.id,urls,names)
+    except Exception:
+        for key in uploaded_keys:
+            try:
+                s3.delete_object(key)
+            except Exception:
+                logger.exception("s3 cleanup failed key=%s",key)
+        if created_new:
+            await Inbox_Messages.delete_by_id(session,row.id)
+        raise
+
+
 def extract_phone(text:str) -> str|None:
     m=_PHONE.search(text or "")
     if not m:
@@ -148,22 +224,34 @@ def extract_phone(text:str) -> str|None:
 
 
 async def extract_resume_text(file_paths:list[str]) -> tuple[str,str]:
-    candidates=[resolve_attachment_path(p) for p in (file_paths or []) if p and p.strip()]
+    """Extract text from S3 URLs or leftover local PDF paths."""
-    existing=[p for p in candidates if p.is_file() and p.suffix.lower()==".pdf"]
+    refs=[p.strip() for p in (file_paths or []) if p and p.strip()]
-    if not existing:
+    if not refs:
-        return "","no PDF attachment to extract (.doc/.docx not supported)"
+        return "","no PDF attachment to extract"
 
     texts=[]
     errors=[]
-    for path in existing:
+    for ref in refs:
+        name=Path(ref.replace("\\","/")).name or "resume.pdf"
+        is_url=ref.lower().startswith("http://") or ref.lower().startswith("https://")
+        if not is_url and not name.lower().endswith(".pdf"):
+            continue
+        if is_url and ".pdf" not in ref.lower() and not name.lower().endswith(".pdf"):
+            # still try — key may omit extension rarely
+            pass
         try:
-            raw=path.read_bytes()
+            raw=await asyncio.to_thread(load_file_bytes,ref)
-            result=await FileRead(session=None,filename=path.name,file=raw).read_file()
+            if raw is None:
+                errors.append(f"{name}: could not load file (S3 Access Denied or missing)")
+                continue
+            result=await FileRead(session=None,filename=name if name.lower().endswith(".pdf") else f"{name}.pdf",file=raw).read_file()
             text=(result.get("text") or "").strip()
             if text:
                 texts.append(text)
+            else:
+                errors.append(f"{name}: no text extracted")
         except Exception as exc:
-            errors.append(f"{path.name}: {exc}")
+            errors.append(f"{name}: {exc}")
 
     if not texts:
         return "","; ".join(errors) if errors else "no text extracted from PDF"

backend/inbox/serializers.py

@@ -36,7 +36,7 @@ def _attachment_name(message: Inbox_Messages) -> str | None:
     return None
 
 
-def serialize_message(message: Inbox_Messages) -> dict:
+def serialize_message(message: Inbox_Messages, *, linkedin_url=None) -> dict:
     """inbox_messages row -> the shape the #inbox Email tab renders."""
     sender_name = _sender_name(message)
     attachment_name = _attachment_name(message)
@@ -60,6 +60,8 @@ def serialize_message(message: Inbox_Messages) -> dict:
         "message_sent_time": message.message_sent_time,
         "message_reply": message.message_reply,
         "file_path": message.file_path,
+        "linkedin_slug": message.linkedin_slug or None,
+        "linkedin_url": linkedin_url or None,
         "suggested_job_post_ids": list(message.suggested_job_post_ids or []),
         "assigned_job_post_id": str(message.assigned_job_post_id) if message.assigned_job_post_id else None,
         "match_summary": message.match_summary,
@@ -79,7 +81,7 @@ _PROCESSING_LABEL = {
 }
 
 
-def serialize_application(message: Inbox_Messages) -> dict:
+def serialize_application(message: Inbox_Messages, *, linkedin_url=None) -> dict:
     """inbox_messages row -> the shape the #inbox All Applications tab renders.
 
     `position` is the mail subject and `source` is the To address, which is where
@@ -109,6 +111,9 @@ def serialize_application(message: Inbox_Messages) -> dict:
         "resume_status": _RESUME_STATUS.get(message.match_status, "Pending"),
         "attachment": _attachment_name(message),
         "has_attachment": message.attachment,
+        "file_path": message.file_path,
+        "linkedin_slug": message.linkedin_slug or None,
+        "linkedin_url": linkedin_url or None,
         "resume_text": message.resume_text,
         "suggested_job_post_ids": list(message.suggested_job_post_ids or []),
         "assigned_job_post_id": str(message.assigned_job_post_id) if message.assigned_job_post_id else None,

backend/inbox/tasks.py

@@ -94,6 +94,7 @@ async def match_inbox_message(record_id:str,force:bool=False) -> dict:
 
         paths=[p.strip() for p in row.file_path.split(",") if p.strip()]
         subject=row.message_subject or ""
+        body=row.message_body or ""
         row.match_status="processing"
         row.match_error=None
         row.matched_at=datetime.now(timezone.utc)
@@ -117,7 +118,9 @@ async def match_inbox_message(record_id:str,force:bool=False) -> dict:
     if status=="failed":
         raise RuntimeError(result.get("error") or "agent returned failed status")
 
-    current_employment,education,current_title=await run_employment_agent(resume_text=text)
+    current_employment,education,current_title,linkedin_url=await run_employment_agent(
+        resume_text=text if not body else f"{text}\n\n{body}",
+    )
 
     async with session_scope() as session:
         await Inbox_Messages.set_match_result(
@@ -129,6 +132,7 @@ async def match_inbox_message(record_id:str,force:bool=False) -> dict:
             current_employment=current_employment,
             current_title=current_title,
             candidate_education=education,
+            linkedin_url=linkedin_url,
             suggested_job_post_ids=result.get("suggested_job_post_ids") or [],
             summary=result.get("summary") or "",
             reasoning=result.get("reasoning") or "",
@@ -159,4 +163,5 @@ async def match_inbox_message(record_id:str,force:bool=False) -> dict:
         "current_employment":current_employment,
         "current_title":current_title,
         "education":education,
+        "linkedin_url":linkedin_url,
     }

backend/inbox/views.py

@@ -4,11 +4,12 @@ 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,MailboxSyncRun
+from inbox.models import Inbox_Messages,Inbox_Message_Triage,MailboxSyncRun,Inbox
-from inbox.file_decoder import decode_attachment
+from inbox.file_decoder import extract_pdf_attachments
 from inbox.serializers import serialize_application, serialize_message, serialize_triage, serialize_mailbox_sync_run
 from inbox.plugins import (
     EMAIL_API_TOKEN,
+    attach_email_pdfs_to_s3,
     fetch_message_read_status,
     load_message_files,
     request_email_confirmation,
@@ -184,15 +185,11 @@ class Email:
         `decision` is the pre-computed verdict from triage_round; without one this
         classifies inline, so a single-message call still works.
 
-        Ordering is deliberate. The verdict comes BEFORE decode_attachment: a rejected
+        Ordering is deliberate. The verdict comes BEFORE PDF extract / S3 upload: a
-        mail must not write a file into decoded_attachments (nothing on this path ever
+        rejected mail must not create a candidate Users row or queue confirmation mail.
-        deletes one, and _write uses the basename only, so a vendor "resume.pdf" would
+        Flow: extract PDF bytes in memory → insert inbox_messages → link sender →
-        clobber a candidate's stored CV), and must not reach _link_sender, which would
+        upload Email/{id}/{user_id}/file.pdf → store permanent S3 URL on file_path.
-        create a candidate Users row and queue a confirmation mail for a stranger.
+        If S3 fails on a brand-new row, the table entry is deleted (atomicity).
-
-        The gate lives here, not in Inbox_Messages.insert_email, so
-        FileRead.ingest_upload bypasses it for free — that path fabricates an EMPTY body
-        and would be a guaranteed false negative under a subject+body classifier.
         """
         try:
             if decision is None:
@@ -208,8 +205,17 @@ class Email:
                 return {"message_id":str(message_id),"skipped":"not_application",
                         "reason":decision.get("reason") or "","status":decision.get("status") or ""}
 
-            re_create_file=await decode_attachment(data.get("attachments"))
+            upstream_id=data.get("id")
-            row,new_user_email=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file)
+            already=await Inbox_Messages.get_by_upstream_id(self.session,upstream_id) if upstream_id else None
+            pdfs=extract_pdf_attachments(data.get("attachments"))
+            # Insert first (no file_path yet) so S3 keys can use the table PK.
+            row,new_user_email=await Inbox_Messages.insert_email(
+                session=self.session,email_data=data,file_path=None,
+            )
+            if pdfs:
+                row=await attach_email_pdfs_to_s3(
+                    self.session,row,pdfs,created_new=(already is None),
+                )
             if decision.get("fresh"):
                 await self.record_triage(data,decision,ingested=True)
             if row.attachment and row.file_path and row.match_status is None:
@@ -228,9 +234,10 @@ class Email:
 
     async def get_inbox_messages(self,top,skip,search=None):
         messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search)
+        urls=await Inbox.linkedin_urls_by_message_ids(self.session,[m.id for m in messages])
         items=[]
         for m in messages:
-            item=serialize_message(m)
+            item=serialize_message(m,linkedin_url=urls.get(m.id))
             files=load_message_files(m)
             if files:
                 item["files"]=files
@@ -241,7 +248,8 @@ class Email:
         message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
         if not message:
             raise HTTPException(status_code=404,detail="Message not found")
-        item=serialize_message(message)
+        urls=await Inbox.linkedin_urls_by_message_ids(self.session,[message.id])
+        item=serialize_message(message,linkedin_url=urls.get(message.id))
         files=load_message_files(message)
         if files:
             item["files"]=files
@@ -272,13 +280,15 @@ class Email:
             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,is_duplicate=is_duplicate)
-        return [serialize_application(m) for m in messages]
+        urls=await Inbox.linkedin_urls_by_message_ids(self.session,[m.id for m in messages])
+        return [serialize_application(m,linkedin_url=urls.get(m.id)) for m in messages]
 
     async def get_application_by_id(self,record_id):
         message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
         if not message:
             raise HTTPException(status_code=404,detail="Application not found")
-        return serialize_application(message)
+        urls=await Inbox.linkedin_urls_by_message_ids(self.session,[message.id])
+        return serialize_application(message,linkedin_url=urls.get(message.id))
 
     async def queue_rematch(self,record_id):
         message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
@@ -572,8 +582,15 @@ class Email:
         user_id=(current_user or {}).get("id")
         if is_application and not row.ingested:
             data=await self.fetch_message(row.message_id)
-            re_create_file=await decode_attachment(data.get("attachments"))
+            already=await Inbox_Messages.get_by_upstream_id(self.session,row.message_id)
-            message,new_user_email=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file)
+            pdfs=extract_pdf_attachments(data.get("attachments"))
+            message,new_user_email=await Inbox_Messages.insert_email(
+                session=self.session,email_data=data,file_path=None,
+            )
+            if pdfs:
+                message=await attach_email_pdfs_to_s3(
+                    self.session,message,pdfs,created_new=(already is None),
+                )
             await Inbox_Message_Triage.mark_ingested(self.session,row.message_id,True)
             if message.attachment and message.file_path and message.match_status is None:
                 await self.enqueue_matching([str(message.id)],force=False)

backend/job/candidate/models.py

@@ -9,7 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 from sqlmodel import Field, Relationship, SQLModel, select
 
-from linkedin_utils import primary_slug_from_text
+from linkedin_utils import NO_SLUG, slug_from_url
 
 if TYPE_CHECKING:
     from inbox.models import Inbox
@@ -38,6 +38,9 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
     # not yet scanned — see linkedin_utils). Same contract as
     # inbox_messages.linkedin_slug; Find Talent matches on it.
     linkedin_slug: str | None = Field(default=None, index=True)
+    # Canonical profile URL for the LinkedIn button. Written at CV ingest;
+    # fetch reads this, it does not re-parse full_text.
+    linkedin_url: str | None = Field(default=None)
     current_company: str = Field(default="")
     # Candidate's role at that company (e.g. "Senior Merchandiser"). Distinct
     # from job_posts.title — that is the role they applied to, not their own.
@@ -80,6 +83,8 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
                     cls.experience,
                     cls.platform,
                     cls.apply_via,
+                    cls.linkedin_url,
+                    Users.linkedin_url.label("user_linkedin_url"),
                     cls.created_at,
                     cls.updated_at,
                     AtsResults.id.label("ats_result_id"),
@@ -139,6 +144,7 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
                     "experience":row["experience"] or None,
                     "platform":row["platform"] or None,
                     "apply_via":row["apply_via"] or None,
+                    "linkedin_url":row["linkedin_url"] or row["user_linkedin_url"] 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,
@@ -189,6 +195,14 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
         email=(fields.get("candidate_email") or "").strip().lower()
         name=(fields.get("candidate_name") or "").strip() or email
         default_pw=os.getenv("DEFAULT_CANDIDATE_PASSWORD","Utopia!@#")
+        full_text=fields.get("full_text") or ""
+        linkedin_url=(fields.get("linkedin_url") or "").strip() or None
+        if linkedin_url:
+            linkedin_slug=slug_from_url(linkedin_url) or NO_SLUG
+        elif full_text:
+            linkedin_slug=NO_SLUG
+        else:
+            linkedin_slug=None
 
         user=await Users.get_user_by_email(session,email)
         if not user:
@@ -200,15 +214,19 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
                 "password":hash_password(default_pw),
                 "is_active":True,
                 "is_deleted":False,
+                "linkedin_url":linkedin_url,
             })
+        elif linkedin_url:
+            await Users.set_linkedin_url_if_empty(session,user_id=user.id,url=linkedin_url)
 
         row=cls(
             candidate_email=email,
             candidate_name=name,
             candidate_phone=(fields.get("candidate_phone") or "").strip(),
             job_post_id=cls._as_uuid(fields.get("job_post_id")),
-            full_text=fields.get("full_text") or "",
+            full_text=full_text,
-            linkedin_slug=primary_slug_from_text(fields.get("full_text") or ""),
+            linkedin_slug=linkedin_slug,
+            linkedin_url=linkedin_url,
             current_company=(fields.get("current_company") or "").strip(),
             current_position=(fields.get("current_position") or "").strip(),
             apply_via=(fields.get("apply_via") or "manual_upload").strip() or "manual_upload",
@@ -331,6 +349,31 @@ class Manual_UPLOAD_CANDIDATE(SQLModel, table=True):
                 out[key] = label
         return out
 
+    @classmethod
+    async def file_paths_by_user_ids(cls, session: AsyncSession, user_ids) -> dict[str, str]:
+        """Newest stored CV path per user — search Open resume when there is no inbox row."""
+        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.file_path)
+            .where(cls.user_id.in_(parsed))
+            .order_by(cls.created_at.desc())
+        )
+        out: dict[str, str] = {}
+        for user_id, file_path in result.all():
+            key = str(user_id)
+            if key in out:
+                continue
+            first = (file_path or "").strip()
+            if first:
+                out[key] = first
+        return out
+
 
 class Candidates(SQLModel, table=True):
 
@@ -342,7 +385,8 @@ class Candidates(SQLModel, table=True):
     job_id: uuid.UUID = Field(foreign_key="job_posts.id", index=True)
     source: str = Field(default="upload")  # "upload" | "inbox" origin metadata, not an FK
     filename: str
-    file_path: str | None = Field(default=None)  # decoded-attachment path (inbox only)
+    file_path: str | None = Field(default=None)  # permanent S3 URL (same as manual_upload_candidate / inbox)
+
     content_sha256: str | None = Field(default=None, index=True)
     candidate_email: str | None = Field(default=None)
     candidate_name: str | None = Field(default=None)
@@ -353,6 +397,8 @@ class Candidates(SQLModel, table=True):
     matched_keywords: list[str] = Field(default_factory=list, sa_type=JSON)
     missing_keywords: list[str] = Field(default_factory=list, sa_type=JSON)
     summary_critique: str | None = Field(default=None)
+    # Public LinkedIn URL extracted from the scored CV. Fetch reads this column.
+    linkedin_url: str | None = Field(default=None)
 
     status: str  # "completed" | "failed"
     error_code: str | None = Field(default=None)
@@ -470,6 +516,31 @@ class Candidates(SQLModel, table=True):
         await session.refresh(existing)
         return existing
 
+    @classmethod
+    async def sync_s3_file_path(cls, session: AsyncSession, email, job_id, file_path):
+        """Stamp the Manual/Email S3 URL onto every Candidates row for this email+job.
+
+        Same link as manual_upload_candidate.file_path — scoring may create the
+        Candidates row after Add Candidate, so both create and score call this.
+        """
+        url=(file_path or "").strip()
+        normalized=(email or "").strip().lower()
+        jid=cls._as_uuid(job_id)
+        if not url or not normalized or jid is None:
+            return 0
+        result=await session.execute(
+            select(cls).where(func.lower(cls.candidate_email)==normalized,cls.job_id==jid)
+        )
+        rows=list(result.scalars().all())
+        if not rows:
+            return 0
+        for row in rows:
+            row.file_path=url
+            row.updated_at=_now()
+            session.add(row)
+        await session.commit()
+        return len(rows)
+
 
 class Interviews(SQLModel, table=True):
     __tablename__ = "interviews"

backend/job/candidate/plugins.py

@@ -118,6 +118,7 @@ def candidate_failed_fields(source, code, message):
         "matched_keywords": [],
         "missing_keywords": [],
         "summary_critique": None,
+        "linkedin_url": None,
     }
 
 
@@ -133,11 +134,58 @@ def candidate_completed_fields(source, result):
         "matched_keywords": result.matched_keywords,
         "missing_keywords": result.missing_keywords,
         "summary_critique": result.summary_critique,
+        "linkedin_url": None,
         "error_code": None,
         "error_message": None,
     }
 
 
+def extract_pdf_link_uris(reader) -> list[str]:
+    """Clickable /URI annotations that pypdf's extract_text() never returns.
+
+    Designer CVs put LinkedIn (and portfolio) behind an icon; the URL lives on
+    the annotation, not in the text layer. Appending these after page text is
+    what lets linkedin_utils see a profile the recruiter can open.
+    """
+    found: list[str] = []
+    seen: set[str] = set()
+    try:
+        pages = reader.pages
+    except Exception:
+        return found
+    for page in pages:
+        try:
+            annots = page.get("/Annots")
+            if annots is None:
+                continue
+            if hasattr(annots, "get_object"):
+                annots = annots.get_object()
+        except Exception:
+            continue
+        if not annots:
+            continue
+        for annot in annots:
+            try:
+                obj = annot.get_object() if hasattr(annot, "get_object") else annot
+                action = obj.get("/A") if obj is not None else None
+                if action is not None and hasattr(action, "get_object"):
+                    action = action.get_object()
+                uri = None
+                if action is not None:
+                    uri = action.get("/URI")
+                if uri is None and obj is not None:
+                    uri = obj.get("/URI")
+                if uri is None:
+                    continue
+                value = str(uri).strip()
+                if value and value not in seen:
+                    seen.add(value)
+                    found.append(value)
+            except Exception:
+                continue
+    return found
+
+
 @normalize_unicode
 @despace_line
 def normalize_spaced_text(text) -> str:

backend/job/candidate/serializers.py

@@ -2,6 +2,12 @@ from inbox.models import Inbox
 from typing import Any,List,Dict
 
 from job.candidate.plugins import documents_from_message, source_from_message_to
+
+
+def _first_file_path(value):
+    if not value:
+        return None
+    return str(value).split(",")[0].strip() or None
 from job.interviews.serializers import serialize_interview
 from job.activity.serializers import serialize_activity
 from job.feedback.serializers import serialize_feedback
@@ -24,6 +30,7 @@ def serialize_candidate(row) -> dict:
         "matched_keywords": list(row.matched_keywords or []),
         "missing_keywords": list(row.missing_keywords or []),
         "summary_critique": row.summary_critique,
+        "linkedin_url": row.linkedin_url or None,
         "status": row.status,
         "error_code": row.error_code,
         "error_message": row.error_message,
@@ -42,6 +49,7 @@ def serialize_manual_upload_candidate(row) -> Dict[str,Any]:
         "candidate_phone":row.candidate_phone,
         "job_post_id":str(row.job_post_id) if row.job_post_id else None,
         "full_text":row.full_text,
+        "linkedin_url":row.linkedin_url or None,
         "current_company":row.current_company,
         "current_position":row.current_position,
         "apply_via":row.apply_via,
@@ -77,6 +85,7 @@ def serialize_candidate_profile(
         "candidate_id": None,
         "name": user.name if user else None,
         "email": user.email if user else None,
+        "linkedin_url": (user.linkedin_url if user else None) or None,
         "is_active": user.is_active if user else None,
         "message_id": str(link.message_id) if link.message_id else None,
         "created_at": link.created_at.isoformat() if link.created_at else None,
@@ -92,6 +101,7 @@ def serialize_candidate_profile(
         "match_status": message.match_status if message else None,
         "match_error": message.match_error if message else None,
         "matched_at": message.matched_at.isoformat() if message and message.matched_at else None,
+        "file_path": _first_file_path(message.file_path if message else None),
         "job_posts": [],
     }
     if not detail:
@@ -145,6 +155,7 @@ def serialize_manual_candidate_profile(row, user, job_post) -> Dict[str, Any]:
         "candidate_id": None,
         "name": (user.name if user else None) or row.candidate_name or None,
         "email": (user.email if user else None) or row.candidate_email or None,
+        "linkedin_url": (user.linkedin_url if user else None) or row.linkedin_url or None,
         "is_active": user.is_active if user else None,
         "message_id": None,
         "created_at": created,
@@ -169,6 +180,7 @@ def serialize_manual_candidate_profile(row, user, job_post) -> Dict[str, Any]:
         "stage": row.status or None,
         "source": (row.platform or "").strip() or None,
         "applied": created,
+        "file_path": file_path,
         "documents": documents,
         "recruiter": job_payload.get("created_by_name") if job_payload else None,
         "recruiter_id": job_payload.get("created_by") if job_payload else None,

backend/job/candidate/views.py

@@ -21,6 +21,7 @@ from job.candidate.plugins import (
     candidate_failed_fields,
     contained_download_path,
     documents_from_message,
+    extract_pdf_link_uris,
     get_scorer,
     get_scoring_settings,
     normalize_spaced_text,
@@ -42,6 +43,20 @@ MANUAL_UPLOAD_TO_ADDRESS=os.getenv(
     "MANUAL_UPLOAD_TO_ADDRESS","manual-cv-upload@hr-ats.local"
 )
 
+
+async def parse_linkedin_url_from_cv(resume_text) -> str | None:
+    """Employment-agent `linkedin_url` key from parsed CV text. None if absent or the call fails."""
+    text=(resume_text or "").strip()
+    if not text:
+        return None
+    try:
+        from employment_agent.execute_agent import run_employment_agent
+        *_,url=await run_employment_agent(resume_text=text)
+        return url
+    except Exception:
+        logger.exception("employment agent linkedin_url parse failed")
+        return None
+
 class FileRead:
     def __init__(self,session:AsyncSession,filename=None,file=None):
         self.session=session
@@ -54,10 +69,18 @@ class FileRead:
             if reader.is_encrypted:
                 raise HTTPException(400, "PDF is password protected")
             pages = [(page.extract_text() or "") for page in reader.pages]
+            text = normalize_spaced_text("\n".join(pages))
+            # Icon-only LinkedIn buttons never appear in extract_text(); the
+            # URL is on the annotation. Append so the employment agent can
+            # return linkedin_url as its own parsed key.
+            uris = extract_pdf_link_uris(reader)
+            if uris:
+                extra = "\n".join(uris)
+                text = f"{text}\n\n{extra}".strip() if text else extra
             return {
                 "filename": self.filename,
                 "num_pages": len(reader.pages),
-                "text": normalize_spaced_text("\n".join(pages)),
+                "text": text,
             }
         except HTTPException:
             raise
@@ -76,63 +99,35 @@ class FileRead:
             raise HTTPException(status_code=400,detail=str(e))
 
     async def save_manual_upload(self):
-        """Write the uploaded CV under inbox/decoded_attachments.
+        """Deprecated — Manual CVs go to S3 via create_candidate (no local disk)."""
-
+        raise HTTPException(
-        Returns ``{"file_name", "file_path"}``: the recruiter-facing original
+            status_code=410,
-        name, and the absolute path actually written.
+            detail="Local CV storage was removed; use create_candidate (S3 Manual/{id}/{user_id}/)",
-
+        )
-        Those two differ deliberately. decode_attachment writes ``Path(name).name``
-        with plain ``write_bytes`` — no collision handling — so two candidates
-        uploading "resume.pdf" would silently clobber each other and the first
-        row's file_path would then serve the second candidate's CV. Prefixing the
-        stored basename with a uuid makes every upload its own file, while
-        file_name keeps what the recruiter recognises. resolve_attachment_path
-        handles the result either way: the stored absolute path wins, and its
-        basename-under-attachments fallback still finds the prefixed name.
-        """
-        from inbox.file_decoder import AttachmentDecodeError,decode_attachment
-
-        # Separators normalized before taking the basename: a Windows client can
-        # send "C:\Users\x\cv.pdf", whose Path(...).name on Linux is the whole
-        # string. Same reasoning as inbox.plugins.resolve_attachment_path.
-        original=Path((self.filename or "resume.pdf").replace("\\","/")).name or "resume.pdf"
-        stored=f"{uuid.uuid4().hex}-{original}"
-        try:
-            paths=await decode_attachment([{
-                "name":stored,
-                "contentBytes":base64.b64encode(self.file).decode("ascii"),
-            }])
-        except AttachmentDecodeError as e:
-            raise HTTPException(status_code=400,detail=str(e))
-        if not paths:
-            # decode_attachment skips rather than raises on an unsupported
-            # extension, so an empty list is the only signal that nothing landed.
-            raise HTTPException(status_code=400,detail="attachment could not be saved")
-        return {"file_name":original,"file_path":paths[0]}
 
     @staticmethod
     def discard_upload(file_path):
-        """Best-effort removal of a saved CV whose row never got created.
+        """Best-effort removal of a leftover local CV (legacy rows only)."""
-
-        Called on the failure path so a rejected request (a missing email, a DB
-        error) does not leave an orphan PDF behind. Failure to delete is logged
-        and swallowed — it must never mask the error that got us here.
-        """
         if not file_path:
             return
+        if str(file_path).lower().startswith("http://") or str(file_path).lower().startswith("https://"):
+            return
         try:
             Path(file_path).unlink(missing_ok=True)
         except OSError as e:
             logger.warning("could not remove orphaned upload %s: %s",file_path,e)
 
     async def ingest_upload(self,candidate_email=None,candidate_name=None,current_user=None):
-        """Persist a recruiter-uploaded CV with full email-ingestion parity."""
+        """Persist a recruiter-uploaded CV with full email-ingestion parity (S3)."""
-        from inbox.file_decoder import AttachmentDecodeError,decode_attachment
+        from inbox.file_decoder import extract_pdf_attachments
         from inbox.cv_tasks import match_uploaded_cv
+        from inbox.plugins import attach_email_pdfs_to_s3
         from inbox.views import Email
+        from s3.plugins import S3ServiceError,assert_pdf
 
         parsed=await self.read_file()
         text=parsed.get("text") or ""
+        parsed_linkedin=await parse_linkedin_url_from_cv(text)
         detected,emails_found=extract_candidate_email(text)
         supplied=(candidate_email or "").strip().lower() or None
         email=supplied or detected
@@ -152,14 +147,18 @@ class FileRead:
 
         filename=self.filename or "resume.pdf"
         try:
-            paths=await decode_attachment([{
+            assert_pdf(filename,"application/pdf")
-                "name":filename,
+        except S3ServiceError as e:
-                "contentBytes":base64.b64encode(self.file).decode("ascii"),
+            raise HTTPException(status_code=e.status_code,detail=e.message) from e
-            }])
+
-        except AttachmentDecodeError as e:
+        # In-memory only — no decoded_attachments write.
-            raise HTTPException(status_code=400,detail=str(e))
+        import base64 as _b64
-        if not paths:
+        pdfs=extract_pdf_attachments([{
-            raise HTTPException(status_code=400,detail="attachment could not be saved")
+            "name":filename,
+            "contentBytes":_b64.b64encode(self.file).decode("ascii"),
+        }])
+        if not pdfs:
+            raise HTTPException(status_code=400,detail="Only PDF resumes are allowed")
 
         now=datetime.now(timezone.utc).isoformat()
         email_data={
@@ -178,8 +177,19 @@ class FileRead:
             "receivedDateTime":now,
         }
         row,new_user_email=await Inbox_Messages.insert_email(
-            self.session,email_data,file_path=paths,
+            self.session,email_data,file_path=None,
         )
+        try:
+            row=await attach_email_pdfs_to_s3(self.session,row,pdfs,created_new=True)
+        except Exception as e:
+            raise HTTPException(status_code=502,detail=f"S3 upload failed: {e}") from e
+
+        if parsed_linkedin:
+            user_id=await Inbox_Messages.get_linked_user_id(self.session,row.id)
+            if user_id and await Users.set_linkedin_url_if_empty(
+                self.session,user_id=user_id,url=parsed_linkedin,
+            ):
+                await self.session.commit()
 
         created_at=datetime.now(timezone.utc).isoformat()
         task=await match_uploaded_cv.kicker().with_labels(
@@ -198,8 +208,6 @@ class FileRead:
                 logger.warning("account setup mail failed for %s: %s",new_user_email,e)
                 account_setup=[{"email":new_user_email,"sent":False}]
 
-        # Inbox link may not exist yet (match task creates it later); resolve
-        # by email because insert_email creates the Users row synchronously.
         user=await Users.get_user_by_email(self.session,email)
         if user:
             await HistoryRecorder(self.session).record(
@@ -228,7 +236,7 @@ class FileRead:
         }
 
     async def match_inbox_cv(self,inbox_message_id,current_user=None):
-        from inbox.plugins import resolve_attachment_path
+        from inbox.plugins import load_file_bytes
         from inbox.tasks import match_inbox_message
 
         row=await Inbox_Messages.get_inbox_message_by_id(self.session,inbox_message_id)
@@ -237,13 +245,17 @@ class FileRead:
         if not row.attachment or not row.file_path:
             raise HTTPException(status_code=400,detail="your file isnt in the system")
 
-        found=None
+        found_name=None
         for path_str in (p.strip() for p in row.file_path.split(",") if p.strip()):
-            path=resolve_attachment_path(path_str)
+            # S3 URL or local — presence of bytes (or a https URL we already stored) counts.
-            if path.is_file():
+            if path_str.lower().startswith("http://") or path_str.lower().startswith("https://"):
-                found=path
+                found_name=Path(path_str.replace("\\","/")).name or "resume.pdf"
                 break
-        if found is None:
+            raw=load_file_bytes(path_str)
+            if raw is not None:
+                found_name=Path(path_str.replace("\\","/")).name or "resume.pdf"
+                break
+        if found_name is None:
             raise HTTPException(status_code=400,detail="your file isnt in the system")
 
         created_at=datetime.now(timezone.utc).isoformat()
@@ -253,7 +265,7 @@ class FileRead:
             queue="inbox",
         ).kiq(str(row.id),force=True)
 
-        file_name=(row.file_name or "").split(",")[0].strip() or found.name
+        file_name=(row.file_name or "").split(",")[0].strip() or found_name
         await HistoryRecorder(self.session).record(
             HistoryEvent.CANDIDATE_IMPORTED.value,
             current_user=current_user,message_id=inbox_message_id,
@@ -311,10 +323,8 @@ class CandidateScoring:
         return await self._score_and_persist(job_id,sources,"upload",current_user)
 
     async def score_inbox(self,job_id,message_ids,current_user):
-        """Score the decoded attachments of inbox messages (PK uuids, not Graph ids)."""
+        """Score PDF attachments of inbox messages (S3 URLs or legacy local paths)."""
-        # Local import: inbox.plugins imports this module (FileRead), so a top-level
+        from inbox.plugins import load_file_bytes
-        # import would be circular — same pattern as match_inbox_cv above.
-        from inbox.plugins import resolve_attachment_path
 
         sources=[]
         for mid in message_ids:
@@ -323,28 +333,31 @@ class CandidateScoring:
                 raise HTTPException(status_code=404,detail=f"Inbox message {mid} not found")
             if not row.file_path:
                 continue
-            for path_str in (p.strip() for p in row.file_path.split(",") if p.strip()):
+            names=[n.strip() for n in (row.file_name or "").split(",") if n.strip()]
-                path=resolve_attachment_path(path_str)
+            for idx,path_str in enumerate(p.strip() for p in row.file_path.split(",") if p.strip()):
+                name=names[idx] if idx<len(names) else Path(path_str.replace("\\","/")).name or "resume.pdf"
                 source={
-                    "filename":path.name,
+                    "filename":name,
                     "data":None,
-                    "file_path":str(path),
+                    "file_path":path_str,
-                    "inbox_message_id":row.id,  # call-scoped; not persisted on Candidates
+                    "inbox_message_id":row.id,
                     "candidate_email":(row.message_from or "").strip().lower() or None,
                     "precheck":None,
                 }
-                suffix=path.suffix.lower()
+                lower=name.lower()
-                if suffix in (".doc",".docx"):
+                if lower.endswith(".doc") or lower.endswith(".docx"):
                     source["precheck"]=(ErrorCode.UNSUPPORTED_FILE_TYPE,"DOC/DOCX extraction is not supported yet.")
-                elif suffix!=".pdf":
+                elif not lower.endswith(".pdf") and ".pdf" not in path_str.lower():
                     source["precheck"]=(ErrorCode.UNSUPPORTED_FILE_TYPE,"Only PDF resumes are supported.")
-                elif not path.is_file():
-                    source["precheck"]=(FILE_NOT_FOUND,"The decoded attachment is missing on disk.")
                 else:
                     try:
-                        source["data"]=await asyncio.to_thread(path.read_bytes)
+                        data=await asyncio.to_thread(load_file_bytes,path_str)
-                    except OSError:
+                    except Exception:
-                        source["precheck"]=(FILE_NOT_FOUND,"The decoded attachment could not be read.")
+                        data=None
+                    if data is None:
+                        source["precheck"]=(FILE_NOT_FOUND,"The CV could not be loaded from S3 (check GetObject / public read).")
+                    else:
+                        source["data"]=data
                 sources.append(source)
         if not sources:
             raise HTTPException(status_code=400,detail="No attachments found for the given message(s)")
@@ -376,6 +389,19 @@ class CandidateScoring:
         if len(jd)>settings.max_jd_chars:
             raise HTTPException(status_code=422,detail="The job post is too large to score against")
         fields_by_slot=await self._score_sources(sources,jd,settings)
+        # Prefer the Manual S3 URL for this email+job when scoring from a raw upload
+        # (Add Candidate scores right after create — same link as manual_upload_candidate).
+        for slot,source in enumerate(sources):
+            fields=fields_by_slot.get(slot) or {}
+            if (fields.get("file_path") or source.get("file_path") or "").strip():
+                continue
+            email=(fields.get("candidate_email") or source.get("candidate_email") or "").strip().lower()
+            if not email:
+                continue
+            manual=await Manual_UPLOAD_CANDIDATE.get_by_email_and_job(self.session,email,job.id)
+            if manual and (manual.file_path or "").strip():
+                fields["file_path"]=manual.file_path.strip()
+                source["file_path"]=manual.file_path.strip()
         common={
             "job_id":job.id,
             "source":source_kind,
@@ -384,7 +410,19 @@ class CandidateScoring:
         }
         rows=[]
         for slot in range(len(sources)):
-            rows.append(await Candidates.upsert_candidate(self.session,{**fields_by_slot[slot],**common}))
+            fields={**fields_by_slot[slot],**common}
+            email=(fields.get("candidate_email") or "").strip().lower()
+            if email and not fields.get("linkedin_url"):
+                user=await Users.get_user_by_email(self.session,email)
+                if user and (user.linkedin_url or "").strip():
+                    fields["linkedin_url"]=user.linkedin_url
+            row=await Candidates.upsert_candidate(self.session,fields)
+            if row.linkedin_url and row.candidate_email:
+                if await Users.set_linkedin_url_if_empty(
+                    self.session,email=row.candidate_email,url=row.linkedin_url,
+                ):
+                    await self.session.commit()
+            rows.append(row)
         await self._sync_ats_results(source_kind,job,rows,sources,current_user)
         rows.sort(key=lambda r:(0,-(r.match_score or 0)) if r.status=="completed" else (1,0))
         return [serialize_candidate(row) for row in rows]
@@ -420,7 +458,7 @@ class CandidateScoring:
             scorer=get_scorer(),
             concurrency=settings.scoring_concurrency,
         )
-        for (slot,_),result in zip(extracted,scored,strict=True):
+        for (slot,_resume),result in zip(extracted,scored,strict=True):
             source=sources[slot]
             if isinstance(result,CompletedCandidate):
                 fields_by_slot[slot]=candidate_completed_fields(source,result)
@@ -565,6 +603,8 @@ class CandidateView:
                 if not file_bytes:
                     raise HTTPException(status_code=422,detail="file is empty")
 
+            parsed_linkedin=await parse_linkedin_url_from_cv(full_text)
+
             data={
                 "candidate_email":email,
                 "candidate_name":(candidate_name or "").strip(),
@@ -581,6 +621,7 @@ class CandidateView:
                 # path filled after S3 succeeds; never leave a local orphan path here
                 "file_path":(file_path or "").strip() if file_bytes is None else "",
                 "full_text":full_text or "",
+                "linkedin_url":parsed_linkedin,
                 "created_by":current_user,
             }
             row=await Manual_UPLOAD_CANDIDATE.create_manual_upload_candidate(session=self.session,fields=data)
@@ -606,6 +647,16 @@ class CandidateView:
                 row=await Manual_UPLOAD_CANDIDATE.set_file_path(
                     self.session,row.id,uploaded["url"],file_name=uploaded.get("filename") or original_name,
                 )
+                # Same permanent URL on candidates rows for this email+job (if scored already).
+                try:
+                    await Candidates.sync_s3_file_path(
+                        self.session,
+                        email=row.candidate_email,
+                        job_id=row.job_post_id,
+                        file_path=row.file_path,
+                    )
+                except Exception:
+                    logger.exception("candidates.file_path sync failed for manual %s",row.id)
 
             await HistoryRecorder(self.session).record(
                 HistoryEvent.CANDIDATE_CREATED.value,
@@ -705,6 +756,7 @@ class CandidateView:
                     "job_posts":payload.get("job_posts") or [],
                     "assigned_job_post":payload.get("assigned_job_post"),
                     "source":payload.get("source"),
+                    "file_path":payload.get("file_path"),
                     "ai_score":None,
                     "recommendation":None,
                 })

backend/job/job_post/models.py

@@ -167,43 +167,6 @@ class JobPosts(SQLModel, table=True):
         result = await session.execute(statement)
         return list(result.scalars().all()), total
 
-    @classmethod
-    async def recruiter_names(cls, session: AsyncSession, recruiter_ids) -> dict[str, str]:
-        """Resolve {recruiter_id: name} for a page of rows in a single query."""
-        # Local import and COLUMN select, both load-bearing: users.models imports
-        # this module at its top, so a module-level import here is a startup cycle;
-        # and a Users *entity* would drag in its five selectin relations for what is
-        # a two-column lookup.
-        from users.models import Users
-
-        uids = {u for u in (recruiter_ids or []) if u}
-        if not uids:
-            return {}
-        result = await session.execute(
-            select(Users.id, Users.name).where(Users.id.in_(uids))
-        )
-        return {str(uid): name for uid, name in result.all()}
-
-    @classmethod
-    async def applicant_counts(cls, session: AsyncSession, job_post_ids) -> dict[str, int]:
-        """Resolve {job_post_id: applicant_count} for a page of rows in a single query.
-
-        Counts Inbox_Messages rows, not Inbox rows: one message fans out to several
-        Inbox rows (one per recipient), so counting Inbox would over-count.
-        Local import matches recruiter_names — job_post.models ↔ inbox.models is a cycle.
-        """
-        from inbox.models import Inbox_Messages
-
-        uids = {u for u in (job_post_ids or []) if u}
-        if not uids:
-            return {}
-        result = await session.execute(
-            select(Inbox_Messages.assigned_job_post_id, func.count().label("applicants"))
-            .where(Inbox_Messages.assigned_job_post_id.in_(uids))
-            .group_by(Inbox_Messages.assigned_job_post_id)
-        )
-        return {str(job_id): int(n) for job_id, n in result.all()}
-
     @classmethod
     async def insert_job_post(cls, session: AsyncSession, fields: dict):
         row = cls(**fields)

backend/job/job_post/views.py

@@ -8,7 +8,9 @@ from dotenv import load_dotenv
 from fastapi import HTTPException
 from sqlalchemy.ext.asyncio import AsyncSession
 from pydantic import BaseModel, model_validator
+from inbox.models import Inbox_Messages
 from job.job_post.models import JobPosts,SocialPlatform
+from users.models import Users
 from job.job_post.plugins import (
     BufferError,
     create_buffer_post,
@@ -173,10 +175,10 @@ class JobPost:
             department=department,requisition_status=requisition_status,
             employment_type=employment_type,
         )
-        names=await JobPosts.recruiter_names(
+        names=await Users.names_by_ids(
             self.session,[r.current_recruiter_id for r in rows],
         )
-        counts=await JobPosts.applicant_counts(self.session,[r.id for r in rows])
+        counts=await Inbox_Messages.counts_by_job_post_ids(self.session,[r.id for r in rows])
         return [
             serialize_job_row(
                 r,
@@ -187,7 +189,7 @@ class JobPost:
         ],total
 
     async def _job_row(self,row):
-        names=await JobPosts.recruiter_names(
+        names=await Users.names_by_ids(
             self.session,[row.current_recruiter_id] if row.current_recruiter_id else [],
         )
         return serialize_job_row(row,recruiter_name=names.get(str(row.current_recruiter_id)))

backend/linkedin_utils.py

@@ -13,10 +13,31 @@ import re
 from urllib.parse import unquote
 
 # CV text arrives from PDF extraction: URLs may carry percent-escapes, no
-# scheme ("linkedin.com/in/jane-doe"), or trailing sentence punctuation glued
+# scheme ("linkedin.com/in/jane-doe"), trailing sentence punctuation glued on by
-# on by layout. /pub/ is the legacy public-profile path some older CVs still
+# layout, or line-wraps inside the path ("linkedin.com/in/\njane-doe").
-# carry.
+# /pub/ is the legacy public-profile path; /mwlite/in/ is the mobile web path.
-_SLUG_RE = re.compile(r"linkedin\.com/(?:in|pub)/([A-Za-z0-9\-_.%]+)", re.IGNORECASE)
+_SLUG_RE = re.compile(
+    r"linkedin\.com/(?:in|pub|mwlite/in)/([A-Za-z0-9\-_.%]+)",
+    re.IGNORECASE,
+)
+
+# pypdf wraps URLs across lines / glyph gaps. Flatten those runs before matching
+# so "linkedin.com/in/\n jane-doe" still yields a slug.
+_LINKEDIN_RUN_RE = re.compile(
+    r"(?:https?://)?(?:(?:[a-z0-9-]+\.)*)linkedin\.com(?:\s*/\s*[A-Za-z0-9\-_.%]*)+",
+    re.IGNORECASE,
+)
+
+# Clickable CV icons often store the URL only in an HTML href or a PDF
+# annotation, not in the visible text layer.
+_HREF_RE = re.compile(
+    r"""href\s*=\s*["']([^"'>\s]*(?:linkedin\.com|lnkd\.in)[^"']*)["']""",
+    re.IGNORECASE,
+)
+
+# Short links from LinkedIn's own share button. Not a match key (no /in/<slug>)
+# but enough to open a profile from the inbox button.
+_LNKD_RE = re.compile(r"lnkd\.in/([A-Za-z0-9_-]+)", re.IGNORECASE)
 
 # Sentinel stored on application rows: NULL means "never scanned", the empty
 # string means "scanned, no link found". The distinction is what lets the lazy
@@ -36,16 +57,29 @@ def slug_from_url(url) -> str | None:
     """Slug from an already-normalized profile URL (talent_profiles.linkedin_url)."""
     if not url:
         return None
-    match = _SLUG_RE.search(str(url))
+    match = _SLUG_RE.search(_flatten_linkedin_runs(str(url)))
     return normalize_slug(match.group(1)) if match else None
 
 
+def _flatten_linkedin_runs(text: str) -> str:
+    """Remove whitespace inside linkedin.com/... runs so wrapped PDFs still match."""
+    if not text:
+        return ""
+    return _LINKEDIN_RUN_RE.sub(lambda m: re.sub(r"\s+", "", m.group(0)), text)
+
+
+def _haystack(text) -> str:
+    """Flatten wrapped LinkedIn URLs and splice href= targets into the scan text."""
+    raw = text or ""
+    hrefs = "\n".join(_HREF_RE.findall(raw))
+    blob = f"{raw}\n{hrefs}" if hrefs else raw
+    return _flatten_linkedin_runs(blob)
+
+
 def slugs_from_text(text) -> list[str]:
     """Every distinct slug mentioned in a CV, in order of first appearance."""
-    if not text:
-        return []
     found: list[str] = []
-    for match in _SLUG_RE.finditer(text):
+    for match in _SLUG_RE.finditer(_haystack(text)):
         slug = normalize_slug(match.group(1))
         if slug and slug not in found:
             found.append(slug)
@@ -56,3 +90,18 @@ def primary_slug_from_text(text) -> str:
     """The slug to persist on an application row; NO_SLUG when the CV has none."""
     slugs = slugs_from_text(text)
     return slugs[0] if slugs else NO_SLUG
+
+
+def profile_url_from_text(text) -> str | None:
+    """Public profile URL for the inbox LinkedIn button, or None.
+
+    Prefers /in/<slug> (and /pub/, /mwlite/in/). Falls back to lnkd.in short
+    links which open the profile but are not a Find Talent match key.
+    """
+    slug = primary_slug_from_text(text)
+    if slug:
+        return f"https://www.linkedin.com/in/{slug}"
+    short = _LNKD_RE.search(_haystack(text))
+    if short:
+        return f"https://lnkd.in/{short.group(1)}"
+    return None

backend/migrations/manual/010_linkedin_url.sql

@@ -0,0 +1,38 @@
+-- 010_linkedin_url.sql
+-- Persist the public LinkedIn profile URL extracted from a CV at ingest time
+-- on users, manual_upload_candidate, and candidates. Fetch reads this column
+-- instead of re-parsing resume text. Applied at startup by
+-- alembic_setup.run_manual_sql().
+
+ALTER TABLE app.users
+    ADD COLUMN IF NOT EXISTS linkedin_url TEXT;
+
+ALTER TABLE app.manual_upload_candidate
+    ADD COLUMN IF NOT EXISTS linkedin_url TEXT;
+
+ALTER TABLE app.candidates
+    ADD COLUMN IF NOT EXISTS linkedin_url TEXT;
+
+-- Backfill from already-extracted /in/<slug> values.
+UPDATE app.manual_upload_candidate
+SET linkedin_url = 'https://www.linkedin.com/in/' || linkedin_slug
+WHERE linkedin_url IS NULL
+  AND linkedin_slug IS NOT NULL
+  AND linkedin_slug <> '';
+
+UPDATE app.users AS u
+SET linkedin_url = m.linkedin_url
+FROM app.manual_upload_candidate AS m
+WHERE u.id = m.user_id
+  AND u.linkedin_url IS NULL
+  AND m.linkedin_url IS NOT NULL
+  AND m.linkedin_url <> '';
+
+UPDATE app.users AS u
+SET linkedin_url = 'https://www.linkedin.com/in/' || m.linkedin_slug
+FROM app.inbox AS i
+JOIN app.inbox_messages AS m ON m.id = i.message_id
+WHERE i.user_id = u.id
+  AND u.linkedin_url IS NULL
+  AND m.linkedin_slug IS NOT NULL
+  AND m.linkedin_slug <> '';

backend/s3/app.py

@@ -35,7 +35,7 @@ async def upload_s3_file(
     owner_id: str=Form(...),
     current_user: dict=Depends(require_permission(PermissionTag.CANDIDATES_CREATE,PermissionTag.SETTINGS_EDIT,require_all=False)),
 ):
-    """PDF only. Requires an existing table row — key is {source}/{record_id}/{owner_id}/{file}.pdf."""
+    """PDF only. Private PutObject under {source}/{record_id}/{owner_id}/{file}.pdf."""
     try:
         service=S3Storage()
         data=await service.upload_for_record(file,source=source,record_id=record_id,owner_id=owner_id)
@@ -51,7 +51,7 @@ async def fetch_s3_url(
     key: str=Query(...),
     current_user: dict=Depends(require_permission(PermissionTag.CANDIDATES_VIEW,PermissionTag.SETTINGS_VIEW,require_all=False)),
 ):
-    """Recompute the permanent URL for an existing key (no S3 round trip)."""
+    """Stable private object address stored in file_path (not anonymously openable)."""
     try:
         service=S3Storage()
         data=await service.object_url(key)
@@ -62,6 +62,38 @@ async def fetch_s3_url(
         raise HTTPException(status_code=500,detail=str(e))
 
 
+@router.get("/s3/open")
+async def open_s3_file(
+    key: str=Query(...,description="S3 key or stored file_path URL"),
+    expires_in: int | None=Query(None,ge=60,le=604800),
+    current_user: dict=Depends(require_permission(PermissionTag.CANDIDATES_VIEW,PermissionTag.SETTINGS_VIEW,PermissionTag.INBOX_VIEW,require_all=False)),
+):
+    """Short-lived presigned GET for a private CV — open this URL in the browser."""
+    try:
+        service=S3Storage()
+        data=await service.open_url(key,expires_in=expires_in)
+        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("/s3/download")
+async def download_s3_file(
+    key: str=Query(...,description="S3 key or stored file_path URL"),
+    current_user: dict=Depends(require_permission(PermissionTag.CANDIDATES_VIEW,PermissionTag.SETTINGS_VIEW,require_all=False)),
+):
+    """Stream a private PDF through the API (IAM GetObject — no public bucket)."""
+    try:
+        service=S3Storage()
+        return await service.download_file(key)
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
 @router.post("/s3/delete")
 async def delete_s3_file(
     payload: DeleteObjectBody,

backend/s3/plugins.py

@@ -1,10 +1,12 @@
-"""S3 helpers — boto3 client class, upload/delete, permanent object URLs.
+"""S3 helpers — boto3 client class, upload/delete, private-object access.
 
 No FastAPI imports (house rule). Raise S3ServiceError; s3/views.py maps to HTTPException.
 
-Permanent links: we NEVER return expiring presigned URLs. The URL is the virtual-hosted
+CVs are confidential: objects stay private (no Principal "*" bucket policy).
-HTTPS object address, which stays valid until the object is deleted (or the bucket
+DB ``file_path`` stores a stable object address (virtual-hosted HTTPS form of the key)
-policy stops public GetObject).
+so the same path survives forever until the object is deleted. That address is NOT
+meant to be opened anonymously — open via authenticated download or a short-lived
+presigned GET (see S3.presigned_get_url / GET /s3/open).
 
 CV keys are record-scoped (atomicity): DB row is created first, then upload uses that id:
 
@@ -25,10 +27,11 @@ from pathlib import Path
 
 import boto3
 from botocore.client import BaseClient
+from botocore.config import Config
 from botocore.exceptions import BotoCoreError,ClientError
 from dotenv import load_dotenv
 
-load_dotenv()
+load_dotenv(override=True)
 
 logger=logging.getLogger("s3.plugins")
 
@@ -36,11 +39,12 @@ AWS_ACCESS_KEY_ID=os.getenv("AWS_ACCESS_KEY_ID","").strip()
 AWS_SECRET_ACCESS_KEY=os.getenv("AWS_SECRET_ACCESS_KEY","").strip()
 AWS_REGION=(os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") or "us-east-2").strip()
 S3_BUCKET=os.getenv("S3_BUCKET","").strip()
-# Optional CDN / custom domain. Blank → https://{bucket}.s3.{region}.amazonaws.com/{key}
+# Optional CDN / custom domain for stable identity URLs only (still private).
 S3_PUBLIC_BASE_URL=os.getenv("S3_PUBLIC_BASE_URL","").strip().rstrip("/")
-
+# Short-lived open links for recruiters (seconds). Max 604800 (7d) with IAM user keys.
-# modern buckets often have ACLs disabled; leave blank and rely on bucket policy.
+S3_PRESIGN_EXPIRES_SECONDS=int(os.getenv("S3_PRESIGN_EXPIRES_SECONDS") or "900")
-S3_OBJECT_ACL=os.getenv("S3_OBJECT_ACL","").strip()  # e.g. public-read
+# Leave blank — never use public-read ACL for confidential CVs.
+S3_OBJECT_ACL=os.getenv("S3_OBJECT_ACL","").strip()
 
 _SAFE_NAME=re.compile(r"[^A-Za-z0-9._-]+")
 _PDF_MIME=frozenset({"application/pdf","application/x-pdf"})
@@ -81,8 +85,6 @@ def assert_pdf(filename: str,content_type: str | None=None) -> str:
     if not safe.lower().endswith(".pdf"):
         raise S3ServiceError("Only PDF files are allowed",status_code=415)
     mime=(content_type or "").strip().lower().split(";")[0].strip()
-    # browsers sometimes send application/octet-stream for PDFs — allow that
-    # only when the extension already passed; reject every other non-PDF MIME.
     if mime and mime not in _PDF_MIME and mime!="application/octet-stream":
         raise S3ServiceError(f"Only PDF MIME types are allowed (got {mime})",status_code=415)
     return safe
@@ -92,7 +94,6 @@ def normalize_source(source: str) -> str:
     raw=(source or "").strip()
     if not raw:
         raise S3ServiceError("source is required (Email|Manual|Form)",status_code=422)
-    # accept case-insensitive input, store canonical folder casing
     for name in S3Source.ALL:
         if raw.lower()==name.lower():
             return name
@@ -103,7 +104,7 @@ def normalize_source(source: str) -> str:
 
 
 class S3:
-    """One boto3 client + bucket config — upload / delete / URL / health share this."""
+    """One boto3 client + bucket config — private objects, auth download / short presign."""
 
     def __init__(self,client: BaseClient | None=None):
         self._require_config()
@@ -111,11 +112,15 @@ class S3:
         self.region=AWS_REGION
         self.public_base_url=S3_PUBLIC_BASE_URL
         self.object_acl=S3_OBJECT_ACL
+        self.presign_expires=max(60,min(S3_PRESIGN_EXPIRES_SECONDS,604800))
+        # Regional endpoint + SigV4 — required for private-bucket presigns outside us-east-1.
         self.client=client or boto3.client(
             "s3",
             region_name=self.region,
+            endpoint_url=f"https://s3.{self.region}.amazonaws.com",
             aws_access_key_id=AWS_ACCESS_KEY_ID,
             aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
+            config=Config(signature_version="s3v4",s3={"addressing_style":"virtual"}),
         )
 
     @staticmethod
@@ -132,7 +137,6 @@ class S3:
             )
 
     def _raise_boto(self,exc,action,key=None,status_code=502):
-        """Map ClientError / BotoCoreError → S3ServiceError (single place)."""
         if isinstance(exc,ClientError):
             code=(exc.response or {}).get("Error",{}).get("Code") or ""
             logger.exception("s3 %s failed key=%s code=%s",action,key,code)
@@ -159,8 +163,8 @@ class S3:
         safe=assert_pdf(filename)
         return f"{folder}/{rid}/{oid}/{safe}"
 
-    def permanent_object_url(self,key: str) -> str:
+    def object_url(self,key: str) -> str:
-        """Stable HTTPS URL for a public object — does not expire."""
+        """Stable object address for DB file_path — private, not anonymously openable."""
         object_key=(key or "").lstrip("/")
         if not object_key:
             raise S3ServiceError("object key is required",status_code=422)
@@ -170,6 +174,32 @@ class S3:
             raise S3ServiceError("S3_BUCKET is not configured",status_code=500)
         return f"https://{self.bucket}.s3.{self.region}.amazonaws.com/{object_key}"
 
+    # Back-compat alias used by older call sites
+    permanent_object_url=object_url
+
+    def presigned_get_url(self,key_or_url: str,expires_in: int | None=None) -> dict:
+        """Short-lived HTTPS GET for a private object — browser-openable after auth gate."""
+        object_key=self.key_from_url(key_or_url)
+        if not object_key:
+            raise S3ServiceError("object key is required",status_code=422)
+        ttl=expires_in if expires_in is not None else self.presign_expires
+        ttl=max(60,min(int(ttl),604800))
+        try:
+            name=Path(object_key).name or "resume.pdf"
+            url=self.client.generate_presigned_url(
+                "get_object",
+                Params={
+                    "Bucket":self.bucket,
+                    "Key":object_key,
+                    "ResponseContentType":"application/pdf",
+                    "ResponseContentDisposition":f'inline; filename="{name}"',
+                },
+                ExpiresIn=ttl,
+            )
+        except (ClientError,BotoCoreError) as e:
+            self._raise_boto(e,"presign",key=object_key)
+        return {"key":object_key,"url":url,"expires_in":ttl}
+
     def upload_bytes(
         self,
         body: bytes,
@@ -178,7 +208,7 @@ class S3:
         content_type: str | None=None,
         key: str | None=None,
     ) -> dict:
-        """PutObject + permanent URL. Prefer upload_for_record for CV flows."""
+        """PutObject + stable object_url for DB. Prefer upload_for_record for CV flows."""
         if body is None:
             raise S3ServiceError("file body is required",status_code=422)
         if not key:
@@ -190,7 +220,7 @@ class S3:
         object_key=key.lstrip("/")
         ctype=content_type if (content_type or "").strip().lower().startswith("application/pdf") else "application/pdf"
         extra={}
-        if self.object_acl:
+        if self.object_acl and self.object_acl.strip().lower()!="public-read":
             extra["ACL"]=self.object_acl
         try:
             self.client.put_object(
@@ -202,7 +232,7 @@ class S3:
             )
         except (ClientError,BotoCoreError) as e:
             self._raise_boto(e,"upload",key=object_key)
-        url=self.permanent_object_url(object_key)
+        url=self.object_url(object_key)
         return {
             "bucket":self.bucket,
             "key":object_key,
@@ -238,9 +268,49 @@ class S3:
         result["owner_id"]=str(owner_id)
         return result
 
+    @staticmethod
+    def is_http_url(value: str) -> bool:
+        v=(value or "").strip().lower()
+        return v.startswith("https://") or v.startswith("http://")
+
+    def key_from_url(self,url: str) -> str:
+        """Strip virtual-hosted / path-style S3 URL down to the object key."""
+        raw=(url or "").strip()
+        if not raw:
+            raise S3ServiceError("url is required",status_code=422)
+        if not self.is_http_url(raw):
+            return raw.lstrip("/")
+        from urllib.parse import urlparse,unquote
+        parsed=urlparse(raw)
+        path=unquote((parsed.path or "").lstrip("/"))
+        host=(parsed.netloc or "").lower()
+        if host.startswith(f"{self.bucket.lower()}.s3."):
+            return path
+        if host.startswith("s3.") or host.startswith("s3-"):
+            prefix=f"{self.bucket}/"
+            if path.startswith(prefix):
+                return path[len(prefix):]
+            parts=path.split("/",1)
+            if len(parts)==2 and parts[0]==self.bucket:
+                return parts[1]
+        if self.public_base_url and raw.startswith(self.public_base_url+"/"):
+            return raw[len(self.public_base_url)+1:]
+        return path
+
+    def download_bytes(self,key_or_url: str) -> bytes:
+        """Authenticated GetObject — matching / app download for private objects."""
+        object_key=self.key_from_url(key_or_url)
+        if not object_key:
+            raise S3ServiceError("object key is required",status_code=422)
+        try:
+            obj=self.client.get_object(Bucket=self.bucket,Key=object_key)
+            return obj["Body"].read()
+        except (ClientError,BotoCoreError) as e:
+            self._raise_boto(e,"download",key=object_key,status_code=403 if isinstance(e,ClientError) else 502)
+
     def delete_object(self,key: str) -> dict:
-        """DeleteObject — after this the permanent URL 404s."""
+        """DeleteObject — after this the stable address is dead."""
-        object_key=(key or "").lstrip("/")
+        object_key=self.key_from_url(key) if self.is_http_url(key) else (key or "").lstrip("/")
         if not object_key:
             raise S3ServiceError("object key is required",status_code=422)
         try:
@@ -264,5 +334,7 @@ class S3:
             "bucket":self.bucket,
             "region":self.region,
             "status":"ok",
-            "public_base_url":base,
+            "object_base_url":base,
+            "access":"private",
+            "presign_expires_seconds":self.presign_expires,
         }

backend/s3/serializers.py

@@ -2,7 +2,7 @@
 
 
 def serialize_upload(result: dict) -> dict:
-    """upload result → API dict (permanent url, never a presign)."""
+    """upload result → API dict. ``url`` is the stable private object address for DB."""
     return {
         "bucket": result.get("bucket"),
         "key": result.get("key"),
@@ -13,6 +13,16 @@ def serialize_upload(result: dict) -> dict:
         "source": result.get("source"),
         "record_id": result.get("record_id"),
         "owner_id": result.get("owner_id"),
+        "access": "private",
+    }
+
+
+def serialize_open(result: dict) -> dict:
+    """Short-lived presigned GET for opening a private CV in the browser."""
+    return {
+        "key": result.get("key"),
+        "url": result.get("url"),
+        "expires_in": result.get("expires_in"),
     }
 
 
@@ -29,5 +39,7 @@ def serialize_health(result: dict) -> dict:
         "status": result.get("status") or "ok",
         "bucket": result.get("bucket"),
         "region": result.get("region"),
-        "public_base_url": result.get("public_base_url"),
+        "object_base_url": result.get("object_base_url") or result.get("public_base_url"),
+        "access": result.get("access") or "private",
+        "presign_expires_seconds": result.get("presign_expires_seconds"),
     }

backend/s3/views.py

@@ -1,13 +1,16 @@
-"""S3 storage service — upload / delete / health over the plugins S3 class."""
+"""S3 storage service — private objects; auth download / short-lived open URLs."""
+
+from pathlib import Path
 
 from fastapi import HTTPException,UploadFile
+from fastapi.responses import Response
 
 from s3.plugins import S3,S3ServiceError,assert_pdf
-from s3.serializers import serialize_delete,serialize_health,serialize_upload
+from s3.serializers import serialize_delete,serialize_health,serialize_open,serialize_upload
 
 
 class S3Storage:
-    """No DB session — pure object storage against the configured bucket."""
+    """No DB session — pure object storage against the configured private bucket."""
 
     def __init__(self):
         self.s3=S3()
@@ -70,10 +73,38 @@ class S3Storage:
             self._map(e)
 
     async def object_url(self,key):
+        """Stable DB identity address (private — not for anonymous open)."""
         if not key or not str(key).strip():
             raise HTTPException(status_code=422,detail="key is required")
         try:
-            url=self.s3.permanent_object_url(str(key).strip())
+            raw=str(key).strip()
-            return {"key":str(key).strip(),"url":url}
+            object_key=self.s3.key_from_url(raw)
+            url=self.s3.object_url(object_key)
+            return {"key":object_key,"url":url,"access":"private"}
+        except S3ServiceError as e:
+            self._map(e)
+
+    async def open_url(self,key,expires_in=None):
+        """Short-lived presigned GET — use this when a recruiter needs to open the CV."""
+        if not key or not str(key).strip():
+            raise HTTPException(status_code=422,detail="key is required")
+        try:
+            return serialize_open(self.s3.presigned_get_url(str(key).strip(),expires_in=expires_in))
+        except S3ServiceError as e:
+            self._map(e)
+
+    async def download_file(self,key):
+        """Authenticated stream of a private PDF (no public bucket needed)."""
+        if not key or not str(key).strip():
+            raise HTTPException(status_code=422,detail="key is required")
+        try:
+            object_key=self.s3.key_from_url(str(key).strip())
+            body=self.s3.download_bytes(object_key)
+            name=Path(object_key).name or "resume.pdf"
+            return Response(
+                content=body,
+                media_type="application/pdf",
+                headers={"Content-Disposition":f'inline; filename="{name}"'},
+            )
         except S3ServiceError as e:
             self._map(e)

backend/search/serializers.py

@@ -8,12 +8,13 @@ def serialize_search_job(row) -> dict:
     }
 
 
-def serialize_search_candidate(user_id, name, email, inbox_id=None) -> dict:
+def serialize_search_candidate(user_id, name, email, inbox_id=None, file_path=None) -> dict:
     return {
         "id": str(user_id) if user_id else None,
         "name": name,
         "email": email,
         "inbox_id": inbox_id,
+        "file_path": file_path or None,
     }
 
 

backend/search/views.py

@@ -1,8 +1,7 @@
-from sqlalchemy import or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import selectinload
 
 from inbox.models import Inbox
+from job.candidate.models import Manual_UPLOAD_CANDIDATE
 from job.job_post.models import JobPosts
 from role.models import EnumRoles, Roles
 from search.serializers import (
@@ -19,86 +18,52 @@ MANAGERS_CAP = 3
 
 
 class Search:
-    def __init__(self, session: AsyncSession):
+    def __init__(self,session:AsyncSession):
-        self.session = session
+        self.session=session
 
-    async def fetch(self, q, limit, current_user):
+    async def fetch(self,q,limit,current_user):
-        query = (q or "").strip()
+        query=(q or "").strip()
-        granted = current_user.get("permissions") or []
+        granted=current_user.get("permissions") or []
-        jobs = []
+        jobs=[]
-        candidates = []
+        candidates=[]
-        managers = []
+        managers=[]
         if query:
-            if has_permission(granted, PermissionTag.JOBS_VIEW):
+            if has_permission(granted,PermissionTag.JOBS_VIEW):
-                jobs = await self._jobs(query, min(limit, JOBS_CAP))
+                jobs=await self._jobs(query,min(limit,JOBS_CAP))
-            if has_permission(granted, PermissionTag.CANDIDATES_VIEW):
+            if has_permission(granted,PermissionTag.CANDIDATES_VIEW):
-                candidates = await self._candidates(query, min(limit, CANDIDATES_CAP))
+                candidates=await self._candidates(query,min(limit,CANDIDATES_CAP))
-            managers = await self._managers(query, min(limit, MANAGERS_CAP))
+            managers=await self._managers(query,min(limit,MANAGERS_CAP))
-        data = {"jobs": jobs, "candidates": candidates, "managers": managers}
+        data={"jobs":jobs,"candidates":candidates,"managers":managers}
-        total = len(jobs) + len(candidates) + len(managers)
+        total=len(jobs)+len(candidates)+len(managers)
-        return data, total
+        return data,total
 
-    async def _jobs(self, query, cap):
+    async def _jobs(self,query,cap):
-        like = f"%{query}%"
+        rows,_total=await JobPosts.fetch_job_posts(
-        statement = (
+            self.session,search=query,top=cap,skip=0,active_only=False,include_deleted=False,
-            select(JobPosts)
-            .where(
-                JobPosts.is_deleted == False,  # noqa: E712
-                or_(
-                    JobPosts.title.ilike(like),
-                    JobPosts.location.ilike(like),
-                    JobPosts.department.ilike(like),
-                ),
-            )
-            .order_by(JobPosts.created_at.desc())
-            .limit(cap)
         )
-        result = await self.session.execute(statement)
+        return [serialize_search_job(r) for r in rows]
-        return [serialize_search_job(r) for r in result.scalars().all()]
 
-    async def _candidates(self, query, cap):
+    async def _candidates(self,query,cap):
-        like = f"%{query}%"
+        role=await Roles.get_role_by_name(self.session,EnumRoles.CANDIDATE.value)
-        statement = (
-            select(Users)
-            .join(Roles, Users.role_id == Roles.id)
-            .where(
-                Roles.role_name == EnumRoles.CANDIDATE.value,
-                Users.is_deleted == False,  # noqa: E712
-                or_(Users.name.ilike(like), Users.email.ilike(like)),
-            )
-            .order_by(Users.created_at.desc())
-            .limit(cap)
-        )
-        users = list((await self.session.execute(statement)).scalars().all())
-        inbox_by_user = {}
-        if users:
-            inbox_q = (
-                select(Inbox.user_id, Inbox.id)
-                .where(Inbox.user_id.in_([u.id for u in users]))
-                .order_by(Inbox.created_at.desc())
-            )
-            for user_id, inbox_id in (await self.session.execute(inbox_q)).all():
-                inbox_by_user.setdefault(user_id, inbox_id)
-        return [
-            serialize_search_candidate(u.id, u.name, u.email, inbox_by_user.get(u.id))
-            for u in users
-        ]
-
-    async def _managers(self, query, cap):
-        like = f"%{query}%"
-        role = await Roles.get_role_by_name(self.session, EnumRoles.HIRING_MANAGER.value)
         if role is None:
             return []
-        statement = (
+        users=list(await Users.get_users(self.session,top=cap,search=query,role_id=role.id))
-            select(Users)
+        uids=[u.id for u in users]
-            .options(selectinload(Users.role))
+        inbox_hits=await Inbox.newest_cv_by_user_ids(self.session,uids)
-            .where(
+        manual_paths=await Manual_UPLOAD_CANDIDATE.file_paths_by_user_ids(self.session,uids)
-                Users.role_id == role.id,
+        rows=[]
-                Users.is_deleted == False,  # noqa: E712
+        for u in users:
-                or_(Users.name.ilike(like), Users.email.ilike(like)),
+            key=str(u.id)
-            )
+            hit=inbox_hits.get(key) or {}
-            .order_by(Users.created_at.desc())
+            rows.append(serialize_search_candidate(
-            .limit(cap)
+                u.id,u.name,u.email,hit.get("inbox_id"),
-        )
+                hit.get("file_path") or manual_paths.get(key),
-        result = await self.session.execute(statement)
+            ))
-        return [serialize_search_manager(u) for u in result.scalars().all()]
+        return rows
+
+    async def _managers(self,query,cap):
+        role=await Roles.get_role_by_name(self.session,EnumRoles.HIRING_MANAGER.value)
+        if role is None:
+            return []
+        users=await Users.get_users(self.session,top=cap,search=query,role_id=role.id)
+        return [serialize_search_manager(u) for u in users]

backend/talent/matching.py

@@ -38,7 +38,10 @@ async def _backfill_slugs(session: AsyncSession) -> None:
         .limit(BACKFILL_BATCH)
     )
     for row in (await session.execute(inbox_q)).scalars().all():
-        row.linkedin_slug = primary_slug_from_text(row.resume_text)
+        haystack = row.resume_text or ""
+        if row.message_body:
+            haystack = f"{haystack}\n{row.message_body}"
+        row.linkedin_slug = primary_slug_from_text(haystack)
         session.add(row)
         changed = True
 

backend/tests/test_employment_agent.py

@@ -0,0 +1,68 @@
+"""employment_agent parse_employment_response — linkedin_url is an agent key."""
+
+from __future__ import annotations
+
+from employment_agent.decorators import parse_employment_response
+from employment_agent.prompt import EDUCATION, NO_COMPANY, NO_LINKEDIN
+
+
+def test_parses_linkedin_url_key_separately():
+    company, education, title, url = parse_employment_response(
+        {
+            "current_employment": "Acme",
+            "education": "BS CS",
+            "current_title": "Engineer",
+            "linkedin_url": "https://www.linkedin.com/in/jane-doe",
+        },
+        "Acme BS CS Engineer",
+    )
+    assert company == "Acme"
+    assert education == "BS CS"
+    assert title == "Engineer"
+    assert url == "https://www.linkedin.com/in/jane-doe"
+
+
+def test_sentinel_and_non_linkedin_are_dropped():
+    *_, url = parse_employment_response(
+        {
+            "current_employment": NO_COMPANY,
+            "education": EDUCATION,
+            "current_title": "x",
+            "linkedin_url": NO_LINKEDIN,
+        },
+        "",
+    )
+    assert url is None
+    *_, github = parse_employment_response(
+        {
+            "current_employment": NO_COMPANY,
+            "education": EDUCATION,
+            "current_title": "x",
+            "linkedin_url": "https://github.com/jane",
+        },
+        "",
+    )
+    assert github is None
+
+
+def test_adds_scheme_and_rejects_company_page():
+    *_, url = parse_employment_response(
+        {
+            "current_employment": NO_COMPANY,
+            "education": EDUCATION,
+            "current_title": "x",
+            "linkedin_url": "www.linkedin.com/in/jane-doe",
+        },
+        "",
+    )
+    assert url == "https://www.linkedin.com/in/jane-doe"
+    *_, company = parse_employment_response(
+        {
+            "current_employment": NO_COMPANY,
+            "education": EDUCATION,
+            "current_title": "x",
+            "linkedin_url": "https://www.linkedin.com/company/acme",
+        },
+        "",
+    )
+    assert company is None

backend/tests/test_linkedin_matching.py

@@ -36,6 +36,29 @@ def test_extracts_bare_and_schemed_links():
     assert slugs_from_text(text2) == ["ali-raza-8a1b2c"]
 
 
+def test_wrapped_and_spaced_pdf_urls():
+    # pypdf wraps the path; glyph-padded CVs insert spaces around slashes.
+    assert slugs_from_text("linkedin.com/in/\njane-doe") == ["jane-doe"]
+    assert slugs_from_text("linkedin.com / in / jane-doe") == ["jane-doe"]
+    assert slugs_from_text("https://pk.linkedin.com/in/jane-doe") == ["jane-doe"]
+
+
+def test_html_href_and_mobile_path():
+    html = '<a href="https://www.linkedin.com/in/jane-doe">LinkedIn</a>'
+    assert slugs_from_text(html) == ["jane-doe"]
+    assert slugs_from_text("See linkedin.com/mwlite/in/jane-doe") == ["jane-doe"]
+
+
+def test_profile_url_from_text_prefers_slug_then_short_link():
+    from linkedin_utils import profile_url_from_text
+
+    assert profile_url_from_text("linkedin.com/in/jane-doe") == (
+        "https://www.linkedin.com/in/jane-doe"
+    )
+    assert profile_url_from_text("Contact: lnkd.in/abc12XY") == "https://lnkd.in/abc12XY"
+    assert profile_url_from_text("no profile here") is None
+
+
 def test_percent_encoding_and_trailing_punctuation():
     # PDF extraction often percent-encodes hyphens and glues sentence dots on.
     assert slugs_from_text("see linkedin.com/in/jane%2Ddoe.") == ["jane-doe"]

backend/users/models.py

@@ -57,6 +57,9 @@ class Users(SQLModel, table=True):
     )
 
     password: str
+    # Public profile URL extracted from a CV at ingest. NULL until a CV
+    # mentions LinkedIn; never overwrite a stored value with empty.
+    linkedin_url: str | None = Field(default=None)
     created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
     updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
     is_active: bool = Field(default=False)
@@ -111,6 +114,22 @@ class Users(SQLModel, table=True):
         result = await session.execute(statement)
         return result.scalars().all()
 
+    @classmethod
+    async def names_by_ids(cls, session: AsyncSession, user_ids) -> dict[str, str]:
+        """Resolve {user_id: name} in a single query.
+
+        COLUMN select, not the Users entity: `select(cls)` would pull the five
+        selectin relations (role, job_posts, inbox, feedback, notes) for a
+        two-column lookup.
+        """
+        uids = {u for u in (user_ids or []) if u}
+        if not uids:
+            return {}
+        result = await session.execute(
+            select(cls.id, cls.name).where(cls.id.in_(uids))
+        )
+        return {str(uid): name for uid, name in result.all()}
+
     @classmethod
     async def get_user_by_id(cls, session: AsyncSession, record_id: str):
         uid = cls._as_uuid(record_id)
@@ -138,6 +157,30 @@ class Users(SQLModel, table=True):
         result = await session.execute(statement)
         return result.scalar_one()
 
+    @classmethod
+    async def set_linkedin_url_if_empty(cls, session: AsyncSession, *, user_id=None, email=None, url=None) -> bool:
+        """Write linkedin_url only when the user has none yet. Caller commits."""
+        value = (url or "").strip() or None
+        if not value:
+            return False
+        statement = select(cls)
+        if user_id is not None:
+            uid = cls._as_uuid(user_id)
+            if uid is None:
+                return False
+            statement = statement.where(cls.id == uid)
+        elif email:
+            statement = statement.where(func.lower(cls.email) == str(email).strip().lower())
+        else:
+            return False
+        user = (await session.execute(statement)).scalars().first()
+        if user is None or (user.linkedin_url or "").strip():
+            return False
+        user.linkedin_url = value
+        user.updated_at = _now()
+        session.add(user)
+        return True
+
     @classmethod
     async def insert_user(cls, session: AsyncSession, fields: dict):
         """`fields["password"]` is expected to be hashed already — see users.plugins."""

backend/users/serializers.py

@@ -21,6 +21,7 @@ def serialize_user(
         "role_id": user.role_id,
         "role_name": role_name,
         "role_description": role.description if role is not None else None,
+        "linkedin_url": user.linkedin_url or None,
         "is_active": user.is_active,
         "is_deleted": user.is_deleted,
         "created_at": user.created_at.isoformat() if user.created_at else None,

frontend/src/api/candidates.js

@@ -78,6 +78,7 @@ export function toCandidateView(row) {
     jobId: row.job_id,
     name,
     filename: row.filename,
+    filePath: row.file_path || null,
     source: row.source, // 'upload' | 'inbox'
     currentTitle: row.job_title ?? null,
     currentCompany: row.current_company ?? null,

frontend/src/api/s3.js

@@ -0,0 +1,52 @@
+import { request } from '../lib/apiClient'
+
+/**
+ * Short-lived presigned GET for a private CV. Needs candidates.view,
+ * settings.view, or inbox.view. `key` is the stored file_path (S3 URL or object key).
+ *
+ * Returns `{ data: { key, url, expires_in } }` — open `data.url` in a new tab.
+ */
+export function openUrl(key, { expiresIn } = {}) {
+  return request('/s3/open', {
+    params: { key, expires_in: expiresIn },
+  })
+}
+
+/** First comma-separated stored path — inbox_messages.file_path can list several. */
+export function firstKey(filePath) {
+  return (filePath || '').split(',')[0].trim() || null
+}
+
+/** S3 object address (virtual-hosted URL) or record-scoped key Email|Manual|Form/... */
+export function isS3Ref(value) {
+  const raw = firstKey(value)
+  if (!raw) return false
+  if (/^https?:\/\//i.test(raw)) {
+    return /\.s3[.-]/i.test(raw) || /\/\/s3[.-]/i.test(raw)
+  }
+  return /^(Email|Manual|Form)\//i.test(raw)
+}
+
+/**
+ * Fresh presign on every click. The signed URL opens in a new tab so the
+ * browser's built-in PDF viewer renders it. Non-S3 http (Drive / Sheet links)
+ * open as-is. Pass `tab` from a synchronous window.open to beat the pop-up blocker.
+ */
+export async function openPdf(filePath, { tab } = {}) {
+  const key = firstKey(filePath)
+  if (!key) throw new Error('No resume file on this application')
+  let url
+  if (isS3Ref(key) || !/^https?:\/\//i.test(key)) {
+    const res = await openUrl(key)
+    url = res?.data?.url
+    if (!url) throw new Error('Could not open resume')
+  } else {
+    url = key
+  }
+  if (tab && !tab.closed) tab.location.replace(url)
+  else {
+    const opened = window.open(url, '_blank', 'noopener,noreferrer')
+    if (!opened) throw new Error('Pop-up blocked — allow pop-ups to view the PDF')
+  }
+  return url
+}

frontend/src/screens/CandidateProfile.jsx

@@ -300,6 +300,17 @@ export default function CandidateProfile({
             {(live?.source || c.source) && <Badge className="b-gray">{live?.source || c.source}</Badge>}
             {expChip && <span className="badge b-plain b-indigo badge-plain">{expChip}</span>}
           </div>
+          {live?.linkedin_url && (
+            <a
+              className="btn btn-secondary btn-sm"
+              href={live.linkedin_url}
+              target="_blank"
+              rel="noopener noreferrer"
+              style={{ marginTop: 10 }}
+            >
+              <Icon name="linkedin" /> LinkedIn
+            </a>
+          )}
         </div>
         {/* No score anywhere -> the whole block goes, rather than a ring drawn
             around a blank. Seed-backed callers still pass a number and are

frontend/src/screens/Inbox.jsx

@@ -20,6 +20,7 @@ import { qk } from '../lib/queryKeys'
 import { friendlyAuthError } from '../lib/errors'
 import * as inboxApi from '../api/inbox'
 import * as sheetApi from '../api/sheet'
+import * as s3Api from '../api/s3'
 import {
   atsRecommendationClass, avatarColor, fmtDate, initials as initialsOf,
   inboxSources, sourceMeta,
@@ -286,6 +287,9 @@ async function fetchMessageDetail(recordId) {
     bcc: row.message_bcc || '',
     sentAt: parseDate(row.message_sent_time),
     files: Array.isArray(row.files) ? row.files : [],
+    filePath: row.file_path || '',
+    linkedinSlug: row.linkedin_slug || '',
+    linkedinUrl: row.linkedin_url || '',
     matchStatus: row.match_status || null,
     matchSummary: row.match_summary || '',
     matchReasoning: row.match_reasoning || '',
@@ -329,6 +333,9 @@ async function fetchApplications(params) {
         resumeStatus: row.resume_status || 'Pending',
         attachment: row.attachment,
         hasAttachment: Boolean(row.has_attachment),
+        filePath: row.file_path || '',
+        linkedinSlug: row.linkedin_slug || '',
+        linkedinUrl: row.linkedin_url || '',
         resumeText: row.resume_text || '',
         atsScore: row.ats_score,
         phone: row.phone,
@@ -671,7 +678,6 @@ export default function Inbox() {
   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)
 
@@ -1130,7 +1136,6 @@ export default function Inbox() {
                   busy={setState.isPending || markDuplicate.isPending}
                   canEdit={canEdit}
                   toast={toast}
-                  onPreview={() => setPreviewing(selected)}
                   onImport={() => importItem(selected)}
                   onMove={() => moveToPipeline(selected)}
                   onNote={() => setNoting(selected)}
@@ -1142,30 +1147,6 @@ export default function Inbox() {
           </div>
       </div>
 
-      {previewing && (
-        <Modal
-          title={previewing.attachment}
-          subtitle={`Resume preview · ${previewing.name}`}
-          size="modal-lg"
-          onClose={() => setPreviewing(null)}
-          footer={
-            <>
-              <button className="btn btn-secondary" onClick={() => setPreviewing(null)}>Close</button>
-              <button
-                className="btn btn-primary"
-                onClick={() => { const it = previewing; setPreviewing(null); importItem(it) }}
-              >
-                <Icon name="user-plus" /> Import Candidate
-              </button>
-            </>
-          }
-        >
-          <pre className="resume-thumb" style={{ maxHeight: 'none', fontSize: 12 }}>
-            {previewing.resumeText || 'Resume text not extracted yet — the matching task has not run for this application.'}
-          </pre>
-        </Modal>
-      )}
-
       {assigning && (
         <AssignRecruiter
           item={assigning}
@@ -1215,6 +1196,19 @@ function externalHref(url) {
   return `https://${raw}`
 }
 
+/** Persistable /in/<slug> → a public profile URL. Empty string means scanned, none found. */
+function linkedinHrefFromSlug(slug) {
+  const cleaned = (slug || '').trim()
+  if (!cleaned) return null
+  return `https://www.linkedin.com/in/${cleaned}`
+}
+
+function firstResumeKey(item) {
+  const fromFiles = (item?.files || []).map((f) => f.url).find(Boolean)
+  if (fromFiles) return fromFiles
+  return s3Api.firstKey(item?.filePath)
+}
+
 /**
  * Sheet form applicant detail — profile grids + resume/LinkedIn links +
  * title-matched job selection (position_applied_for ↔ job_posts.title) +
@@ -1454,7 +1448,7 @@ function FormApplicantDetail({
           </div>
           {matchCards.length === 0 && !manualPost ? (
             <EmptyState icon="alert" title="No matching roles">
-              No job post title matches this position. Choose a role manually.
+              <p>No job post title matches this position. Choose a role manually.</p>
               <div style={{ display: 'flex', gap: 8, justifyContent: 'center', flexWrap: 'wrap', marginTop: 10 }}>
                 <button
                   className="btn btn-primary btn-sm"
@@ -1555,7 +1549,7 @@ function FormApplicantDetail({
 }
 
 function ApplicationDetail({
-  item: i, loading, busy, canEdit, toast, onPreview, onImport, onMove, onNote, onReject, onToggleDuplicate,
+  item: i, loading, busy, canEdit, toast, onImport, onMove, onNote, onReject, onToggleDuplicate,
 }) {
   const qc = useQueryClient()
   const recLabel = i.atsScore >= 82 ? 'Strong Match' : i.atsScore >= 65 ? 'Potential Match' : 'Weak Match'
@@ -1622,6 +1616,21 @@ function ApplicationDetail({
   const assigned = i.assignedPost
   const panelBusy = busy || assignMutation.isPending || rematchMutation.isPending
   const canAssign = canEdit && selection && selection !== i.assignedId && !assignMutation.isPending
+  const resumeKey = firstResumeKey(i)
+  const canOpenResume = Boolean(resumeKey)
+  const profileHref = i.linkedinUrl || linkedinHrefFromSlug(i.linkedinSlug)
+
+  const openResume = useMutation({
+    mutationFn: async (tab) => {
+      try {
+        return await s3Api.openPdf(firstResumeKey(i), { tab })
+      } catch (err) {
+        if (tab && !tab.closed) tab.close()
+        throw err
+      }
+    },
+    onError: (err) => toast(friendlyAuthError(err, 'Could not open resume'), 'error'),
+  })
 
   async function handleMove() {
     if (busy || alreadyProcessed) return
@@ -1669,6 +1678,25 @@ function ApplicationDetail({
         )}
       </div>
 
+      {(resumeKey || i.hasAttachment || profileHref) && (
+        <div className="flex gap-8" style={{ flexWrap: 'wrap', marginBottom: 18 }}>
+          {(resumeKey || i.hasAttachment) && (
+            <button
+              className="btn btn-primary btn-sm"
+              disabled={openResume.isPending || !canOpenResume}
+              onClick={() => openResume.mutate(window.open('about:blank', '_blank'))}
+            >
+              <Icon name="paperclip" /> Open resume
+            </button>
+          )}
+          {profileHref && (
+            <a className="btn btn-secondary btn-sm" href={profileHref} target="_blank" rel="noopener noreferrer">
+              <Icon name="linkedin" /> LinkedIn
+            </a>
+          )}
+        </div>
+      )}
+
       <div className="info-grid" style={{ marginBottom: 20 }}>
         <div className="info-item"><div className="il">Email</div><div className="iv">{orDash(i.email)}</div></div>
         <div className="info-item"><div className="il">Phone</div><div className="iv">{orDash(i.phone)}</div></div>
@@ -1755,25 +1783,6 @@ function ApplicationDetail({
               )}
             </div>
           )}
-
-          {i.hasAttachment && (
-            <div className="card" style={{ boxShadow: 'none', background: 'var(--bg-sunken)' }}>
-              <div className="card-body">
-                <div className="flex items-center gap-12" style={{ justifyContent: 'space-between', marginBottom: 12 }}>
-                  <div className="fw-600">
-                    <Icon name="paperclip" /> {orDash(i.attachment)}
-                    {i.files?.[0]?.size != null && (
-                      <span className="cell-sub"> · {Math.round(i.files[0].size / 1024)} KB</span>
-                    )}
-                  </div>
-                  <button className="btn btn-secondary btn-sm" onClick={onPreview}><Icon name="eye" /> Preview</button>
-                </div>
-                <pre className="resume-thumb">
-                  {resumeText || 'Resume text not extracted yet — the matching task has not run for this application.'}
-                </pre>
-              </div>
-            </div>
-          )}
         </div>
 
         <div role="radiogroup" aria-label="Suggested roles" style={{ flex: '0 1 360px', minWidth: 260 }}>
@@ -1814,6 +1823,7 @@ function ApplicationDetail({
           <div className="fw-600" style={{ marginBottom: 8 }}>Suggested roles</div>
           {suggestionCards.length === 0 && !manualPost ? (
             <EmptyState icon="alert" title="No suggested roles">
+              <p>No job post was suggested. Choose a role manually.</p>
               <div style={{ display: 'flex', gap: 8, justifyContent: 'center', flexWrap: 'wrap', marginTop: 10 }}>
                 <button
                   className="btn btn-secondary btn-sm"

frontend/src/screens/Matching.jsx

@@ -672,6 +672,7 @@ function MatchingWorkspace({
           <div className="fw-600" style={{ marginBottom: 8 }}>Suggested roles</div>
           {suggestionCards.length === 0 && !manualPost ? (
             <EmptyState icon="alert" title="No suggested roles">
+              <p>No job post was suggested. Choose a role manually.</p>
               <div style={{ display: 'flex', gap: 8, justifyContent: 'center', flexWrap: 'wrap', marginTop: 10 }}>
                 <button
                   className="btn btn-secondary btn-sm"

frontend/src/styles/styles.css

@@ -771,6 +771,9 @@ canvas { width: 100%; max-width: 100%; display: block; }
 .empty-state { text-align: center; padding: 60px 20px; color: var(--text-3); }
 .empty-state svg { width: 48px; height: 48px; margin-bottom: 14px; opacity: .5; }
 .empty-state h3 { font-size: 18px; color: var(--text-2); margin-bottom: 6px; }
+.empty-state p { margin: 0; }
+.empty-state-body { margin-top: 4px; }
+.empty-state-body p { margin: 0 0 10px; }
 
 .avatar-stack { display: flex; }
 .avatar-stack .avatar { width: 30px; height: 30px; font-size: 11px; border: 2px solid var(--bg-elev); margin-left: -8px; }

frontend/src/ui/primitives.jsx

@@ -73,11 +73,13 @@ export function ProgressBar({ pct, className }) {
 }
 
 export function EmptyState({ icon = 'search', title = 'No results found', children }) {
+  const body = children ?? 'Try adjusting your filters or search.'
+  const isSimple = body == null || typeof body === 'string' || typeof body === 'number'
   return (
     <div className="empty-state">
       <Icon name={icon} />
       <h3>{title}</h3>
-      <p>{children || 'Try adjusting your filters or search.'}</p>
+      {isSimple ? <p>{body}</p> : <div className="empty-state-body">{body}</div>}
     </div>
   )
 }