fix(rss): hand the pipeline plain text instead of feed HTML

docs/architecture.md

@@ -549,7 +549,7 @@ table `schema.sql` creates must appear in the rows below, so a new table cannot
 
 | Datum | Today | Destination | Notes |
 |---|---|---|---|
-| Article store | **D1 `stored_articles`** | same | `url` PRIMARY KEY is the dedup key; a colliding newsletter issue is re-keyed to `newsletter:{id}`. Besides a run's save, the onboarding votes step (§6) stores each card the reader marked and then accepts or rejects it with `triaged_at`, as a `/labels` answer is written; a card left alone gets no row and stays in the buffer for the first run |
+| Article store | **D1 `stored_articles`** | same | `content` is written as plain text for every source (newsletter bodies already were; an RSS entry's HTML is converted by `CloudflareRssSource`, or by `RssSource` on the direct-polling path, through `adapters/fetch/html_text.py`). Rows stored before RSS content was converted keep their HTML: `INSERT OR IGNORE` never rewrites them, and a pending row or a rerun turns them back into articles as they are. `url` PRIMARY KEY is the dedup key; a colliding newsletter issue is re-keyed to `newsletter:{id}`. Besides a run's save, the onboarding votes step (§6) stores each card the reader marked and then accepts or rejects it with `triaged_at`, as a `/labels` answer is written; a card left alone gets no row and stays in the buffer for the first run |
 | Tag vocabulary | **D1 `tags`** | same | Normalized tags emitted by clustering and scoring |
 | Article tags | **D1 `article_tags`** | same | URL-keyed article membership in the tag vocabulary |
 | Stories | **D1 `stories`** | same | Pre-truncation news clusters, keyed `{date}-{period}-{urlhash}` (content-derived from member URLs), replaced per window |
@@ -559,7 +559,7 @@ table `schema.sql` creates must appear in the rows below, so a new table cannot
 | Topic tags | **D1 `topic_tags`**; `topics.json` for `json` | same | A topic's normalized tags with their kind, `entity`, `domain` or `region`: the tags its newest save gave it, so a rerun that tags a topic differently replaces them. Today's tags carry no kind, so the stand-in stores every one as `domain` (`STAND_IN_TAG_KIND`) |
 | Topic votes | **D1 `topic_votes`**; `topics.json` for `json` | same | Append-only, keyed by topic and click time: the direction (1 or -1). Written by the vote sync (`sync_promotions`): a voted URL that an issue put under a topic becomes one vote on that topic, a group's per-URL posts collapsing to one row timed by the promote Worker's `ts`. A same-direction vote on the same topic within `ONE_CLICK_WINDOW` (60 s) of a recorded one, with no opposite vote between them, is taken as that click again and not written, since a sync can ACK part of one click's posts before the rest arrive. A sync whose topic-vote write fails still stamps and ACKs its votes, since the stamp is the vote's record and a held ACK would re-stamp the same URLs every run; that sync's topic votes are lost, and `run_summary`'s `topic_votes_failed` says why. A URL no topic holds (an issue published before topics, the raw page) writes no row |
 | Article image URL | **D1 `articles.image_url`** and **`stored_articles.image_url`** (the `json` store keeps it in the article), carried on each Stories item and each Topics member in the issue's `DigestContent` | same | The picture a feed entry names for its article, as a URL only: `media:thumbnail`, `media:content` or an image `enclosure`, else the first `<img>` in the entry's content. Parsed by `workers/rss` when it buffers the entry, or by `RssSource` on the direct-polling path with the same order and rules (`tests/test_feed_image_parity.py` holds the two to one URL), and carried with the article to the store, so a rerun shows the image the article had. The image bytes are never fetched or kept: the digest page loads them from the feed's host in the reader's browser, with no referrer (UI spec §4 *story image*). No article page is fetched for an `og:image` (owner, 2026-10-31): a feed's own picture costs no new outbound request. A newsletter issue has no feed, so it has no image. The digest pipeline stamps it on Stories items and on news clusters' members only, leaving out a picture two or more of one source's articles in the issue share (`domain/story_images.py`); a Topics block shows its first member's that is left (`domain/layers.py`), and an issue stored before Topics showed pictures carries none on its members, so a rerender of it shows no Topics picture where a rerun does. Headlines and the mail show none |
-| RSS buffer | **D1 `articles`** | same | Same database, different lifecycle: disposable, 8-day retention counted from buffer entry |
+| RSS buffer | **D1 `articles`** | same | Same database, different lifecycle: disposable, 8-day retention counted from buffer entry. Its `content` is the feed's HTML, which the Worker reads the story image from; `CloudflareRssSource` turns it into plain text (trafilatura) on the way out, so what it writes to `stored_articles.content` is text (older rows: see Article store) |
 | Feed health | **D1 `feed_health`** | same | One row per RSS feed the Worker has polled: the current failure streak, the last error and when, and the last success. Written by `workers/rss` on every poll, read by `cyris doctor` and `/settings` together with the newest `stored_articles.first_seen_at` per source, which needs no state of its own; `cyris run` neither reads nor writes it. Both `schema.sql` and the Worker create it. A row for a retired source stays behind, harmless, since both readers start from `sources`. D1 only: a `json` deployment polls no Worker |
 | Source definitions | **D1 `sources`**; `sources.yaml` for a `json` deployment | same | One home per backend. Both cyris and `workers/rss` read the table; an empty one is a deployment waiting for onboarding, so it polls no feed and `cyris run` ends having fetched nothing, with no failure alert (`Config.awaiting_onboarding`) |
 | Runtime settings | **D1 `settings`**; `cyris.toml` for a `json` deployment | same | Grade D. One home per backend, no value in code, and a missing key stops the run — see §5 |

pyproject.toml

@@ -20,6 +20,7 @@ dependencies = [
     "pyyaml>=6.0",
     "feedparser>=6.0.14",
     "blake3>=1.0.9",
+    "trafilatura>=2.3.1",
 ]
 
 [project.scripts]

src/cyris/adapters/fetch/html_text.py

@@ -0,0 +1,72 @@
+"""Feed-entry HTML to the body text prompts and language detection read."""
+
+from __future__ import annotations
+
+import html
+import logging
+import re
+from html.parser import HTMLParser
+
+import trafilatura
+
+# trafilatura logs a bare teaser it declines to parse at ERROR, once per entry, though
+# `plain_text` handles that case; the run log is read for real errors.
+logging.getLogger("trafilatura").setLevel(logging.CRITICAL)
+
+# What an unterminated tag at the end of the input starts with; a bare `<` in prose
+# (`x <3`, `a < b`) does not match.
+_OPEN_TAG = re.compile(r"<[A-Za-z/!?]")
+
+
+class _TextCollector(HTMLParser):
+    """Collects the text between tags, with character references decoded."""
+
+    def __init__(self) -> None:
+        super().__init__(convert_charrefs=True)
+        self.parts: list[str] = []
+
+    def handle_data(self, data: str) -> None:
+        self.parts.append(data)
+
+    # A tag stands for a word break: `</p><p>` must not glue two paragraphs into one word.
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        self.parts.append(" ")
+
+    def handle_endtag(self, tag: str) -> None:
+        self.parts.append(" ")
+
+
+def _strip_tags(markup: str) -> str:
+    collector = _TextCollector()
+    try:
+        collector.feed(markup)
+    except AssertionError:
+        # CPython 3.12 before 3.12.16 asserts on `<![` followed by a non-name; the base
+        # image's patch level is not pinned, and one entry must not fail the whole read.
+        return " ".join(re.sub(r"<[^>]+>", " ", html.unescape(markup)).split())
+    # A feed that cuts its summary mid-tag leaves the tag unterminated in the parser's
+    # buffer: drop it rather than let `close` flush it as text.
+    if not _OPEN_TAG.match(collector.rawdata):
+        collector.close()
+    return " ".join("".join(collector.parts).split())
+
+
+def plain_text(markup: str) -> str:
+    """The body text of a feed entry's HTML, for prompts and language detection.
+
+    Both feed adapters call this: everything downstream wants words, and a post that
+    opens with a few thousand characters of `<picture srcset>` otherwise fills the whole
+    snippet. `RssSource` reads the story image out of the raw HTML first;
+    `CloudflareRssSource` gets `image_url` from the Worker and never needs the markup.
+    """
+    if not markup:
+        return ""
+    extracted = trafilatura.extract(
+        markup, favor_recall=True, include_comments=False, include_tables=True
+    )
+    if extracted:
+        return extracted.strip()
+    # trafilatura returns None for a fragment too short or too bare to look like an
+    # article (a one-line teaser); an empty content would read as "no body" to the
+    # summary prompt, so keep the words the old way instead.
+    return _strip_tags(markup)

src/cyris/adapters/fetch/rss_source.py

@@ -22,6 +22,7 @@
 import feedparser
 
 from cyris.adapters.fetch.email_parser import strip_tracking_params
+from cyris.adapters.fetch.html_text import plain_text
 from cyris.adapters.http_client import HttpClient
 from cyris.domain.models import Article, Lane, SourceConfig
 
@@ -50,6 +51,54 @@ def _entry_published(entry: Mapping[str, object]) -> datetime | None:
     return None
 
 
+def _articles_from_feed(
+    body: bytes, source: SourceConfig, after: datetime, before: datetime
+) -> list[Article]:
+    """Parse a feed body and map the entries inside the window onto Articles.
+
+    Blocking: `fetch_articles` runs it in a worker thread.
+    """
+    # feedparser.parse(url) would do its own blocking urllib fetch — hand it the body
+    # so the shared client's User-Agent and redirect handling apply. As a stream:
+    # bare bytes that name a path are opened as that local file.
+    parsed = feedparser.parse(io.BytesIO(body))
+
+    articles = []
+    for entry in parsed.entries:
+        published_at = _entry_published(entry)
+        if published_at is None:
+            logger.debug("Entry without date skipped: %s", getattr(entry, "link", ""))
+            continue
+        if not (after <= published_at < before):
+            continue
+
+        # Without stripping, the same article arrives under two URLs and
+        # defeats the store's URL primary key.
+        link = getattr(entry, "link", "")
+        url = strip_tracking_params(link)
+        if not url:
+            continue
+
+        content = _entry_content(entry)
+
+        articles.append(
+            Article(
+                id=getattr(entry, "id", "") or url,
+                # A CDATA title keeps its entities as text, and the page escapes it again.
+                title=html.unescape(getattr(entry, "title", "")),
+                url=url,
+                content=plain_text(content),
+                author=getattr(entry, "author", None) or None,
+                published_at=published_at,
+                source_name=source.name,
+                source_lane=source.lane or Lane.SKIM,
+                source_tags=source.tags,
+                image_url=_entry_image(entry, link),
+            )
+        )
+    return articles
+
+
 class RssSource:
     """FetchSource that fetches and parses feed URLs from sources.yaml."""
 
@@ -93,45 +142,9 @@ async def _fetch_feed(
     ) -> list[Article]:
         response = await self._http.get(source.url or "")
         response.raise_for_status()
-        # feedparser.parse(url) would do its own blocking urllib fetch — hand it the body
-        # so the shared client's User-Agent and redirect handling apply. As a stream:
-        # bare bytes that name a path are opened as that local file.
-        parsed = await asyncio.to_thread(feedparser.parse, io.BytesIO(response.content))
-
-        articles = []
-        for entry in parsed.entries:
-            published_at = _entry_published(entry)
-            if published_at is None:
-                logger.debug("Entry without date skipped: %s", getattr(entry, "link", ""))
-                continue
-            if not (after <= published_at < before):
-                continue
-
-            # Without stripping, the same article arrives under two URLs and
-            # defeats the store's URL primary key.
-            link = getattr(entry, "link", "")
-            url = strip_tracking_params(link)
-            if not url:
-                continue
-
-            content = _entry_content(entry)
-
-            articles.append(
-                Article(
-                    id=getattr(entry, "id", "") or url,
-                    # A CDATA title keeps its entities as text, and the page escapes it again.
-                    title=html.unescape(getattr(entry, "title", "")),
-                    url=url,
-                    content=content,
-                    author=getattr(entry, "author", None) or None,
-                    published_at=published_at,
-                    source_name=source.name,
-                    source_lane=source.lane or Lane.SKIM,
-                    source_tags=source.tags,
-                    image_url=_entry_image(entry, link),
-                )
-            )
-        return articles
+        # Parsing and the HTML-to-text conversion both burn CPU, so they share one
+        # thread call and the event loop stays free for the other feeds.
+        return await asyncio.to_thread(_articles_from_feed, response.content, source, after, before)
 
     async def health_check(self) -> bool:
         return True

src/cyris/adapters/fetch/rss_worker_source.py

@@ -7,13 +7,15 @@
 
 from __future__ import annotations
 
+import asyncio
 import html
 import logging
 import math
 from datetime import UTC, datetime
 
 import httpx
 
+from cyris.adapters.fetch.html_text import plain_text
 from cyris.domain.models import Article, Lane, SourceConfig
 
 logger = logging.getLogger(__name__)
@@ -47,6 +49,36 @@ def _worker_bound(value: datetime) -> str:
     return value.astimezone(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
 
 
+def _articles_from_rows(rows: list[dict], sources: dict[str, SourceConfig]) -> list[Article]:
+    """Map the Worker's rows onto Articles, skipping a malformed row.
+
+    Blocking: `fetch_articles` runs it in a worker thread.
+    """
+    articles = []
+    for row in rows:
+        source = sources.get(row.get("source_name", ""))
+        try:
+            articles.append(
+                Article(
+                    id=row.get("guid") or row["url"],
+                    # The Worker's parser, like feedparser, leaves a CDATA title's
+                    # entities as text, and the page would escape them again.
+                    title=html.unescape(row.get("title", "")),
+                    url=row["url"],
+                    content=plain_text(row.get("content", "")),
+                    author=row.get("author") or None,
+                    published_at=row["published_at"],
+                    source_name=row.get("source_name", ""),
+                    source_lane=source.lane if source else Lane.SKIM,
+                    source_tags=source.tags if source else [],
+                    image_url=row.get("image_url") or None,
+                )
+            )
+        except (KeyError, ValueError):
+            logger.warning("Skipping malformed row: %s", row.get("url"), exc_info=True)
+    return articles
+
+
 class CloudflareRssSource:
     """Read feed entries buffered in the Cloudflare RSS Worker's D1."""
 
@@ -87,28 +119,9 @@ async def fetch_articles(
                 WORKER_ROW_CEILING,
             )
 
-        articles = []
-        for row in rows:
-            source = sources.get(row.get("source_name", ""))
-            try:
-                articles.append(
-                    Article(
-                        id=row.get("guid") or row["url"],
-                        # The Worker's parser, like feedparser, leaves a CDATA title's
-                        # entities as text, and the page would escape them again.
-                        title=html.unescape(row.get("title", "")),
-                        url=row["url"],
-                        content=row.get("content", ""),
-                        author=row.get("author") or None,
-                        published_at=row["published_at"],
-                        source_name=row.get("source_name", ""),
-                        source_lane=source.lane if source else Lane.SKIM,
-                        source_tags=source.tags if source else [],
-                        image_url=row.get("image_url") or None,
-                    )
-                )
-            except (KeyError, ValueError):
-                logger.warning("Skipping malformed row: %s", row.get("url"), exc_info=True)
+        # `plain_text` parses each entry and a read holds up to WORKER_ROW_CEILING
+        # rows: build the whole batch off the event loop in one hop.
+        articles = await asyncio.to_thread(_articles_from_rows, rows, sources)
 
         logger.info("Read %d buffered entries from the RSS worker", len(articles))
         return articles

src/cyris/service_layer/degrade.py

@@ -40,8 +40,10 @@ def strip_dateline(text: str) -> str:
 def excerpt(content: str, length: int = 200, *, drop_dateline: bool = False) -> str:
     """Plain-text excerpt from the start of the content — the no-LLM stand-in for a summary.
 
-    Feed content is raw HTML; strip tags and decode entities so fallbacks
-    never leak markup into the digest. `drop_dateline` strips a leading wire
+    Content is plain text, or HTML for a row stored before feed content was
+    converted to text; stripping tags and decoding entities keeps markup out of the
+    digest for both, at a cost on plain text: a literal `<word>` (`Vec<String>`) is
+    dropped and an already-decoded `&amp;` is decoded again. `drop_dateline` strips a leading wire
     dateline from the plain text before it is cut, so it costs none of `length`.
     """
     text = re.sub(r"<[^>]+>", " ", content or "")

tests/captioned_post.py

@@ -0,0 +1,24 @@
+"""A synthetic Substack-shaped feed body: a long picture block, then the paragraphs.
+
+Both feed adapters must hand the pipeline the first paragraph, not the markup
+that precedes it (`tests/test_rss_worker_source.py`, `tests/test_rss_source.py`).
+"""
+
+FIRST_SENTENCE = "A copier repeats what it was shown, and a craftsperson answers the room."
+SRCSET = "".join(
+    f'<source type="image/webp" srcset="https://img.test/w_{w}.webp {w}w">'
+    for w in range(100, 1500, 25)
+)
+CAPTIONED_IMAGE = (
+    '<div class="captioned-image-container"><figure>'
+    '<a class="image-link" target="_blank" href="https://img.test/full.png"><picture>'
+    + SRCSET
+    + '<img src="https://img.test/w_1456.png" alt="" width="1456" height="819">'
+    "</picture></a></figure></div>"
+)
+
+POST_HTML = (
+    CAPTIONED_IMAGE
+    + f"<p>{FIRST_SENTENCE} It keeps the form and loses the reason.</p>"
+    + "<p>The second paragraph says why the reason matters more than the form.</p>"
+)