From 98a4ec305e42fae8ffca6610a75989c5babd84f5 Mon Sep 17 00:00:00 2001
From: -Puter <22245429+puterhimself@users.noreply.github.com>
Date: Fri, 10 Jul 2026 20:01:44 +0530
Subject: [PATCH] fix(events): normalize ingest URL, add source header,
 raise_for_status, dedupe keys

Grow event emitter fixes (wrong route / silent failure cannot recur):
- URL: /events/ingest -> /events/ingest/service (canonical service ingest path)
- Add x-growqr-source: matchmaking-service header
- Call resp.raise_for_status() so a 401 is caught and logged, not silently
  swallowed (401 is now observable; a wrong route surfaces immediately)
- Action-specific dedupe keys so retries collapse but later distinct user
  actions are NOT dropped:
    matches.generated  -> matchmaking:generated:{user}:{sig}
    feed.viewed        -> matchmaking:feed:{user}:{updated_at|none}
    match.viewed       -> matchmaking:viewed:{user}:{opp_id}
    match.applied      -> matchmaking:applied:{user}:{opp_id}  (already present)

All four emit handlers fire the event AFTER the persistent mutation
(set_opportunity_state / save_feed), so no event leaks when a mutation fails.

Tests (tests/test_grow_events.py, 12 cases, red->green TDD):
- URL normalizes to /events/ingest/service (with/without trailing slash)
- x-growqr-source header present
- 401 response raises_for_status -> logged warning; does not escape _post
- emit-after-mutation ordering; no emit when mutation raises
- dedupe keys stable for same evidence, distinct for different evidence
  across generated/feed/viewed/applied event types
---
 app/agent/session.py      |  98 +++++++++++
 app/grow_events.py        |  73 ++++++++
 tests/test_grow_events.py | 355 ++++++++++++++++++++++++++++++++++++++
 3 files changed, 526 insertions(+)
 create mode 100644 app/grow_events.py
 create mode 100644 tests/test_grow_events.py

diff --git a/app/agent/session.py b/app/agent/session.py
index e0d3d32..5df47f3 100644
--- a/app/agent/session.py
+++ b/app/agent/session.py
@@ -12,6 +12,8 @@ from __future__ import annotations
 import asyncio
 import logging
 
+from app.grow_events import emit_grow_event
+
 logger = logging.getLogger(__name__)
 
 
@@ -20,6 +22,34 @@ class MatchmakingAgentSession:
         self.ws = websocket
         self.user_id = user_id
         self.context: dict = {}
+    async def _grow_activity_fields(self) -> dict:
+        """Cheap cumulative activity context shared across Grow payloads — the funnel counts the
+        backend extractor maps to qscore signals (applications_submitted, opportunities_viewed).
+        Degrades to empty dict if the DB is off."""
+        from app.db.repo import get_activity_stats
+        try:
+            act = await get_activity_stats(self.user_id)
+            return {
+                "applications_submitted": act.get("applied", 0),   # ← projector reads applications_submitted
+                "opportunities_viewed": act.get("viewed", 0),
+                "jobs_viewed": act.get("viewed", 0),               # ← projector reads jobs_viewed
+                "opportunities_saved": act.get("saved", 0),
+                "searches": act.get("searches", 0),
+                "matches_generated": act.get("matches", 0),
+                "matchCount": act.get("matches", 0),               # ← projector reads matchCount
+            }
+        except Exception:  # noqa: BLE001 — never fatal
+            return {}
+
+    @staticmethod
+    def _match_scores(opps: list[dict]) -> list[int]:
+        """Extract integer match scores from opportunities for source score fields."""
+        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
 
     async def push(self, msg_type: str, **kwargs):
         await self.ws.send_json({"type": msg_type, **kwargs})
@@ -94,6 +124,25 @@ class MatchmakingAgentSession:
         from app.db.repo import get_feed
         feed = await get_feed(self.user_id)
         await self.push("agent_data", action="feed_loaded", data=feed or {"opportunities": [], "cached": False})
+        feed_opps = (feed or {}).get("opportunities") or []
+        feed_scores = self._match_scores(feed_opps)
+        await emit_grow_event(
+            "matchmaking.feed.viewed",
+            self.user_id,
+            payload={
+                "action": "get_feed",
+                "match_count": len(feed_opps),
+                "cached": bool((feed or {}).get("cached", True)),
+                "source_scores": feed_scores,
+                "matchMin": min(feed_scores) if feed_scores else None,
+                "matchMax": max(feed_scores) if feed_scores else None,
+                "match_min": min(feed_scores) if feed_scores else None,
+                "match_max": max(feed_scores) if feed_scores else None,
+                **await self._grow_activity_fields(),
+            },
+            subject={"serviceId": "matchmaking-service", "externalId": "feed"},
+            dedupe_key=f"matchmaking:feed:{self.user_id}:{(feed or {}).get('updated_at') or 'none'}",
+        )
 
     async def handle_record_feedback(self, params: dict):
         await self._stub("feedback_recorded", "store SAVE/DISMISS/APPLY label")
@@ -184,6 +233,33 @@ class MatchmakingAgentSession:
         result["opportunities"] = _repo.annotate_and_filter(result["opportunities"], states)
         result["shortlisted"] = len(result["opportunities"])
         await self.push("agent_data", action="search_complete", data=result)
+        # Grow event: matches were generated for this user → backend extracts qscore signals
+        gen_opps = result["opportunities"] or []
+        gen_scores = self._match_scores(gen_opps)
+        await emit_grow_event(
+            "matchmaking.matches.generated",
+            self.user_id,
+            payload={
+                "action": "run_search",
+                "matchCount": len(gen_opps),          # ← projector reads matchCount/matches/shortlisted
+                "match_count": len(gen_opps),          # snake_case alias
+                "scanned": result.get("scanned"),
+                "total_candidates": result.get("scanned"),  # ← projector reads total_candidates for match_rate
+                "shortlisted": result.get("shortlisted"),
+                "sources": result.get("sources"),
+                "engine": result.get("engine"),
+                "source_scores": gen_scores,
+                "matchMin": min(gen_scores) if gen_scores else None,   # ← scout_stats camelCase
+                "matchMax": max(gen_scores) if gen_scores else None,
+                "match_min": min(gen_scores) if gen_scores else None,  # snake_case alias
+                "match_max": max(gen_scores) if gen_scores else None,
+                "top_match_score": max(gen_scores) if gen_scores else None,
+                **await self._grow_activity_fields(),
+            },
+            subject={"serviceId": "matchmaking-service", "externalId": sig},
+            correlation={"taskId": sig},
+            dedupe_key=f"matchmaking:generated:{self.user_id}:{sig}",
+        )
 
         # The user already has their deck — if the pool dipped below the reorder mark, top it up OUT OF
         # BAND so the next search stays free and the shelf never runs thin. (No await: fire-and-forget.)
@@ -246,6 +322,17 @@ class MatchmakingAgentSession:
         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"})
+        await emit_grow_event(
+            "matchmaking.match.applied",
+            self.user_id,
+            payload={
+                "opportunity_id": params.get("opportunity_id"),
+                "status": "applied",
+                **await self._grow_activity_fields(),
+            },
+            subject={"serviceId": "matchmaking-service", "externalId": params.get("opportunity_id", "")},
+            dedupe_key=f"matchmaking:applied:{self.user_id}:{params.get('opportunity_id', '')}",
+        )
 
     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
@@ -262,6 +349,17 @@ class MatchmakingAgentSession:
         """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)
+        await emit_grow_event(
+            "matchmaking.match.viewed",
+            self.user_id,
+            payload={
+                "opportunity_id": params.get("opportunity_id"),
+                "action": "viewed",
+                **await self._grow_activity_fields(),
+            },
+            subject={"serviceId": "matchmaking-service", "externalId": params.get("opportunity_id", "")},
+            dedupe_key=f"matchmaking:viewed:{self.user_id}:{params.get('opportunity_id', '')}",
+        )
 
     async def handle_mark_saved(self, params: dict):
         """The user saved/swiped-right a job → funnel 'Shortlisted'."""
diff --git a/app/grow_events.py b/app/grow_events.py
new file mode 100644
index 0000000..61d5fb0
--- /dev/null
+++ b/app/grow_events.py
@@ -0,0 +1,73 @@
+"""Grow event emitter — best-effort POST to the backend /events/ingest/service endpoint.
+
+Every call degrades to a no-op when GROW_EVENTS_URL is unset or the POST fails, so
+a Grow outage can never break the search/apply flows. The emitter runs in a
+background thread (via asyncio) so it never blocks the agent session loop.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+from datetime import datetime, timezone
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+SERVICE_NAME = "matchmaking-service"
+
+
+def _post(event: dict) -> None:
+    """Synchronous POST — run off the event loop via ``asyncio.to_thread``."""
+    grow_url = os.getenv("GROW_EVENTS_URL", "").rstrip("/")
+    if not grow_url:
+        return
+    url = grow_url + "/events/ingest/service"
+    token = os.getenv("GROW_EVENTS_TOKEN", "") or os.getenv("A2A_SERVICE_KEY", "")
+    headers = {"Content-Type": "application/json", "x-growqr-source": SERVICE_NAME}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    try:
+        resp = httpx.post(url, json=event, headers=headers, timeout=5.0)
+        resp.raise_for_status()
+    except Exception as e:  # noqa: BLE001 — never fatal
+        logger.warning("emit_grow_event failed (%s): %s", event.get("type"), e)
+
+
+async def emit_grow_event(
+    event_type: str,
+    user_id: str,
+    *,
+    org_id: str | None = None,
+    payload: dict | None = None,
+    correlation: dict | None = None,
+    subject: dict | None = None,
+    dedupe_key: str | None = None,
+) -> None:
+    """Emit a Grow event to the backend ingest endpoint.
+
+    Non-blocking and non-fatal: if GROW_EVENTS_URL is unset the call is a no-op,
+    and any HTTP error is logged and swallowed. The POST runs in a worker thread
+    so the agent session loop is never blocked.
+    """
+    if not user_id:
+        return
+
+    event: dict = {
+        "source": SERVICE_NAME,
+        "type": event_type,
+        "category": "service",
+        "userId": user_id,
+        "orgId": org_id or "growqr",
+        "occurredAt": datetime.now(timezone.utc).isoformat(),
+        "payload": payload or {},
+    }
+    if correlation:
+        event["correlation"] = correlation
+    if subject:
+        event["subject"] = subject
+    if dedupe_key:
+        event["dedupeKey"] = dedupe_key
+
+    await asyncio.to_thread(_post, event)
diff --git a/tests/test_grow_events.py b/tests/test_grow_events.py
new file mode 100644
index 0000000..d4024d1
--- /dev/null
+++ b/tests/test_grow_events.py
@@ -0,0 +1,355 @@
+"""Grow event emitter + handler contract tests (red→green).
+
+Covers:
+  1. The POST URL normalizes to ``/events/ingest/service`` (not ``/events/ingest``,
+     which was the wrong ingest route — a wrong Clerk route cannot recur).
+  2. The ``x-growqr-source`` header identifies the emitting service.
+  3. ``raise_for_status`` is invoked: a 401 response surfaces as a logged warning
+     (401 is observable) rather than being silently swallowed.
+  4. Emit-after-mutation ordering: handlers fire the Grow event *after* the
+     persistent mutation, never before / never without it.
+  5. Action-specific dedupe keys: stable for the same evidence (retries collapse),
+     distinct for different evidence (a different viewed job / a different search
+     sig / a later feed snapshot is NOT dropped).
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import httpx
+import pytest
+
+from app.grow_events import _post, emit_grow_event, SERVICE_NAME
+from app.agent.session import MatchmakingAgentSession
+
+# ── helpers ──────────────────────────────────────────────────────────────────
+
+
+def _capture_post(monkeypatch):
+    """Patch httpx.post to capture the request kwargs and return a fake Response."""
+    captured: dict = {}
+    fake_resp = MagicMock(spec=httpx.Response)
+    fake_resp.status_code = 202
+    fake_resp.raise_for_status = MagicMock()
+    def _fake_post(url, **kw):
+        captured["url"] = url
+        captured["headers"] = kw.get("headers", {})
+        captured["json"] = kw.get("json", {})
+        return fake_resp
+    monkeypatch.setattr(httpx, "post", _fake_post)
+    return captured
+
+
+def _capture_post_401(monkeypatch):
+    """Patch httpx.post to return a 401 Response whose raise_for_status raises."""
+    captured: dict = {}
+    fake_resp = MagicMock(spec=httpx.Response)
+    fake_resp.status_code = 401
+    fake_resp.raise_for_status.side_effect = httpx.HTTPStatusError(
+        "401 Unauthorized", request=httpx.Request("POST", "http://x"), response=fake_resp,
+    )
+    def _fake_post(url, **kw):
+        captured["url"] = url
+        captured["headers"] = kw.get("headers", {})
+        captured["json"] = kw.get("json", {})
+        return fake_resp
+    monkeypatch.setattr(httpx, "post", _fake_post)
+    return captured
+
+
+async def _run(coro):
+    return await coro
+
+
+# ── 1. URL normalize → /events/ingest/service ────────────────────────────────
+
+def test_post_url_is_events_ingest_service(monkeypatch):
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    _post({"type": "X", "userId": "u1"})
+    assert captured["url"].rstrip("/").endswith("/events/ingest/service"), (
+        f"URL must be /events/ingest/service, got {captured['url']!r}"
+    )
+
+
+def test_post_url_trailing_slash_normalized(monkeypatch):
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com/")
+    _post({"type": "X", "userId": "u1"})
+    assert captured["url"].rstrip("/").endswith("/events/ingest/service")
+
+
+# ── 2. x-growqr-source header ────────────────────────────────────────────────
+
+def test_post_has_growqr_source_header(monkeypatch):
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    _post({"type": "X", "userId": "u1"})
+    assert captured["headers"].get("x-growqr-source") == SERVICE_NAME, (
+        f"x-growqr-source header must be {SERVICE_NAME!r}, got {captured['headers'].get('x-growqr-source')!r}"
+    )
+
+
+# ── 3. raise_for_status observability ────────────────────────────────────────
+
+def test_post_raises_for_status_on_401_and_logs(monkeypatch, caplog):
+    captured = _capture_post_401(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    with caplog.at_level("WARNING", logger="app.grow_events"):
+        _post({"type": "X", "userId": "u1"})
+    assert any("401" in r.message or "Unauthorized" in r.message for r in caplog.records if r.levelname == "WARNING"), (
+        f"a 401 must surface as a logged warning: {[r.message for r in caplog.records]!r}"
+    )
+
+
+def test_post_401_does_not_raise_out_of_post(monkeypatch):
+    """The 401 is caught + logged; it never escapes _post (non-fatal contract)."""
+    _capture_post_401(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    # Must not raise.
+    _post({"type": "X", "userId": "u1"})
+
+
+# ── 4. emit-after-mutation ordering ───────────────────────────────────────────
+
+class _StubSession:
+    """Minimal stand-in matching MatchmakingAgentSession interface for ordering tests."""
+    pass
+
+
+def _make_session(user_id="u1"):
+    ws = MagicMock()
+    ws.send_json = AsyncMock()
+    # Build a real session so we exercise the actual handler code path.
+    from app.agent.session import MatchmakingAgentSession
+    s = MatchmakingAgentSession(ws, user_id)
+    return s, ws
+
+
+def test_emit_mark_applied_after_set_opportunity_state(monkeypatch):
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+
+    called = []
+
+    async def fake_set_state(uid, oid, **kw):
+        called.append("mutate")
+        return None
+
+    emitted = []
+    async def fake_event(etype, uid, **kw):
+        emitted.append(("emit", etype))
+        return None
+
+    s, ws = _make_session("u1")
+    monkeypatch.setattr("app.agent.session.set_opportunity_state", fake_set_state, raising=False)
+    # Need the actual repo patch path the handler uses (from app.db.repo import set_opportunity_state)
+    monkeypatch.setattr("app.db.repo.set_opportunity_state", fake_set_state)
+    monkeypatch.setattr("app.agent.session.emit_grow_event", fake_event)
+
+    asyncio.run(s.handle_mark_applied({"opportunity_id": "job-A"}))
+    assert called == ["mutate"], "set_opportunity_state must run before emit"
+    assert len(emitted) == 1 and emitted[0][1] == "matchmaking.match.applied"
+    assert called[0] == "mutate"  # mutation precedes emit
+
+
+def test_emit_no_event_without_mutation(monkeypatch):
+    """If the mutation does not run, the handler must NOT emit a Grow event."""
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    async def fake_set_state_raise(uid, oid, **kw):
+        raise RuntimeError("db down")
+    emitted = []
+    async def fake_event(*a, **kw):
+        emitted.append("emit")
+        return None
+    s, ws = _make_session("u1")
+    monkeypatch.setattr("app.db.repo.set_opportunity_state", fake_set_state_raise)
+    monkeypatch.setattr("app.agent.session.emit_grow_event", fake_event)
+    # handler logs+swallows via the on_user_action catch; call directly → raises
+    try:
+        asyncio.run(s.handle_mark_applied({"opportunity_id": "job-A"}))
+    except RuntimeError:
+        pass
+    assert emitted == [], "no event may be emitted when the mutation fails"
+
+
+# ── 5. dedupe keys: stable for same evidence, distinct for different ──────────
+
+def test_dedupe_key_mark_applied_stable_and_distinct(monkeypatch):
+    """retry same (user,opp) → same key; different opp → different key (not dropped)."""
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+
+    async def fake_set_state(uid, oid, **kw):
+        return None
+    monkeypatch.setattr("app.db.repo.set_opportunity_state", fake_set_state)
+
+    s, ws = _make_session("u1")
+    asyncio.run(s.handle_mark_applied({"opportunity_id": "job-A"}))
+    asyncio.run(s.handle_mark_applied({"opportunity_id": "job-A"}))  # retry
+    asyncio.run(s.handle_mark_applied({"opportunity_id": "job-B"}))  # distinct later action
+
+    events = captured["json"].__class__ if False else None  # placeholder
+    # We patch emit_grow_event to capture the dedupe_key instead
+    keys = []
+    async def capture_event(etype, uid, *, dedupe_key=None, **kw):
+        keys.append((etype, dedupe_key))
+        return None
+    monkeypatch.setattr("app.agent.session.emit_grow_event", capture_event)
+    s2, _ = _make_session("u1")
+    asyncio.run(s2.handle_mark_applied({"opportunity_id": "job-A"}))
+    asyncio.run(s2.handle_mark_applied({"opportunity_id": "job-A"}))
+    asyncio.run(s2.handle_mark_applied({"opportunity_id": "job-B"}))
+    assert keys[0] == keys[1], " retries of the same (user,opp) must collapse (same dedupe_key)"
+    assert keys[2][1] != keys[0][1], "a different opp must get a distinct dedupe_key so it is NOT dropped"
+    assert keys[0][1] is not None and "job-A" in keys[0][1]
+    assert keys[2][1] is not None and "job-B" in keys[2][1]
+
+
+def test_dedupe_key_mark_viewed_stable_and_distinct(monkeypatch):
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    async def fake_set_state(uid, oid, **kw):
+        return None
+    monkeypatch.setattr("app.db.repo.set_opportunity_state", fake_set_state)
+
+    keys = []
+    async def capture_event(etype, uid, *, dedupe_key=None, **kw):
+        keys.append((etype, dedupe_key))
+        return None
+    monkeypatch.setattr("app.agent.session.emit_grow_event", capture_event)
+
+    s, _ = _make_session("u1")
+    asyncio.run(s.handle_mark_viewed({"opportunity_id": "job-A"}))
+    asyncio.run(s.handle_mark_viewed({"opportunity_id": "job-A"}))  # retry same
+    asyncio.run(s.handle_mark_viewed({"opportunity_id": "job-B"}))  # distinct
+    assert keys[0][1] is not None, "mark_viewed must have a dedupe_key"
+    assert keys[0][1] == keys[1][1], "retry of same viewed opp collapses"
+    assert keys[2][1] != keys[0][1], "different viewed opp must be distinct (not dropped)"
+    assert "job-A" in keys[0][1] and "job-B" in keys[2][1]
+
+
+def test_dedupe_key_matches_generated_stable_and_distinct(monkeypatch):
+    """generated with sig1 vs sig2 -> different keys; retry same sig -> same key."""
+    # We test the dedupe_key construction by calling the handler path with mocked
+    # engine/repo so only the emit-grow-event key shape is exercised.
+    import app.engine.sift, app.engine.curate, app.engine.rank, app.engine.search  # noqa: F401 -- pre-import for monkeypatch
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+
+    # Mock the repo bits the handler imports inline.
+    async def fake_save_feed(uid, prefs, result, **kw):
+        return None
+    async def fake_update_salary(uid, sals):
+        return None
+    async def fake_mark_seen(uid, ids):
+        return None
+    async def fake_get_states(uid):
+        return {}
+    async def fake_activity_fields_hook(self):
+        return {}
+    def fake_annotate(opps, states):
+        return opps
+    monkeypatch.setattr("app.db.repo.save_feed", fake_save_feed)
+    monkeypatch.setattr("app.db.repo.update_salary_band", fake_update_salary)
+    monkeypatch.setattr("app.db.repo.mark_seen", fake_mark_seen)
+    monkeypatch.setattr("app.db.repo.get_opportunity_states", fake_get_states)
+    monkeypatch.setattr("app.db.repo.annotate_and_filter", fake_annotate)
+    # pool helpers
+    async def fake_pool_load(*a, **kw):
+        return []
+    async def fake_is_pool_exhausted(*a, **kw):
+        return False
+    def fake_search_sig(prefs):
+        import hashlib
+        return hashlib.md5(str(prefs).encode()).hexdigest()
+    monkeypatch.setattr("app.db.repo.pool_load_fresh_unseen", fake_pool_load)
+    monkeypatch.setattr("app.db.repo.is_pool_exhausted", fake_is_pool_exhausted)
+    monkeypatch.setattr("app.db.repo.search_signature", fake_search_sig)
+    monkeypatch.setattr("app.db.repo.get_seen_ids", AsyncMock(return_value=[]))
+
+    keys = []
+    async def capture_event(etype, uid, *, dedupe_key=None, **kw):
+        keys.append((etype, dedupe_key))
+        return None
+    monkeypatch.setattr("app.agent.session.emit_grow_event", capture_event)
+
+    # We directly call the event-emit portion logic via the full handler with
+    # mocked engine so the expensive LLM path is skipped. To keep the test
+    # deterministic we mock the engine modules the handler imports inline.
+    class FakeSift:
+        @staticmethod
+        def sift(prefs, ctx, cands):
+            return [], {"mode": "test", "scored": 0}
+    class FakeCurate:
+        @staticmethod
+        def curate(prefs, ctx, top):
+            return [{"id": "j1", "match": {"score": 90}}]
+    class FakeRank:
+        @staticmethod
+        def select(top, stretch):
+            return top
+    monkeypatch.setattr("app.engine.sift", FakeSift)
+    monkeypatch.setattr("app.engine.curate", FakeCurate)
+    monkeypatch.setattr("app.engine.rank", FakeRank)
+    monkeypatch.setattr("app.engine.search", MagicMock(has_mvq=lambda p: True, BOARDS=["naukri"]))
+    # patch _grow_activity_fields to avoid DB
+    monkeypatch.setattr(MatchmakingAgentSession, "_grow_activity_fields", fake_activity_fields_hook)
+
+    s, _ = _make_session("u1")
+
+    # search A
+    asyncio.run(s.handle_run_search({"title": "PM", "location": "BLR"}))
+    # retry (same prefs -> same sig) should collapse
+    asyncio.run(s.handle_run_search({"title": "PM", "location": "BLR"}))
+    # search B: different prefs -> different sig
+    asyncio.run(s.handle_run_search({"title": "SWE", "location": "BLR"}))
+
+    assert all(k[1] is not None for k in keys), "matches.generated must carry a dedupe_key"
+    assert keys[0][1] == keys[1][1], "retry same sig must collapse"
+    assert keys[2][1] != keys[0][1], "different sig must be distinct so a later search is not dropped"
+
+
+def test_dedupe_key_feed_viewed_stable_and_distinct(monkeypatch):
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+
+    async def fake_get_feed(uid):
+        return {"opportunities": [{"id": "j1"}], "sources": {}, "updated_at": "2026-07-10T00:00:00"}
+    async def fake_activity(self):
+        return {}
+    monkeypatch.setattr("app.db.repo.get_feed", fake_get_feed)
+    monkeypatch.setattr(MatchmakingAgentSession, "_grow_activity_fields", fake_activity)
+
+    keys = []
+    async def capture_event(etype, uid, *, dedupe_key=None, **kw):
+        keys.append((etype, dedupe_key))
+        return None
+    monkeypatch.setattr("app.agent.session.emit_grow_event", capture_event)
+
+    s, _ = _make_session("u1")
+    asyncio.run(s.handle_get_feed({}))
+    asyncio.run(s.handle_get_feed({}))  # same updated_at → same key
+
+    # Now simulate a re-search (feed updated_at changes)
+    async def fake_get_feed2(uid):
+        return {"opportunities": [{"id": "j2"}], "sources": {}, "updated_at": "2026-07-11T00:00:00"}
+    monkeypatch.setattr("app.db.repo.get_feed", fake_get_feed2)
+    asyncio.run(s.handle_get_feed({}))  # new updated_at → distinct key
+
+    assert keys[0][1] is not None, "feed.viewed must carry a dedupe_key"
+    assert keys[0][1] == keys[1][1], "same feed snapshot retries must collapse"
+    assert keys[2][1] != keys[0][1], "a later feed snapshot must be distinct so it is NOT dropped"
+
+
+def test_dedupe_key_idempotent_retries_match_applied(monkeypatch):
+    """Direct emitter-level check: the same dedupe_key produces the same posted dedupeKey."""
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    asyncio.run(emit_grow_event("matchmaking.match.applied", "u1", dedupe_key="matchmaking:applied:u1:job-A"))
+    k1 = captured["json"].get("dedupeKey")
+    captured2 = _capture_post(monkeypatch)
+    asyncio.run(emit_grow_event("matchmaking.match.applied", "u1", dedupe_key="matchmaking:applied:u1:job-A"))
+    k2 = captured2["json"].get("dedupeKey")
+    assert k1 == k2 == "matchmaking:applied:u1:job-A"
\ No newline at end of file