feat(seo): aviso de colisión de tema en /generate blog en

Al publicar el draft EN en Ghost, el título propuesto se compara contra
los posts published+scheduled del sitio (corpus vía Admin API — incluye
la cola programada, justo el caso del doble Kecksburg del 2026-07-10) con
el topic_collision vendorizado. Si colisiona, el notice de Telegram lleva
un bloque "🚨 Posible colisión de tema" en las tres rutas (live, dryrun,
bare). Nunca bloquea: el draft se crea igual, el humano decide (fusionar,
retitular o enlazar a propósito). Solo lang=en (stopwords inglesas).
Títulos ajenos saneados de entidades Markdown (regla de _safe_send).
Aislamiento: cualquier fallo del check → notice sin bloque y warning en
logs, jamás rompe la publicación.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.14-slim
+FROM python:3.12-slim
 
 WORKDIR /app
 

KNOWN-ISSUES.md

@@ -60,3 +60,24 @@ hit a consent wall from EU IPs, and the inner `AU_yqL` id is only resolvable via
 Google's private batchexecute API. Do not retry. The news seed uses Bing News
 RSS instead (`ENABLE_NEWS_SEED`, real publisher URL in the `?url=` param of
 apiclick.aspx — unwrapped by `_unwrap_news_link`).
+
+## Large sources can OOM-kill the pod
+
+On 2026-07-10 the pod was OOMKilled (memory limit was 1Gi) mid-research: a
+batch of 20 concurrent sources hit a 98k-word document plus several large PDFs
+at once, and pdfplumber's parse spiked RAM past the limit. The in-memory
+research task died with the pod and its session sat in `running` forever.
+
+Mitigations now in place:
+
+- Memory limit raised to 2Gi (`k8s-manifests/researchowl/deployment.yaml`).
+- PDFs capped at 15MB (was 50MB), checked both via Content-Length and actual
+  body size; pdfplumber runs in `run_in_executor` (it is sync + CPU-heavy and
+  also froze the event loop, same class of bug as DDGS) and flushes its page
+  cache per page.
+- Extracted content is truncated to `max_content_length` (300k chars) before
+  hitting `source_contents`.
+- On startup the bot marks orphaned `running` sessions as `interrupted`.
+
+If a research still dies, the scraped sources survive in the DB: `/process`
+re-chunks and scores them without re-scraping.

src/bot/bot.py

@@ -851,6 +851,27 @@ async def cmd_help(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
 
 # ─── Bot setup ────────────────────────────────────────────────────────────────
 
+async def _mark_interrupted_on_startup(app: Application) -> None:
+    """Las tareas de research viven solo en memoria (_active_tasks): un
+    reinicio del pod las mata sin tocar la DB, y sus sesiones quedan en
+    'running' para siempre — parecen activas en /status y get_active_session.
+    """
+    db_conn = await get_db()
+    try:
+        cursor = await db_conn.execute(
+            "UPDATE research_sessions SET status = ?, updated_at = ? WHERE status = ?",
+            (ResearchStatus.INTERRUPTED, time.time(), ResearchStatus.RUNNING),
+        )
+        await db_conn.commit()
+        if cursor.rowcount:
+            logger.info("Orphaned running sessions marked interrupted",
+                        count=cursor.rowcount)
+    except Exception as e:
+        logger.warning("Interrupted-mark failed — bot continues", error=str(e))
+    finally:
+        await db_conn.close()
+
+
 async def _purge_on_startup(app: Application) -> None:
     db_conn = await get_db()
     try:
@@ -994,6 +1015,7 @@ async def _start_scheduler(app: Application) -> None:
 
 
 async def _on_startup(app: Application) -> None:
+    await _mark_interrupted_on_startup(app)
     await _purge_on_startup(app)
     await _start_scheduler(app)
 

src/config.py

@@ -47,6 +47,8 @@ class Settings(BaseSettings):
     request_timeout: int = Field(30)
     request_delay: float = Field(1.0)          # seconds between requests
     min_content_length: int = Field(200)  # chars
+    # Libros/dumps enteros (100k+ palabras) inflan RAM y DB sin aportar al RAG
+    max_content_length: int = Field(300_000)  # chars
 
     # Fuentes opcionales — desactivadas por defecto: la IP del homelab está
     # bloqueada por Reddit (403) y YouTube (transcripts vacíos), eran peso muerto.

src/db/database.py

@@ -18,6 +18,7 @@ class ResearchStatus(str, Enum):
     SATURATED = "saturated"
     FINISHED = "finished"
     ERROR = "error"
+    INTERRUPTED = "interrupted"  # el pod se reinició con el research en marcha
 
 
 class OutputType(str, Enum):

src/generator/generator.py

@@ -612,6 +612,7 @@ class OutputGenerator:
             return ""
         title = _extract_title(full_output) or topic
         mode = _resolve_seo_mode(seo_override)
+        collision_note = await self._collision_note(lang, title)
 
         if mode in ("on", "dryrun"):
             try:
@@ -636,13 +637,17 @@ class OutputGenerator:
                     # surface the proposal so Jose can inspect before trusting writes.
                     result = await ghost.publish_draft(title, full_output)
                     post = result["posts"][0]
-                    self.last_publish_notice = _seo_dryrun_message(ghost, post, seo, inserted_pairs)
+                    self.last_publish_notice = (
+                        _seo_dryrun_message(ghost, post, seo, inserted_pairs)
+                        + collision_note)
                 else:
                     result = await ghost.publish_draft(
                         title, full_output, tags=seo["tags"], seo=seo,
                         body_html=linked_html)
                     post = result["posts"][0]
-                    self.last_publish_notice = _seo_live_message(ghost, post, seo, inserted_pairs)
+                    self.last_publish_notice = (
+                        _seo_live_message(ghost, post, seo, inserted_pairs)
+                        + collision_note)
                 logger.info("Auto-published blog to Ghost",
                             mode=mode, post_id=post["id"], links=len(inserted_pairs))
                 return ""
@@ -661,11 +666,27 @@ class OutputGenerator:
             result = await ghost.publish_draft(title, full_output)
             post = result["posts"][0]
             logger.info("Auto-published blog to Ghost (bare)", post_id=post["id"])
-            return _bare_ghost_notice(ghost, post)
+            return _bare_ghost_notice(ghost, post) + collision_note
         except Exception as e:
             logger.warning("Auto-publish to Ghost failed", error=str(e))
             return ""
 
+    async def _collision_note(self, lang: str, title: str) -> str:
+        """Aviso de canibalización del título propuesto contra los posts
+        published+scheduled del sitio. Solo EN (las stopwords del motor son
+        inglesas). Nunca lanza y nunca bloquea: el draft se publica igual y
+        el aviso viaja en el notice de Telegram.
+        """
+        if lang != "en":
+            return ""
+        try:
+            from src.seo.autofill import collision_notice, fetch_collision_corpus
+            corpus = await fetch_collision_corpus(lang)
+            return collision_notice(title, corpus) or ""
+        except Exception as e:
+            logger.warning("Topic collision check failed — skipped", error=str(e))
+            return ""
+
     async def generate(self, session_id: int, output_type: OutputType,
                        progress_callback=None, lang: str = "es",
                        seo_override: str | None = None) -> str:

src/scraper/exhaustive.py

@@ -628,6 +628,11 @@ class ExhaustiveScraper:
                                         error="Content too short or empty")
             return
 
+        if len(content) > settings.max_content_length:
+            logger.info("Content truncated", source_id=source_id,
+                        original_length=len(content), url=url[:60])
+            content = content[:settings.max_content_length]
+
         word_count = len(content.split())
 
         await self.db.save_source_content(source_id, content)
@@ -819,30 +824,49 @@ class ExhaustiveScraper:
                     entries=len(entries), added=added)
         return added
 
+    # pdfplumber es síncrono y CPU-intensivo: parsear inline congela el event
+    # loop, y con PDFs grandes el pico de RAM puede matar el pod (OOM con
+    # límite de 1-2Gi). Ejecutar SIEMPRE vía run_in_executor.
+
+    @staticmethod
+    def _parse_pdf_sync(path: str) -> str:
+        import pdfplumber
+
+        with pdfplumber.open(path) as pdf:
+            pages = []
+            for page in pdf.pages[:50]:  # max 50 pages
+                pages.append(page.extract_text() or "")
+                page.flush_cache()  # pdfplumber cachea objetos de página: liberar
+            return "\n\n".join(pages)
+
     async def _extract_pdf(self, url: str) -> tuple[Optional[str], Optional[str]]:
         """Download and extract PDF text"""
-        import pdfplumber
         import tempfile
         import os
 
+        max_pdf_bytes = 15 * 1024 * 1024  # varios PDFs concurrentes en RAM: cap agresivo
+
         http = await self._get_http()
         try:
             async with http.get(url) as resp:
                 if resp.status != 200:
                     return None, None
                 content_length = int(resp.headers.get("content-length", 0))
-                if content_length > 50 * 1024 * 1024:  # skip PDFs > 50MB
+                if content_length > max_pdf_bytes:
                     return None, None
                 pdf_bytes = await resp.read()
+                # Sin Content-Length el check anterior no protege
+                if len(pdf_bytes) > max_pdf_bytes:
+                    return None, None
 
             with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f:
                 f.write(pdf_bytes)
                 tmp_path = f.name
+            del pdf_bytes
 
             try:
-                with pdfplumber.open(tmp_path) as pdf:
+                loop = asyncio.get_running_loop()
-                    pages = [p.extract_text() or "" for p in pdf.pages[:50]]  # max 50 pages
+                text = await loop.run_in_executor(None, self._parse_pdf_sync, tmp_path)
-                    text = "\n\n".join(pages)
                 return text, url.split("/")[-1]
             finally:
                 os.unlink(tmp_path)

src/seo/.rules.sha256

@@ -1 +1 @@
-b8faa93b1f7727d3870e18f69b68283d9a97ed9da819ef0cfa79a60cc2c4ab70
+73872aaf4079e5a0b690fbbe7de7e7967edc2297c8bad79142308714452c60ef

src/seo/autofill.py

@@ -650,3 +650,62 @@ def insert_internal_links(
                          phrase=phrase, slug=slug)
 
     return "".join(tokens), inserted_pairs
+
+# ─── 4. Topic collision (aviso pre-publish) ──────────────────────────────────
+
+_MD_UNSAFE = re.compile(r"[*_`\[\]]")
+
+
+def _slugify_title(title: str) -> str:
+    """Aproximación del slug que Ghost generará del título — el draft aún no
+    tiene slug real, y topic_collision usa el slug como una de sus señales."""
+    return re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")
+
+
+async def fetch_collision_corpus(lang: str) -> list[dict]:
+    """Posts published+scheduled del sitio (id, slug, title, status) para el
+    check de colisión de tema. A diferencia de fetch_published_menu incluye
+    los PROGRAMADOS: chocar con la cola de vacaciones es justo el caso a
+    cazar (2026-07-10: segundo Kecksburg publicado con otro ya en cola).
+    Aislamiento total: cualquier fallo → [] y log, nunca raise.
+    """
+    try:
+        # Lazy import to avoid a heavy/circular import at module load.
+        from src.generator.generator import GhostPublisher
+
+        pub = GhostPublisher(lang=lang)
+        if not pub.is_configured():
+            return []
+        data = await pub._admin_get(
+            "posts/?filter=status:[published,scheduled]"
+            "&fields=id,slug,title,status&limit=all",
+            timeout=30,
+        )
+        if data is None:
+            return []
+        corpus = [p for p in data.get("posts", []) if p.get("slug")]
+        logger.info("seo.collision: corpus fetched", lang=lang, count=len(corpus))
+        return corpus
+    except Exception as e:  # noqa: BLE001 — isolation guarantee
+        logger.warning("seo.collision: corpus fetch failed", lang=lang, error=str(e))
+        return []
+
+
+def collision_notice(title: str, corpus: list[dict]) -> str | None:
+    """Aviso (Markdown seguro para Telegram) si el título propuesto colisiona
+    con un post existente, vía el motor vendorizado R.topic_collision.
+    None si no hay colisión. Los títulos ajenos se sanean de entidades
+    Markdown para no romper el parseo del mensaje (ver regla de _safe_send).
+    """
+    if not corpus:
+        return None
+    candidate = {"id": None, "title": title, "slug": _slugify_title(title)}
+    violations = R.topic_collision(candidate, corpus)
+    if not violations:
+        return None
+    lines = [_MD_UNSAFE.sub("", v.message) for v in violations[:3]]
+    return (
+        "\n\n🚨 *Posible colisión de tema* — el draft se ha creado igualmente:\n"
+        + "\n".join(f"• {line}" for line in lines)
+        + "\nAntes de publicar: fusionar, retitular a otro ángulo o enlazar a propósito."
+    )

src/seo/rules.py

@@ -195,6 +195,112 @@ def r_jsonld(p):
     return [Violation("jsonld.missing", MED, "no BlogPosting JSON-LD", "add JSON-LD")]
 
 
+# ---- topic collision (corpus-aware; NOT in RULES) ---------------------------
+# Two posts about the same case cannibalize each other in the SERP (2026-07-10:
+# a second Kecksburg post was published while another sat scheduled; a "When
+# Nuclear ... Went/Go Silent" near-twin title was already queued). RULES functions
+# are (post) -> violations; this one also needs the rest of the site, so callers
+# (seo_validate.py) pass the corpus explicitly: published + scheduled posts as
+# dicts with at least {id, title, slug}.
+
+TOPIC_STOPWORDS = {
+    # english glue
+    "the", "a", "an", "of", "and", "in", "at", "on", "to", "that", "what",
+    "when", "who", "why", "how", "its", "his", "her", "their", "our", "one",
+    "still", "cant", "couldnt", "went", "go", "goes", "most", "from", "with",
+    "they", "them", "these", "this", "are", "were", "was", "is", "be", "been",
+    "has", "have", "had", "but", "for", "all", "than", "then", "ever", "never",
+    # domain-generic (present in half the catalog — carry no case identity)
+    "ufo", "ufos", "uap", "uaps", "incident", "incidents", "case", "cases",
+    "file", "files", "mystery", "declassified", "declassification", "pentagon",
+    "government", "military", "congress", "secret", "program", "investigation",
+    "evidence", "witness", "witnesses", "document", "documents", "documented",
+    "unexplained", "encounter", "sighting", "sightings", "alien", "aliens",
+    "phenomena", "aerial", "unidentified", "extraordinary", "americas",
+    "american", "video", "footage",
+}
+# 0.70 calibrated 2026-07-10: the "When Nuclear Weapons Go / Arsenal Went
+# Silent" near-twin pair scores 0.742 (char-level penalizes weapons/arsenal);
+# the closest legit-distinct pair in the catalog scores 0.65.
+TITLE_HOOK_SIM_MIN = 0.70   # SequenceMatcher on the pre-colon hook
+SLUG_JACCARD_MIN   = 0.5    # shared slug-token ratio
+# Years >= this are "news era", not case identity: every contemporary post
+# carries the current year (PURSUE 2026, Grusch 2026...) without being the same
+# story. Case years in the catalog run 1947-2019.
+NEWS_YEAR_MIN = 2020
+
+_YEAR_RE = re.compile(r"\b(19|20)\d{2}\b")
+
+
+def _tokens(text):
+    return set(re.findall(r"[a-z0-9]+", _s(text).lower()))
+
+
+def _case_years(title, slug):
+    """Historical case years (pre news-era) found in title+slug."""
+    return {m.group(0) for m in _YEAR_RE.finditer(title + " " + slug)
+            if int(m.group(0)) < NEWS_YEAR_MIN}
+
+
+def _sig_tokens(title, slug):
+    """Case-identity tokens: title+slug minus glue/domain words, years and
+    fragments shorter than 3 chars (possessive 's', initials...)."""
+    toks = _tokens(title) | _tokens(slug.replace("-", " "))
+    return {t for t in toks
+            if len(t) >= 3 and t not in TOPIC_STOPWORDS and not _YEAR_RE.fullmatch(t)}
+
+
+def _hook(title):
+    return _s(title).split(":")[0].strip().lower()
+
+
+def topic_collision(post, corpus):
+    """Compare one candidate post against the site corpus → list[Violation].
+
+    Fires when the candidate and an existing post look like the same story:
+      - share a case year AND a case-identity token (Kecksburg+1965), or
+      - their pre-colon title hooks read nearly the same, or
+      - their slugs share most of their tokens.
+    """
+    from difflib import SequenceMatcher
+
+    out = []
+    c_years = _case_years(_s(post.get("title")), _s(post.get("slug")))
+    c_sig = _sig_tokens(post.get("title"), _s(post.get("slug")))
+    c_hook = _hook(post.get("title"))
+    c_slug_toks = _tokens(_s(post.get("slug")).replace("-", " "))
+
+    for other in corpus:
+        if other.get("id") == post.get("id"):
+            continue
+        o_title, o_slug = _s(other.get("title")), _s(other.get("slug"))
+        o_years = _case_years(o_title, o_slug)
+        o_sig = _sig_tokens(o_title, o_slug)
+
+        reasons = []
+        if (c_years & o_years) and (c_sig & o_sig):
+            shared = ", ".join(sorted(c_sig & o_sig)[:3] + sorted(c_years & o_years))
+            reasons.append((HIGH, f"same case + year ({shared})"))
+        hook_sim = SequenceMatcher(None, c_hook, _hook(o_title)).ratio()
+        if c_hook and hook_sim >= TITLE_HOOK_SIM_MIN:
+            reasons.append((MED, f"title hooks {hook_sim:.0%} similar"))
+        o_slug_toks = _tokens(o_slug.replace("-", " "))
+        union = c_slug_toks | o_slug_toks
+        if union:
+            jac = len(c_slug_toks & o_slug_toks) / len(union)
+            if jac >= SLUG_JACCARD_MIN:
+                reasons.append((MED, f"slugs {jac:.0%} overlapping"))
+
+        if reasons:
+            sev = max(s for s, _ in reasons)
+            why = "; ".join(r for _, r in reasons)
+            out.append(Violation(
+                "topic.collision", sev,
+                f"collides with [{other.get('status', '?')}] \"{o_title[:60]}\" — {why}",
+                "merge, retitle to a distinct angle, or interlink deliberately"))
+    return out
+
+
 RULES = [
     r_meta_title,
     r_meta_description,

tests/test_seo_autofill.py

@@ -37,3 +37,47 @@ def test_system_prompt_lists_allowed_tags_per_lang():
     assert 'never use "investigacion"' in en
     assert 'never use "investigacion"' not in es
     assert "ONLY from this exact list" in es
+
+
+# ─── topic collision ─────────────────────────────────────────────────────────
+
+from src.seo.autofill import collision_notice, _slugify_title
+
+CORPUS = [
+    {"id": "1", "status": "published", "slug": "kecksburg-1965-acorn-ufo-missing-nasa-files",
+     "title": 'Kecksburg 1965: The Acorn-Shaped Object, the Missing NASA Files, and "Pennsylvania\'s Roswell"'},
+    {"id": "2", "status": "scheduled", "slug": "uss-russell-2019-pyramid-uap-channel-islands",
+     "title": "USS Russell 2019: The Pyramid UAP Video and the Channel Islands Drone Swarm"},
+]
+
+
+def test_collision_fires_on_same_case_and_year():
+    note = collision_notice("Kecksburg 1965: New Acorn Evidence", CORPUS)
+    assert note is not None
+    assert "kecksburg" in note.lower()
+    assert "1965" in note
+
+
+def test_collision_none_on_distinct_case():
+    assert collision_notice("Tehran 1976: The Jet-Disabling Encounter", CORPUS) is None
+
+
+def test_collision_none_on_empty_corpus():
+    assert collision_notice("Kecksburg 1965: Anything", []) is None
+
+
+def test_collision_note_is_markdown_safe():
+    corpus = [{"id": "9", "status": "published", "slug": "weird-1990-case",
+               "title": "Weird *1990* [Case] with_underscores and `ticks`"}]
+    note = collision_notice("Weird 1990: Case Revisited", corpus)
+    assert note is not None
+    # las entidades Markdown de títulos ajenos se sanean (solo quedan las nuestras)
+    bullets = [line for line in note.split("\n") if line.startswith("• ")]
+    assert bullets
+    for line in bullets:
+        for ch in "*_`[]":
+            assert ch not in line
+
+
+def test_slugify_title():
+    assert _slugify_title("USS Russell 2019: The Pyramid UAP!") == "uss-russell-2019-the-pyramid-uap"