Real Scout dashboard backend + per-board cursor rework

- get_scout_stats aggregator (app/engine/stats): assembles the REAL Summary metrics —
  funnel + cohort engagement rank + active-window (activity timestamps by hour) from our DB,
  accumulating salary band (avg of each deck's peak, ₹L), match/competition stats from the
  feed, Momentum/QX + Q-Score trend (qscore-service), day streak (user-service). Honest:
  unsourced cards return None so the UI omits/locks them, never faked. posted_date extractor.
- Activity tracking: viewed/saved flags + search_count → funnel (Matches→Viewed→Shortlisted→
  Applied) + engagement percentile. matchesFound = all-time count.
- Per-board search cursors {board: page} (replaces the single cursor): only boards that truly
  paginate (LinkedIn) get one; cursor = LAST page fetched (1st search of a new query → 1).
  Resets on query change OR >24h (boards refresh ~daily). Dropped Naukri incremental/stateKey
  (opaque, exhausting, cross-account dedup state) — dedup is the PER-USER seen-net only.
- tests: stats helpers (posting-age, histogram, active-window, engagement score).

app/agent/session.py

@@ -51,6 +51,10 @@ class MatchmakingAgentSession:
             "restore_opportunity": self.handle_restore_opportunity,
             "mark_applied": self.handle_mark_applied,
             "save_apply_docs": self.handle_save_apply_docs,
+            # activity tracking → the dashboard funnel + engagement rank
+            "mark_viewed": self.handle_mark_viewed,
+            "mark_saved": self.handle_mark_saved,
+            "get_scout_stats": self.handle_get_scout_stats,
         }
         handler = handlers.get(action)
         if not handler:
@@ -91,14 +95,13 @@ 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.
+        # Per-board search cursors: re-running the SAME query advances EACH board's own page so every
+        # run fetches GENUINELY NEW jobs. Resets when the query changes OR >24h (boards refresh ~daily).
+        # Dedup is the per-user seen-net below — never cross-user, no opaque actor-side state.
         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(":", "_")
+        cursors = await _repo.get_search_cursors(self.user_id, sig)
+        prefs["_cursors"] = cursors
         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).
@@ -124,11 +127,19 @@ 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.
+        # Persist the RAW deck + ADVANCE every board that ran (its page +1) so the next run of this
+        # query fetches new jobs. Per-opportunity state (applied/dismissed/docs) lives separately.
         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)
+                       if k not in ("user_context", "_fresh", "_cursors")}
+        # Store the LAST page fetched per paginating board (last + 1 = the page we just pulled). So a
+        # NEW query's first search stores 1 (page 1), not 2. Naukri/Foundit have no page param → no
+        # cursor (they dedup via the per-user seen-net).
+        next_cursors = {b: int(cursors.get(b, 0)) + 1
+                        for b in (sweep.get("sources") or {}) if b in search.PAGINATING_BOARDS}
+        await _repo.save_feed(self.user_id, clean_prefs, result, query_sig=sig, cursors=next_cursors)
+        # Accumulate the salary band — each deck contributes its peak salary, averaged over decks
+        # (survives decks that disclose nothing, like Naukri "Not disclosed").
+        await _repo.update_salary_band(self.user_id, [o.get("salary_lpa") for o in (result["opportunities"] or [])])
         # 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.
@@ -187,3 +198,23 @@ class MatchmakingAgentSession:
         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})
+
+    # ── activity tracking (dashboard funnel + engagement rank) ──
+    async def handle_mark_viewed(self, params: dict):
+        """The user opened a job → funnel 'Viewed'. Fire-and-forget (no response needed)."""
+        from app.db.repo import set_opportunity_state
+        await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), viewed=True)
+
+    async def handle_mark_saved(self, params: dict):
+        """The user saved/swiped-right a job → funnel 'Shortlisted'."""
+        from app.db.repo import set_opportunity_state
+        await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), saved=True)
+
+    async def handle_get_scout_stats(self, params: dict):
+        """Assemble the REAL Summary dashboard metrics (Phase 2 fills the cross-service bits)."""
+        from app.engine import stats
+        data = await stats.build_scout_stats(
+            self.user_id,
+            user_uuid=params.get("user_uuid"),
+        )
+        await self.push("agent_data", action="scout_stats", data=data)

app/config.py

@@ -63,9 +63,15 @@ class Settings(BaseSettings):
     ATS_COMPANIES: str = "stripe,databricks,gitlab,figma,ramp,notion,razorpay,zerodha"
     INDEED_MAX_JOBS: int = 30
 
-    # QScore (competence proxy) — consumed, not computed
+    # QScore (competence proxy) — consumed, not computed. Dashboard Momentum/QX reads it via REST.
     QSCORE_BASE_URL: str = "http://localhost:8004"
     QSCORE_AUTH_TOKEN: str | None = None
+    QSCORE_ORG_ID: str = "growqr"
+    # user-service — dashboard Day-streak reads metadata.current_streak via GET /api/state/{clerk_id}
+    USER_SERVICE_BASE_URL: str = "http://localhost:8003"
+    # dashboard-service — Q-Score trend series via GET /api/qscore-history/{clerk_id}
+    DASHBOARD_SERVICE_BASE_URL: str = "http://localhost:8005"
+    A2A_OUTBOUND_KEY: str = "dev-a2a-key"   # Bearer token presented to sibling services
 
     # Storage for feedback/labels (NOT the old corpus). Optional until the learning loop lands.
     DATABASE_URL: str | None = None

app/db/models.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import JSON, Boolean, DateTime, Integer, String, func
+from sqlalchemy import JSON, Boolean, DateTime, Float, Integer, String, func
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
 
@@ -22,19 +22,26 @@ 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.
+    # PER-BOARD search cursors — {board: LAST page fetched}. Next page = last + 1, so a NEW query's
+    # first search reads page 1 and stores 1. Resets to {} when the query signature changes (or >24h).
+    # Dedup is purely per-user (the seen-net below) — never cross-user, never opaque actor-side state.
     query_sig: Mapped[str] = mapped_column(String, default="")
-    cursor_page: Mapped[int] = mapped_column(Integer, default=1)
+    cursors: Mapped[dict] = mapped_column(JSON, default=dict)
+    search_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0")  # total searches run (engagement)
+    # Salary band = AVERAGE of each deck's HIGHEST disclosed salary (₹L). Each deck contributes one
+    # peak; the band is the running mean of those peaks. Survives decks that disclose nothing.
+    salary_peak_sum: Mapped[float] = mapped_column(Float, default=0.0, server_default="0")
+    salary_peak_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
     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)."""
+    """Per-(user, opportunity) state that SURVIVES deck refreshes. Persists: `applied`/`dismissed`
+    status, the resume-builder document ids already generated (never re-pay to re-tailor), and the
+    PROGRESSIVE activity flags that feed the dashboard funnel + engagement rank. A job can be viewed
+    AND saved AND applied at once, so viewed/saved are booleans (not the single-valued status)."""
     __tablename__ = "opportunity_state"
 
     user_id: Mapped[str] = mapped_column(String, primary_key=True)
@@ -43,7 +50,9 @@ class OpportunityState(Base):
     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
+    seen: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")    # shown in a deck → exclude from new runs
+    viewed: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")  # opened the job → funnel "Viewed"
+    saved: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")   # swiped right → funnel "Shortlisted"
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
     )

app/db/repo.py

@@ -6,7 +6,7 @@ import hashlib
 import json
 import logging
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from app.db.models import Base, OpportunityState, UserFeed
@@ -14,8 +14,8 @@ 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")
+# fields a caller may set on an opportunity's state (status + the generated document ids + activity flags)
+_STATE_FIELDS = ("status", "tailored_resume_id", "tailored_version_id", "cover_letter_id", "seen", "viewed", "saved")
 
 # the search-defining pref fields — same values → same signature → cursor advances (new jobs);
 # change any → new signature → cursor resets to page 1.
@@ -43,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, *, 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."""
+async def save_feed(user_id: str, prefs: dict, result: dict, *, query_sig: str = "", cursors: dict | None = None) -> None:
+    """Upsert the user's last completed search + the advanced per-board cursors. No-op if the DB is off."""
     factory = session_factory()
     if factory is None or not user_id:
         return
@@ -56,14 +56,17 @@ async def save_feed(user_id: str, prefs: dict, result: dict, *, query_sig: str =
         "engine": result.get("engine") or "",
         "scanned": int(result.get("scanned") or 0),
         "query_sig": query_sig,
-        "cursor_page": cursor_page,
+        "cursors": cursors or {},
+        "search_count": 1,
     }
     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", "query_sig", "cursor_page")},
+                set_={**{k: row[k] for k in ("prefs", "opportunities", "sources", "engine", "scanned", "query_sig", "cursors")},
+                      # engagement: count every search run (insert starts at 1, each re-run +1)
+                      "search_count": UserFeed.search_count + 1},
             )
             await s.execute(stmt)
             await s.commit()
@@ -71,6 +74,42 @@ async def save_feed(user_id: str, prefs: dict, result: dict, *, query_sig: str =
         logger.warning("save_feed failed (non-fatal): %s", e)
 
 
+async def update_salary_band(user_id: str, salaries: list[float]) -> None:
+    """Accumulate the salary band: each deck contributes its HIGHEST disclosed salary (a 'peak'); the
+    band is the running AVERAGE of those peaks (₹L). No-op when the deck disclosed nothing (persists)."""
+    sal = [float(x) for x in salaries if x]
+    factory = session_factory()
+    if factory is None or not user_id or not sal:
+        return
+    hi = max(sal)
+    try:
+        async with factory() as s:
+            row = (await s.execute(select(UserFeed).where(UserFeed.user_id == user_id))).scalar_one_or_none()
+            if row is None:
+                return
+            row.salary_peak_sum = (row.salary_peak_sum or 0.0) + hi   # this deck's peak
+            row.salary_peak_count = (row.salary_peak_count or 0) + 1
+            await s.commit()
+    except Exception as e:  # noqa: BLE001
+        logger.warning("update_salary_band failed (non-fatal): %s", e)
+
+
+async def get_salary_band(user_id: str) -> float | None:
+    """The salary band = average of each deck's peak salary (₹L), or None if no deck disclosed one."""
+    factory = session_factory()
+    if factory is None or not user_id:
+        return None
+    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_salary_band failed (non-fatal): %s", e)
+        return None
+    if not row or not row.salary_peak_count:
+        return None
+    return round((row.salary_peak_sum or 0.0) / row.salary_peak_count, 1)
+
+
 async def get_feed(user_id: str) -> dict | None:
     """The user's last cached feed as a search_complete-shaped dict, or None."""
     factory = session_factory()
@@ -145,6 +184,101 @@ async def get_opportunity_states(user_id: str) -> dict[str, dict]:
     }
 
 
+# engagement score weights — applies count most, then saves, views, searches (honest "working it")
+def _activity_score(viewed: int, saved: int, applied: int, searches: int) -> int:
+    return applied * 3 + saved * 2 + viewed * 1 + searches * 1
+
+
+async def get_activity_stats(user_id: str) -> dict:
+    """Real dashboard activity for `user_id`: the funnel counts (viewed/saved/applied), how many
+    documents Mira tailored, search count, and the COHORT engagement percentile (this user's activity
+    vs every Scout user in the matchmaking DB). All honest, DB-derived. Degrades to zeros if DB is off."""
+    factory = session_factory()
+    empty = {"matches": 0, "viewed": 0, "saved": 0, "applied": 0, "tailored": 0, "searches": 0, "percentile": 0, "cohort_size": 0}
+    if factory is None or not user_id:
+        return empty
+    try:
+        async with factory() as s:
+            # per-user activity across ALL users (one grouped pass → funnel for this user + cohort rank)
+            os_rows = (await s.execute(
+                select(
+                    OpportunityState.user_id,
+                    func.count().filter(OpportunityState.seen.is_(True)).label("matches"),
+                    func.count().filter(OpportunityState.viewed.is_(True)).label("viewed"),
+                    func.count().filter(OpportunityState.saved.is_(True)).label("saved"),
+                    func.count().filter(OpportunityState.status == "applied").label("applied"),
+                    func.count().filter(
+                        OpportunityState.tailored_version_id.isnot(None)
+                        | OpportunityState.cover_letter_id.isnot(None)
+                    ).label("tailored"),
+                ).group_by(OpportunityState.user_id)
+            )).all()
+            feed_rows = (await s.execute(select(UserFeed.user_id, UserFeed.search_count))).all()
+    except Exception as e:  # noqa: BLE001
+        logger.warning("get_activity_stats failed (non-fatal): %s", e)
+        return empty
+
+    searches_by = {r.user_id: int(r.search_count or 0) for r in feed_rows}
+    me, scores = dict(empty), []
+    users = {r.user_id for r in os_rows} | set(searches_by)
+    by_os = {r.user_id: r for r in os_rows}
+    for uid in users:
+        r = by_os.get(uid)
+        mt, v, sv, ap, tl = (int(r.matches), int(r.viewed), int(r.saved), int(r.applied), int(r.tailored)) if r else (0, 0, 0, 0, 0)
+        sc = searches_by.get(uid, 0)
+        scores.append(_activity_score(v, sv, ap, sc))
+        if uid == user_id:
+            me = {"matches": mt, "viewed": v, "saved": sv, "applied": ap, "tailored": tl, "searches": sc}
+    my_score = _activity_score(me["viewed"], me["saved"], me["applied"], me["searches"])
+    cohort = len(scores) or 1
+    # percentile = share of the cohort you're STRICTLY ahead of (so rank-1-of-3 → 67th → "top 33%",
+    # not "top 0%"). UI shows top (100 - percentile)%.
+    below = sum(1 for x in scores if x < my_score)
+    me["percentile"] = round(100 * below / cohort)
+    me["cohort_size"] = cohort
+    return me
+
+
+async def get_active_window(user_id: str, *, tz_offset_min: int = 330) -> dict:
+    """WHEN the user works their search — activity timestamps (views/saves/applies + searches)
+    bucketed by local hour-of-day. Returns the peak 3-hour band + a 24-bucket histogram. Real signal,
+    from this DB. tz_offset_min defaults to IST (+5:30); the product is India-centric.
+
+    SQLite/Postgres-agnostic: we read raw UTC timestamps and bucket in Python (so it works in tests too)."""
+    factory = session_factory()
+    empty = {"by_hour": [0] * 24, "peak_start": None, "peak_end": None, "samples": 0}
+    if factory is None or not user_id:
+        return empty
+    try:
+        async with factory() as s:
+            os_ts = (await s.execute(
+                select(OpportunityState.updated_at).where(OpportunityState.user_id == user_id)
+            )).scalars().all()
+            feed_ts = (await s.execute(
+                select(UserFeed.updated_at).where(UserFeed.user_id == user_id)
+            )).scalars().all()
+    except Exception as e:  # noqa: BLE001
+        logger.warning("get_active_window failed (non-fatal): %s", e)
+        return empty
+
+    by_hour = [0] * 24
+    for ts in [*os_ts, *feed_ts]:
+        if ts is None:
+            continue
+        local_min = ts.hour * 60 + ts.minute + tz_offset_min
+        by_hour[(local_min // 60) % 24] += 1
+    samples = sum(by_hour)
+    if not samples:
+        return empty
+    # peak 3-hour window (circular) by summed activity
+    best_h, best_sum = 0, -1
+    for h in range(24):
+        win = by_hour[h] + by_hour[(h + 1) % 24] + by_hour[(h + 2) % 24]
+        if win > best_sum:
+            best_h, best_sum = h, win
+    return {"by_hour": by_hour, "peak_start": best_h, "peak_end": (best_h + 3) % 24, "samples": samples}
+
+
 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."""
@@ -163,21 +297,31 @@ def annotate_and_filter(opportunities: list[dict], states: dict[str, dict]) -> l
     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."""
+_CURSOR_TTL_HOURS = 24   # job boards refresh ~daily (live scrape, 7-day newest-first window) → reset every 24h
+
+
+async def get_search_cursors(user_id: str, query_sig: str) -> dict:
+    """Per-board start pages for this query, e.g. {'linkedin': 3}. RESETS to empty (every board → page 1)
+    when EITHER the query changed OR the last search was >24h ago (so we re-surface the freshest daily
+    postings instead of marching into stale depth). Each board advances its OWN page; dedup is the
+    per-user seen-net, never cross-user."""
+    from datetime import datetime, timezone
     factory = session_factory()
     if factory is None or not user_id:
-        return 1
+        return {}
     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
+        logger.warning("get_search_cursors failed (non-fatal): %s", e)
+        return {}
     if not row or row.query_sig != query_sig:
-        return 1                                  # new/changed query → fresh cursor
-    return int(row.cursor_page or 1)
+        return {}                                 # new/changed query → all boards start at page 1
+    if row.updated_at:
+        age_h = (datetime.now(timezone.utc) - row.updated_at).total_seconds() / 3600
+        if age_h >= _CURSOR_TTL_HOURS:
+            return {}                             # >24h stale → reset to catch the newest jobs
+    return dict(row.cursors or {})
 
 
 async def get_seen_ids(user_id: str) -> set[str]:

app/engine/board_adapters/adapters.py

@@ -67,13 +67,14 @@ 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,
-                            state_key: str | None = None) -> dict:
+def build_naukri_feed_input(prefs: Prefs, *, max_jobs: int = 50, fetch_details: bool = True) -> dict:
     # City-filtered at the board (hyper-relevant + cheap). fetch_details=True surfaces the
-    # offsite `applyRedirectUrl`; skipReposts dedups for cost. Narrowing (dept/industry/exp)
+    # offsite `applyRedirectUrl`; skipReposts dedups WITHIN a run. Narrowing (dept/industry/exp)
     # stays off here — that's engine ranking later.
+    # NOTE: no `incremental`/`stateKey` — that put opaque, exhausting, cross-account dedup state on
+    # Apify. Cross-deck dedup is the PER-USER seen-net in our DB. (The feed actor has no page/offset.)
     city = next((lp["city"] for lp in C.parsed_locations(prefs) if lp["city"]), "")
-    out: dict[str, Any] = {
+    return {
         "keyword": C.title_of(prefs) or C.build_keyword(prefs),
         "location": city,                # precise city, e.g. "New Delhi"
         "maxResults": max_jobs,
@@ -81,13 +82,6 @@ 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) ──

app/engine/normalize.py

@@ -75,6 +75,23 @@ def _logo_url(it: dict) -> str | None:
     return None
 
 
+# posting-date field across boards (Naukri createdDate · LinkedIn postedDate · Foundit date_posted).
+# Feeds the dashboard "Apply window" (freshness urgency) + honest "posted Nd ago" age signals.
+_POSTED_KEYS = ("postedDate", "createdDate", "date_posted", "postedAt", "listedAt", "datePosted", "published")
+
+
+def _posted_date(it: dict) -> str | None:
+    scopes = [it, it.get("jobDetails") or {}]
+    for sc in scopes:
+        if not isinstance(sc, dict):
+            continue
+        for k in _POSTED_KEYS:
+            v = sc.get(k)
+            if isinstance(v, str) and v[:4].isdigit():   # ISO-ish "2026-06-18T…"
+                return v
+    return None
+
+
 def _digits(s: str) -> int | None:
     nums = re.sub(r"[^\d]", "", s or "")
     return int(nums) if nums else None
@@ -146,6 +163,7 @@ def naukri_to_scoutjob(it: dict, board: str = "Naukri") -> dict | None:
         "title": title,
         "organization": org,
         "logo": _logo(org, _logo_url(it)),
+        "posted_date": _posted_date(it),
         "location_city": city,
         "location_country": "India",
         "matchScore": 0,                       # placeholder — set after merge
@@ -187,6 +205,7 @@ def indeed_to_scoutjob(it: dict, board: str = "Indeed") -> dict | None:
         "title": title,
         "organization": org,
         "logo": _logo(org, _logo_url(it)),
+        "posted_date": _posted_date(it),
         "location_city": city,
         "location_country": "India",
         "matchScore": 0,
@@ -237,6 +256,7 @@ def linkedin_to_scoutjob(it: dict, board: str = "LinkedIn") -> dict | None:
         "title": title,
         "organization": org,
         "logo": _logo(org, _logo_url(it)),
+        "posted_date": _posted_date(it),
         "location_city": city,
         "location_country": country,
         "location_mode": mode,
@@ -270,6 +290,7 @@ def ats_to_scoutjob(it: dict, board: str = "Direct") -> dict | None:
         "title": title,
         "organization": org,
         "logo": _logo(org, _logo_url(it)),
+        "posted_date": _posted_date(it),
         "location_city": city,
         "location_country": None,
         "location_mode": mode,
@@ -328,6 +349,7 @@ def naukri_feed_to_scoutjob(it: dict, board: str = "Naukri") -> dict | None:
         "title": title.title() if title.isupper() else title,
         "organization": org,
         "logo": _logo(org, _logo_url(it)),
+        "posted_date": _posted_date(it),
         "location_city": city,
         "location_country": "India",
         "location_mode": _WORKMODE.get((it.get("wfhType") or "").lower()),
@@ -375,6 +397,7 @@ def foundit_to_scoutjob(it: dict, board: str = "Foundit") -> dict | None:
         "title": title,
         "organization": org,
         "logo": _logo(org, _logo_url(it)),
+        "posted_date": _posted_date(it),
         "location_city": city,
         "location_country": "India",
         "matchScore": 0,

app/engine/search.py

@@ -21,19 +21,25 @@ def _ats_companies() -> list[dict]:
     return [{"company": c.strip()} for c in get_settings().ATS_COMPANIES.split(",") if c.strip()]
 
 
+# Boards whose actor takes a real page/offset → they get a per-board cursor that advances each run.
+# Everyone else (Naukri/Foundit feeds — no page param) relies purely on the per-user seen-net; we do
+# NOT track a meaningless cursor for them.
+PAGINATING_BOARDS = {"linkedin", "indeed"}
+
 # board key → (actor_id, build_input, normalizer). "Balanced" India stack = naukri+foundit+linkedin,
 # 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,
-                                                   state_key=p.get("_naukri_state")),
+               # no page/offset param → per-user seen-net handles dedup (NOT actor-side incremental).
+               lambda p: A.build_naukri_feed_input(p, max_jobs=get_settings().NAUKRI_MAX_JOBS),
                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",
+                 # cursor stores the LAST page fetched (0 = none yet) → next page = last + 1.
                  lambda p: A.build_linkedin_input(p, recall=True, max_items=get_settings().LINKEDIN_MAX_JOBS,
-                                                  page=int(p.get("_page") or 1)),
+                                                  page=int((p.get("_cursors") or {}).get("linkedin", 0)) + 1),
                  N.linkedin_to_scoutjob),
     # ── registered fallbacks / other lanes (off unless added to BOARDS_ENABLED) ──
     "ats": ("bovi~greenhouse-lever-ashby-job-scraper",

app/engine/stats/__init__.py

@@ -0,0 +1,114 @@
+"""Scout Summary dashboard — assemble REAL metrics (no fabricated numbers).
+
+Honest sources (SCOUT_UI_SPEC):
+- feed stats (matches / top / range / salary / competition)  ← the persisted feed
+- funnel + engagement rank                                   ← opportunity_state activity (this DB)
+- Apply window (freshness urgency)                           ← listing posting-age
+- Momentum/QX + Readiness trend                              ← qscore-service   (Phase 1)
+- Day streak                                                 ← dashboard-service (Phase 1)
+
+Anything without a real source is omitted — the UI cuts or locks it, never fakes it.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import datetime, timezone
+
+from app.db import repo as _repo
+from app.engine.stats import clients as _clients
+
+logger = logging.getLogger(__name__)
+
+
+def _match_scores(opps: list[dict]) -> list[int]:
+    out = []
+    for o in opps:
+        s = (o.get("match") or {}).get("score") or o.get("matchScore")
+        if isinstance(s, (int, float)) and s > 0:
+            out.append(int(s))
+    return out
+
+
+def _histogram(scores: list[int], lo: int = 60, hi: int = 100, bins: int = 11) -> list[int]:
+    h = [0] * bins
+    span = (hi - lo) / bins
+    for s in scores:
+        i = min(bins - 1, max(0, int((s - lo) / span)))
+        h[i] += 1
+    return h
+
+
+def _days_since(iso: str | None) -> int | None:
+    if not iso:
+        return None
+    try:
+        dt = datetime.fromisoformat(iso.replace("Z", "+00:00"))
+        return max(0, (datetime.now(timezone.utc) - dt).days)
+    except (ValueError, TypeError):
+        return None
+
+
+def _apply_window(opps: list[dict]) -> dict:
+    """Freshness urgency from listing posting-age (replaces the old 'golden window'). Fresher
+    deck → apply now. Honest: derived from the real posting dates the actors returned."""
+    ages = [d for o in opps if (d := _days_since(o.get("posted_date"))) is not None]
+    if not ages:
+        return {"median_age_days": None, "fresh_share": None, "urgency": None}
+    ages.sort()
+    median = ages[len(ages) // 2]
+    fresh = sum(1 for a in ages if a <= 3) / len(ages)        # posted within 3 days
+    urgency = "high" if median <= 3 else "medium" if median <= 7 else "low"
+    return {"median_age_days": median, "fresh_share": round(fresh, 2), "urgency": urgency, "dated": len(ages)}
+
+
+async def build_scout_stats(user_id: str, *, user_uuid: str | None = None) -> dict:
+    """The real Summary payload. Feed + activity are wired now; qscore/streak land in Phase 1."""
+    # local DB reads + cross-service reads (QX, QX-trend, streak) — all concurrent, each degrades to None
+    feed, act, active, salary_band, qx, qx_trend, streak = await asyncio.gather(
+        _repo.get_feed(user_id),
+        _repo.get_activity_stats(user_id),
+        _repo.get_active_window(user_id),
+        _repo.get_salary_band(user_id),
+        _clients.fetch_qx(user_uuid),
+        _clients.fetch_qx_trend(user_uuid),
+        _clients.fetch_streak(user_id),
+    )
+    feed = feed or {}
+    opps = feed.get("opportunities") or []
+
+    scores = _match_scores(opps)
+    apps = [j.get("applicants") for j in opps if j.get("applicants")]
+    matches = len(opps)
+
+    return {
+        # ── feed-derived (real) ──
+        # all-time roles Scout has matched you with (consistent with the funnel's "Matches"), not just
+        # the latest deck — that's the headline "Matches found" number.
+        "matchesFound": max(act["matches"], matches),
+        "topMatch": max(scores) if scores else None,
+        "matchMin": min(scores) if scores else None,
+        "matchMax": max(scores) if scores else None,
+        "matchHist": _histogram(scores) if scores else [],
+        # salary band = average of each deck's peak salary, accumulated across decks (₹L)
+        "salaryMaxL": salary_band,
+        "avgApplicants": round(sum(apps) / len(apps)) if apps else None,
+        "activeWindow": active if active.get("samples") else None,   # WHEN the user works their search
+        # ── activity-derived (real, this DB) — funnel is all-time so the stages are consistent ──
+        "funnel": [
+            {"label": "Matches", "value": max(act["matches"], matches), "tracked": True},
+            {"label": "Viewed", "value": act["viewed"], "tracked": True},
+            {"label": "Shortlisted", "value": act["saved"], "tracked": True},
+            {"label": "Applied", "value": act["applied"], "tracked": True},
+        ],
+        "tailoredResumes": act["tailored"],
+        "searches": act["searches"],
+        "percentile": act["percentile"],     # cohort engagement percentile (top 100-percentile %)
+        "cohortSize": act["cohort_size"],
+        "searchStage": "Applying" if act["applied"] else "Reviewing" if act["saved"] else "Scanning",
+        # ── cross-service (real; None → the UI omits the card, never fakes it) ──
+        "qxNow": (qx or {}).get("qx"),
+        "quotients": (qx or {}).get("quotients"),
+        "qxTrend": qx_trend,            # Q-Score series (dashboard-service) → Readiness-trend sparkline
+        "dayStreak": streak,
+    }

app/engine/stats/clients.py

@@ -0,0 +1,75 @@
+"""Cross-service reads for the dashboard — QScore (Momentum/QX) + user-service (Day streak).
+
+Direct REST, Clerk-mesh-authed. Both degrade to None on any failure (timeout / 404 / service down) —
+a missing KPI is omitted by the UI, never faked, and never crashes the dashboard.
+"""
+from __future__ import annotations
+
+import logging
+
+import httpx
+
+from app.config import get_settings
+
+logger = logging.getLogger(__name__)
+
+
+async def fetch_qx(user_uuid: str | None) -> dict | None:
+    """QScore for the user → {qx, quotients}. GET /v1/qscore/{uuid}?org_id (qscore-service)."""
+    if not user_uuid:
+        return None
+    s = get_settings()
+    try:
+        async with httpx.AsyncClient(timeout=4.0) as c:
+            r = await c.get(f"{s.QSCORE_BASE_URL}/v1/qscore/{user_uuid}", params={"org_id": s.QSCORE_ORG_ID})
+        if r.status_code == 404:        # no score computed yet — honest absence
+            return None
+        r.raise_for_status()
+        d = r.json()
+        qx = d.get("q_score")
+        return {"qx": round(float(qx), 1), "quotients": d.get("quotients") or {}} if qx is not None else None
+    except (httpx.HTTPError, ValueError) as e:
+        logger.warning("fetch_qx failed (non-fatal): %s", e)
+        return None
+
+
+async def fetch_qx_trend(user_uuid: str | None) -> list[float] | None:
+    """Q-Score trend (oldest→newest) ← qscore-service GET /v1/qscore/{uuid}/history. Same source as
+    the live QX, so the trend's latest point matches the Momentum value (consistent, not stale).
+    None on failure → the UI keeps the Readiness card locked."""
+    if not user_uuid:
+        return None
+    s = get_settings()
+    try:
+        async with httpx.AsyncClient(timeout=4.0) as c:
+            r = await c.get(
+                f"{s.QSCORE_BASE_URL}/v1/qscore/{user_uuid}/history",
+                params={"org_id": s.QSCORE_ORG_ID, "points": 10},
+            )
+        if r.status_code == 404:
+            return None
+        r.raise_for_status()
+        pts = [round(float(p["q_score"]), 1) for p in (r.json().get("points") or []) if p.get("q_score") is not None]
+        return pts if len(pts) >= 2 else None     # need ≥2 points for a trend
+    except (httpx.HTTPError, ValueError, TypeError, KeyError) as e:
+        logger.warning("fetch_qx_trend failed (non-fatal): %s", e)
+        return None
+
+
+async def fetch_streak(clerk_id: str | None) -> int | None:
+    """Day streak ← user-service metadata.current_streak. GET /api/state/{clerk_id} (Bearer A2A)."""
+    if not clerk_id:
+        return None
+    s = get_settings()
+    try:
+        async with httpx.AsyncClient(timeout=4.0) as c:
+            r = await c.get(
+                f"{s.USER_SERVICE_BASE_URL}/api/state/{clerk_id}",
+                headers={"Authorization": f"Bearer {s.A2A_OUTBOUND_KEY}"},
+            )
+        r.raise_for_status()
+        streak = (r.json().get("metadata") or {}).get("current_streak")
+        return int(streak) if streak is not None else None
+    except (httpx.HTTPError, ValueError, TypeError) as e:
+        logger.warning("fetch_streak failed (non-fatal): %s", e)
+        return None

tests/test_stats.py

@@ -0,0 +1,33 @@
+"""Dashboard stats — pure-helper contract tests (no DB / no network)."""
+from app.engine import stats as S
+from app.engine.normalize import _posted_date
+from app.db.repo import _activity_score
+
+
+def test_posted_date_per_board():
+    assert _posted_date({"createdDate": "2026-06-18T10:37:17Z"}) == "2026-06-18T10:37:17Z"   # Naukri
+    assert _posted_date({"postedDate": "2026-06-18T12:46:39Z"}) == "2026-06-18T12:46:39Z"     # LinkedIn
+    assert _posted_date({"date_posted": "2026-06-17T17:38:14Z"}) == "2026-06-17T17:38:14Z"    # Foundit
+    assert _posted_date({"jobDetails": {"createdDate": "2026-06-18T00:00:00Z"}}) is not None   # nested
+    assert _posted_date({"nope": "x"}) is None
+
+
+def test_match_scores_and_histogram():
+    opps = [{"match": {"score": 84}}, {"matchScore": 72}, {"match": {"score": 0}}, {}]
+    scores = S._match_scores(opps)
+    assert sorted(scores) == [72, 84]                 # 0 and missing dropped
+    h = S._histogram(scores)
+    assert sum(h) == 2 and len(h) == 11
+
+
+def test_apply_window_freshness():
+    # all None when no posting dates (honest — never fabricated)
+    assert S._apply_window([{}, {"posted_date": None}])["urgency"] is None
+    w = S._apply_window([{"posted_date": "2026-06-19T00:00:00Z"}, {"posted_date": "2026-06-15T00:00:00Z"}])
+    assert w["dated"] == 2 and w["median_age_days"] is not None and w["urgency"] in ("high", "medium", "low")
+
+
+def test_activity_score_weighting():
+    # applies weighted highest, then saves, views, searches
+    assert _activity_score(0, 0, 1, 0) > _activity_score(0, 1, 0, 0) > _activity_score(1, 0, 0, 0)
+    assert _activity_score(0, 0, 0, 0) == 0