Manual-apply backend: per-opportunity state, unique-decks engine, offsite-first + logos

- OpportunityState model + repo: persist applied / dismissed / restored status and the
  generated resume-builder doc ids (never re-pay to re-tailor). annotate_and_filter drops
  dismissed + annotates applied/docs on read; handlers for dismiss/restore/mark_applied/
  save_apply_docs.
- Unique decks per run (spend only for NEW jobs, no cache, no re-fetch+dedup band-aid):
  per-(user,query) search cursor → LinkedIn page++ and Naukri incremental+stateKey (async
  run path in apify_client, since the actor's crawl exceeds the run-sync window); seen-net
  excludes already-shown ids (covers Foundit, which can't paginate).
- normalize: every board offsite-first (apply_url prefers the employer/ATS redirect over the
  board listing) + offsite_apply flag; real company logos (_logo_url across Naukri logoPath /
  LinkedIn company.logo / etc.).

app/agent/session.py

@@ -46,6 +46,11 @@ class MatchmakingAgentSession:
             "tailor_resume": self.handle_tailor_resume,
             "submit_application": self.handle_submit_application,
             "get_apply_proof": self.handle_get_apply_proof,
+            # per-opportunity state — persists across deck refreshes (applied / dismissed / saved docs)
+            "dismiss_opportunity": self.handle_dismiss_opportunity,
+            "restore_opportunity": self.handle_restore_opportunity,
+            "mark_applied": self.handle_mark_applied,
+            "save_apply_docs": self.handle_save_apply_docs,
         }
         handler = handlers.get(action)
         if not handler:
@@ -86,7 +91,19 @@ class MatchmakingAgentSession:
         await self.push("agent_thinking",
                         message="Fetching fresh roles across job boards…" if fresh
                         else "Scanning live roles across job boards…")
+        # Search cursor: re-running the SAME query advances the page (LinkedIn) + reuses the Naukri
+        # incremental stateKey, so each run fetches GENUINELY NEW jobs — not the same set re-charged.
+        from app.db import repo as _repo
+        sig = _repo.search_signature(prefs)
+        page = await _repo.get_search_cursor(self.user_id, sig)
+        prefs["_page"] = page
+        # colon-free key (state-store record id charset is alphanumeric/_-.)
+        prefs["_naukri_state"] = f"{self.user_id}_{sig}".replace(":", "_")
         sweep = await search.run_sweep(prefs, fresh=fresh)
+        # Safety net: drop anything already shown / applied / dismissed (covers boards that can't
+        # paginate, so the same posting never reappears even if a board re-returns it).
+        seen = await _repo.get_seen_ids(self.user_id)
+        candidates = [o for o in sweep["opportunities"] if o.get("id") not in seen]
         # Engine: assemble FULL context (profile + prefs). Cheap rankers SIFT 90 → top ~18 (white-box ‖
         # embedding vibe); then Opus reads only those and curates the honest shortlist + report cards.
         from app.engine import curate as _curate
@@ -96,7 +113,7 @@ class MatchmakingAgentSession:
         await self.push("agent_thinking", message="Scoring roles against your profile…")
         # Run the blocking LLM work (embeddings + Opus) OFF the event loop, or the long sync OpenAI
         # call freezes the loop and the Redis response can't publish (→ the loader hangs forever).
-        top, dbg = await asyncio.to_thread(_sift.sift, prefs, user_context, sweep["opportunities"])
+        top, dbg = await asyncio.to_thread(_sift.sift, prefs, user_context, candidates)
         await self.push("agent_thinking", message="Scout is reading your top roles…")
         curated = await asyncio.to_thread(_curate.curate, prefs, user_context, top)   # Opus's call is final
         if curated is None:                                       # safety net: sift + templated cards
@@ -107,11 +124,18 @@ class MatchmakingAgentSession:
         result = {"opportunities": curated, "sources": sweep["sources"],
                   "shortlisted": len(curated), "scanned": dbg["scored"], "engine": engine,
                   "has_profile": bool(user_context and user_context.get("skills"))}
+        # Persist the RAW deck + ADVANCE the cursor (page+1) so the next run of this query fetches new
+        # jobs. Per-opportunity state (applied/dismissed/saved docs) lives separately, applied on read.
+        clean_prefs = {k: v for k, v in prefs.items()
+                       if k not in ("user_context", "_fresh", "_page", "_naukri_state")}
+        await _repo.save_feed(self.user_id, clean_prefs, result, query_sig=sig, cursor_page=page + 1)
+        # Mark the shown deck as SEEN so it never reappears in a future run.
+        await _repo.mark_seen(self.user_id, [o.get("id") for o in (result["opportunities"] or []) if o.get("id")])
+        # Hand the UI a state-aware view: dismissed dropped, applied + saved-doc ids annotated.
+        states = await _repo.get_opportunity_states(self.user_id)
+        result["opportunities"] = _repo.annotate_and_filter(result["opportunities"], states)
+        result["shortlisted"] = len(result["opportunities"])
         await self.push("agent_data", action="search_complete", data=result)
-        # Persist the feed so it survives navigation/refresh (replayed by handle_get_feed).
-        from app.db.repo import save_feed
-        clean_prefs = {k: v for k, v in prefs.items() if k not in ("user_context", "_fresh")}
-        await save_feed(self.user_id, clean_prefs, result)
 
     async def handle_suggest_bubbles(self, params: dict):
         """Fine-tune bubbling — stage-aware (broad→narrow→role) bubbles from the picks so far."""
@@ -131,3 +155,35 @@ class MatchmakingAgentSession:
 
     async def handle_get_apply_proof(self, params: dict):
         await self._stub("apply_proof", "return captured confirmation screenshots")
+
+    # ── per-opportunity state (persists across deck refreshes) ──
+    async def handle_dismiss_opportunity(self, params: dict):
+        """Mark a posting stale/dismissed → it never resurfaces in the deck."""
+        from app.db.repo import set_opportunity_state
+        await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), status="dismissed")
+        await self.push("agent_data", action="opportunity_state_saved",
+                        data={"opportunity_id": params.get("opportunity_id"), "status": "dismissed"})
+
+    async def handle_restore_opportunity(self, params: dict):
+        """Undo a dismiss/pass → clear the status so the role returns to the deck."""
+        from app.db.repo import set_opportunity_state
+        await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), status=None)
+        await self.push("agent_data", action="opportunity_state_saved",
+                        data={"opportunity_id": params.get("opportunity_id"), "status": None})
+
+    async def handle_mark_applied(self, params: dict):
+        """Record that the user applied to a role (sticks across reloads)."""
+        from app.db.repo import set_opportunity_state
+        await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), status="applied")
+        await self.push("agent_data", action="opportunity_state_saved",
+                        data={"opportunity_id": params.get("opportunity_id"), "status": "applied"})
+
+    async def handle_save_apply_docs(self, params: dict):
+        """Remember the resume-builder document ids generated for a role so we never re-pay to
+        re-tailor / re-generate them. Only the provided ids are written."""
+        from app.db.repo import set_opportunity_state
+        fields = {k: params[k] for k in ("tailored_resume_id", "tailored_version_id", "cover_letter_id")
+                  if params.get(k)}
+        await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), **fields)
+        await self.push("agent_data", action="opportunity_state_saved",
+                        data={"opportunity_id": params.get("opportunity_id"), **fields})

app/db/models.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import JSON, DateTime, Integer, String, func
+from sqlalchemy import JSON, Boolean, DateTime, Integer, String, func
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
 
@@ -22,6 +22,28 @@ class UserFeed(Base):
     sources: Mapped[dict] = mapped_column(JSON, default=dict)          # per-board counts
     engine: Mapped[str] = mapped_column(String, default="")           # "opus" | "fallback:…"
     scanned: Mapped[int] = mapped_column(Integer, default=0)
+    # Search cursor — advances per re-run of the SAME query so each run fetches NEW jobs (LinkedIn
+    # page++, Naukri incremental stateKey). Resets when the query signature changes.
+    query_sig: Mapped[str] = mapped_column(String, default="")
+    cursor_page: Mapped[int] = mapped_column(Integer, default=1)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
+    )
+
+
+class OpportunityState(Base):
+    """Per-(user, opportunity) state that SURVIVES deck refreshes. Persists three things:
+    `applied`/`dismissed` status, and the resume-builder document ids already generated for the
+    role (so we never re-tailor / re-generate a paid document twice). Keyed by (user, opportunity)."""
+    __tablename__ = "opportunity_state"
+
+    user_id: Mapped[str] = mapped_column(String, primary_key=True)
+    opportunity_id: Mapped[str] = mapped_column(String, primary_key=True)
+    status: Mapped[str | None] = mapped_column(String, nullable=True)              # 'applied' | 'dismissed' | None
+    tailored_resume_id: Mapped[str | None] = mapped_column(String, nullable=True)  # resume-builder resume id
+    tailored_version_id: Mapped[str | None] = mapped_column(String, nullable=True) # the tailored version
+    cover_letter_id: Mapped[str | None] = mapped_column(String, nullable=True)     # the generated cover letter
+    seen: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")  # shown in a deck → exclude from new runs
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
     )

app/db/repo.py

@@ -2,16 +2,32 @@
 when the DB is off or unreachable (persistence is optional; the engine must never crash on it)."""
 from __future__ import annotations
 
+import hashlib
+import json
 import logging
 
 from sqlalchemy import select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 
-from app.db.models import Base, UserFeed
+from app.db.models import Base, OpportunityState, UserFeed
 from app.db.session import engine, session_factory
 
 logger = logging.getLogger(__name__)
 
+# fields a caller may set on an opportunity's state (status + the generated document ids)
+_STATE_FIELDS = ("status", "tailored_resume_id", "tailored_version_id", "cover_letter_id")
+
+# the search-defining pref fields — same values → same signature → cursor advances (new jobs);
+# change any → new signature → cursor resets to page 1.
+_SIG_FIELDS = ("title", "role", "location", "industry", "experience", "years", "work_mode", "remote")
+
+
+def search_signature(prefs: dict) -> str:
+    """Stable short hash of the search-defining prefs (ignores transient flags / user_context)."""
+    payload = {k: prefs.get(k) for k in _SIG_FIELDS}
+    blob = json.dumps(payload, sort_keys=True, default=str)
+    return hashlib.sha1(blob.encode()).hexdigest()[:16]
+
 
 async def init_db() -> None:
     """Create tables if the DB is configured (best-effort; logged, never fatal)."""
@@ -27,8 +43,8 @@ async def init_db() -> None:
         logger.warning("feed store init failed (persistence off): %s", e)
 
 
-async def save_feed(user_id: str, prefs: dict, result: dict) -> None:
-    """Upsert the user's last completed search. No-op if the DB is off."""
+async def save_feed(user_id: str, prefs: dict, result: dict, *, query_sig: str = "", cursor_page: int = 1) -> None:
+    """Upsert the user's last completed search + the advanced search cursor. No-op if the DB is off."""
     factory = session_factory()
     if factory is None or not user_id:
         return
@@ -39,13 +55,15 @@ async def save_feed(user_id: str, prefs: dict, result: dict) -> None:
         "sources": result.get("sources") or {},
         "engine": result.get("engine") or "",
         "scanned": int(result.get("scanned") or 0),
+        "query_sig": query_sig,
+        "cursor_page": cursor_page,
     }
     try:
         async with factory() as s:
             stmt = pg_insert(UserFeed).values(**row)
             stmt = stmt.on_conflict_do_update(
                 index_elements=[UserFeed.user_id],
-                set_={k: row[k] for k in ("prefs", "opportunities", "sources", "engine", "scanned")},
+                set_={k: row[k] for k in ("prefs", "opportunities", "sources", "engine", "scanned", "query_sig", "cursor_page")},
             )
             await s.execute(stmt)
             await s.commit()
@@ -66,13 +84,135 @@ async def get_feed(user_id: str) -> dict | None:
         return None
     if not row:
         return None
+    states = await get_opportunity_states(user_id)
+    opps = annotate_and_filter(row.opportunities or [], states)
     return {
-        "opportunities": row.opportunities or [],
+        "opportunities": opps,
         "sources": row.sources or {},
         "prefs": row.prefs or {},
         "engine": row.engine,
         "scanned": row.scanned,
-        "shortlisted": len(row.opportunities or []),
+        "shortlisted": len(opps),
         "cached": True,
         "updated_at": row.updated_at.isoformat() if row.updated_at else None,
     }
+
+
+async def set_opportunity_state(user_id: str, opportunity_id: str, **fields) -> None:
+    """Upsert per-opportunity state (status and/or document ids). No-op if the DB is off.
+    Only the provided fields are written, so partial updates don't clobber existing values."""
+    factory = session_factory()
+    if factory is None or not user_id or not opportunity_id:
+        return
+    vals = {k: fields[k] for k in _STATE_FIELDS if k in fields}
+    if not vals:
+        return
+    row = {"user_id": user_id, "opportunity_id": opportunity_id, **vals}
+    try:
+        async with factory() as s:
+            stmt = pg_insert(OpportunityState).values(**row)
+            stmt = stmt.on_conflict_do_update(
+                index_elements=[OpportunityState.user_id, OpportunityState.opportunity_id],
+                set_=vals,
+            )
+            await s.execute(stmt)
+            await s.commit()
+    except Exception as e:  # noqa: BLE001
+        logger.warning("set_opportunity_state failed (non-fatal): %s", e)
+
+
+async def get_opportunity_states(user_id: str) -> dict[str, dict]:
+    """{opportunity_id: {status, tailored_resume_id, tailored_version_id, cover_letter_id}}."""
+    factory = session_factory()
+    if factory is None or not user_id:
+        return {}
+    try:
+        async with factory() as s:
+            rows = (await s.execute(
+                select(OpportunityState).where(OpportunityState.user_id == user_id)
+            )).scalars().all()
+    except Exception as e:  # noqa: BLE001
+        logger.warning("get_opportunity_states failed (non-fatal): %s", e)
+        return {}
+    return {
+        r.opportunity_id: {
+            "status": r.status,
+            "tailored_resume_id": r.tailored_resume_id,
+            "tailored_version_id": r.tailored_version_id,
+            "cover_letter_id": r.cover_letter_id,
+        }
+        for r in rows
+    }
+
+
+def annotate_and_filter(opportunities: list[dict], states: dict[str, dict]) -> list[dict]:
+    """Drop dismissed opportunities and annotate the rest with their persisted state (applied flag
+    + already-generated document ids) so the UI can restore without re-tailoring."""
+    out: list[dict] = []
+    for o in opportunities:
+        st = states.get(o.get("id"))
+        if st and st.get("status") == "dismissed":
+            continue                                   # never resurface
+        if st:
+            o = {**o,
+                 "applied": st.get("status") == "applied",
+                 "tailored_resume_id": st.get("tailored_resume_id"),
+                 "tailored_version_id": st.get("tailored_version_id"),
+                 "cover_letter_id": st.get("cover_letter_id")}
+        out.append(o)
+    return out
+
+
+async def get_search_cursor(user_id: str, query_sig: str) -> int:
+    """The next LinkedIn start-page for this query. Same signature as last run → advance; a new
+    query (different signature) → start at page 1. Returns 1 when the DB is off or query changed."""
+    factory = session_factory()
+    if factory is None or not user_id:
+        return 1
+    try:
+        async with factory() as s:
+            row = (await s.execute(select(UserFeed).where(UserFeed.user_id == user_id))).scalar_one_or_none()
+    except Exception as e:  # noqa: BLE001
+        logger.warning("get_search_cursor failed (non-fatal): %s", e)
+        return 1
+    if not row or row.query_sig != query_sig:
+        return 1                                  # new/changed query → fresh cursor
+    return int(row.cursor_page or 1)
+
+
+async def get_seen_ids(user_id: str) -> set[str]:
+    """Every opportunity the user has already been shown / acted on — excluded from new runs."""
+    factory = session_factory()
+    if factory is None or not user_id:
+        return set()
+    try:
+        async with factory() as s:
+            rows = (await s.execute(
+                select(OpportunityState.opportunity_id).where(OpportunityState.user_id == user_id)
+            )).scalars().all()
+        return set(rows)
+    except Exception as e:  # noqa: BLE001
+        logger.warning("get_seen_ids failed (non-fatal): %s", e)
+        return set()
+
+
+async def mark_seen(user_id: str, opportunity_ids: list[str]) -> None:
+    """Record that these opportunities were shown in a deck (so future runs skip them). No-op on DB off.
+    Sets seen=true without touching any existing status/document ids."""
+    factory = session_factory()
+    if factory is None or not user_id or not opportunity_ids:
+        return
+    rows = [{"user_id": user_id, "opportunity_id": oid, "seen": True} for oid in opportunity_ids if oid]
+    if not rows:
+        return
+    try:
+        async with factory() as s:
+            stmt = pg_insert(OpportunityState).values(rows)
+            stmt = stmt.on_conflict_do_update(
+                index_elements=[OpportunityState.user_id, OpportunityState.opportunity_id],
+                set_={"seen": True},
+            )
+            await s.execute(stmt)
+            await s.commit()
+    except Exception as e:  # noqa: BLE001
+        logger.warning("mark_seen failed (non-fatal): %s", e)

app/engine/apify_client.py

@@ -47,6 +47,44 @@ def _cache_write(path: Path, items: list[dict]) -> None:
 # would double the cost for no gain (an empty result is usually a real "0 matches").
 _MAX_ATTEMPTS = 2
 _BACKOFF = 2.0
+_RUNS_BASE = "https://api.apify.com/v2"
+_POLL = 3.0  # async-run poll interval (s)
+
+
+async def _run_actor_async(actor_id: str, run_input: dict, *, timeout: float, token: str) -> list[dict]:
+    """Start an ASYNC run, poll to completion, return its dataset items. Needed for incremental /
+    stateful actors (state persisted across runs via `stateKey`) whose crawl exceeds the run-sync
+    window — that window aborts the run mid-crawl. No cache: the actor itself is the dedup."""
+    async with httpx.AsyncClient(timeout=60.0) as client:
+        r = await client.post(f"{_BASE}/{actor_id}/runs", params={"token": token}, json=run_input)
+        r.raise_for_status()
+        run = r.json().get("data", {})
+        run_id = run.get("id")
+        ds_id = run.get("defaultDatasetId")
+        if not run_id:
+            return []
+        elapsed = 0.0
+        while elapsed < timeout:
+            await asyncio.sleep(_POLL)
+            elapsed += _POLL
+            s = await client.get(f"{_RUNS_BASE}/actor-runs/{run_id}", params={"token": token})
+            data = s.json().get("data", {})
+            status = data.get("status")
+            ds_id = data.get("defaultDatasetId") or ds_id
+            if status == "SUCCEEDED":
+                break
+            if status in ("FAILED", "ABORTED", "TIMED-OUT", "TIMING-OUT"):
+                logger.warning("apify async run %s ended %s", actor_id, status)
+                return []
+        else:
+            logger.warning("apify async run %s did not finish within %ss", actor_id, timeout)
+            return []
+        d = await client.get(f"{_RUNS_BASE}/datasets/{ds_id}/items", params={"token": token, "format": "json"})
+        d.raise_for_status()
+        items = d.json()
+        n = len(items) if isinstance(items, list) else 0
+        logger.info("apify async run %s SUCCEEDED — %d items (incremental)", actor_id, n)
+        return items if isinstance(items, list) else []
 
 
 async def run_actor(actor_id: str, run_input: dict, *, timeout: float = 240.0,
@@ -70,6 +108,11 @@ async def run_actor(actor_id: str, run_input: dict, *, timeout: float = 240.0,
     token = get_settings().APIFY_TOKEN
     if not token:
         raise RuntimeError("APIFY_TOKEN not configured")
+
+    # Incremental/stateful actors need an async run (the sync window aborts the longer crawl).
+    if run_input.get("incremental"):
+        return await _run_actor_async(actor_id, run_input, timeout=timeout, token=token)
+
     url = f"{_BASE}/{actor_id}/run-sync-get-dataset-items"
 
     async with httpx.AsyncClient(timeout=timeout) as client:

app/engine/board_adapters/adapters.py

@@ -67,12 +67,13 @@ def build_naukri_input(prefs: Prefs, *, max_jobs: int = 80, freshness: str = "7"
 
 
 # ── Naukri (blackfalcondata feed — city-precise, real offsite applyRedirectUrl) ──
-def build_naukri_feed_input(prefs: Prefs, *, max_jobs: int = 50, fetch_details: bool = True) -> dict:
+def build_naukri_feed_input(prefs: Prefs, *, max_jobs: int = 50, fetch_details: bool = True,
+                            state_key: str | None = None) -> dict:
     # City-filtered at the board (hyper-relevant + cheap). fetch_details=True surfaces the
     # offsite `applyRedirectUrl`; skipReposts dedups for cost. Narrowing (dept/industry/exp)
     # stays off here — that's engine ranking later.
     city = next((lp["city"] for lp in C.parsed_locations(prefs) if lp["city"]), "")
-    return {
+    out: dict[str, Any] = {
         "keyword": C.title_of(prefs) or C.build_keyword(prefs),
         "location": city,                # precise city, e.g. "New Delhi"
         "maxResults": max_jobs,
@@ -80,6 +81,13 @@ def build_naukri_feed_input(prefs: Prefs, *, max_jobs: int = 50, fetch_details:
         "skipReposts": True,
         "sortBy": "date",
     }
+    # Incremental mode: the actor persists what it has emitted under `stateKey` (a named state store)
+    # and returns only NEW / changed jobs on each run — so re-running the same search costs us only
+    # genuinely new postings, with ZERO re-fetch. Requires an ASYNC run (apify_client routes it there).
+    if state_key:
+        out["incremental"] = True
+        out["stateKey"] = state_key
+    return out
 
 
 # ── Foundit / Monster India (city-precise; apply_url is the offsite redirect) ──
@@ -103,7 +111,7 @@ def build_foundit_input(prefs: Prefs, *, results_wanted: int = 50) -> dict:
 
 # ── LinkedIn (takes the band, not years; ₹ comp incompatible) ────────────────
 def build_linkedin_input(prefs: Prefs, *, max_items: int = 80, posted: str = "week",
-                         recall: bool = False) -> dict:
+                         recall: bool = False, page: int = 1) -> dict:
     # recall=True (retrieve stage): title + COUNTRY-level location only — drop the narrowing
     # filters (easyApply / workplaceType / experienceLevel / industryIds / freshness), which
     # become ENGINE ranking later. City-level location + a wrong industryId zero the actor out.
@@ -111,6 +119,7 @@ def build_linkedin_input(prefs: Prefs, *, max_items: int = 80, posted: str = "we
         "jobTitles": [C.title_of(prefs)],
         "maxItems": max_items,
         "sortBy": "date",
+        "page": page,                    # Start Page Number — advance per re-run for NEW jobs
     }
 
     cities: list[str] = []          # bare city names ("New Delhi") — harvestapi's location format

app/engine/normalize.py

@@ -46,11 +46,33 @@ def _compact(d: dict) -> dict:
     return {k: v for k, v in d.items() if v not in (None, "", [], {})}
 
 
-def _logo(org: str) -> dict:
+def _logo(org: str, url: str | None = None) -> dict:
     org = (org or "").strip()
     label = org[0].upper() if org else "?"
     bg = _PALETTE[sum(map(ord, org)) % len(_PALETTE)] if org else "#9CA3AF"
-    return {"label": label, "bg": bg}
+    d = {"label": label, "bg": bg}
+    # Real company logo from the actor (used by the UI; the label/bg stay as the fallback).
+    if url and isinstance(url, str) and url.startswith("http"):
+        d["url"] = url
+    return d
+
+
+# actor field names for the company logo, across boards + nesting (Naukri=logoPath,
+# LinkedIn=company.logo, Indeed=companyLogo, …). First http(s) value wins; None → letter badge.
+_LOGO_KEYS = ("logoPath", "logoPathV3", "logo", "companyLogo", "company_logo", "logoUrl", "image", "companyImage")
+
+
+def _logo_url(it: dict) -> str | None:
+    scopes = [it, it.get("company") or {}, it.get("jobDetails") or {},
+              (it.get("jobDetails") or {}).get("companyDetail") or {}]
+    for sc in scopes:
+        if not isinstance(sc, dict):
+            continue
+        for k in _LOGO_KEYS:
+            v = sc.get(k)
+            if isinstance(v, str) and v.startswith("http"):
+                return v
+    return None
 
 
 def _digits(s: str) -> int | None:
@@ -114,11 +136,16 @@ def naukri_to_scoutjob(it: dict, board: str = "Naukri") -> dict | None:
     sd = jd.get("salaryDetail") or {}
     if isinstance(sd, dict) and not sd.get("hideSalary") and sd.get("maximumSalary"):
         lpa, pay = _salary_from_amount(int(sd["maximumSalary"]), per_year=True)
+    # Offsite employer/ATS apply link first; the Naukri listing (staticUrl) is the fallback.
+    redirect = (
+        jd.get("applyRedirectUrl") or jd.get("companyApplyUrl")
+        or it.get("applyRedirectUrl") or it.get("companyApplyUrl")
+    )
     return {
         "id": f"naukri-{jd.get('jobId')}",
         "title": title,
         "organization": org,
-        "logo": _logo(org),
+        "logo": _logo(org, _logo_url(it)),
         "location_city": city,
         "location_country": "India",
         "matchScore": 0,                       # placeholder — set after merge
@@ -129,7 +156,8 @@ def naukri_to_scoutjob(it: dict, board: str = "Naukri") -> dict | None:
         "tags": skills[:3],
         "applicants": jd.get("applyCount"),
         "note": f"Live from {board}",
-        "apply_url": jd.get("staticUrl"),
+        "apply_url": redirect or jd.get("staticUrl"),   # offsite employer link first, Naukri listing fallback
+        "offsite_apply": bool(redirect),
         "qx": 10,
     }
 
@@ -151,14 +179,14 @@ def indeed_to_scoutjob(it: dict, board: str = "Indeed") -> dict | None:
         pass
     sal = it.get("salary") or ""
     lpa, pay = _salary_from_amount(_digits(sal), per_year="year" in sal.lower()) if sal else (None, "Not disclosed")
-    apply_url = it.get("externalApplyLink")
-    if not apply_url or apply_url == "None":
-        apply_url = it.get("url")
+    ext = it.get("externalApplyLink")
+    offsite = bool(ext and ext != "None")          # externalApplyLink = real offsite employer link
+    apply_url = ext if offsite else it.get("url")  # offsite first, Indeed listing fallback
     return {
         "id": f"indeed-{it.get('id')}",
         "title": title,
         "organization": org,
-        "logo": _logo(org),
+        "logo": _logo(org, _logo_url(it)),
         "location_city": city,
         "location_country": "India",
         "matchScore": 0,
@@ -169,6 +197,7 @@ def indeed_to_scoutjob(it: dict, board: str = "Indeed") -> dict | None:
         "applicants": None,
         "note": f"Live from {board}",
         "apply_url": apply_url,
+        "offsite_apply": offsite,
         "qx": 10,
     }
 
@@ -195,11 +224,19 @@ def linkedin_to_scoutjob(it: dict, board: str = "LinkedIn") -> dict | None:
         "skills": industries,
         "company_about": _plain(co.get("description"), 700) if isinstance(co, dict) else "",
     })
+    # Prefer the OFFSITE employer/ATS apply link over the LinkedIn listing — the
+    # board URL can go stale, the employer's own apply page is the durable target.
+    _apply_method = it.get("applyMethod") or {}
+    offsite = (
+        it.get("companyApplyUrl")
+        or (_apply_method.get("companyApplyUrl") if isinstance(_apply_method, dict) else None)
+        or it.get("applyRedirectUrl")
+    )
     return {
         "id": f"linkedin-{it.get('id')}",
         "title": title,
         "organization": org,
-        "logo": _logo(org),
+        "logo": _logo(org, _logo_url(it)),
         "location_city": city,
         "location_country": country,
         "location_mode": mode,
@@ -211,7 +248,8 @@ def linkedin_to_scoutjob(it: dict, board: str = "LinkedIn") -> dict | None:
         "tags": industries[:3],
         "applicants": it.get("applicants"),
         "note": f"Live from {board}",
-        "apply_url": it.get("linkedinUrl"),
+        "apply_url": offsite or it.get("linkedinUrl"),   # offsite employer link first, board link fallback
+        "offsite_apply": bool(offsite),
         "details": details,
         "qx": 10,
     }
@@ -231,7 +269,7 @@ def ats_to_scoutjob(it: dict, board: str = "Direct") -> dict | None:
         "id": f"ats-{it.get('global_id') or it.get('job_id')}",
         "title": title,
         "organization": org,
-        "logo": _logo(org),
+        "logo": _logo(org, _logo_url(it)),
         "location_city": city,
         "location_country": None,
         "location_mode": mode,
@@ -244,6 +282,7 @@ def ats_to_scoutjob(it: dict, board: str = "Direct") -> dict | None:
         "applicants": None,
         "note": f"Live from {board}",  # direct employer posting → real offsite apply link
         "apply_url": it.get("apply_url") or it.get("url"),
+        "offsite_apply": bool(it.get("apply_url") or it.get("url")),   # ATS direct posting = employer's own apply page
         "qx": 10,
     }
 
@@ -288,7 +327,7 @@ def naukri_feed_to_scoutjob(it: dict, board: str = "Naukri") -> dict | None:
         "id": f"naukri-{it.get('jobId')}",
         "title": title.title() if title.isupper() else title,
         "organization": org,
-        "logo": _logo(org),
+        "logo": _logo(org, _logo_url(it)),
         "location_city": city,
         "location_country": "India",
         "location_mode": _WORKMODE.get((it.get("wfhType") or "").lower()),
@@ -335,7 +374,7 @@ def foundit_to_scoutjob(it: dict, board: str = "Foundit") -> dict | None:
         "id": f"foundit-{it.get('job_id')}",
         "title": title,
         "organization": org,
-        "logo": _logo(org),
+        "logo": _logo(org, _logo_url(it)),
         "location_city": city,
         "location_country": "India",
         "matchScore": 0,

app/engine/search.py

@@ -25,13 +25,15 @@ def _ats_companies() -> list[dict]:
 # all city-filtered at the board. The rest stay registered (code ready) but off unless enabled.
 BOARDS = {
     "naukri": ("blackfalcondata~naukri-jobs-feed",
-               lambda p: A.build_naukri_feed_input(p, max_jobs=get_settings().NAUKRI_MAX_JOBS),
+               lambda p: A.build_naukri_feed_input(p, max_jobs=get_settings().NAUKRI_MAX_JOBS,
+                                                   state_key=p.get("_naukri_state")),
                N.naukri_feed_to_scoutjob),
     "foundit": ("shahidirfan~Foundit-Jobs-Scraper",
                 lambda p: A.build_foundit_input(p, results_wanted=get_settings().FOUNDIT_MAX_JOBS),
                 N.foundit_to_scoutjob),
     "linkedin": ("harvestapi~linkedin-job-search",
-                 lambda p: A.build_linkedin_input(p, recall=True, max_items=get_settings().LINKEDIN_MAX_JOBS),
+                 lambda p: A.build_linkedin_input(p, recall=True, max_items=get_settings().LINKEDIN_MAX_JOBS,
+                                                  page=int(p.get("_page") or 1)),
                  N.linkedin_to_scoutjob),
     # ── registered fallbacks / other lanes (off unless added to BOARDS_ENABLED) ──
     "ats": ("bovi~greenhouse-lever-ashby-job-scraper",