b1e9cdd182
- A1: fire the hard filter (Apify is precise); embedding vibe-ranker (embed.py) blends with the floor-free white-box (rank.py) to sift ~90 -> top 18 - A2: Opus curator (curate.py) reads the 18 -> honest <=count + Gemini-voiced report cards; graceful fallback to the white-box + templated cards - LLM via opencode.ai/zen gateway (llm.py): Opus chat + direct-OpenAI embeddings - Run LLM work off the event loop (asyncio.to_thread) so the Redis response publishes - C1: dedicated Postgres (app/db/) persists the per-user feed; get_scout_feed replays it so matches survive navigation/refresh - match contract + report-card fields (schema.py); skills.py; tests/
26 lines
1.2 KiB
Python
26 lines
1.2 KiB
Python
"""A1 — sift: rank-fusion math + graceful degradation when the vibe-engineer is unavailable.
|
|
(Offline — no network; the embedding path is exercised in the cached e2e verification.)"""
|
|
from app.engine import sift as S
|
|
|
|
|
|
def test_ranks_normalization():
|
|
r = S._ranks({"a": 0.9, "b": 0.5, "c": 0.1})
|
|
assert r["a"] == 1.0 and r["c"] == 0.0 and 0.0 < r["b"] < 1.0
|
|
assert S._ranks({"x": 0.5}) == {"x": 1.0} # single item → top
|
|
|
|
|
|
def test_sift_degrades_to_whitebox_when_no_vibe(monkeypatch):
|
|
monkeypatch.setattr(S._embed, "vibe_scores", lambda *a, **k: None) # embeddings unavailable
|
|
jobs = [{"id": str(i), "title": "Product Manager", "organization": "X",
|
|
"location_city": "New Delhi", "location_country": "India",
|
|
"details": {"description": "product manager roadmap"}} for i in range(30)]
|
|
prefs = {"title": "Product Manager", "location": ["New Delhi · India"]}
|
|
top, dbg = S.sift(prefs, None, jobs, k=18)
|
|
assert dbg["mode"] == "whitebox-only" and len(top) == 18
|
|
assert all("match" in j for j in top) # white-box match block rides along
|
|
|
|
|
|
def test_sift_empty_input():
|
|
top, dbg = S.sift({"title": "PM"}, None, [], k=18)
|
|
assert top == [] and dbg["mode"] == "empty"
|