# Online Ranking Implementation Plan **Goal:** Serve the pairing graph to shoppers, taking recommendation coverage from 7 products to 201, without ever serving something a merchant has not approved. **Architecture:** One new read model in `pairing/queries.py`, which already owns every read of the neighbour table, and one rewritten branch inside `decide()`. Nothing else moves. `catalog/products.py` is not touched — its `related_products` stays as the fallback for tenants who have never run pairing. **Tech Stack:** FastAPI, psycopg2, pytest. **Spec:** `docs/specs/2026-08-11-online-ranking-design.md` **Depends on:** Phases 0–3, all built and live-verified. The live graph holds 1,888 pairs over 201 anchors. ## Global Constraints - **Never commit, never push, never stage.** The user commits their own work. Every task ends with `git status --short` and a report. - **No assistant attribution** in any file, message, or commit. - **Comments explain WHY, not WHAT.** - **THE RULE THIS PHASE EXISTS TO ENFORCE: nothing a merchant has not approved may reach a shopper.** Reuse `is_servable` from `pairing/decisions.py`; do not reimplement or approximate it. Servability is evaluated at serve time, never trusted from pairing time. - **No model calls anywhere reachable from `decide()`.** No embeddings, no chat completions, no `ai_tokens` spend. This path runs on every page view of every visitor on an unauthenticated endpoint. Read the module docstring in `pairing/recommendations.py` — it is emphatic about this, and there is a test asserting it. - **`decide()` must never raise.** A suggestion the visitor never asked for must never surface as an error in their browser. The existing try/except contract stays. - **Do not modify `app/services/catalog/products.py`.** Not `related_products`, not `compute_relatedness`, not `to_card`. They are the fallback and the card shape. - **Do not change the widget's response contract.** `recommend`, `reason`, `message`, `products` keep their meanings. New fields on a card are additive only. - **When servability cannot be determined, show less, not more.** Any uncertainty resolves to not serving. - Per-tenant tables are `strategist_*`; interpolate identifiers with `sql.Identifier`. - Match conventions: module-level `logger`, snake_case, double quotes, 4-space indent, imports stdlib → third-party → `app.*`. - Tests: `.venv/bin/python -m pytest`. Baselines: `tests/unit/` **608**, `tests/integration/` **121**. - **Run the two suites in the FOREGROUND and SEPARATELY.** Never in one pytest process — the unit conftest sets dummy DB credentials. Never in the background. --- ## File Structure | File | Responsibility | |---|---| | `app/services/pairing/queries.py` (modify) | `servable_neighbours()` — the one query that feeds serving | | `app/services/pairing/recommendations.py` (modify) | `decide()` reads the graph, falls back, attributes cards | | `tests/integration/test_servable_neighbours.py` (new) | The four ways an unwanted product could reach a shopper | | `tests/unit/test_recommendations_graph.py` (new) | Blend, ordering, fallback, no-model-calls | --- ### Task 1: The serving query **Files:** - Modify: `app/services/pairing/queries.py` - Test: `tests/integration/test_servable_neighbours.py` **Interfaces:** - Consumes: `get_db_connection`, `is_servable`, `load_decisions`. - Produces: `SERVED_PAIR_TYPES: tuple` = `("similar", "complement")`, and `servable_neighbours(tenant_id: str, product_key: str, limit: int = 3) -> list[dict]` — whole product rows, each with `pair_type` and `pair_score` attached, ordered by score descending. **Why here:** this module already owns every read of `strategist_product_neighbors` and already imports `is_servable`. A pairing query inside the product store would invert the dependency for nothing. **Why whole rows:** `decide()` passes them straight to `to_card`, which expects a product row. Returning keys would force a second lookup on the hottest path in the system. - [ ] **Step 1: Write the failing test** Create `tests/integration/test_servable_neighbours.py`: ```python import pytest from app.services.infra.database import bootstrap_tenant, get_db_connection from app.services.pairing.decisions import APPROVAL_THRESHOLD, record from app.services.pairing.queries import SERVED_PAIR_TYPES, servable_neighbours def _exec(schema, statement, params=None): conn = get_db_connection() try: with conn.cursor() as cur: cur.execute(statement.replace("{S}", f'"{schema}"'), params or ()) conn.commit() finally: conn.close() def _product(schema, key, name, in_stock=True, missing=None): _exec(schema, "INSERT INTO {S}.strategist_products (product_key, name, product_url, " "price_cents, currency, in_stock, missing_fields) " "VALUES (%s,%s,%s,1000,'USD',%s,%s)", (key, name, f"https://x.example/{key}", in_stock, missing or [])) def _pair(schema, anchor, neighbor, pair_type="similar", score=0.9, confidence=0.9): _exec(schema, "INSERT INTO {S}.strategist_product_neighbors (anchor_key, " "neighbor_key, pair_type, score, confidence, source, reasons) " "VALUES (%s,%s,%s,%s,%s,'embedding','[]')", (anchor, neighbor, pair_type, score, confidence)) @pytest.fixture def graph(temp_tenant): bootstrap_tenant(temp_tenant) _product(temp_tenant, "a", "Anchor") _product(temp_tenant, "b", "Servable Neighbour") _pair(temp_tenant, "a", "b") return temp_tenant def _keys(rows): return [r["product_key"] for r in rows] def test_a_servable_neighbour_is_returned(graph): rows = servable_neighbours(graph, "a") assert _keys(rows) == ["b"] assert rows[0]["pair_type"] == "similar" assert rows[0]["pair_score"] == pytest.approx(0.9) def test_a_rejected_pair_is_never_served(graph): # THE test of this phase. The approval queue means nothing if serving # ignores it. record(graph, [{"anchor_key": "a", "neighbor_key": "b", "pair_type": "similar", "decision": "rejected"}]) assert servable_neighbours(graph, "a") == [] def test_an_unapproved_low_confidence_pair_is_never_served(temp_tenant): bootstrap_tenant(temp_tenant) _product(temp_tenant, "a", "Anchor") _product(temp_tenant, "b", "Unsure Neighbour") _pair(temp_tenant, "a", "b", confidence=APPROVAL_THRESHOLD - 0.2) assert servable_neighbours(temp_tenant, "a") == [] def test_an_approved_low_confidence_pair_is_served(temp_tenant): bootstrap_tenant(temp_tenant) _product(temp_tenant, "a", "Anchor") _product(temp_tenant, "b", "Approved Neighbour") _pair(temp_tenant, "a", "b", confidence=APPROVAL_THRESHOLD - 0.2) record(temp_tenant, [{"anchor_key": "a", "neighbor_key": "b", "pair_type": "similar", "decision": "approved"}]) assert _keys(servable_neighbours(temp_tenant, "a")) == ["b"] def test_an_out_of_stock_neighbour_is_never_served(temp_tenant): # Checked now rather than at pairing time: a product can sell out an hour # after the graph was built. bootstrap_tenant(temp_tenant) _product(temp_tenant, "a", "Anchor") _product(temp_tenant, "b", "Sold Out", in_stock=False) _pair(temp_tenant, "a", "b") assert servable_neighbours(temp_tenant, "a") == [] def test_a_flagged_neighbour_is_never_served(temp_tenant): bootstrap_tenant(temp_tenant) _product(temp_tenant, "a", "Anchor") _product(temp_tenant, "b", "No URL", missing=["product_url"]) _pair(temp_tenant, "a", "b") assert servable_neighbours(temp_tenant, "a") == [] def test_only_the_served_types_come_back(temp_tenant): bootstrap_tenant(temp_tenant) _product(temp_tenant, "a", "Anchor") for i, pair_type in enumerate(("similar", "complement", "upsell", "bundle")): _product(temp_tenant, f"n{i}", f"Neighbour {i}") _pair(temp_tenant, "a", f"n{i}", pair_type=pair_type) types = {r["pair_type"] for r in servable_neighbours(temp_tenant, "a", limit=10)} assert types == set(SERVED_PAIR_TYPES) assert "bundle" not in types assert "upsell" not in types def test_results_are_ordered_by_score(temp_tenant): bootstrap_tenant(temp_tenant) _product(temp_tenant, "a", "Anchor") for key, score in (("low", 0.7), ("high", 0.95), ("mid", 0.8)): _product(temp_tenant, key, key) _pair(temp_tenant, "a", key, score=score) assert _keys(servable_neighbours(temp_tenant, "a", limit=3)) == \ ["high", "mid", "low"] def test_ordering_is_stable_when_scores_tie(temp_tenant): # An unstable order reads as a bug and makes Phase 5's measurement noisier. bootstrap_tenant(temp_tenant) _product(temp_tenant, "a", "Anchor") for key in ("n1", "n2", "n3"): _product(temp_tenant, key, key) _pair(temp_tenant, "a", key, score=0.8) assert _keys(servable_neighbours(temp_tenant, "a", limit=3)) == \ _keys(servable_neighbours(temp_tenant, "a", limit=3)) def test_the_limit_is_respected(temp_tenant): bootstrap_tenant(temp_tenant) _product(temp_tenant, "a", "Anchor") for i in range(8): _product(temp_tenant, f"n{i}", f"N{i}") _pair(temp_tenant, "a", f"n{i}", score=0.9 - i * 0.01) assert len(servable_neighbours(temp_tenant, "a", limit=3)) == 3 def test_an_anchor_with_no_graph_returns_nothing(temp_tenant): bootstrap_tenant(temp_tenant) _product(temp_tenant, "a", "Anchor") assert servable_neighbours(temp_tenant, "a") == [] def test_a_missing_neighbour_row_is_dropped(temp_tenant): # The catalogue changes between pairing runs. bootstrap_tenant(temp_tenant) _product(temp_tenant, "a", "Anchor") _pair(temp_tenant, "a", "ghost") assert servable_neighbours(temp_tenant, "a") == [] ``` - [ ] **Step 2: Run test to verify it fails** Run: `.venv/bin/python -m pytest tests/integration/test_servable_neighbours.py -v` Expected: FAIL — `servable_neighbours` does not exist. - [ ] **Step 3: Write the implementation** Add to `app/services/pairing/queries.py`: ```python # Upsell and bundle are computed and visible in the admin tool but never served. # Upsell only makes sense when a shopper is comparing, and the live review found # the surviving bundles were lexical coincidences ("Pea" matching "Peeler"). SERVED_PAIR_TYPES = ("similar", "complement") def servable_neighbours(tenant_id: str, product_key: str, limit: int = 3) -> list: """Whole product rows a shopper may be shown for this anchor. Stock and missing_fields are filtered HERE rather than trusted from pairing time, because a product can sell out an hour after the graph was built. """ conn = get_db_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(sql.SQL(""" SELECT p.*, n.pair_type, n.score AS pair_score, n.confidence AS pair_confidence, n.source AS pair_source, n.neighbor_key, n.anchor_key FROM {}.strategist_product_neighbors n JOIN {}.strategist_products p ON p.product_key = n.neighbor_key WHERE n.anchor_key = %s AND n.pair_type = ANY(%s) AND p.in_stock AND COALESCE(array_length(p.missing_fields, 1), 0) = 0 ORDER BY n.score DESC, n.neighbor_key """).format(sql.Identifier(tenant_id), sql.Identifier(tenant_id)), (product_key, list(SERVED_PAIR_TYPES))) rows = [dict(r) for r in cur.fetchall()] finally: conn.close() decisions = load_decisions(tenant_id) servable = [] for row in rows: pair = {"confidence": row["pair_confidence"], "source": row["pair_source"]} decision = decisions.get( (row["anchor_key"], row["neighbor_key"], row["pair_type"])) if not is_servable(pair, decision): continue servable.append(row) if len(servable) >= limit: break return servable ``` Note `ORDER BY n.score DESC, n.neighbor_key` — the second key is what makes the order stable when scores tie. - [ ] **Step 4: Run test to verify it passes** Run: `.venv/bin/python -m pytest tests/integration/test_servable_neighbours.py -v` Expected: PASS, 12 tests - [ ] **Step 5: Report changed files** ```bash git status --short ``` Do not commit and do not stage. --- ### Task 2: Serve the graph **Files:** - Modify: `app/services/pairing/recommendations.py` - Test: `tests/unit/test_recommendations_graph.py` **Interfaces:** - Consumes: `servable_neighbours`, `SERVED_PAIR_TYPES`, plus the existing `find_by_url`, `match_products`, `related_products`, `to_card`, `get_tool_settings`. - Produces: `decide()` unchanged in signature and contract; each returned card gains `pair_type` and `pair_score`. **The whole change is one branch.** Where `decide()` calls `related_products`, it now calls `servable_neighbours` and falls back to `related_products` when the graph has nothing. Everything else in the function stays. - [ ] **Step 1: Write the failing test** Create `tests/unit/test_recommendations_graph.py`: ```python import pytest import app.services.pairing.recommendations as mod TENANT = "org_test" PAGE = {"url": "https://shop.example/p/anchor", "dwell_seconds": 60} ANCHOR = {"product_key": "a", "name": "Anchor", "product_url": PAGE["url"], "ctas": [], "options": []} def _neighbour(key, pair_type="similar", score=0.9): return {"product_key": key, "name": key.title(), "ctas": [], "options": [], "product_url": f"https://shop.example/p/{key}", "pair_type": pair_type, "pair_score": score} @pytest.fixture(autouse=True) def wiring(monkeypatch): monkeypatch.setattr(mod, "get_tool_settings", lambda t: {"features": {"product_recommendation": True}}) monkeypatch.setattr(mod, "find_by_url", lambda t, u: dict(ANCHOR)) monkeypatch.setattr(mod, "related_products", lambda t, row, limit=3: []) monkeypatch.setattr(mod, "servable_neighbours", lambda t, key, limit=3: []) def test_graph_neighbours_are_served(monkeypatch): monkeypatch.setattr(mod, "servable_neighbours", lambda t, key, limit=3: [_neighbour("b"), _neighbour("c")]) result = mod.decide(TENANT, "dwell", PAGE) assert result["recommend"] is True assert [p["product_key"] for p in result["products"]] == ["b", "c"] def test_every_card_carries_its_attribution(monkeypatch): # Phase 5 cannot measure which pair types earn their place without this. monkeypatch.setattr(mod, "servable_neighbours", lambda t, key, limit=3: [_neighbour("b", "complement", 0.77)]) card = mod.decide(TENANT, "dwell", PAGE)["products"][0] assert card["pair_type"] == "complement" assert card["pair_score"] == pytest.approx(0.77) def test_the_anchor_is_never_recommended_to_itself(monkeypatch): monkeypatch.setattr(mod, "servable_neighbours", lambda t, key, limit=3: [_neighbour("a"), _neighbour("b")]) keys = [p["product_key"] for p in mod.decide(TENANT, "dwell", PAGE)["products"]] assert "a" not in keys def test_at_most_three_cards(monkeypatch): monkeypatch.setattr(mod, "servable_neighbours", lambda t, key, limit=3: [_neighbour(k) for k in "bcdef"]) assert len(mod.decide(TENANT, "dwell", PAGE)["products"]) <= 3 def test_an_empty_graph_falls_back_to_related_keys(monkeypatch): # A tenant who has never run pairing must be no worse off than today. monkeypatch.setattr(mod, "servable_neighbours", lambda t, key, limit=3: []) monkeypatch.setattr(mod, "related_products", lambda t, row, limit=3: [_neighbour("legacy")]) keys = [p["product_key"] for p in mod.decide(TENANT, "dwell", PAGE)["products"]] assert keys == ["legacy"] def test_the_graph_wins_when_both_exist(monkeypatch): monkeypatch.setattr(mod, "servable_neighbours", lambda t, key, limit=3: [_neighbour("graph")]) monkeypatch.setattr(mod, "related_products", lambda t, row, limit=3: [_neighbour("legacy")]) keys = [p["product_key"] for p in mod.decide(TENANT, "dwell", PAGE)["products"]] assert keys == ["graph"] def test_neither_source_means_no_recommendation(): result = mod.decide(TENANT, "dwell", PAGE) assert result["recommend"] is False assert result["reason"] == "no_match" def test_a_failing_lookup_never_raises(monkeypatch): def boom(*a, **kw): raise RuntimeError("database down") monkeypatch.setattr(mod, "servable_neighbours", boom) result = mod.decide(TENANT, "dwell", PAGE) assert result["recommend"] is False def test_short_dwell_still_recommends_nothing(monkeypatch): monkeypatch.setattr(mod, "servable_neighbours", lambda t, key, limit=3: [_neighbour("b")]) result = mod.decide(TENANT, "dwell", {"url": PAGE["url"], "dwell_seconds": 5}) assert result["recommend"] is False def test_the_feature_flag_still_wins(monkeypatch): monkeypatch.setattr(mod, "get_tool_settings", lambda t: {"features": {"product_recommendation": False}}) monkeypatch.setattr(mod, "servable_neighbours", lambda t, key, limit=3: [_neighbour("b")]) assert mod.decide(TENANT, "dwell", PAGE)["reason"] == "feature_off" def test_the_search_branch_still_uses_text_matching(monkeypatch): # Not the embedding search: that spends ai_tokens on an unauthenticated # endpoint firing on every page view. called = {} monkeypatch.setattr(mod, "match_products", lambda t, q, limit=4: called.setdefault("q", q) or [_neighbour("hit")]) monkeypatch.setattr(mod, "servable_neighbours", lambda t, key, limit=3: [_neighbour("graph")]) result = mod.decide(TENANT, "search", {"url": "https://shop.example/search?q=boots"}) assert called["q"] == "boots" assert [p["product_key"] for p in result["products"]] == ["hit"] def test_nothing_reachable_from_decide_calls_a_model(monkeypatch): # The hard constraint of this whole path: it runs on every page view of # every visitor on an unauthenticated endpoint. import app.core.llm_client as llm def forbidden(*a, **kw): raise AssertionError("decide() must never call a model") monkeypatch.setattr(llm.client.chat.completions, "create", forbidden, raising=False) monkeypatch.setattr(llm.client.embeddings, "create", forbidden, raising=False) monkeypatch.setattr(mod, "servable_neighbours", lambda t, key, limit=3: [_neighbour("b")]) assert mod.decide(TENANT, "dwell", PAGE)["recommend"] is True ``` - [ ] **Step 2: Run test to verify it fails** Run: `.venv/bin/python -m pytest tests/unit/test_recommendations_graph.py -v` Expected: FAIL — `servable_neighbours` is not imported into the module. - [ ] **Step 3: Write the implementation** In `app/services/pairing/recommendations.py`, add the import: ```python from app.services.pairing.queries import servable_neighbours ``` Replace the `else` branch that calls `related_products` with: ```python else: # The graph first; related_keys only for tenants who have never run # pairing, so this change is safe to deploy before every tenant has # been paired and rollback is one constant rather than a revert. candidates = servable_neighbours(tenant_id, current["product_key"], limit=4) if not candidates: candidates = related_products(tenant_id, current, limit=4) ``` And where the cards are built, carry the attribution through: ```python return {"recommend": True, "message": MESSAGES.get(event, MESSAGES["dwell"]), "products": [_attributed(r) for r in others]} ``` with, near the other helpers: ```python def _attributed(row: dict) -> dict: """to_card is shared with the chat path and must not learn about pairing, so the attribution is attached here. Phase 5 needs it to work out which pair types earn their place.""" card = to_card(row) if row.get("pair_type"): card["pair_type"] = row["pair_type"] card["pair_score"] = row.get("pair_score") return card ``` Update the module docstring: the paragraph describing `related_products` as the source of neighbours is now wrong, and the docstring is the first thing the next person reads. Say that the graph is the source, that `related_keys` is the fallback, and keep every existing sentence about not calling a model. - [ ] **Step 4: Run test to verify it passes** Run: `.venv/bin/python -m pytest tests/unit/test_recommendations_graph.py -v` Expected: PASS, 12 tests - [ ] **Step 5: Check the existing recommendation tests still hold** ```bash .venv/bin/python -m pytest tests/unit/test_recommendation_isolation.py tests/unit/test_recommendations.py -v ``` These patch `app.services.pairing.recommendations.find_by_url` and friends by string. If one fails because it asserted on `related_products` being the source, that is a real contract change and it must be updated deliberately — read the test, decide whether the new behaviour is correct, and update the test rather than weakening the assertion. Report any test you changed and why. - [ ] **Step 6: Run both suites** Run `.venv/bin/python -m pytest tests/unit/ -q`, then `.venv/bin/python -m pytest tests/integration/ -q`, sequentially in the foreground. Expected: no regressions against 608 / 121. - [ ] **Step 7: Report changed files** ```bash git status --short ``` Do not commit and do not stage. --- ### Task 3: Live verification **Files:** none — verification only. **Prerequisites:** uvicorn on port 8001, the live tenant with 1,888 pairs over 201 anchors. - [ ] **Step 1: Record what the widget can do today** ```bash PYTHONPATH=. .venv/bin/python -c " from app.services.infra.database import get_db_connection T='org_8c32bf3e-6a18-4739-9b1c-94c0cf11125f' c=get_db_connection(); cur=c.cursor() cur.execute('SELECT count(*) FROM \"'+T+'\".strategist_products WHERE array_length(related_keys,1) > 0') print('coverage via related_keys:', cur.fetchone()[0]) cur.execute('SELECT count(DISTINCT anchor_key) FROM \"'+T+'\".strategist_product_neighbors') print('coverage via the graph:', cur.fetchone()[0]); c.close()" ``` Expected: 7 and 201. That gap is the phase. - [ ] **Step 2: Ask for a real recommendation** Start uvicorn on 8001, then pick a real smartphone URL from the catalogue and post a `dwell` event: ```bash PYTHONPATH=. .venv/bin/python -c " from app.services.pairing.queries import list_products p = list_products('org_8c32bf3e-6a18-4739-9b1c-94c0cf11125f', category='smartphones', limit=1)['products'][0] print(p['product_key'])" ``` Then fetch that product's `product_url` and post: ```bash curl -s -X POST http://localhost:8001/recommendations/event \ -H 'Content-Type: application/json' \ -d '{"tenant_id":"org_8c32bf3e-6a18-4739-9b1c-94c0cf11125f", "visitor_id":"live-check","event":"dwell", "page":{"url":"","dwell_seconds":60}}' \ | python3 -m json.tool ``` Expected: `recommend: true` with up to three cards, each carrying `pair_type` and `pair_score`. **Read the product names and say whether a shopper would find them sensible** — that judgement is the deliverable, not the status code. - [ ] **Step 3: Prove the approval rule reaches the shopper** Take one `neighbor_key` from the response and reject it: ```bash curl -s -X POST http://localhost:8001/catalog/pairings/decide \ -H 'Content-Type: application/json' \ -d '{"tenant_id":"org_8c32bf3e-6a18-4739-9b1c-94c0cf11125f", "decisions":[{"anchor_key":"","neighbor_key":"", "pair_type":"","decision":"rejected", "decided_by":"live-check"}]}' ``` Re-post the same `dwell` event **without re-running the pairing job**. That product must be gone from the response. This is the phase's reason for existing: a merchant's rejection has to reach the shopper immediately, not after the next pairing run. - [ ] **Step 4: Confirm an out-of-stock product cannot be served** Pick a served neighbour, set `in_stock = false` directly in the database, re-post the event, and confirm it disappears. Set it back afterwards and say so in the report. - [ ] **Step 5: Confirm the cost** Time ten consecutive `dwell` events against the same URL. Report the median. It must be tens of milliseconds; this endpoint fires on every page view of every visitor. Also confirm from the uvicorn log that no embedding or chat request was made during those ten calls. - [ ] **Step 6: Report** Summarise: changed files, test counts, coverage before and after, the actual product names a shopper would be shown and whether they are sensible, proof that the rejection took effect immediately, proof that out-of-stock is filtered, and the median latency. Remind the user to commit. --- ## Follow-on work (not in this plan) - **Phase 5:** measurement — impressions and clicks per `pair_type`, which the attribution field added here makes possible, and which is what would let the approval threshold be tuned on evidence. - **Phase 6:** per-variant stock, accessory-target extraction, colour coverage, order history. - **Per-event mapping:** if measurement shows that serving alternatives after `add_to_cart` suppresses conversion, §3 of the spec is where to start. - **Cooldown:** a visitor moving through several pages is suggested something on each one. True today, unchanged here, and belongs with measurement.