fix(dataflows): report a failed Reddit fetch as unavailable, not silence

- a failed fetch and an empty search both returned [], so a 429 rendered as
  'no posts found' and the sentiment analyst read throttling as real silence;
  when every subreddit was throttled the summary asserted it outright
- a failed fetch now returns None and renders as unavailable, and the summary
  only claims silence for subreddits actually searched
- raise the headerless-429 back-off to 60s, which is where a retry starts
  succeeding; pay it at most once per run so three throttled subreddits do not
  stall the analysis, and match the Retry-After cap to it #1295

tests/test_reddit_fallback.py

@@ -86,9 +86,9 @@ class TestRssParsing:
         assert posts[0]["created_utc"] > 0
         assert "datacenter unit" in posts[0]["selftext"]
 
-    def test_malformed_xml_fails_open(self):
+    def test_malformed_xml_reports_unavailable(self):
         with patch.object(reddit, "urlopen", return_value=_resp(lambda: b"<<not xml>>")):
-            assert reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) == []
+            assert reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) is None
 
 
 @pytest.mark.unit
@@ -139,7 +139,7 @@ class TestRss429Backoff:
              patch.object(reddit.time, "sleep"):
             posts = reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0)
         assert op.call_count == 2          # one retry, then gives up cleanly
-        assert posts == []
+        assert posts is None
 
     def test_retry_after_header_is_honoured(self):
         err = HTTPError("url", 429, "Too Many Requests", {"Retry-After": "12"}, None)
@@ -166,7 +166,7 @@ class TestRss429Backoff:
             reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0)
         slept.assert_called_once()
         (wait,), _ = slept.call_args
-        assert 4.0 <= wait <= 6.0  # 5s +/-20% jitter
+        assert 48.0 <= wait <= 72.0  # 60s +/-20% jitter
 
 
 @pytest.mark.unit
@@ -174,9 +174,9 @@ class TestChunkedTransferErrorsHandled:
     """IncompleteRead/RemoteDisconnected come from http.client and are NOT
     OSErrors, so they were previously uncaught and crashed the pipeline (#1024)."""
 
-    def test_rss_incomplete_read_degrades_to_empty(self):
+    def test_rss_incomplete_read_reports_unavailable(self):
         with patch.object(reddit, "urlopen", return_value=_raise(http.client.IncompleteRead(b""))):
-            assert reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) == []
+            assert reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) is None
 
     def test_json_incomplete_read_falls_back_to_rss(self):
         with patch.object(reddit, "urlopen", return_value=_raise(http.client.IncompleteRead(b""))), \
@@ -190,7 +190,7 @@ class TestChunkedTransferErrorsHandled:
         big = _resp(lambda: b"x" * 100)
         with patch.object(reddit, "_MAX_FEED_BYTES", 10), \
              patch.object(reddit, "urlopen", return_value=big):
-            assert reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) == []
+            assert reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) is None
 
 
 @pytest.mark.unit
@@ -228,7 +228,7 @@ class TestCryptoSearchTerm:
     def _captured_ticker(self, ticker):
         seen = {}
 
-        def fake_fetch(t, sub, limit, timeout):
+        def fake_fetch(t, sub, limit, timeout, **kwargs):
             seen["ticker"] = t
             return []
 
@@ -241,3 +241,64 @@ class TestCryptoSearchTerm:
 
     def test_equity_passes_through(self):
         assert self._captured_ticker("NVDA") == "NVDA"
+
+
+@pytest.mark.unit
+class TestFailedFetchIsNotSilence:
+    """A throttled fetch must not be rendered as "no posts found" (#1295).
+
+    Returning [] for both a failed request and a genuinely empty search made the
+    sentiment analyst read rate limiting as real silence ("r/stocks and
+    r/investing are silent"), which is a signal that was never observed.
+    """
+
+    _POST = {
+        "title": "NVDA pops", "score": None, "num_comments": None,
+        "created_utc": reddit._iso_to_timestamp("2026-05-20T14:30:00Z"),
+        "selftext": "", "source": "rss",
+    }
+
+    def _run(self, results):
+        """Drive fetch_reddit_posts with a per-subreddit result sequence."""
+        subs = tuple(f"s{i}" for i in range(len(results)))
+        with patch.object(reddit, "_fetch_subreddit", side_effect=list(results)):
+            return reddit.fetch_reddit_posts(
+                "NVDA", subreddits=subs, inter_request_delay=0
+            )
+
+    def test_failed_subreddit_is_marked_unavailable_not_empty(self):
+        out = self._run([None, [self._POST]])
+        assert "unavailable" in out
+        assert "no posts found" not in out.split("unavailable")[0]
+
+    def test_all_sources_failing_does_not_claim_no_posts(self):
+        out = self._run([None, None])
+        assert "Reddit unavailable" in out
+        assert "no Reddit posts found" not in out
+
+    def test_mixed_failure_and_empty_only_claims_silence_for_searched_subs(self):
+        # s0 failed, s1 genuinely returned nothing: the "no posts" claim must
+        # cover only s1, with s0 reported separately as unavailable.
+        out = self._run([None, []])
+        assert "r/s1" in out.split("unavailable (fetch failed)")[0]
+        assert "unavailable (fetch failed): r/s0" in out
+
+    def test_genuine_empty_still_reports_no_posts(self):
+        out = self._run([[], []])
+        assert "no Reddit posts found" in out
+        assert "unavailable" not in out
+
+    def test_retry_is_not_spent_again_after_a_failure(self):
+        # The 60s back-off must be paid at most once per run, so subsequent
+        # subreddits are fetched with retry disabled rather than stalling.
+        seen = []
+
+        def record(t, sub, limit, timeout, _retry=True):
+            seen.append(_retry)
+            return None
+
+        with patch.object(reddit, "_fetch_subreddit", side_effect=record):
+            reddit.fetch_reddit_posts(
+                "NVDA", subreddits=("a", "b", "c"), inter_request_delay=0
+            )
+        assert seen == [True, False, False]

tradingagents/dataflows/reddit.py

@@ -10,6 +10,10 @@ off once (honouring ``Retry-After``). RSS lacks score / comment counts, so those
 posts are marked and the formatter omits the metrics rather than printing fake
 zeros.
 
+A fetch that fails is reported as ``<unavailable>``, never as "no posts found":
+the two are different claims, and passing a rate-limited fetch off as silence
+hands the sentiment analyst a signal that was never observed (#1295).
+
 No API key required. Returns formatted plaintext blocks ready for prompt
 injection and degrades gracefully — returns a placeholder string rather than
 raising, so callers never special-case missing data.
@@ -102,9 +106,12 @@ def _strip_html(content: str) -> str:
     return " ".join(html.unescape(text).split())
 
 
-# Headerless-429 backoff when Reddit gives no Retry-After. Jittered so several
+# Headerless-429 backoff when Reddit gives no Retry-After. Measured against
-# analyses sharing an IP don't retry in lockstep and re-collide on the limit.
+# /r/{sub}/search.rss, a retry still 429s at 8s, 10s and 30s of spacing and
-_RETRY_FALLBACK_SECONDS = 5.0
+# succeeds at 60s, so a shorter wait spends the one retry on a request that
+# cannot succeed (#1295). Jittered so several analyses sharing an IP don't
+# retry in lockstep and re-collide on the limit.
+_RETRY_FALLBACK_SECONDS = 60.0
 
 
 def _jitter(seconds: float, frac: float = 0.2) -> float:
@@ -114,14 +121,18 @@ def _jitter(seconds: float, frac: float = 0.2) -> float:
 
 
 def _retry_after_seconds(exc: HTTPError) -> float | None:
-    """Seconds to wait from a 429's ``Retry-After`` header, capped at 30s.
+    """Seconds to wait from a 429's ``Retry-After`` header, capped at 60s.
+
+    The cap matches ``_RETRY_FALLBACK_SECONDS``: honouring less than we would
+    wait on our own would spend the one retry on a request we already know is
+    too early.
 
     Returns ``None`` only when the header is absent or unparseable; a valid
     ``Retry-After: 0`` returns ``0.0`` (retry at once), not ``None``.
     """
     try:
         val = exc.headers.get("Retry-After") if getattr(exc, "headers", None) else None
-        return min(float(val), 30.0) if val is not None else None
+        return min(float(val), 60.0) if val is not None else None
     except (ValueError, TypeError, AttributeError):
         return None
 
@@ -149,13 +160,18 @@ def _fetch_subreddit_rss(
     limit: int,
     timeout: float,
     _retry: bool = True,
-) -> list[dict]:
+) -> list[dict] | None:
     """Default path: parse the public Atom search feed for a subreddit.
 
     Carries no score / comment counts, so those fields are left None and the
     post is tagged ``source="rss"`` for honest display. On a 429 (Reddit's
     per-IP rate limit) we back off once — honouring ``Retry-After`` when
     present — before giving up, so a transient burst doesn't blank the feed.
+
+    Returns ``[]`` when the search ran and matched nothing, and ``None`` when
+    the fetch itself failed. The caller must keep these apart: rendering a
+    failed fetch as "no posts found" hands the sentiment analyst an absence of
+    discussion that was never observed (#1295).
     """
     url = _RSS.format(sub=sub, qs=_search_qs(ticker, limit))
     req = Request(url, headers={"User-Agent": _UA})
@@ -175,12 +191,12 @@ def _fetch_subreddit_rss(
             time.sleep(wait)
             return _fetch_subreddit_rss(ticker, sub, limit, timeout, _retry=False)
         logger.warning("Reddit RSS fetch failed for r/%s · %s: %s", sub, ticker, exc)
-        return []
+        return None
     except (OSError, http.client.HTTPException, ET.ParseError) as exc:
         # OSError covers URLError/TimeoutError/connection resets; HTTPException
         # covers chunked-transfer errors (IncompleteRead/BadStatusLine, #1024).
         logger.warning("Reddit RSS fetch failed for r/%s · %s: %s", sub, ticker, exc)
-        return []
+        return None
 
     posts = []
     for entry in root.findall("atom:entry", _ATOM_NS)[:limit]:
@@ -234,14 +250,15 @@ def _fetch_subreddit(
     sub: str,
     limit: int,
     timeout: float,
-) -> list[dict]:
+    _retry: bool = True,
-    """Fetch one subreddit, RSS-first.
+) -> list[dict] | None:
+    """Fetch one subreddit, RSS-first. ``None`` means the fetch failed.
 
     The JSON search endpoint is reliably WAF-blocked (403) for public clients,
     so we go straight to the RSS feed — which serves our identified User-Agent
     reliably — halving our request volume against Reddit's per-IP rate limit.
     """
-    return _fetch_subreddit_rss(ticker, sub, limit, timeout)
+    return _fetch_subreddit_rss(ticker, sub, limit, timeout, _retry=_retry)
 
 
 def fetch_reddit_posts(
@@ -267,13 +284,26 @@ def fetch_reddit_posts(
     # Crypto reaches us as a Yahoo pair (BTC-USD); search Reddit for the base
     # ("BTC") so the query actually matches discussion instead of near-nothing.
     ticker = crypto_base(ticker) or ticker
+    subreddits = list(subreddits)
     blocks = []
     total_posts = 0
+    unavailable = []
+    allow_retry = True
     for i, sub in enumerate(subreddits):
         if i > 0 and inter_request_delay:
             time.sleep(_jitter(inter_request_delay))
-        posts = _within_window(_fetch_subreddit(ticker, sub, limit_per_sub, timeout),
+        fetched = _fetch_subreddit(ticker, sub, limit_per_sub, timeout, _retry=allow_retry)
-                               start_date, end_date)
+        if fetched is None:
+            # A failed fetch is not an absence of discussion, so it must not be
+            # rendered as "no posts found" (#1295). One failure also means the
+            # per-IP budget is likely gone, so skip the (now 60s) back-off on
+            # the remaining subreddits rather than stalling the run on retries
+            # that cannot succeed; #1286 tracks coordinating this properly.
+            allow_retry = False
+            unavailable.append(sub)
+            blocks.append(f"r/{sub}: <unavailable: fetch failed, not an absence of posts>")
+            continue
+        posts = _within_window(fetched, start_date, end_date)
         total_posts += len(posts)
         if not posts:
             blocks.append(f"r/{sub}: <no posts found mentioning {ticker.upper()} in the past 7 days>")
@@ -306,8 +336,23 @@ def fetch_reddit_posts(
         blocks.append("\n".join(lines))
 
     if total_posts == 0:
-        return (
+        searched = [s for s in subreddits if s not in unavailable]
+        if not searched:
+            # Every source failed: claiming "no posts" here would assert a
+            # silence we never observed.
+            return (
+                f"<Reddit unavailable: every source failed to fetch "
+                f"({', '.join(f'r/{s}' for s in unavailable)}); this is not an "
+                f"absence of discussion>"
+            )
+        summary = (
             f"<no Reddit posts found mentioning {ticker.upper()} across "
-            f"{', '.join(f'r/{s}' for s in subreddits)} in the past 7 days>"
+            f"{', '.join(f'r/{s}' for s in searched)} in the past 7 days>"
         )
+        if unavailable:
+            summary += (
+                f"\n<unavailable (fetch failed): "
+                f"{', '.join(f'r/{s}' for s in unavailable)}>"
+            )
+        return summary
     return "\n\n".join(blocks)