feat(reddit): arctic-shift fallback for shreddit listing lanes (#992)

Cherry-picks arctic listing fallback from #960 and keeps failed shreddit lanes honest when arctic recovers only part of the request.

Co-authored-by: technicianofthesacred <technicianofthesacred@users.noreply.github.com>

changelog.d/+reddit-arctic-listing-fallback.added.md

@@ -0,0 +1 @@
+Reddit keyless discovery now falls back to the arctic-shift archive when the shreddit listing partials return nothing — hosts on datacenter egress (where Reddit 403s `/svc/shreddit`) keep scored Reddit discovery, score backfill, and discover-mode listings instead of reporting `auth-failed`.

skills/last30days/scripts/lib/reddit_arctic.py

@@ -16,16 +16,32 @@ count rather than failing the Reddit source.
 
 import sys
 import time
-from typing import Dict, List
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
 
 from . import http
 
 API = "https://arctic-shift.photon-reddit.com/api/posts/ids"
+SEARCH_API = "https://arctic-shift.photon-reddit.com/api/posts/search"
 BATCH = 50          # ids per request
 TIMEOUT = 15
 MAX_BATCHES = 3     # cap total requests per run (bounds latency + rate-limit risk)
 PACE_SECONDS = 0.4  # gap between batches; arctic-shift answers 422 "slow down"
 CACHE_MAX = 4096    # hard size bound so the in-run memo can never grow unbounded
+# Listing-lane knobs. Base limits mirror reddit_listing's DEPTH_LIMITS so callers
+# get the same per-depth volume. The supplement multiplier is applied when the
+# caller requested multiple sorts (top/hot/new) — arctic-shift has no sort lanes,
+# so we fetch more posts to increase the chance of covering what the failed
+# shreddit lanes would have returned.
+#
+# KNOWN LIMITATION: Arctic-shift is recency-only (sort=desc). It has no top/hot/
+# new/rising lanes — failed shreddit sort lanes are supplemented with recent
+# posts, not lane-specific results. This is a fundamental backend constraint.
+_LISTING_DEPTH_LIMITS = {"quick": 10, "default": 25, "deep": 50}
+_LISTING_SUPPLEMENT_MULTIPLIER = 2  # fetch 2x posts when supplementing multi-sort requests
+# Total deadline for listing fetches to prevent unbounded stalls when many
+# subreddits are requested and arctic is slow/unreachable.
+_LISTING_DEADLINE_SECONDS = 45  # ~3 subs at 15s timeout each
 # In-run memo: base36 id -> {score, num_comments}. Module-level so repeated
 # fetch_scores calls within one `/last30days` run (e.g. across subqueries) reuse
 # results, but capped at CACHE_MAX entries (never reached in a normal CLI run).
@@ -90,3 +106,128 @@ def fetch_scores(post_ids: List[str]) -> Dict[str, Dict[str, int]]:
                 _cache[rid] = entry
             out[rid] = entry
     return out
+
+
+def _epoch_to_date(value: Any) -> Optional[str]:
+    """Epoch seconds -> YYYY-MM-DD (UTC), or None on garbage."""
+    try:
+        return datetime.fromtimestamp(int(value), tz=timezone.utc).date().isoformat()
+    except (TypeError, ValueError, OSError):
+        return None
+
+
+def _normalize_listing_row(row: Dict[str, Any], query: str = "") -> Dict[str, Any]:
+    """Normalize an arctic-shift post row to reddit_listing.parse_cards shape.
+
+    Mirrors the shreddit card schema (title/url/score/num_comments/subreddit/
+    created_utc/author/selftext/date/engagement/relevance/metadata.post_id) so
+    reddit_keyless can consume either backend interchangeably.
+    """
+    from .relevance import token_overlap_relevance
+
+    pid = str(row.get("id") or "").removeprefix("t3_")
+    permalink = row.get("permalink") or ""
+    title = row.get("title") or ""
+    try:
+        score = int(row.get("score") or 0)
+    except (TypeError, ValueError):
+        score = 0
+    try:
+        num_comments = int(row.get("num_comments") or 0)
+    except (TypeError, ValueError):
+        num_comments = 0
+    author = row.get("author") or "[deleted]"
+    if author in ("[deleted]", "[removed]"):
+        author = "[deleted]"
+    url = f"https://www.reddit.com{permalink}" if permalink.startswith("/") else (permalink or "")
+    return {
+        "id": "",
+        "title": title,
+        "url": url,
+        "score": score,
+        "num_comments": num_comments,
+        "subreddit": row.get("subreddit") or "",
+        "created_utc": row.get("created_utc"),
+        "author": author,
+        "selftext": row.get("selftext") or "",
+        "date": _epoch_to_date(row.get("created_utc")),
+        "engagement": {"score": score, "num_comments": num_comments, "upvote_ratio": None},
+        "relevance": round(token_overlap_relevance(query, title), 3) if query else 0.0,
+        "why_relevant": "Reddit listing (arctic-shift)",
+        "metadata": {"post_id": pid},
+    }
+
+
+def fetch_listings(
+    subreddits: List[str],
+    depth: str = "default",
+    query: str = "",
+    sorts: Optional[List[str]] = None,
+    timeframe: str = "month",
+    limit: Optional[int] = None,
+) -> List[Dict[str, Any]]:
+    """Scored subreddit listings from the arctic-shift archive, keyless.
+
+    Drop-in fallback/supplement for ``reddit_listing.fetch_listings`` (shreddit
+    partials), which datacenter IPs get HTTP 403 on. Arctic-shift serves recent
+    posts with real score/num_comments from any IP.
+
+    Arctic-shift has no top/hot/new lanes, only recency. When ``sorts`` contains
+    multiple entries (e.g., dedicated lanes requesting top+hot+new), we fetch
+    more posts per subreddit to partially compensate for the missing lane
+    coverage — the caller's engagement ranking does the final sorting.
+
+    Best-effort, never raises: returns ``[]`` on any failure.
+    """
+    if not subreddits:
+        return []
+    base = limit or _LISTING_DEPTH_LIMITS.get(depth, _LISTING_DEPTH_LIMITS["default"])
+    # When multiple sorts were requested, fetch more posts to compensate for
+    # arctic-shift's lack of sort lanes.
+    n = base * _LISTING_SUPPLEMENT_MULTIPLIER if sorts and len(sorts) > 1 else base
+    out: List[Dict[str, Any]] = []
+    # Process all requested subreddits with pacing and a total deadline to
+    # prevent unbounded stalls when arctic is slow or unreachable.
+    deadline = time.time() + _LISTING_DEADLINE_SECONDS
+    fetched_count = 0
+    for sub in subreddits:
+        if time.time() >= deadline:
+            _log(f"listing deadline reached after {fetched_count} subs; skipping remaining")
+            break
+        sub = sub.removeprefix("r/").strip()
+        if not sub or sub.lower() == "all":
+            continue
+        if fetched_count:
+            time.sleep(PACE_SECONDS)
+        fetched_count += 1
+        try:
+            # Use retries=1 (single attempt) so retries don't exceed our deadline.
+            # The deadline handles overall timing; per-request retries would
+            # multiply the delay unpredictably.
+            data = http.get(
+                f"{SEARCH_API}?subreddit={sub}&limit={n}&sort=desc",
+                headers={"User-Agent": http.BROWSER_USER_AGENT},
+                timeout=TIMEOUT,
+                retries=1,
+            )
+        except Exception as e:  # network error / non-200 — degrade, never raise
+            _log(f"listing search failed r/{sub}: {e}")
+            continue
+        rows = (data or {}).get("data")
+        if not isinstance(rows, list):
+            _log(f"unexpected listing response for r/{sub}: {str(data)[:80]}")
+            continue
+        for row in rows:
+            if not isinstance(row, dict):
+                continue
+            post = _normalize_listing_row(row, query)
+            if post["url"]:
+                out.append(post)
+
+    seen: set = set()
+    unique: List[Dict[str, Any]] = []
+    for p in out:
+        if p["url"] not in seen:
+            seen.add(p["url"])
+            unique.append(p)
+    return unique

skills/last30days/scripts/lib/reddit_keyless.py

@@ -72,6 +72,51 @@ def _apply_scores(post: Dict[str, Any], scored: Dict[str, int]) -> None:
     post["engagement"]["num_comments"] = scored["num_comments"]
 
 
+def _scored_listings(
+    subreddits: List[str],
+    depth: str = "default",
+    query: str = "",
+    sorts: Optional[List[str]] = None,
+) -> List[Dict[str, Any]]:
+    """Scored subreddit listings: shreddit partials, arctic-shift supplement.
+
+    The shreddit ``community-more-posts`` partials 403 from datacenter IPs
+    (and any host Reddit decides to block). Shreddit is tried first; arctic-
+    shift supplements with any posts shreddit missed. Individual sort lanes
+    can fail silently (shreddit's ``fetch_listings`` flattens results without
+    exposing per-sort status), so arctic is called for all requested subreddits
+    and merged via deduplication. This ensures fresh posts sought through
+    ``hot`` or ``new`` are recovered even when only ``top`` succeeded. Never
+    raises.
+    """
+    posts = reddit_listing.fetch_listings(subreddits, depth=depth, query=query, sorts=sorts)
+
+    # Supplement with arctic for all requested subreddits. Shreddit's per-sort
+    # success/failure is opaque, so arctic provides coverage for any failed
+    # sort lanes (e.g., hot/new failing while top succeeded). Deduplication
+    # ensures no redundant posts when shreddit fully succeeded.
+    if subreddits:
+        try:
+            arctic_posts = reddit_arctic.fetch_listings(
+                subreddits, depth=depth, query=query, sorts=sorts
+            )
+        except Exception as exc:  # the fallback must never break the pipeline
+            _log(f"arctic-shift listing supplement failed: {exc}")
+            arctic_posts = []
+        if arctic_posts:
+            # Merge and dedupe by URL — shreddit posts take priority.
+            seen = {p["url"] for p in posts}
+            added = 0
+            for p in arctic_posts:
+                if p["url"] not in seen:
+                    seen.add(p["url"])
+                    posts.append(p)
+                    added += 1
+            if added:
+                _log(f"arctic-shift supplement: {added} new posts from {len(arctic_posts)} arctic results")
+    return posts
+
+
 def _discover(
     topic: str,
     depth: str,
@@ -83,7 +128,7 @@ def _discover(
     # an on-topic post whose title lacks the entity name is never dropped.
     dedicated_posts: List[Dict[str, Any]] = []
     if dedicated_subreddits:
-        dedicated_posts = reddit_listing.fetch_listings(
+        dedicated_posts = _scored_listings(
             dedicated_subreddits, depth=depth, query=topic, sorts=DEDICATED_SORTS
         )
         for p in dedicated_posts:
@@ -98,7 +143,7 @@ def _discover(
     if subreddits:
         # Targeted run: the caller chose these subreddits, so their listing cards
         # are on-topic — include them as scored discovery AND as a score source.
-        listing_posts = reddit_listing.fetch_listings(subreddits, depth=depth, query=topic)
+        listing_posts = _scored_listings(subreddits, depth=depth, query=topic)
         score_source = listing_posts
     else:
         # Bare global run: subreddits derived from noisy RSS results are NOT
@@ -107,7 +152,7 @@ def _discover(
         # would flood results with high-upvote but irrelevant posts.
         listing_posts = []
         derived = _top_subreddits(rss_posts)
-        score_source = reddit_listing.fetch_listings(derived, depth=depth, query=topic)
+        score_source = _scored_listings(derived, depth=depth, query=topic)
     _log(
         f"Tier 1 (RSS) {len(rss_posts)} posts; "
         f"{'listing discovery ' + str(len(listing_posts)) if subreddits else 'score-only'}; "

skills/last30days/scripts/lib/reddit_listing.py

@@ -18,10 +18,16 @@ import re
 import sys
 from datetime import datetime, timezone
 from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Set
 
 from . import http
-from .relevance import token_overlap_relevance
+from .relevance import token_overlap_relevance, tokenize
+
+# Generic domain terms that are excluded from the keyword gate — matches
+# pipeline._DISCOVERY_GENERIC_DOMAIN_TERMS (duplicated to avoid circular import).
+_DISCOVERY_GENERIC_DOMAIN_TERMS: Set[str] = {
+    "ai", "artificial", "intelligence", "tech", "technology", "trending", "trend",
+}
 
 # Listing sorts pulled per subreddit, by depth.
 LISTING_SORTS = {
@@ -42,6 +48,26 @@ def _log(msg: str) -> None:
     sys.stderr.flush()
 
 
+def _matches_discovery_domain(domain: str, text: str) -> bool:
+    """Require a distinctive domain term, not a generic token such as ``AI``.
+
+    Duplicated from pipeline._matches_discovery_domain to avoid circular imports.
+    The rule must stay in sync: pipeline.py owns the authoritative version and
+    test_reddit_listing.py verifies parity.
+    """
+    def terms(value: str) -> Set[str]:
+        words: Set[str] = set()
+        for word in tokenize(value):
+            words.add(word)
+            if len(word) > 4 and word.endswith("s") and not word.endswith("ss"):
+                words.add(word[:-1])
+        return words
+
+    domain_terms = terms(domain)
+    anchors = domain_terms - _DISCOVERY_GENERIC_DOMAIN_TERMS
+    return bool((anchors or domain_terms) & terms(text))
+
+
 def _attr(tag: str, name: str) -> Optional[str]:
     m = re.search(rf'\b{name}="([^"]*)"', tag)
     return _html.unescape(m.group(1)) if m else None
@@ -73,6 +99,21 @@ def _post_id(permalink: str) -> str:
     return m.group(1) if m else ""
 
 
+_ERROR_PATTERN = re.compile(r"^r/(\S+)\s+(\S+):", re.IGNORECASE)
+
+
+def _shreddit_error_recovered(error: str, successes: Set[tuple[str, str]]) -> bool:
+    """Return True if the error's (sub, sort) pair is in the successes set.
+
+    Error format: "r/{sub} {sort}: {message}".
+    """
+    m = _ERROR_PATTERN.match(error)
+    if not m:
+        return False
+    sub, sort = m.group(1).lower(), m.group(2).lower()
+    return (sub, sort) in successes
+
+
 def parse_cards(html_text: str, query: str = "") -> List[Dict[str, Any]]:
     """Parse <shreddit-post> cards into normalized post dicts with real scores."""
     posts: List[Dict[str, Any]] = []
@@ -211,12 +252,22 @@ def fetch_discovery_listings(
     query: str,
     depth: str = "default",
 ) -> Dict[str, Any]:
-    """Fetch rising/top-week listings while preserving per-feed failures."""
+    """Fetch rising/top-week listings while preserving per-feed failures.
+
+    When shreddit fails and arctic-shift recovers, errors are cleared only for
+    subreddits whose posts survive the keyword gate. If query is empty (global
+    ``--discover`` with no domain), the gate is skipped and any arctic result
+    counts as recovery.
+    """
     if not subreddits:
         return {"items": [], "errors": []}
     jobs = [(subreddit, sort) for subreddit in subreddits for sort in ("rising", "top")]
     items: List[Dict[str, Any]] = []
     errors: List[str] = []
+    # Track which (sub, sort) pairs shreddit successfully delivered posts for.
+    # Used to decide which errors to clear — Arctic can supplement but cannot
+    # "recover" a failed hot/top/new/rising lane (it's recency-only).
+    shreddit_successes: Set[tuple[str, str]] = set()
     with ThreadPoolExecutor(max_workers=min(MAX_WORKERS, len(jobs)) or 1) as executor:
         # submit_with_context, not executor.submit: a plain submit starts the
         # worker with an empty context, dropping the pipeline's
@@ -237,6 +288,9 @@ def fetch_discovery_listings(
             items.extend(fetched)
             if error:
                 errors.append(f"r/{subreddit} {sort}: {error}")
+            elif fetched:
+                # Shreddit succeeded for this (sub, sort) lane.
+                shreddit_successes.add((subreddit.lower(), sort.lower()))
 
     seen: set[str] = set()
     unique = []
@@ -245,6 +299,47 @@ def fetch_discovery_listings(
             continue
         seen.add(item["url"])
         unique.append(item)
+
+    # Supplement with arctic-shift for all requested subreddits. Shreddit's
+    # per-sort success/failure is opaque (individual rising/top lanes can fail
+    # while others succeed), so arctic provides coverage for any failed lanes.
+    # Deduplication ensures no redundant posts when shreddit fully succeeded.
+    from . import reddit_arctic
+    arctic_items = reddit_arctic.fetch_listings(
+        subreddits, depth=depth, query=query, sorts=("rising", "top")
+    )
+    if arctic_items:
+        _log(f"discovery arctic supplement: {len(arctic_items)} posts")
+        # Apply the same keyword gate that pipeline._fetch_discovery_source
+        # uses downstream. When query is empty (global --discover), skip the
+        # gate — there's no keyword to match, and the river feed IS the signal.
+        if query:
+            arctic_items = [
+                item for item in arctic_items
+                if _matches_discovery_domain(
+                    query,
+                    f"{item.get('title') or ''} {item.get('selftext') or ''}",
+                )
+            ]
+        # Merge arctic items into unique list, deduping by URL.
+        added = 0
+        for item in arctic_items:
+            if item["url"] not in seen:
+                seen.add(item["url"])
+                unique.append(item)
+                added += 1
+        if added:
+            _log(f"discovery arctic supplement added {added} new posts")
+
+    # Clear errors only for (sub, sort) pairs where shreddit succeeded.
+    # Arctic supplements recency posts but cannot "recover" a failed hot/top/
+    # rising lane — it has no sort lanes. Errors for failed shreddit lanes are
+    # preserved even when another sort for the same subreddit succeeded.
+    if errors and shreddit_successes:
+        errors = [
+            e for e in errors
+            if not _shreddit_error_recovered(e, shreddit_successes)
+        ]
     return {"items": unique, "errors": errors}
 
 

tests/test_discover_mode.py

@@ -408,21 +408,34 @@ def test_discovery_listing_block_is_reported_as_rate_limited():
 
 
 def test_reddit_discovery_adapter_preserves_partial_feed_errors():
+    """When one shreddit sort lane fails and another succeeds, the failed lane's error is kept.
+
+    Errors are cleared per (sub, sort) pair, not per subreddit. Arctic-shift cannot
+    recover a specific sort lane since it's recency-only.
+    """
     item = {
         "url": "https://reddit.com/r/example/comments/1",
         "title": "AI agent launch",
+        "subreddit": "AI_Agents",  # Required for error-clearing logic.
     }
     with mock.patch.object(
         reddit_listing,
         "_fetch_one_with_status",
         side_effect=[([], "rising timed out"), ([item], None)],
+    ), mock.patch(
+        "lib.reddit_arctic.fetch_listings",
+        return_value=[],  # Arctic supplement returns nothing.
     ):
         result = reddit_listing.fetch_discovery_listings(
             ["AI_Agents"], query="AI agents",
         )
 
     assert result["items"] == [item]
-    assert result["errors"] == ["r/AI_Agents rising: rising timed out"]
+    # Shreddit top succeeded → no error for top.
+    # Shreddit rising failed → error for rising is preserved.
+    # Error-clearing is per (sub, sort) pair, not per subreddit.
+    assert len(result["errors"]) == 1
+    assert "rising" in result["errors"][0].lower()
 
 
 def test_discovery_cli_json_contract_and_mutual_exclusion():

tests/test_reddit_arctic.py

@@ -60,6 +60,104 @@ class TestFetchScores:
         assert g.call_count == 1
         assert set(out) == {"a", "b"}
 
+
+def _listing_row(pid="abc123", title="matcha farm tour", score=406, ncmt=88,
+                 created=1783000000, subreddit="tea", permalink="/r/tea/comments/abc123/x/"):
+    return {
+        "id": pid, "title": title, "score": score, "num_comments": ncmt,
+        "created_utc": created, "subreddit": subreddit,
+        "permalink": permalink, "author": "u", "selftext": "",
+    }
+
+
+class TestFetchListings:
+    """fetch_listings serves scored subreddit listings from the archive —
+    the keyless fallback for hosts where shreddit partials 403."""
+
+    def test_returns_normalized_scored_posts(self):
+        with mock.patch.object(
+            reddit_arctic.http, "get",
+            return_value=_resp([_listing_row()]),
+        ) as g:
+            out = reddit_arctic.fetch_listings(["tea"], query="matcha")
+        assert len(out) == 1
+        post = out[0]
+        assert post["title"] == "matcha farm tour"
+        assert post["score"] == 406
+        assert post["engagement"]["score"] == 406
+        assert post["num_comments"] == 88
+        assert post["subreddit"] == "tea"
+        assert post["metadata"]["post_id"] == "abc123"  # t3_ prefix stripped
+        assert post["date"] == "2026-07-02"  # created_utc -> YYYY-MM-DD
+        assert post["url"].startswith("https://www.reddit.com/r/tea/comments/")
+        assert post["why_relevant"] == "Reddit listing (arctic-shift)"
+        # one call per subreddit, recent-first, depth-default volume
+        assert "subreddit=tea" in g.call_args[0][0]
+        assert "limit=25" in g.call_args[0][0]
+
+    def test_strips_r_prefix_and_skips_all(self):
+        with mock.patch.object(reddit_arctic.http, "get",
+                               return_value=_resp([_listing_row(subreddit="tea")])) as g:
+            out = reddit_arctic.fetch_listings(["r/tea", "all", ""])
+        assert len(out) == 1
+        assert g.call_count == 1  # only r/tea fetched; "all"/"" skipped
+
+    def test_processes_all_subreddits_no_cap(self):
+        """All requested subreddits are processed (no hard cap)."""
+        subs = [f"sub{i}" for i in range(20)]
+        with mock.patch.object(reddit_arctic.http, "get",
+                               return_value=_resp([_listing_row()])) as g:
+            reddit_arctic.fetch_listings(subs)
+        # All 20 should be fetched (no cap).
+        assert g.call_count == 20
+
+    def test_depth_controls_volume(self):
+        for depth, want in (("quick", 10), ("default", 25), ("deep", 50)):
+            with mock.patch.object(reddit_arctic.http, "get",
+                                   return_value=_resp([_listing_row()])) as g:
+                reddit_arctic.fetch_listings(["tea"], depth=depth)
+            assert f"limit={want}" in g.call_args[0][0], depth
+
+    def test_multi_sort_request_increases_limit(self):
+        """When multiple sorts are requested, fetch 2x posts to compensate."""
+        with mock.patch.object(reddit_arctic.http, "get",
+                               return_value=_resp([_listing_row()])) as g:
+            reddit_arctic.fetch_listings(["tea"], depth="default", sorts=["top", "hot", "new"])
+        # default depth = 25, with 3 sorts → 25 * 2 = 50
+        assert "limit=50" in g.call_args[0][0]
+
+    def test_single_sort_uses_base_limit(self):
+        """Single sort uses base limit, no multiplier."""
+        with mock.patch.object(reddit_arctic.http, "get",
+                               return_value=_resp([_listing_row()])) as g:
+            reddit_arctic.fetch_listings(["tea"], depth="default", sorts=["top"])
+        assert "limit=25" in g.call_args[0][0]
+
+    def test_dedupes_by_url(self):
+        rows = [_listing_row(), _listing_row()]
+        with mock.patch.object(reddit_arctic.http, "get", return_value=_resp(rows)):
+            out = reddit_arctic.fetch_listings(["tea"])
+        assert len(out) == 1
+
+    def test_network_error_degrades_to_empty(self):
+        with mock.patch.object(reddit_arctic.http, "get", side_effect=Exception("boom")):
+            assert reddit_arctic.fetch_listings(["tea"]) == []
+
+    def test_rate_limit_response_degrades_to_empty(self):
+        with mock.patch.object(reddit_arctic.http, "get",
+                               return_value={"error": "Timeout. Maybe slow down a bit"}):
+            assert reddit_arctic.fetch_listings(["tea"]) == []
+
+    def test_empty_subreddits_returns_empty(self):
+        assert reddit_arctic.fetch_listings([]) == []
+
+    def test_removed_author_normalized(self):
+        row = _listing_row()
+        row["author"] = "[deleted]"
+        with mock.patch.object(reddit_arctic.http, "get", return_value=_resp([row])):
+            out = reddit_arctic.fetch_listings(["tea"])
+        assert out[0]["author"] == "[deleted]"
+
     def test_cache_is_size_bounded(self):
         # The in-run memo never grows past CACHE_MAX; scores are still returned
         # for the current call once the cache is full.

tests/test_reddit_dedicated.py

@@ -40,6 +40,7 @@ def _no_enrich():
 class TestDedicatedLane:
     def test_dedicated_listings_pulled_with_top_hot_new_and_marked(self):
         ded = _listing_post(1, 2643, "What the actual fuck is this ye?")
+        ded["subreddit"] = "Kanye"  # Match the requested dedicated subreddit.
         captured = {}
 
         def fake_fetch(subs, depth="default", query="", sorts=None):
@@ -51,6 +52,8 @@ class TestDedicatedLane:
 
         with mock.patch.object(reddit_keyless.reddit_listing, "fetch_listings",
                                side_effect=fake_fetch), \
+             mock.patch.object(reddit_keyless.reddit_arctic, "fetch_listings",
+                               return_value=[]), \
              mock.patch.object(reddit_keyless.reddit_rss, "search_rss", return_value=[]):
             out = reddit_keyless._discover("Kanye West", "default", None,
                                            dedicated_subreddits=["Kanye"])
@@ -76,7 +79,9 @@ class TestDedicatedLane:
         # A thread present in both the dedicated lane and a broad listing keeps
         # its dedicated (floor-exempt) flag — dedicated is merged first.
         shared_ded = _listing_post(1, 500, "fresh thread", rel=0.0)
+        shared_ded["subreddit"] = "Kanye"  # Match the requested dedicated subreddit.
         shared_broad = _listing_post(1, 500, "fresh thread", rel=0.0)  # same url/id
+        shared_broad["subreddit"] = "hiphopheads"  # Match the requested broad subreddit.
 
         def fake_fetch(subs, depth="default", query="", sorts=None):
             return [shared_ded] if sorts == reddit_keyless.DEDICATED_SORTS else [shared_broad]

tests/test_reddit_keyless.py

@@ -33,6 +33,8 @@ class TestDiscovery:
         with mock.patch.object(reddit_keyless.reddit_rss, "search_rss",
                                return_value=[_post(1), _post(2)]) as rss, \
              mock.patch.object(reddit_keyless.reddit_listing, "fetch_listings",
+                               return_value=[]), \
+             mock.patch.object(reddit_keyless.reddit_arctic, "fetch_listings",
                                return_value=[]):
             out = reddit_keyless._discover("topic", "default", ["test"])
         assert len(out) == 2
@@ -42,10 +44,13 @@ class TestDiscovery:
         # RSS finds post 1 (no score); listing card for post 1 carries the score.
         rss_post = _post(1)
         listing_post = _scored(1, score=52692, ncmt=1743)
+        listing_post["subreddit"] = "test"  # Match the requested subreddit.
         with mock.patch.object(reddit_keyless.reddit_rss, "search_rss",
                                return_value=[rss_post]), \
              mock.patch.object(reddit_keyless.reddit_listing, "fetch_listings",
-                               return_value=[listing_post]):
+                               return_value=[listing_post]), \
+             mock.patch.object(reddit_keyless.reddit_arctic, "fetch_listings",
+                               return_value=[]):  # No arctic supplement.
             out = reddit_keyless._discover("topic", "default", ["test"])
         # listing post (scored) is kept; RSS dup of same url is dropped
         assert len(out) == 1
@@ -253,3 +258,94 @@ class TestSlotPriority:
         with mock.patch("lib.rerank._primary_entity", side_effect=Exception("boom")):
             out = reddit_keyless._slot_priority("openclaw", posts)
         assert out == posts
+
+
+class TestScoredListingsFallback:
+    """_scored_listings falls back to the arctic-shift archive when the
+    shreddit listing partials return nothing (datacenter egress 403)."""
+
+    def test_arctic_fallback_when_shreddit_empty(self):
+        arctic_post = _scored(1, score=406)
+        with mock.patch.object(reddit_keyless.reddit_listing, "fetch_listings",
+                               return_value=[]), \
+             mock.patch.object(reddit_keyless.reddit_arctic, "fetch_listings",
+                               return_value=[arctic_post]) as arctic:
+            out = reddit_keyless._scored_listings(["tea"], depth="quick", query="matcha")
+        assert out == [arctic_post]
+        arctic.assert_called_once_with(["tea"], depth="quick", query="matcha", sorts=None)
+
+    def test_shreddit_and_arctic_both_called_deduped(self):
+        """Shreddit and arctic are both called; arctic supplements missing posts."""
+        shreddit_post = _scored(1, score=42)
+        shreddit_post["subreddit"] = "tea"
+        arctic_post = _scored(2, score=100)
+        arctic_post["subreddit"] = "tea"
+        with mock.patch.object(reddit_keyless.reddit_listing, "fetch_listings",
+                               return_value=[shreddit_post]), \
+             mock.patch.object(reddit_keyless.reddit_arctic, "fetch_listings",
+                               return_value=[arctic_post]) as arctic:
+            out = reddit_keyless._scored_listings(["tea"], depth="quick", query="matcha")
+        # Both shreddit and arctic posts should be in the result (deduped by URL).
+        assert len(out) == 2
+        urls = {p["url"] for p in out}
+        assert shreddit_post["url"] in urls
+        assert arctic_post["url"] in urls
+        arctic.assert_called_once()
+
+    def test_both_empty_returns_empty(self):
+        with mock.patch.object(reddit_keyless.reddit_listing, "fetch_listings",
+                               return_value=[]), \
+             mock.patch.object(reddit_keyless.reddit_arctic, "fetch_listings",
+                               return_value=[]):
+            out = reddit_keyless._scored_listings(["tea"], depth="quick", query="matcha")
+        assert out == []
+
+    def test_never_raises_when_arctic_fails(self):
+        with mock.patch.object(reddit_keyless.reddit_listing, "fetch_listings",
+                               return_value=[]), \
+             mock.patch.object(reddit_keyless.reddit_arctic, "fetch_listings",
+                               side_effect=Exception("boom")):
+            out = reddit_keyless._scored_listings(["tea"], depth="quick", query="matcha")
+        assert out == []
+
+    def test_dedicated_sorts_passed_through(self):
+        with mock.patch.object(reddit_keyless.reddit_listing, "fetch_listings",
+                               return_value=[]), \
+             mock.patch.object(reddit_keyless.reddit_arctic, "fetch_listings",
+                               return_value=[]) as arctic:
+            reddit_keyless._scored_listings(
+                ["Kanye"], depth="default", query="Kanye", sorts=["top", "hot", "new"]
+            )
+        arctic.assert_called_once_with(
+            ["Kanye"], depth="default", query="Kanye", sorts=["top", "hot", "new"]
+        )
+
+    def test_arctic_supplements_all_subreddits(self):
+        """Arctic is called for all subreddits to supplement any failed sort lanes."""
+        shreddit_post = _scored(1, score=100)
+        shreddit_post["subreddit"] = "tea"
+        arctic_post_tea = _scored(2, score=200)
+        arctic_post_tea["subreddit"] = "tea"
+        arctic_post_coffee = _scored(3, score=150)
+        arctic_post_coffee["subreddit"] = "coffee"
+
+        def shreddit_side_effect(subs, **kwargs):
+            # Shreddit only returns posts for "tea", not "coffee".
+            return [shreddit_post] if "tea" in subs else []
+
+        with mock.patch.object(reddit_keyless.reddit_listing, "fetch_listings",
+                               side_effect=shreddit_side_effect), \
+             mock.patch.object(reddit_keyless.reddit_arctic, "fetch_listings",
+                               return_value=[arctic_post_tea, arctic_post_coffee]) as arctic:
+            out = reddit_keyless._scored_listings(
+                ["tea", "coffee"], depth="quick", query="beverages"
+            )
+        # Arctic is called for ALL requested subreddits to supplement any failed sorts.
+        arctic.assert_called_once()
+        call_args = arctic.call_args
+        assert set(call_args[0][0]) == {"tea", "coffee"}, "arctic should be called for all subs"
+        # All posts should be in the result (deduped by URL).
+        urls = [p["url"] for p in out]
+        assert shreddit_post["url"] in urls
+        assert arctic_post_tea["url"] in urls
+        assert arctic_post_coffee["url"] in urls

tests/test_reddit_listing.py

@@ -83,3 +83,208 @@ class TestScoreIndex:
         first = next(iter(idx.values()))
         assert set(first.keys()) == {"score", "num_comments"}
         assert any(v["score"] == 52692 for v in idx.values())
+
+
+class TestFetchDiscoveryListingsFallback:
+    """fetch_discovery_listings supplements with arctic-shift. Arctic is recency-only
+    and cannot "recover" failed hot/top/rising lanes — errors are preserved."""
+
+    def test_arctic_supplement_preserves_shreddit_errors(self):
+        """Arctic supplements posts but cannot clear shreddit errors (recency-only)."""
+        arctic_post = {
+            "id": "", "title": "matcha farm tour", "url": "https://www.reddit.com/r/tea/comments/abc/x/",
+            "score": 406, "num_comments": 88, "subreddit": "tea", "created_utc": None,
+            "author": "u", "selftext": "", "date": "2026-07-02",
+            "engagement": {"score": 406, "num_comments": 88, "upvote_ratio": None},
+            "relevance": 0.5, "why_relevant": "Reddit listing (arctic-shift)",
+            "metadata": {"post_id": "abc"},
+        }
+        with mock.patch.object(rl.http, "get_text", return_value=None), \
+             mock.patch("lib.reddit_arctic.fetch_listings", return_value=[arctic_post]) as arctic:
+            result = rl.fetch_discovery_listings(["tea"], query="matcha", depth="quick")
+        assert result["items"] == [arctic_post]
+        # Arctic is recency-only — it can supplement but cannot "recover" a failed
+        # rising/top lane. Errors for failed shreddit lanes are preserved.
+        assert result["errors"]  # shreddit errors preserved
+        arctic.assert_called_once()
+
+    def test_errors_preserved_when_both_shreddit_and_arctic_empty(self):
+        with mock.patch.object(rl.http, "get_text", return_value=None), \
+             mock.patch("lib.reddit_arctic.fetch_listings", return_value=[]):
+            result = rl.fetch_discovery_listings(["tea"], query="matcha", depth="quick")
+        assert result["items"] == []
+        assert result["errors"]  # the shreddit failures still surface
+
+    def test_arctic_supplements_shreddit_success(self):
+        """Arctic is always called to supplement shreddit; results are deduped."""
+        arctic_post = {
+            "id": "", "title": "netherlands tech news extra", "url": "https://www.reddit.com/r/technology/comments/arctic/x/",
+            "score": 100, "num_comments": 10, "subreddit": "technology", "created_utc": None,
+            "author": "u", "selftext": "", "date": "2026-07-02",
+            "engagement": {"score": 100, "num_comments": 10, "upvote_ratio": None},
+            "relevance": 0.5, "why_relevant": "Reddit listing (arctic-shift)",
+            "metadata": {"post_id": "arctic"},
+        }
+        with mock.patch.object(rl.http, "get_text", return_value=_html()), \
+             mock.patch("lib.reddit_arctic.fetch_listings", return_value=[arctic_post]) as arctic:
+            result = rl.fetch_discovery_listings(["technology"], query="netherlands", depth="quick")
+        # Both shreddit cards and arctic supplement should be in the result.
+        urls = {item["url"] for item in result["items"]}
+        assert arctic_post["url"] in urls
+        assert len(result["items"]) > 1  # shreddit + arctic
+        assert result["errors"] == []
+        arctic.assert_called_once()
+
+    def test_no_subreddits_skips_fallback(self):
+        with mock.patch("lib.reddit_arctic.fetch_listings") as arctic:
+            result = rl.fetch_discovery_listings([], query="matcha", depth="quick")
+        assert result == {"items": [], "errors": []}
+        arctic.assert_not_called()
+
+    def test_multi_sub_partial_arctic_keeps_all_shreddit_errors(self):
+        """AE4: arctic supplements but cannot clear shreddit sort-lane errors."""
+        # Request tea and coffee; shreddit fails for both, arctic returns tea posts.
+        arctic_post = {
+            "id": "", "title": "matcha farm tour", "url": "https://www.reddit.com/r/tea/comments/abc/x/",
+            "score": 406, "num_comments": 88, "subreddit": "tea", "created_utc": None,
+            "author": "u", "selftext": "", "date": "2026-07-02",
+            "engagement": {"score": 406, "num_comments": 88, "upvote_ratio": None},
+            "relevance": 0.5, "why_relevant": "Reddit listing (arctic-shift)",
+            "metadata": {"post_id": "abc"},
+        }
+        with mock.patch.object(rl.http, "get_text", return_value=None), \
+             mock.patch("lib.reddit_arctic.fetch_listings", return_value=[arctic_post]):
+            result = rl.fetch_discovery_listings(["tea", "coffee"], query="matcha", depth="quick")
+        assert result["items"] == [arctic_post]
+        # Both subs had shreddit errors. Arctic is recency-only and cannot recover
+        # failed sort lanes, so ALL errors are preserved.
+        coffee_errors = [e for e in result["errors"] if "r/coffee" in e.lower()]
+        assert coffee_errors, "shreddit errors for coffee should be preserved"
+        tea_errors = [e for e in result["errors"] if "r/tea" in e.lower()]
+        assert tea_errors, "arctic supplement cannot clear shreddit sort-lane errors"
+
+    def test_arctic_rows_failing_keyword_gate_keep_errors(self):
+        """AE5: arctic rows that fail the keyword gate do not count as recovery."""
+        # Arctic returns a row from tea, but the title doesn't match the query.
+        offtopic_post = {
+            "id": "", "title": "best oolong guide", "url": "https://www.reddit.com/r/tea/comments/xyz/x/",
+            "score": 999, "num_comments": 200, "subreddit": "tea", "created_utc": None,
+            "author": "u", "selftext": "", "date": "2026-07-02",
+            "engagement": {"score": 999, "num_comments": 200, "upvote_ratio": None},
+            "relevance": 0.0, "why_relevant": "Reddit listing (arctic-shift)",
+            "metadata": {"post_id": "xyz"},
+        }
+        with mock.patch.object(rl.http, "get_text", return_value=None), \
+             mock.patch("lib.reddit_arctic.fetch_listings", return_value=[offtopic_post]):
+            result = rl.fetch_discovery_listings(["tea"], query="matcha", depth="quick")
+        # "matcha" is not in "best oolong guide", so the row fails the keyword gate.
+        # With no surviving rows, arctic didn't effectively recover → errors kept.
+        assert result["items"] == []
+        assert result["errors"], "keyword-rejected rows should not clear errors"
+
+    def test_empty_query_skips_keyword_gate(self):
+        """Global --discover (empty query) skips the keyword gate entirely."""
+        offtopic_post = {
+            "id": "", "title": "best oolong guide", "url": "https://www.reddit.com/r/tea/comments/xyz/x/",
+            "score": 999, "num_comments": 200, "subreddit": "tea", "created_utc": None,
+            "author": "u", "selftext": "", "date": "2026-07-02",
+            "engagement": {"score": 999, "num_comments": 200, "upvote_ratio": None},
+            "relevance": 0.0, "why_relevant": "Reddit listing (arctic-shift)",
+            "metadata": {"post_id": "xyz"},
+        }
+        with mock.patch.object(rl.http, "get_text", return_value=None), \
+             mock.patch("lib.reddit_arctic.fetch_listings", return_value=[offtopic_post]):
+            # Empty query = global discover, no keyword gate.
+            result = rl.fetch_discovery_listings(["tea"], query="", depth="quick")
+        # No keyword gate → post survives. But arctic is recency-only, so shreddit
+        # errors for failed sort lanes are preserved.
+        assert result["items"] == [offtopic_post]
+        assert result["errors"]  # shreddit errors preserved despite arctic supplement
+
+
+class TestSortLaneErrorPreservation:
+    """Verify errors are cleared per (sub, sort) pair, not per subreddit."""
+
+    def test_shreddit_partial_success_keeps_failed_sort_errors(self):
+        """One sort succeeds, another fails → only the failed lane's error is kept.
+
+        This is the key test: shreddit rising succeeds for a sub, shreddit top fails,
+        arctic supplements with recency posts. The failed top error is preserved
+        because arctic is recency-only and cannot claim to have recovered a "top" lane.
+        """
+        # Shreddit returns posts for "rising" but None for "top".
+        def shreddit_response(url, *args, **kwargs):
+            if "rising" in url:
+                return _html()  # success for rising
+            return None  # fail for top
+
+        arctic_post = {
+            "id": "", "title": "extra from arctic", "url": "https://www.reddit.com/r/technology/comments/arctic/x/",
+            "score": 100, "num_comments": 10, "subreddit": "technology", "created_utc": None,
+            "author": "u", "selftext": "", "date": "2026-07-02",
+            "engagement": {"score": 100, "num_comments": 10, "upvote_ratio": None},
+            "relevance": 0.5, "why_relevant": "Reddit listing (arctic-shift)",
+            "metadata": {"post_id": "arctic"},
+        }
+        with mock.patch.object(rl.http, "get_text", side_effect=shreddit_response), \
+             mock.patch("lib.reddit_arctic.fetch_listings", return_value=[arctic_post]):
+            result = rl.fetch_discovery_listings(["technology"], query="netherlands", depth="quick")
+
+        # Shreddit rising succeeded → no error for rising.
+        # Shreddit top failed → error for top is preserved.
+        # Arctic supplement cannot clear shreddit errors (recency-only).
+        rising_errors = [e for e in result["errors"] if "rising" in e.lower()]
+        top_errors = [e for e in result["errors"] if " top:" in e.lower()]
+        assert not rising_errors, "successful rising lane should have no error"
+        assert top_errors, "failed top lane error should be preserved"
+        # Posts from both shreddit and arctic should be in the result.
+        assert len(result["items"]) > 1
+
+    def test_arctic_does_not_recover_shreddit_sort_lanes(self):
+        """Arctic supplement adds posts but cannot clear shreddit sort-lane errors."""
+        arctic_post = {
+            "id": "", "title": "foobar topic", "url": "https://www.reddit.com/r/foobar/comments/abc/x/",
+            "score": 500, "num_comments": 50, "subreddit": "foobar", "created_utc": None,
+            "author": "u", "selftext": "", "date": "2026-07-02",
+            "engagement": {"score": 500, "num_comments": 50, "upvote_ratio": None},
+            "relevance": 0.5, "why_relevant": "Reddit listing (arctic-shift)",
+            "metadata": {"post_id": "abc"},
+        }
+        # Shreddit fails for both subs; arctic supplements for foobar only.
+        with mock.patch.object(rl.http, "get_text", return_value=None), \
+             mock.patch("lib.reddit_arctic.fetch_listings", return_value=[arctic_post]):
+            result = rl.fetch_discovery_listings(["foo", "foobar"], query="topic", depth="quick")
+        # Arctic is recency-only — it cannot recover failed rising/top lanes.
+        # ALL shreddit errors should be preserved.
+        assert result["items"] == [arctic_post]
+        foo_errors = [e for e in result["errors"] if e.lower().startswith("r/foo ")]
+        assert foo_errors, "shreddit errors for r/foo should be preserved"
+        foobar_errors = [e for e in result["errors"] if e.lower().startswith("r/foobar ")]
+        assert foobar_errors, "arctic cannot clear shreddit errors — foobar errors preserved"
+
+
+class TestMatchesDiscoveryDomainParity:
+    """Verify the copied _matches_discovery_domain stays in sync with pipeline.py."""
+
+    def test_parity_with_pipeline_implementation(self):
+        # Import both implementations and verify they agree on test cases.
+        from lib import pipeline
+        test_cases = [
+            ("matcha", "matcha farm tour", True),
+            ("matcha", "best oolong guide", False),
+            ("AI", "New AI model release", True),  # "ai" is generic but no anchors → use domain_terms
+            ("OpenClaw", "My OpenClaw setup", True),
+            ("OpenClaw", "Generic post about nothing", False),
+            ("Stripe payments", "Stripe is great", True),
+            ("bias", "cognitive bias research", True),  # no naive stem corruption
+        ]
+        for domain, text, expected in test_cases:
+            pipeline_result = pipeline._matches_discovery_domain(domain, text)
+            listing_result = rl._matches_discovery_domain(domain, text)
+            assert pipeline_result == listing_result, (
+                f"Parity mismatch for ({domain!r}, {text!r}): "
+                f"pipeline={pipeline_result}, listing={listing_result}"
+            )
+            assert pipeline_result == expected, (
+                f"Expected {expected} for ({domain!r}, {text!r}), got {pipeline_result}"
+            )