feat: add featured/locked columns and auto-rotation logic to news pipeline

- Add `ai_grade`, `featured`, `featured_locked`, `landing_locked`, `author`, `article_published`, `image_url`, `has_unique_image` columns to news table
- Extend `news_images` schema with `alt_text`, `phash`, `width`, `height`, `is_placeholder` columns
- Create `idx_news_featured` index for efficient featured queries
- Update admin toggle endpoints to set `featured_locked`/`landing_locked` when manually toggling
- Propagate `featured` and `image_url` fields through landing page, news list, and detail page rendering
- Update `get_featured_news` to return `featured` and `image_url` fields
- Document in AGENTS.md the full zero-maintenance pipeline: image perceptual hashing, placeholder detection, AI grading on cleaned text, reliability gate, effective score computation, and post-loop landing rotation
- Update README.md service description to reflect automatic image comparison and landing rotation capabilities
- Clarify docs_api.py summaries that toggling featured/landing now locks articles from auto-rotation

tests/api/admin/aiusage/index.py

@@ -0,0 +1,41 @@
+# retoor <retoor@molodetz.nl>
+
+import time
+import requests
+from tests.conftest import BASE_URL
+from devplacepy.database import get_table, refresh_snapshot
+_counter_aiusage = [0]
+JSON_aiusage = {"Accept": "application/json"}
+def _db_user(name):
+    refresh_snapshot()
+    return get_table("users").find_one(username=name)
+def _admin_aiusage(seeded_db):
+    key = _db_user("alice_test")["api_key"]
+    s = requests.Session()
+    s.headers.update({"X-API-KEY": key})
+    return s
+
+
+def test_admin_ai_usage_page_requires_admin(app_server):
+    r = requests.get(f"{BASE_URL}/admin/ai-usage", allow_redirects=False)
+    assert r.status_code in (302, 401, 403)
+
+
+def test_admin_ai_usage_page_returns_html(app_server, seeded_db):
+    admin = _admin_aiusage(seeded_db)
+    r = admin.get(f"{BASE_URL}/admin/ai-usage")
+    assert r.status_code == 200
+    assert "AI usage" in r.text or "ai-usage" in r.text
+
+
+def test_admin_ai_usage_data_requires_auth(app_server):
+    r = requests.get(f"{BASE_URL}/admin/ai-usage/data", headers=JSON_aiusage)
+    assert r.status_code == 401
+
+
+def test_admin_ai_usage_data_returns_json(app_server, seeded_db):
+    admin = _admin_aiusage(seeded_db)
+    r = admin.get(f"{BASE_URL}/admin/ai-usage/data", headers=JSON_aiusage)
+    assert r.status_code == 200
+    data = r.json()
+    assert isinstance(data, dict)

tests/api/admin/gateway/__init__.py

@@ -0,0 +1 @@
+# retoor <retoor@molodetz.nl>

tests/api/admin/gateway/index.py

@@ -0,0 +1,69 @@
+# retoor <retoor@molodetz.nl>
+
+import time
+
+import requests
+
+from devplacepy.database import get_table, refresh_snapshot
+from tests.conftest import BASE_URL
+
+JSON_gateway = {"Accept": "application/json"}
+_counter_gateway = [0]
+
+
+def _db_user_gateway(name):
+    refresh_snapshot()
+    return get_table("users").find_one(username=name)
+
+
+def _unique_gateway(prefix="gw"):
+    _counter_gateway[0] += 1
+    return f"{prefix}{int(time.time() * 1000)}{_counter_gateway[0]}"
+
+
+def admin_session(seeded_db):
+    session = requests.Session()
+    session.headers.update(
+        {"X-API-KEY": _db_user_gateway("alice_test")["api_key"], **JSON_gateway}
+    )
+    return session
+
+
+def member_key():
+    name = _unique_gateway("gwmem")
+    requests.post(
+        f"{BASE_URL}/auth/signup",
+        data={
+            "username": name,
+            "email": f"{name}@t.dev",
+            "password": "secret123",
+            "confirm_password": "secret123",
+        },
+        allow_redirects=True,
+    )
+    return _db_user_gateway(name)["api_key"]
+
+
+def test_gateway_page_requires_admin(seeded_db):
+    assert (
+        requests.get(
+            f"{BASE_URL}/admin/gateway", headers=JSON_gateway, allow_redirects=False
+        ).status_code
+        == 401
+    )
+    key = member_key()
+    assert (
+        requests.get(
+            f"{BASE_URL}/admin/gateway",
+            headers={**JSON_gateway, "X-API-KEY": key},
+            allow_redirects=False,
+        ).status_code
+        == 403
+    )
+
+
+def test_gateway_page_renders_for_admin(seeded_db):
+    admin = admin_session(seeded_db)
+    response = admin.get(f"{BASE_URL}/admin/gateway", headers={})
+    assert response.status_code == 200
+    assert "Gateway routing" in response.text

tests/api/admin/gateway/models.py

@@ -0,0 +1,85 @@
+# retoor <retoor@molodetz.nl>
+
+import requests
+
+from tests.conftest import BASE_URL
+from tests.api.admin.gateway.index import (
+    JSON_gateway,
+    admin_session,
+    member_key,
+    _unique_gateway,
+)
+
+
+def test_models_require_admin(seeded_db):
+    assert (
+        requests.get(
+            f"{BASE_URL}/admin/gateway/models",
+            headers=JSON_gateway,
+            allow_redirects=False,
+        ).status_code
+        == 401
+    )
+    key = member_key()
+    assert (
+        requests.get(
+            f"{BASE_URL}/admin/gateway/models",
+            headers={**JSON_gateway, "X-API-KEY": key},
+            allow_redirects=False,
+        ).status_code
+        == 403
+    )
+    assert (
+        admin_session(seeded_db).get(f"{BASE_URL}/admin/gateway/models").status_code
+        == 200
+    )
+
+
+def test_model_route_create_and_economy(seeded_db):
+    admin = admin_session(seeded_db)
+    source = _unique_gateway("src")
+
+    created = admin.post(
+        f"{BASE_URL}/admin/gateway/models",
+        json={
+            "source_model": source,
+            "provider": "",
+            "target_model": "vendor/y",
+            "kind": "chat",
+            "price_output_per_m": 5.0,
+            "price_cache_miss_per_m": 2.0,
+            "context_window": 32000,
+        },
+    )
+    assert created.status_code == 200, created.text[:300]
+    assert created.json()["model"]["target_model"] == "vendor/y"
+
+    listed = admin.get(f"{BASE_URL}/admin/gateway/models").json()
+    row = next(m for m in listed["models"] if m["source_model"] == source)
+    assert row["kind"] == "chat"
+    assert row["price_output_per_m"] == 5.0
+    assert row["context_window"] == 32000
+
+    deleted = admin.delete(f"{BASE_URL}/admin/gateway/models/{source}")
+    assert deleted.status_code == 200
+    assert admin.delete(f"{BASE_URL}/admin/gateway/models/{source}").status_code == 404
+
+
+def test_model_route_validation(seeded_db):
+    admin = admin_session(seeded_db)
+    missing_target = admin.post(
+        f"{BASE_URL}/admin/gateway/models",
+        json={"source_model": _unique_gateway("src")},
+    )
+    assert missing_target.status_code == 400
+    assert missing_target.json()["ok"] is False
+
+    bad_kind = admin.post(
+        f"{BASE_URL}/admin/gateway/models",
+        json={
+            "source_model": _unique_gateway("src"),
+            "target_model": "vendor/z",
+            "kind": "bogus",
+        },
+    )
+    assert bad_kind.status_code == 400

tests/api/admin/gateway/providers.py

@@ -0,0 +1,84 @@
+# retoor <retoor@molodetz.nl>
+
+import requests
+
+from tests.conftest import BASE_URL
+from tests.api.admin.gateway.index import (
+    JSON_gateway,
+    admin_session,
+    member_key,
+    _unique_gateway,
+)
+
+
+def test_providers_require_admin(seeded_db):
+    assert (
+        requests.get(
+            f"{BASE_URL}/admin/gateway/providers",
+            headers=JSON_gateway,
+            allow_redirects=False,
+        ).status_code
+        == 401
+    )
+    key = member_key()
+    assert (
+        requests.get(
+            f"{BASE_URL}/admin/gateway/providers",
+            headers={**JSON_gateway, "X-API-KEY": key},
+            allow_redirects=False,
+        ).status_code
+        == 403
+    )
+    assert (
+        admin_session(seeded_db)
+        .get(f"{BASE_URL}/admin/gateway/providers")
+        .status_code
+        == 200
+    )
+
+
+def test_provider_create_update_delete(seeded_db):
+    admin = admin_session(seeded_db)
+    name = _unique_gateway("prov").lower()
+
+    created = admin.post(
+        f"{BASE_URL}/admin/gateway/providers",
+        json={
+            "name": name,
+            "base_url": "https://x.example/v1/chat/completions",
+            "api_key": "sk-x",
+            "is_active": True,
+        },
+    )
+    assert created.status_code == 200, created.text[:300]
+    assert created.json()["ok"] is True
+
+    listed = admin.get(f"{BASE_URL}/admin/gateway/providers").json()
+    match = next((p for p in listed["providers"] if p["name"] == name), None)
+    assert match is not None
+    assert match["base_url"] == "https://x.example/v1/chat/completions"
+
+    updated = admin.post(
+        f"{BASE_URL}/admin/gateway/providers",
+        json={"name": name, "base_url": "https://y.example/v1/chat/completions"},
+    )
+    assert updated.status_code == 200
+    relisted = admin.get(f"{BASE_URL}/admin/gateway/providers").json()
+    match = next(p for p in relisted["providers"] if p["name"] == name)
+    assert match["base_url"] == "https://y.example/v1/chat/completions"
+
+    deleted = admin.delete(f"{BASE_URL}/admin/gateway/providers/{name}")
+    assert deleted.status_code == 200 and deleted.json()["ok"] is True
+    assert (
+        admin.delete(f"{BASE_URL}/admin/gateway/providers/{name}").status_code == 404
+    )
+
+
+def test_provider_name_validation(seeded_db):
+    admin = admin_session(seeded_db)
+    bad = admin.post(
+        f"{BASE_URL}/admin/gateway/providers",
+        json={"name": "has spaces!", "base_url": "https://z.example/v1/chat/completions"},
+    )
+    assert bad.status_code == 400
+    assert bad.json()["ok"] is False

tests/e2e/admin/news.py

@@ -3,7 +3,7 @@
 from uuid import uuid4
 from datetime import datetime, timezone
 from tests.conftest import BASE_URL
-from devplacepy.database import get_table
+from devplacepy.database import get_table, refresh_snapshot
 from devplacepy.utils import make_combined_slug
 def seed_admin_news(count=3):
     news_table = get_table("news")
@@ -103,6 +103,19 @@ def test_admin_news_landing_toggle(alice):
         page.locator(toggle_sel).first.get_attribute("class") or ""
     )
     assert is_active != was_active
+    refresh_snapshot()
+    assert get_table("news").count(landing_locked=1) >= 1
+
+
+def test_admin_news_featured_locks_row(alice):
+    page, _ = alice
+    seed_admin_news()
+    page.goto(f"{BASE_URL}/admin/news", wait_until="domcontentloaded")
+    toggle_sel = "form[action$='/toggle'] button.admin-toggle-switch"
+    page.locator(toggle_sel).first.click()
+    page.wait_for_url(f"{BASE_URL}/admin/news", wait_until="domcontentloaded")
+    refresh_snapshot()
+    assert get_table("news").count(featured_locked=1) >= 1
 
 
 def test_admin_news_delete(alice):

tests/unit/services/news.py

@@ -5,10 +5,21 @@ from devplacepy.services import news as news_mod
 from devplacepy.services import base as base_mod
 from devplacepy.services.news import (
     NewsService,
+    ImageCandidate,
+    LANDING_MIN_SCORE,
+    LANDING_MAX,
+    UNIQUE_IMAGE_BONUS,
+    THIN_CONTENT_PENALTY,
     _extract_grade,
     _get_ai_key,
     _get_article_images,
+    _flag_shared_placeholders,
+    _primary_image,
+    clean_news_text,
+    reliability_reason,
+    effective_score,
 )
+from datetime import datetime, timezone
 from devplacepy.database import get_table
 from devplacepy.utils import generate_uid
 from tests.conftest import run_async
@@ -160,10 +171,10 @@ def test_run_once_updates_existing_news_row(local_db, monkeypatch):
     articles = [
         {
             "guid": external_id,
-            "title": "HighArticle",
-            "description": "d",
-            "content": "c",
-            "link": "",
+            "title": "HighArticle deep technical analysis",
+            "description": "d" * 150,
+            "content": "c" * 150,
+            "link": LINK_HIGH,
             "feed_name": "Feed",
             "author": "A",
             "published": "2026-01-01",
@@ -178,7 +189,7 @@ def test_run_once_updates_existing_news_row(local_db, monkeypatch):
     run_async(NewsService().run_once())
 
     row = get_table("news").find_one(uid=existing_uid)
-    assert row["title"] == "HighArticle"
+    assert row["title"] == "HighArticle deep technical analysis"
     assert row["status"] == "published"
     assert get_table("news").count(external_id=external_id) == 1
 
@@ -205,9 +216,9 @@ def test_run_once_publishes_grades_and_is_idempotent(local_db, monkeypatch):
     articles = [
         {
             "guid": g_high,
-            "title": "HighArticle",
-            "description": "d",
-            "content": "c",
+            "title": "HighArticle definitive guide",
+            "description": "d" * 250,
+            "content": "c" * 250,
             "link": LINK_HIGH,
             "feed_name": "Feed",
             "author": "A",
@@ -215,9 +226,9 @@ def test_run_once_publishes_grades_and_is_idempotent(local_db, monkeypatch):
         },
         {
             "guid": g_low,
-            "title": "LowArticle",
-            "description": "d",
-            "content": "c",
+            "title": "LowArticle filler thoughts",
+            "description": "d" * 250,
+            "content": "c" * 250,
             "link": LINK_LOW,
             "feed_name": "Feed",
             "author": "A",
@@ -225,19 +236,19 @@ def test_run_once_publishes_grades_and_is_idempotent(local_db, monkeypatch):
         },
         {
             "guid": g_fail,
-            "title": "FailArticle",
-            "description": "d",
-            "content": "c",
-            "link": "",
+            "title": "FailArticle broken feed",
+            "description": "d" * 250,
+            "content": "c" * 250,
+            "link": LINK_HIGH,
             "feed_name": "Feed",
             "author": "A",
             "published": "2026-01-01",
         },
         {
             "guid": "",
-            "title": "NoGuid",
-            "description": "d",
-            "content": "c",
+            "title": "NoGuid placeholder entry",
+            "description": "d" * 250,
+            "content": "c" * 250,
             "link": "",
             "feed_name": "Feed",
             "author": "A",
@@ -267,3 +278,195 @@ def test_run_once_publishes_grades_and_is_idempotent(local_db, monkeypatch):
     run_async(NewsService().run_once())
     assert news.count(external_id=g_high) == 1
     assert news.count(external_id=g_fail) == 1
+
+
+def test_clean_news_text_strips_reddit_boilerplate():
+    raw = (
+        "Great article body that is long enough. submitted by /u/someone to "
+        "/r/programming [link] [12 comments]"
+    )
+    cleaned = clean_news_text(raw)
+    assert "submitted by" not in cleaned
+    assert "[link]" not in cleaned
+    assert "comments]" not in cleaned
+    assert "Great article body" in cleaned
+    assert "  " not in cleaned
+
+
+def test_clean_news_text_collapses_whitespace_and_html():
+    assert clean_news_text("<p>hello   world</p>\n\n[link]") == "hello world"
+    assert clean_news_text("") == ""
+
+
+def test_reliability_reason_gate():
+    body = "x" * 300
+    assert reliability_reason("A proper title here", body, "https://a.test") == ""
+    assert reliability_reason("short", body, "https://a.test") == "short_title"
+    assert (
+        reliability_reason("A proper title here", "tiny", "https://a.test")
+        == "thin_body"
+    )
+    assert (
+        reliability_reason("A proper title here", body, "") == "invalid_url"
+    )
+    assert (
+        reliability_reason("A proper title here", body, "ftp://a.test")
+        == "invalid_url"
+    )
+    assert (
+        reliability_reason("THIS IS ALL SHOUTING TITLE", body, "https://a.test")
+        == "shouting_title"
+    )
+
+
+def test_effective_score_bonus_and_penalty():
+    assert effective_score(6, True, False) == 6 + UNIQUE_IMAGE_BONUS
+    assert effective_score(6, False, True) == 6 - THIN_CONTENT_PENALTY
+    assert effective_score(6, False, False) == 6
+    assert effective_score(10, True, False) == 10
+    assert effective_score(1, False, True) == 1
+
+
+def _stripes_phash():
+    from PIL import Image, ImageDraw
+    import imagehash
+
+    image = Image.new("RGB", (200, 200), (255, 255, 255))
+    draw = ImageDraw.Draw(image)
+    for x in range(0, 200, 20):
+        draw.rectangle([x, 0, x + 10, 200], fill=(0, 0, 0))
+    return str(imagehash.phash(image))
+
+
+def _circle_phash():
+    from PIL import Image, ImageDraw
+    import imagehash
+
+    image = Image.new("RGB", (200, 200), (255, 255, 255))
+    draw = ImageDraw.Draw(image)
+    draw.ellipse([40, 40, 160, 160], fill=(0, 0, 0))
+    return str(imagehash.phash(image))
+
+
+def test_flag_shared_placeholders_marks_cross_article_duplicates():
+    shared = _stripes_phash()
+    distinct = _circle_phash()
+    by_article = {
+        "a": [ImageCandidate(url="u1", phash=shared, width=200, height=200, is_placeholder=False)],
+        "b": [ImageCandidate(url="u2", phash=shared, width=200, height=200, is_placeholder=False)],
+        "c": [ImageCandidate(url="u3", phash=distinct, width=200, height=200, is_placeholder=False)],
+    }
+    _flag_shared_placeholders(by_article)
+    assert by_article["a"][0].is_placeholder is True
+    assert by_article["b"][0].is_placeholder is True
+    assert by_article["c"][0].is_placeholder is False
+
+
+def test_flag_shared_placeholders_keeps_same_article_duplicates():
+    shared = _stripes_phash()
+    by_article = {
+        "a": [
+            ImageCandidate(url="u1", phash=shared, width=200, height=200, is_placeholder=False),
+            ImageCandidate(url="u2", phash=shared, width=200, height=200, is_placeholder=False),
+        ]
+    }
+    _flag_shared_placeholders(by_article)
+    assert by_article["a"][0].is_placeholder is False
+    assert by_article["a"][1].is_placeholder is False
+
+
+def test_primary_image_picks_first_non_placeholder():
+    candidates = [
+        ImageCandidate(url="ph", is_placeholder=True),
+        ImageCandidate(url="good", is_placeholder=False),
+    ]
+    assert _primary_image(candidates) == "good"
+    assert _primary_image([ImageCandidate(url="x", is_placeholder=True)]) == ""
+
+
+def test_grade_article_full_unique_image_bonus():
+    service = NewsService()
+    article = {
+        "title": "A solid technical headline",
+        "description": "d" * 250,
+        "content": "c" * 250,
+        "link": "https://example.test/post",
+    }
+    unique = [ImageCandidate(url="img", phash="abc", width=400, height=400, is_placeholder=False)]
+    result = service._grade_article_full(article, 7, unique)
+    assert result.valid is True
+    assert result.has_unique_image is True
+    assert result.image_url == "img"
+    assert result.effective_score == 7 + UNIQUE_IMAGE_BONUS
+
+    placeholder = [ImageCandidate(url="img", is_placeholder=True)]
+    result2 = service._grade_article_full(article, 7, placeholder)
+    assert result2.has_unique_image is False
+    assert result2.effective_score == 7
+
+
+def test_grade_article_full_gate_invalidates():
+    service = NewsService()
+    article = {
+        "title": "tiny",
+        "description": "d",
+        "content": "c",
+        "link": "https://example.test/post",
+    }
+    result = service._grade_article_full(article, 9, [])
+    assert result.valid is False
+    assert result.reject_reason == "short_title"
+
+
+def _insert_landing_row(news_table, *, grade, featured, unique, locked=0, landing=0):
+    uid = generate_uid()
+    news_table.insert(
+        {
+            "deleted_at": None,
+            "deleted_by": None,
+            "uid": uid,
+            "external_id": f"land-{uid}",
+            "slug": uid,
+            "title": "Landing candidate",
+            "status": "published",
+            "grade": grade,
+            "featured": featured,
+            "has_unique_image": unique,
+            "landing_locked": locked,
+            "show_on_landing": landing,
+            "synced_at": datetime.now(timezone.utc).isoformat(),
+        }
+    )
+    return uid
+
+
+def test_apply_landing_selection_promotes_top_and_respects_lock(local_db):
+    news_table = get_table("news")
+    news_table.delete()
+    top = _insert_landing_row(news_table, grade=10, featured=1, unique=1)
+    mid = _insert_landing_row(news_table, grade=LANDING_MIN_SCORE, featured=1, unique=1)
+    too_low = _insert_landing_row(news_table, grade=LANDING_MIN_SCORE - 1, featured=1, unique=1)
+    locked = _insert_landing_row(
+        news_table, grade=10, featured=1, unique=1, locked=1, landing=0
+    )
+
+    NewsService()._apply_landing_selection(news_table, 7)
+
+    assert news_table.find_one(uid=top)["show_on_landing"] == 1
+    assert news_table.find_one(uid=mid)["show_on_landing"] == 1
+    assert news_table.find_one(uid=too_low)["show_on_landing"] == 0
+    assert news_table.find_one(uid=locked)["show_on_landing"] == 0
+
+
+def test_apply_landing_selection_caps_at_landing_max(local_db):
+    news_table = get_table("news")
+    news_table.delete()
+    uids = [
+        _insert_landing_row(news_table, grade=10, featured=1, unique=1)
+        for _ in range(LANDING_MAX + 3)
+    ]
+    NewsService()._apply_landing_selection(news_table, 7)
+    promoted = sum(
+        1 for u in uids if news_table.find_one(uid=u)["show_on_landing"] == 1
+    )
+    assert promoted == LANDING_MAX