test(search): quality guard so the ranking layer cannot silently rot

Extends search-stack-visibility with a ranking assertion: for the fixed query
set, no config demote_domains host may appear in the top 3, and the known
non-answers (bestbuy.com, merriam-webster.com) must not be returned at all.

Without this the layer could rot back to raw engine ordering unnoticed - the same
way the endpoint colours silently rotted before 2026-09-26. It reads the demote
list from the SAME config the layer uses, so the guard cannot drift from the
policy it is guarding.

Live: 'ok: no demoted host in the top 3; no banned non-answer returned';
visibility contract still PASSES end to end.

scripts/search-stack-check.py

@@ -114,6 +114,79 @@ def unresponsive_names(pairs: list) -> dict[str, str]:
     return out
 
 
+# ── QUALITY GUARD (search-agent-consumption) ─────────────────────────────────
+# The agent-consumption layer applies a deterministic demote/drop policy. Without
+# an assertion here it could silently rot back to raw engine ordering - the same
+# way the endpoint colours silently rotted before 2026-09-26.
+QUALITY_QUERIES = [
+    "best practices agent context management",
+    "proxmox thin pool metadata exhaustion recovery",
+]
+# A demoted (content-farm) host must never occupy the top 3 for these queries.
+QUALITY_TOP_N = 3
+# Non-answers that must never be returned for these queries at all.
+QUALITY_BANNED_HOSTS = ["bestbuy.com", "merriam-webster.com"]
+
+
+def _consumption_layer_path():
+    here = os.path.dirname(os.path.abspath(__file__))
+    return os.path.join(here, "search-agent-consume.py")
+
+
+def check_ranking_quality() -> list[str]:
+    """Return a list of quality failures; empty means healthy."""
+    import subprocess as _sp
+
+    layer = _consumption_layer_path()
+    if not os.path.exists(layer):
+        return [f"agent-consumption layer missing: {layer}"]
+
+    failures: list[str] = []
+    for query in QUALITY_QUERIES:
+        r = _sp.run([sys.executable, layer, "--no-extract", "--explain", query],
+                    capture_output=True, text=True, timeout=120)
+        if r.returncode != 0:
+            failures.append(f"{query!r}: layer exited {r.returncode} ({r.stderr[:120]})")
+            continue
+        try:
+            data = json.loads(r.stdout)
+        except json.JSONDecodeError:
+            failures.append(f"{query!r}: layer returned unparseable JSON")
+            continue
+
+        results = data.get("results", [])
+        if len(results) < QUALITY_TOP_N:
+            failures.append(f"{query!r}: only {len(results)} results returned")
+            continue
+
+        # load the demote list from the SAME config the layer uses
+        cfg_path = os.path.join(os.path.dirname(layer), "..", "config", "search-ranking.yaml")
+        demoted: set[str] = set()
+        try:
+            sys.path.insert(0, os.path.dirname(layer))
+            import importlib.util as _iu
+            spec = _iu.spec_from_file_location("_sac_cfg", layer)
+            mod = _iu.module_from_spec(spec)
+            spec.loader.exec_module(mod)
+            demoted = set(mod._load_config().get("demote_domains", []) or [])
+        except Exception:  # noqa: BLE001
+            failures.append(f"{query!r}: could not load demote_domains from config")
+
+        for item in results[:QUALITY_TOP_N]:
+            host = (item.get("host") or "")
+            for d in demoted:
+                if host == d or host.endswith("." + d):
+                    failures.append(
+                        f"{query!r}: demoted host {host} in top {QUALITY_TOP_N}"
+                    )
+        for item in results:
+            host = (item.get("host") or "")
+            for b in QUALITY_BANNED_HOSTS:
+                if host == b or host.endswith("." + b):
+                    failures.append(f"{query!r}: non-answer host {host} returned")
+    return failures
+
+
 def main() -> int:
     failures: list[str] = []
     print(f"Search stack check -- {SEARXNG_URL}")
@@ -216,6 +289,16 @@ def main() -> int:
         print(f"  FAIL: {msg}")
         failures.append(msg)
 
+    print("-" * 72)
+    print("RANKING QUALITY (agent-consumption layer)")
+    quality = check_ranking_quality()
+    if quality:
+        for q in quality:
+            print(f"  FAIL: {q}")
+        failures.extend(quality)
+    else:
+        print("  ok: no demoted host in the top 3; no banned non-answer returned")
+
     print("=" * 72)
     if failures:
         print("VERDICT: FAIL")