chore(deps): update dependency scikit-learn to v1.9.0

find_draft_by_title y fetch_published_menu duplicaban línea a línea el GET
admin de Ghost (token, URL, ClientSession, dict de headers, status check,
resp.json), y el dict Authorization/Accept-Version/Accept-Encoding estaba
copiado en 3 sitios — el incidente brotli ya demostró que el fix por copia
se deja sitios. Ahora los headers se construyen en un único punto
(_admin_headers, con SAFE_ACCEPT_ENCODING de config) y el GET en _admin_get;
publish_draft usa los mismos headers. aiohttp pasa a import de módulo en
generator.py (era 'import aiohttp as _aio' repetido dentro de dos métodos).

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

.gitea/workflows/build.yml

@@ -20,6 +20,21 @@ jobs:
         with:
           ssl-verify: false
 
+      - name: Verify vendored SEO engine
+        run: |
+          set -e
+          CANON_URL="http://gitea.gitea.svc.cluster.local:3000/chemavx/chemavx-seo-tools/raw/branch/main/seo_rules.py"
+          curl -fsS -u "chemavx:${{ secrets.CI_TOKEN }}" "$CANON_URL" -o /tmp/canonical_seo_rules.py
+          N=$(grep -n "BEGIN VENDORED seo_rules.py" src/seo/rules.py | head -1 | cut -d: -f1)
+          tail -n +$((N+1)) src/seo/rules.py > /tmp/vendored_body.py
+          if cmp -s /tmp/canonical_seo_rules.py /tmp/vendored_body.py; then
+            echo "OK: vendored SEO engine matches canonical chemavx-seo-tools/seo_rules.py"
+          else
+            echo "::error::SEO ENGINE DRIFT — src/seo/rules.py != canonical seo_rules.py. Run 'make sync-seo' and commit."
+            diff -u /tmp/canonical_seo_rules.py /tmp/vendored_body.py | head -40 || true
+            exit 1
+          fi
+
       - name: Set image tag
         id: tag
         run: echo "TAG=${GITHUB_SHA::8}" >> $GITHUB_OUTPUT

.gitignore

@@ -0,0 +1,35 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.egg-info/
+.eggs/
+build/
+dist/
+
+# Virtualenvs
+.venv/
+venv/
+env/
+
+# Secrets / config local
+.env
+.env.*
+!.env.example
+
+# Datos / runtime
+*.db
+*.db-wal
+*.db-shm
+/data/
+
+# Tests / coverage
+.pytest_cache/
+.coverage
+htmlcov/
+
+# Editor / OS
+.vscode/
+.idea/
+*.swp
+.DS_Store

KNOWN-ISSUES.md

@@ -0,0 +1,62 @@
+# ResearchOwl — Known Issues & Operational Gotchas
+
+## Brotli / aiohttp incompatibility
+
+Do NOT install any brotli backend (`Brotli`, `brotlicffi`) while on aiohttp 3.14.x.
+aiohttp's incremental brotli decompressor is broken and fails intermittently
+depending on stream chunking ("Can not decode content-encoding: br" on VALID
+streams — the same bytes decompress fine offline; httpx is unaffected).
+
+Extra trap: merely *installing* a backend makes aiohttp advertise `br` in the
+default `Accept-Encoding` of every session that doesn't set one explicitly.
+On 2026-07-04 a scraper fix installed brotlicffi and inadvertently enabled br
+project-wide: Ghost/Cloudflare responded in brotli, the decode blew up AFTER
+Ghost had already accepted the POST, and the publish fallback re-published →
+duplicate drafts + silently lost SEO notices.
+
+Fix applied (commits `397546a` + `76c927f` + `7d07375`, guard relocated
+2026-07-05):
+- no brotli backend in requirements.txt (comment there explains why),
+- explicit `Accept-Encoding: gzip, deflate` on all Ghost/autofill aiohttp
+  sessions and on the scraper HEADERS (`57f341f`),
+- duplicate guard INSIDE `GhostPublisher.publish_draft` (covers every caller,
+  including `/publish`): if the POST is accepted (2xx) but reading the response
+  fails, it recovers the just-created draft via `find_draft_by_title(title,
+  since=attempt_start)` instead of raising. The guard only fires after an
+  accepted POST — a pre-POST failure (Ollama down, menu fetch, links) never
+  triggers it, so a stale same-title draft from a previous run can no longer
+  swallow freshly generated content.
+
+Note: with no backend installed there is NO brotli fallback at all — a server
+that responds `Content-Encoding: br` without it being advertised (misbehaving
+CDN) fails undecodable and that source is lost. Accepted trade-off.
+
+The header value lives in one place: `SAFE_ACCEPT_ENCODING` in `src/config.py`.
+Every aiohttp session/request must use it explicitly — never rely on aiohttp's
+default Accept-Encoding, which silently grows `br` if a backend appears.
+
+Re-test with disclosure.org before ever re-enabling br (e.g. after an aiohttp
+upgrade).
+
+## SQLite WAL mode + read-only mounts
+
+The database runs in WAL mode, so even read-only access needs the `-shm` file
+writable, or must open with `sqlite3 "file:...?immutable=1"`. Backups (daily
+CronJob `researchowl-db-backup`, 03:00 Europe/Madrid, PVC `researchowl-backups`,
+7-day retention) inherit WAL mode: to inspect one from a read-only mount use
+`?immutable=1`; to restore, copy it to a writable location first.
+
+## DDG (duckduckgo_search) blocks the event loop
+
+`DDGS()` is synchronous (blocking requests inside). Never call it directly from
+async code — always wrap in `loop.run_in_executor()` (see `_ddg_text_sync` /
+`_ddg_videos_sync` in `src/scraper/exhaustive.py`). Direct calls froze the
+entire Telegram bot during searches until fixed on 2026-07-04 (`8dfd011`).
+
+## Google News RSS is a dead end from this infrastructure
+
+`news.google.com/rss` entry links point to `/rss/articles/CBMi…` redirects that
+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`).

Makefile

@@ -0,0 +1,30 @@
+# ResearchOwl — developer tasks.
+#
+# SEO engine vendoring: src/seo/rules.py is a byte-for-byte copy of the canonical
+# seo_rules.py (repo git.chemavx.xyz/chemavx/chemavx-seo-tools; local working copy
+# ~/seo-tools). These targets keep the vendored copy in sync; CI enforces it too.
+
+SEO_TOOLS_DIR ?= $(HOME)/seo-tools
+CANON         := $(SEO_TOOLS_DIR)/seo_rules.py
+VENDORED      := src/seo/rules.py
+HEADER        := src/seo/_vendor_header.py
+MARKER        := BEGIN VENDORED seo_rules.py
+
+.PHONY: sync-seo check-seo-sync
+
+sync-seo:          ## Re-copy canonical seo_rules.py into the vendored file + record hash
+	@test -f "$(CANON)" || { echo "canonical not found at $(CANON)"; exit 1; }
+	@cat "$(HEADER)" "$(CANON)" > "$(VENDORED)"
+	@sha256sum "$(CANON)" | cut -d' ' -f1 > src/seo/.rules.sha256
+	@echo "synced $(VENDORED) from $(CANON) (sha $$(cat src/seo/.rules.sha256))"
+
+check-seo-sync:    ## Fail if the vendored copy diverges from the local canonical
+	@test -f "$(CANON)" || { echo "canonical not found at $(CANON) — skipping (run on a machine with seo-tools)"; exit 0; }
+	@N=$$(grep -n "$(MARKER)" "$(VENDORED)" | head -1 | cut -d: -f1); \
+	 tail -n +$$((N+1)) "$(VENDORED)" > /tmp/_vendored_body.py; \
+	 if cmp -s /tmp/_vendored_body.py "$(CANON)"; then \
+	   echo "seo engine in sync ($(VENDORED) == $(CANON))"; \
+	 else \
+	   echo "SEO ENGINE DRIFT: $(VENDORED) != $(CANON). Run 'make sync-seo' and commit."; \
+	   diff -u "$(CANON)" /tmp/_vendored_body.py | head -40; exit 1; \
+	 fi

requirements.txt

@@ -1,38 +1,41 @@
 # Core
-fastapi==0.115.0
-uvicorn==0.30.0
-python-telegram-bot==21.5
-httpx==0.27.0
-aiohttp==3.10.0
+python-telegram-bot==22.8
+httpx==0.28.1
+aiohttp==3.14.1
+# NO instalar backend brotli (Brotli/brotlicffi): el decode incremental br de
+# aiohttp 3.14 está roto (falla con streams válidos según el troceo), y con un
+# backend presente aiohttp anuncia br POR DEFECTO en toda sesión sin
+# Accept-Encoding explícito — el 2026-07-04 eso rompió publish_draft/autofill
+# contra Ghost/Cloudflare (drafts duplicados). Sin backend, nada anuncia br.
 
 # Scraping
-beautifulsoup4==4.12.3
-lxml==5.2.2
-trafilatura==1.12.0
-youtube-transcript-api==0.6.2
-pdfplumber==0.11.3
-feedparser==6.0.11
-duckduckgo-search==6.2.6
+beautifulsoup4==4.15.0
+lxml==5.4.0
+trafilatura==1.12.2
+youtube-transcript-api==0.6.3
+pdfplumber==0.11.10
+feedparser==6.0.12
+duckduckgo-search==6.4.2
 
 # Storage & Embeddings
-sqlite-vec==0.1.6
-aiosqlite==0.20.0
+sqlite-vec==0.1.9
+aiosqlite==0.22.1
 
 # Processing
 tiktoken==0.7.0
 numpy==1.26.4
-scikit-learn==1.5.1
+scikit-learn==1.9.0
 
 # Claude API (scoring)
 anthropic>=0.40.0
 
 # PDF export
-markdown==3.7
+markdown==3.10.2
 reportlab==4.2.5
 
 # Utilities
-pydantic==2.8.0
-pydantic-settings==2.4.0
+pydantic==2.13.4
+pydantic-settings==2.14.2
 tenacity==9.0.0
 structlog==24.4.0
-python-dotenv==1.0.1
+python-dotenv==1.2.2

src/bot/bot.py

@@ -5,11 +5,12 @@ Main user interface — all commands handled here
 import asyncio
 import os
 import time
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 from typing import Optional
+from zoneinfo import ZoneInfo
 
 import structlog
-from telegram import Update, Message
+from telegram import Update, Message, LinkPreviewOptions
 from telegram.ext import (
     Application, CommandHandler, MessageHandler,
     filters, ContextTypes
@@ -17,10 +18,11 @@ from telegram.ext import (
 from telegram.constants import ParseMode
 
 from src.config import settings
-from src.db.database import get_db, ResearchDB, ResearchStatus, OutputType
+from src.db.database import get_db, close_db, ResearchDB, ResearchStatus, OutputType
 from src.scraper.exhaustive import ExhaustiveScraper
 from src.processor.processor import OllamaClient, ContentProcessor
 from src.generator.generator import OutputGenerator
+from src.news.monitor import poll_feeds, item_from_row, format_digest
 
 logger = structlog.get_logger()
 
@@ -277,7 +279,11 @@ async def cmd_generate(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
 
     chat_id = update.effective_chat.id
     output_arg = ctx.args[0].lower() if ctx.args else ""
-    lang = "en" if len(ctx.args) > 1 and ctx.args[1].lower() == "en" else "es"
+    rest = [a.lower() for a in ctx.args[1:]]
+    lang = "en" if "en" in rest else "es"
+    # `/generate blog en dry` forces SEO dry-run for this one call (proposes SEO to
+    # Telegram, writes a bare draft). Global default still comes from SEO_AUTOFILL.
+    seo_override = "dryrun" if ("dry" in rest or "dryrun" in rest) else None
 
     type_map = {
         "podcast": OutputType.PODCAST,
@@ -346,7 +352,8 @@ async def cmd_generate(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
         processor = ContentProcessor(db, ollama)
         generator = OutputGenerator(db, ollama, processor)
 
-        output = await generator.generate(session_id, output_type, gen_progress, lang=lang)
+        output = await generator.generate(session_id, output_type, gen_progress,
+                                           lang=lang, seo_override=seo_override)
 
         # Send as file if very long
         if len(output) > 8000:
@@ -383,6 +390,18 @@ async def cmd_generate(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
         else:
             await send_chunked(update.message, output)
 
+        # SEO autofill / dry-run summary — a SEPARATE short message so it is never
+        # buried inside the long .md document. None on the flag-off path.
+        if getattr(generator, "last_publish_notice", None):
+            try:
+                await update.message.reply_text(
+                    generator.last_publish_notice,
+                    parse_mode=ParseMode.MARKDOWN,
+                    link_preview_options=LinkPreviewOptions(is_disabled=True),
+                )
+            except Exception as e:
+                logger.warning("Failed to send SEO summary message", error=str(e))
+
         try:
             stats = await db.get_usage_stats(session_id)
             total_cost = sum(s.get("total_cost", 0) for s in stats)
@@ -485,6 +504,36 @@ async def cmd_outputs(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
         await db_conn.close()
 
 
+async def cmd_news(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
+    """Monitor RSS manual (sin scheduler — F2 lo automatiza). Refresca news_seen
+    (siembra en cold-start) y muestra las novedades matcheadas de las últimas 24h."""
+    if not is_authorized(update.effective_user.id):
+        return
+
+    db_conn = await get_db()
+    db = ResearchDB(db_conn)
+    try:
+        await update.message.reply_text("📡 Buscando novedades UAP/OVNI…")
+        try:
+            await poll_feeds(db, settings)  # refresca tabla (siembra en cold-start)
+        except Exception:
+            logger.exception("news poll failed")  # best-effort: nunca rompe el comando
+
+        recent = await db.get_recent_news(24)
+        if not recent:
+            await update.message.reply_text("Sin novedades UAP/OVNI en las últimas 24h.")
+            return
+
+        items = [item_from_row(r) for r in recent]
+        for chunk in format_digest(items):
+            # Previews activos a propósito (el digest gana con la tarjeta del link).
+            await update.message.reply_text(
+                chunk, link_preview_options=LinkPreviewOptions(is_disabled=False)
+            )
+    finally:
+        await db_conn.close()
+
+
 async def cmd_costs(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
     if not is_authorized(update.effective_user.id):
         return
@@ -536,22 +585,62 @@ async def cmd_costs(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
         await db_conn.close()
 
 
+def _parse_at_time(time_str: str) -> tuple[float, int, int, bool]:
+    """Parse 'HH:MM' in the configured timezone.
+
+    Returns (next_run_at_unix, hour, minute, is_today). If the time has
+    already passed today, schedules for tomorrow (is_today=False).
+    Raises ValueError if the string is not a valid HH:MM time.
+    """
+    parts = time_str.split(":")
+    if len(parts) != 2 or not (parts[0].isdigit() and parts[1].isdigit()):
+        raise ValueError(f"Hora inválida: {time_str!r}")
+    hour, minute = int(parts[0]), int(parts[1])
+    if not (0 <= hour <= 23 and 0 <= minute <= 59):
+        raise ValueError(f"Hora fuera de rango: {time_str!r}")
+
+    tz = ZoneInfo(settings.timezone)
+    now_local = datetime.now(tz)
+    target = now_local.replace(hour=hour, minute=minute, second=0, microsecond=0)
+    is_today = True
+    if target <= now_local:
+        target += timedelta(days=1)
+        is_today = False
+    return target.timestamp(), hour, minute, is_today
+
+
 async def cmd_watch(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
     if not is_authorized(update.effective_user.id):
         return
 
     chat_id = update.effective_chat.id
-    args = ctx.args or []
+    args = list(ctx.args or [])
 
     if not args:
         await update.message.reply_text(
-            "❌ Uso: `/watch <tema> [horas]`\nEjemplo: `/watch Incidente Roswell 24`",
+            "❌ Uso: `/watch <tema> [horas]` o `/watch <tema> --at HH:MM [horas]`\n"
+            "Ejemplo: `/watch Incidente Roswell 24`\n"
+            "Ejemplo: `/watch Incidente Roswell --at 19:30 6`",
             parse_mode=ParseMode.MARKDOWN
         )
         return
 
+    # Extract optional --at HH:MM from anywhere in the args
+    at_time_str: Optional[str] = None
+    if "--at" in args:
+        idx = args.index("--at")
+        if idx + 1 >= len(args):
+            await update.message.reply_text(
+                "❌ `--at` requiere una hora en formato HH:MM. Ejemplo: `--at 19:30`",
+                parse_mode=ParseMode.MARKDOWN
+            )
+            return
+        at_time_str = args[idx + 1]
+        # Remove '--at' and its value from args
+        del args[idx:idx + 2]
+
     interval_hours = 24
-    if args[-1].isdigit():
+    if args and args[-1].isdigit():
         interval_hours = int(args[-1])
         topic = " ".join(args[:-1]).strip()
     else:
@@ -567,14 +656,31 @@ async def cmd_watch(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
         )
         return
 
+    next_run_at: Optional[float] = None
+    when_msg = f"Primera ejecución en ~{interval_hours}h."
+    if at_time_str is not None:
+        try:
+            next_run_at, hour, minute, is_today = _parse_at_time(at_time_str)
+        except ValueError:
+            await update.message.reply_text(
+                "❌ Hora inválida. Usa el formato HH:MM (24h). Ejemplo: `--at 19:30`",
+                parse_mode=ParseMode.MARKDOWN
+            )
+            return
+        day_word = "hoy" if is_today else "mañana"
+        when_msg = (
+            f"Primera ejecución: {day_word} a las {hour:02d}:{minute:02d} "
+            f"({settings.timezone})"
+        )
+
     db_conn = await get_db()
     db = ResearchDB(db_conn)
     try:
         try:
-            await db.add_watch(topic, chat_id, interval_hours)
+            await db.add_watch(topic, chat_id, interval_hours, next_run_at=next_run_at)
             await update.message.reply_text(
                 f"👁 Watching: `{topic}` — cada {interval_hours}h\n"
-                f"Primera ejecución en ~{interval_hours}h.\n"
+                f"{when_msg}\n"
                 f"Usa /watches para ver todos tus temas.",
                 parse_mode=ParseMode.MARKDOWN
             )
@@ -633,14 +739,27 @@ async def cmd_watches(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
             return
 
         now = time.time()
+        tz = ZoneInfo(settings.timezone)
+        today_local = datetime.now(tz).date()
         lines = ["👁 *Tus temas vigilados:*\n"]
         for i, w in enumerate(watches, 1):
             secs_remaining = max(0.0, w["next_run_at"] - now)
             hours_remaining = secs_remaining / 3600
             eta = f"{int(secs_remaining / 60)}min" if hours_remaining < 1 else f"{hours_remaining:.1f}h"
             status = "✅" if w["enabled"] else "⏸"
+
+            nxt = datetime.fromtimestamp(w["next_run_at"], tz)
+            if nxt.date() == today_local:
+                day_word = "hoy"
+            elif nxt.date() == today_local + timedelta(days=1):
+                day_word = "mañana"
+            else:
+                day_word = nxt.strftime("%d/%m")
+            local_time = nxt.strftime("%H:%M")
+
             lines.append(
-                f"{i}. {status} `{w['topic']}` — cada {w['interval_hours']}h · próxima en {eta}"
+                f"{i}. {status} `{w['topic']}` — cada {w['interval_hours']}h · "
+                f"próxima a las {local_time} ({day_word}) · en {eta}"
             )
 
         await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN)
@@ -745,7 +864,27 @@ async def _purge_on_startup(app: Application) -> None:
         await db_conn.close()
 
 
+async def _safe_send(bot, chat_id, text: str):
+    """Envía un mensaje, con fallback a texto plano si falla el parseo Markdown.
+
+    Los resúmenes generados por Claude suelen contener entidades Markdown
+    desbalanceadas (*, _, [, `) que hacen que Telegram rechace el mensaje con
+    BadRequest. Sin este fallback, el envío fallaba en silencio.
+    """
+    try:
+        await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN)
+    except Exception as e:
+        logger.warning("Envío Markdown falló, reintentando en texto plano", error=str(e))
+        try:
+            await bot.send_message(chat_id, text)
+        except Exception as e2:
+            logger.error("Envío en texto plano también falló", chat_id=chat_id, error=str(e2))
+
+
 async def _scheduler_loop(app: Application):
+    # Estado en memoria (no en DB): en un reinicio re-polla una vez, inofensivo
+    # porque solo se empujan items notified=0 y se marcan tras enviarlos.
+    _last_news_poll: Optional[datetime] = None
     while True:
         db_conn = None
         try:
@@ -761,7 +900,7 @@ async def _scheduler_loop(app: Application):
                 _active_sessions[chat_id] = session_id
                 await db.update_watch_run(watch["id"])
 
-                async def _task(c=chat_id, t=topic, s=session_id, w_id=watch["id"]):
+                async def _task(c=chat_id, t=topic, s=session_id):
                     inner_db_conn = await get_db()
                     inner_db = ResearchDB(inner_db_conn)
                     try:
@@ -787,25 +926,58 @@ async def _scheduler_loop(app: Application):
                             )
 
                         if summary:
-                            await app.bot.send_message(
-                                c, summary, parse_mode=ParseMode.MARKDOWN
-                            )
+                            await _safe_send(app.bot, c, summary)
                         else:
-                            await app.bot.send_message(
-                                c,
-                                f"🔄 *{t}* — sin novedades significativas esta vez.",
-                                parse_mode=ParseMode.MARKDOWN
+                            await _safe_send(
+                                app.bot, c,
+                                f"🔄 *{t}* — sin novedades significativas esta vez."
                             )
+                    except Exception as e:
+                        logger.error("Tarea programada falló",
+                                     topic=t, session_id=s, error=str(e))
                     finally:
                         await inner_db_conn.close()
 
                 task = asyncio.create_task(_task())
                 _active_tasks[chat_id] = task
-                await app.bot.send_message(
-                    chat_id,
-                    f"🔄 Investigación automática iniciada: `{topic}`",
-                    parse_mode=ParseMode.MARKDOWN
+                await _safe_send(
+                    app.bot, chat_id,
+                    f"🔄 Investigación automática iniciada: `{topic}`"
                 )
+
+            # --- Monitor de noticias (F2) — inerte si NEWS_ENABLED=False.
+            # Best-effort: un fallo del news-poll NUNCA tumba el scheduler de
+            # watched_topics (va en su propio try/except).
+            if settings.NEWS_ENABLED:
+                now = datetime.now(timezone.utc)
+                interval = timedelta(hours=settings.NEWS_POLL_INTERVAL_HOURS)
+                due_news = (_last_news_poll is None
+                            or (now - _last_news_poll) >= interval)
+                if due_news:
+                    _last_news_poll = now  # marca antes de pollear: evita reintentos en bucle si falla
+                    news_chat_id = settings.news_chat_id
+                    if not news_chat_id:
+                        logger.warning("NEWS_ENABLED pero sin chat destino; salto poll")
+                    else:
+                        try:
+                            await poll_feeds(db, settings)        # inserta novedades notified=0
+                            pending = await db.get_unnotified()   # TODOS los pendientes
+                            if pending:
+                                shown = pending[:settings.NEWS_MAX_ITEMS]
+                                extra = len(pending) - len(shown)
+                                items = [item_from_row(r) for r in shown]
+                                chunks = format_digest(items, header="🛸 Novedades UAP/OVNI")
+                                if extra > 0 and chunks:
+                                    chunks[-1] += f"\n…y {extra} más."
+                                for c in chunks:
+                                    await app.bot.send_message(
+                                        chat_id=news_chat_id, text=c,
+                                        link_preview_options=LinkPreviewOptions(is_disabled=False),
+                                    )
+                                # Marca todos los pendientes (incluidos los "…y N más")
+                                await db.mark_news_notified([r["id"] for r in pending])
+                        except Exception as e:
+                            logger.warning("News poll/notify falló", error=str(e))
         except Exception as e:
             logger.warning("Scheduler loop error", error=str(e))
         finally:
@@ -826,6 +998,10 @@ async def _on_startup(app: Application) -> None:
     await _start_scheduler(app)
 
 
+async def _on_shutdown(app: Application) -> None:
+    await close_db()
+
+
 async def cmd_export(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
     if not is_authorized(update.effective_user.id):
         return
@@ -1172,6 +1348,7 @@ def create_bot() -> Application:
         Application.builder()
         .token(settings.telegram_bot_token)
         .post_init(_on_startup)
+        .post_shutdown(_on_shutdown)
         .build()
     )
 
@@ -1183,6 +1360,7 @@ def create_bot() -> Application:
     app.add_handler(CommandHandler("generate", cmd_generate))
     app.add_handler(CommandHandler("sources", cmd_sources))
     app.add_handler(CommandHandler("outputs", cmd_outputs))
+    app.add_handler(CommandHandler("news", cmd_news))
     app.add_handler(CommandHandler("export", cmd_export))
     app.add_handler(CommandHandler("costs", cmd_costs))
     app.add_handler(CommandHandler("watch", cmd_watch))

src/config.py

@@ -1,49 +1,99 @@
-from pydantic_settings import BaseSettings
+from pydantic_settings import BaseSettings, SettingsConfigDict
 from pydantic import Field
 from typing import Optional
 
+# Accept-Encoding para TODA petición aiohttp del proyecto. Nunca anunciar "br":
+# el descompresor incremental de aiohttp 3.14 está roto, y con un backend brotli
+# instalado aiohttp lo anunciaría POR DEFECTO en cualquier sesión sin
+# Accept-Encoding explícito (incidente 2026-07-04, ver KNOWN-ISSUES.md).
+# Toda sesión/petición nueva debe usar esta constante, no un literal.
+SAFE_ACCEPT_ENCODING = "gzip, deflate"
+
 
 class Settings(BaseSettings):
     # Telegram
-    telegram_bot_token: str = Field(..., env="TELEGRAM_BOT_TOKEN")
-    telegram_allowed_users: str = Field("", env="TELEGRAM_ALLOWED_USERS")  # comma-separated user IDs
+    telegram_bot_token: str = Field(...)
+    telegram_allowed_users: str = Field("")  # comma-separated user IDs
 
     # Ollama
-    ollama_url: str = Field("http://ollama.chemavx.xyz", env="OLLAMA_URL")
-    ollama_model: str = Field("qwen2.5:3b", env="OLLAMA_MODEL")
-    ollama_embed_model: str = Field("qwen2.5:3b", env="OLLAMA_EMBED_MODEL")
+    ollama_url: str = Field("http://ollama.chemavx.xyz")
+    ollama_model: str = Field("qwen2.5:3b")
+    ollama_embed_model: str = Field("qwen2.5:3b")
 
     # Claude fallback (optional)
-    anthropic_api_key: Optional[str] = Field(None, env="ANTHROPIC_API_KEY")
-    claude_model: str = Field("claude-haiku-4-5", env="CLAUDE_MODEL")
+    anthropic_api_key: Optional[str] = Field(None)
+    claude_model: str = Field("claude-haiku-4-5")
 
     # Database
-    db_path: str = Field("/data/researchowl.db", env="DB_PATH")
+    db_path: str = Field("/data/researchowl.db")
 
     # Scraping
-    max_depth: int = Field(3, env="MAX_DEPTH")                      # recursion depth
-    max_sources: int = Field(150, env="MAX_SOURCES")                # hard cap
-    max_pages_per_search: int = Field(5, env="MAX_PAGES_PER_SEARCH")
-    request_timeout: int = Field(30, env="REQUEST_TIMEOUT")
-    request_delay: float = Field(1.0, env="REQUEST_DELAY")          # seconds between requests
-    min_content_length: int = Field(200, env="MIN_CONTENT_LENGTH")  # chars
+    max_depth: int = Field(3)                      # recursion depth
+    max_sources: int = Field(150)                # hard cap
+    max_pages_per_search: int = Field(5)
+    searxng_url: str = Field(
+        "http://searxng-svc.researchowl.svc.cluster.local:8080/search",
+    )
+    # Motores que el scraper pide a SearXNG (todos vienen configurados por los
+    # defaults de la instancia, use_default_settings: true). Verificados en vivo
+    # 2026-07-03: la lista ampliada duplica resultados y dominios únicos (~2x)
+    # por ~+2s/query, y da redundancia cuando brave/DDG caen en rate limit o
+    # CAPTCHA. Excluidos: qwant (access denied), yahoo (roto), wikidata (0
+    # resultados en queries de texto), wayback machine (duplica contenido vivo).
+    searxng_engines: str = Field(
+        "duckduckgo,google,bing,brave,startpage,mojeek,presearch,"
+        "google news,bing news,internet archive scholar"
+    )
+    request_timeout: int = Field(30)
+    request_delay: float = Field(1.0)          # seconds between requests
+    min_content_length: int = Field(200)  # chars
+
+    # Fuentes opcionales — desactivadas por defecto: la IP del homelab está
+    # bloqueada por Reddit (403) y YouTube (transcripts vacíos), eran peso muerto.
+    enable_youtube: bool = Field(False)
+    enable_reddit: bool = Field(False)
+    # Bing News RSS como semilla de actualidad — off hasta activarlo a propósito
+    enable_news_seed: bool = Field(False)
 
     # Processing
-    chunk_size: int = Field(800, env="CHUNK_SIZE")                  # tokens per chunk
-    chunk_overlap: int = Field(100, env="CHUNK_OVERLAP")
-    quality_threshold: float = Field(0.3, env="QUALITY_THRESHOLD")  # 0-1, chunks below discarded
+    chunk_size: int = Field(800)                  # tokens per chunk
+    chunk_overlap: int = Field(100)
+    quality_threshold: float = Field(0.3)  # 0-1, chunks below discarded
 
     # Ghost CMS
-    ghost_url: Optional[str] = Field(None, env="GHOST_URL")
-    ghost_api_key: Optional[str] = Field(None, env="GHOST_API_KEY")
-    ghost_url_en: str = Field("", env="GHOST_URL_EN")
-    ghost_api_key_en: str = Field("", env="GHOST_API_KEY_EN")
+    ghost_url: Optional[str] = Field(None)
+    ghost_api_key: Optional[str] = Field(None)
+    ghost_url_en: str = Field("")
+    ghost_api_key_en: str = Field("")
+
+    # SEO autofill — "off" | "on" | "dryrun" (default off).
+    #   off    = today's exact behavior (bare draft, no second LLM call).
+    #   on     = adds best-effort meta/OG/Twitter/tags/internal-links to the DRAFT
+    #            (status stays "draft" — NEVER auto-publishes; see src/seo/autofill.py).
+    #   dryrun = runs the full pipeline + reports the proposed SEO to Telegram, but
+    #            writes a BARE draft (no seo fields), for inspection before trusting
+    #            the write path. A `/generate blog en dry` arg forces dryrun per-call.
+    seo_autofill: str = Field("off")
+
+    # News monitor (RSS de noticias UAP/OVNI) — inerte hasta NEWS_ENABLED=true.
+    NEWS_ENABLED: bool = Field(False)
+    NEWS_POLL_INTERVAL_HOURS: int = Field(6)
+    # Chat al que el monitor empuja novedades; si None -> 1er TELEGRAM_ALLOWED_USERS
+    # (ver propiedad news_chat_id).
+    NEWS_CHAT_ID: Optional[int] = Field(None)
+    NEWS_KEYWORDS: str = Field(
+        "ovni,ovnis,uap,ufo,desclasificado,desclasificacion,declassified,"
+        "avistamiento,sighting,extraterrestre,non-human,nhi,grusch,"
+        "aaro,pursue,fenomeno aereo,whistleblower",
+    )
+    NEWS_MAX_ITEMS: int = Field(15)
 
     # Alerts
-    cost_alert_threshold: float = Field(0.15, env="COST_ALERT_THRESHOLD")
+    cost_alert_threshold: float = Field(0.15)
 
     # App
-    log_level: str = Field("INFO", env="LOG_LEVEL")
+    log_level: str = Field("INFO")
+    timezone: str = Field("Europe/Madrid")
 
     @property
     def allowed_user_ids(self) -> list[int]:
@@ -51,8 +101,25 @@ class Settings(BaseSettings):
             return []
         return [int(uid.strip()) for uid in self.telegram_allowed_users.split(",") if uid.strip()]
 
-    class Config:
-        env_file = ".env"
+    @property
+    def news_chat_id(self) -> Optional[int]:
+        """Chat destino del monitor de noticias: NEWS_CHAT_ID explícito o, si no
+        está definido, el primer TELEGRAM_ALLOWED_USERS."""
+        if self.NEWS_CHAT_ID is not None:
+            return self.NEWS_CHAT_ID
+        ids = self.allowed_user_ids
+        return ids[0] if ids else None
+
+    @property
+    def seo_autofill_enabled(self) -> bool:
+        """True when autofill should run (either live 'on' or 'dryrun')."""
+        return (self.seo_autofill or "").strip().lower() in ("on", "true", "1", "yes", "dryrun")
+
+    @property
+    def seo_autofill_dryrun(self) -> bool:
+        return (self.seo_autofill or "").strip().lower() == "dryrun"
+
+    model_config = SettingsConfigDict(env_file=".env")
 
 
 settings = Settings()

src/db/database.py

@@ -1,3 +1,4 @@
+import asyncio
 import aiosqlite
 import json
 import time
@@ -114,18 +115,84 @@ CREATE TABLE IF NOT EXISTS watched_topics (
     created_at REAL NOT NULL,
     UNIQUE(topic, chat_id)
 );
+
+CREATE TABLE IF NOT EXISTS news_seen (
+    id INTEGER PRIMARY KEY,
+    source TEXT NOT NULL,
+    guid TEXT NOT NULL,
+    title TEXT,
+    link TEXT,
+    published_at TEXT,
+    matched_keywords TEXT,
+    seen_at TEXT DEFAULT (datetime('now')),
+    notified INTEGER DEFAULT 0,
+    UNIQUE(source, guid)
+);
+
+CREATE INDEX IF NOT EXISTS idx_news_seen_source ON news_seen(source);
+CREATE INDEX IF NOT EXISTS idx_news_seen_seen ON news_seen(seen_at);
 """
 
 
+# Conexión única compartida por proceso. aiosqlite serializa todas las
+# operaciones en el hilo de la conexión, así que compartirla entre coroutines
+# es seguro y además evita la contención del lock de escritura a nivel de
+# fichero que sufrían las conexiones múltiples (scheduler, /compare).
+_shared_conn: Optional[aiosqlite.Connection] = None
+_init_lock = asyncio.Lock()
+
+
+class _SharedConnection:
+    """Proxy sobre la conexión compartida cuyo close() es no-op.
+
+    Permite que los handlers sigan usando el patrón `db = await get_db()` …
+    `finally: await db.close()` sin cambios, mientras por debajo se reutiliza
+    una sola conexión real durante toda la vida del proceso.
+    """
+    def __init__(self, conn: aiosqlite.Connection):
+        self._conn = conn
+
+    def __getattr__(self, name):
+        return getattr(self._conn, name)
+
+    async def close(self):
+        # No cerrar: la conexión es compartida por todo el proceso.
+        pass
+
+
+async def _init_shared() -> aiosqlite.Connection:
+    global _shared_conn
+    if _shared_conn is None:
+        async with _init_lock:
+            if _shared_conn is None:  # doble check tras adquirir el lock
+                Path(settings.db_path).parent.mkdir(parents=True, exist_ok=True)
+                conn = await aiosqlite.connect(settings.db_path)
+                conn.row_factory = aiosqlite.Row
+                await conn.execute("PRAGMA journal_mode=WAL")
+                await conn.execute("PRAGMA synchronous=NORMAL")
+                # Espera hasta 5s si OTRO proceso tiene el lock de escritura
+                # antes de fallar con "database is locked".
+                await conn.execute("PRAGMA busy_timeout=5000")
+                await conn.executescript(SCHEMA)
+                await conn.commit()
+                _shared_conn = conn
+                logger.info("Shared DB connection initialized", path=settings.db_path)
+    return _shared_conn
+
+
 async def get_db() -> aiosqlite.Connection:
-    Path(settings.db_path).parent.mkdir(parents=True, exist_ok=True)
-    db = await aiosqlite.connect(settings.db_path)
-    db.row_factory = aiosqlite.Row
-    await db.execute("PRAGMA journal_mode=WAL")
-    await db.execute("PRAGMA synchronous=NORMAL")
-    await db.executescript(SCHEMA)
-    await db.commit()
-    return db
+    conn = await _init_shared()
+    return _SharedConnection(conn)
+
+
+async def close_db() -> None:
+    """Cierra la conexión compartida (checkpoint WAL). Llamar al apagar el bot."""
+    global _shared_conn
+    if _shared_conn is not None:
+        try:
+            await _shared_conn.close()
+        finally:
+            _shared_conn = None
 
 
 class ResearchDB:
@@ -341,11 +408,26 @@ class ResearchDB:
 
     # --- API Usage ---
 
+    # Precios Claude en USD por 1M de tokens (input, output).
+    # Se busca por substring del id del modelo; fallback a Haiku.
+    _MODEL_PRICING = {
+        "opus":   (15.00, 75.00),
+        "sonnet": (3.00, 15.00),
+        "haiku":  (0.80, 4.00),
+    }
+
+    @classmethod
+    def _price_for_model(cls, model: str) -> tuple[float, float]:
+        m = (model or "").lower()
+        for key, price in cls._MODEL_PRICING.items():
+            if key in m:
+                return price
+        return cls._MODEL_PRICING["haiku"]
+
     async def log_api_call(self, session_id, call_type: str, model: str,
                             input_tokens: int, output_tokens: int):
-        # Precios Claude Haiku (claude-haiku-4-5):
-        # input: $0.80 / 1M tokens  output: $4.00 / 1M tokens
-        cost = (input_tokens * 0.80 + output_tokens * 4.00) / 1_000_000
+        in_price, out_price = self._price_for_model(model)
+        cost = (input_tokens * in_price + output_tokens * out_price) / 1_000_000
         await self.db.execute(
             """INSERT INTO api_usage
                (session_id, call_type, model, input_tokens, output_tokens, cost_usd, created_at)
@@ -378,13 +460,16 @@ class ResearchDB:
 
     # --- Watched Topics ---
 
-    async def add_watch(self, topic: str, chat_id: int, interval_hours: int) -> int:
+    async def add_watch(self, topic: str, chat_id: int, interval_hours: int,
+                        next_run_at: Optional[float] = None) -> int:
         now = time.time()
+        if next_run_at is None:
+            next_run_at = now + interval_hours * 3600
         cursor = await self.db.execute(
             """INSERT OR REPLACE INTO watched_topics
                (topic, chat_id, interval_hours, next_run_at, created_at)
                VALUES (?, ?, ?, ?, ?)""",
-            (topic, chat_id, interval_hours, now + interval_hours * 3600, now)
+            (topic, chat_id, interval_hours, next_run_at, now)
         )
         await self.db.commit()
         return cursor.lastrowid
@@ -427,6 +512,73 @@ class ResearchDB:
         )
         await self.db.commit()
 
+    # --- News monitor ---
+
+    async def source_seeded(self, source: str) -> bool:
+        """True si ya hemos visto algún item de esta fuente (cold-start por fuente:
+        el primer poll siembra sin notificar)."""
+        cursor = await self.db.execute(
+            "SELECT 1 FROM news_seen WHERE source = ? LIMIT 1", (source,)
+        )
+        return await cursor.fetchone() is not None
+
+    async def record_news_item(self, source: str, guid: str, title: str,
+                               link: str, published_at: str, matched: str,
+                               notified: int) -> bool:
+        """INSERT OR IGNORE de un item. Devuelve True solo si es una inserción
+        real (item nuevo); False si ya existía (UNIQUE(source, guid))."""
+        cursor = await self.db.execute(
+            """INSERT OR IGNORE INTO news_seen
+               (source, guid, title, link, published_at, matched_keywords, notified)
+               VALUES (?, ?, ?, ?, ?, ?, ?)""",
+            (source, guid, title, link, published_at, matched, notified)
+        )
+        await self.db.commit()
+        return cursor.rowcount > 0
+
+    async def mark_news_notified(self, ids: list[int]):
+        if not ids:
+            return
+        placeholders = ",".join("?" for _ in ids)
+        await self.db.execute(
+            f"UPDATE news_seen SET notified = 1 WHERE id IN ({placeholders})",
+            tuple(ids)
+        )
+        await self.db.commit()
+
+    async def get_recent_news(self, hours: int = 24) -> list[dict]:
+        """Items matcheados cuya seen_at (o published_at parseable) cae en la
+        ventana. Orden por published_at desc, con seen_at como desempate."""
+        window = f"-{int(hours)} hours"
+        cursor = await self.db.execute(
+            """SELECT * FROM news_seen
+               WHERE seen_at >= datetime('now', ?)
+                  OR (published_at IS NOT NULL AND published_at != ''
+                      AND datetime(published_at) >= datetime('now', ?))
+               ORDER BY published_at DESC, seen_at DESC
+               LIMIT 50""",
+            (window, window)
+        )
+        rows = await cursor.fetchall()
+        return [dict(r) for r in rows]
+
+    async def get_unnotified(self, limit: Optional[int] = None) -> list[dict]:
+        """Items aún sin notificar (notified=0), más recientes primero. El
+        scheduler (F2) los empuja a Telegram y luego los marca con
+        mark_news_notified. Aplica LIMIT solo si `limit` no es None."""
+        sql = (
+            "SELECT id, source, title, link, published_at, matched_keywords "
+            "FROM news_seen WHERE notified = 0 "
+            "ORDER BY published_at DESC NULLS LAST, seen_at DESC"
+        )
+        params: tuple = ()
+        if limit is not None:
+            sql += " LIMIT ?"
+            params = (int(limit),)
+        cursor = await self.db.execute(sql, params)
+        rows = await cursor.fetchall()
+        return [dict(r) for r in rows]
+
     # --- Maintenance ---
 
     async def purge_old_sessions(self, max_age_days: int = 30) -> dict:
@@ -439,7 +591,7 @@ class ResearchDB:
         )
         session_ids = [row[0] for row in await cursor.fetchall()]
 
-        counts = {"sessions": 0, "sources": 0, "chunks": 0, "outputs": 0}
+        counts = {"sessions": 0, "sources": 0, "chunks": 0, "outputs": 0, "api_usage": 0}
 
         for sid in session_ids:
             await self.db.execute(
@@ -450,6 +602,8 @@ class ResearchDB:
             counts["chunks"] += cur.rowcount
             cur = await self.db.execute("DELETE FROM outputs WHERE session_id = ?", (sid,))
             counts["outputs"] += cur.rowcount
+            cur = await self.db.execute("DELETE FROM api_usage WHERE session_id = ?", (sid,))
+            counts["api_usage"] += cur.rowcount
             cur = await self.db.execute("DELETE FROM sources WHERE session_id = ?", (sid,))
             counts["sources"] += cur.rowcount
             cur = await self.db.execute("DELETE FROM research_sessions WHERE id = ?", (sid,))

src/generator/generator.py

@@ -9,9 +9,11 @@ import json
 import re
 import time
 
+import aiohttp
 import structlog
 
-from src.config import settings
+from src.config import settings, SAFE_ACCEPT_ENCODING
+from src.llm import get_anthropic_client
 from src.processor.processor import OllamaClient, ContentProcessor
 from src.db.database import ResearchDB, OutputType
 
@@ -277,8 +279,134 @@ def _strip_researchowl_header(content: str) -> str:
     return content
 
 
+def _seo_checklist(slug: str) -> str:
+    """Static pre-publish SEO reminder appended to the draft-published notice.
+    NOT a validator call: the fresh draft has no meta yet, so checking now would
+    be all-fail noise. Jose runs `seo-check <slug>` after filling these in."""
+    return (
+        "\n\n⚠️ Antes de publicar, añade: meta title, meta description (≤145), "
+        "custom excerpt, OG/Twitter, feature image + alt (campo Alt, NO Caption), "
+        "2-3 internal links (donde sea natural)\n"
+        f"Luego valida:  seo-check {slug}"
+    )
+
+
+# Language-aware default tag when no seo/tags are supplied. Fixes the historical
+# hardcoded "investigacion" that mis-tagged EN posts; EN canonical is "investigation".
+_DEFAULT_TAG = {"en": "investigation", "es": "investigacion"}
+
+
+def _resolve_seo_mode(override: str | None = None) -> str:
+    """Effective SEO autofill mode: 'off' | 'on' | 'dryrun'.
+    A per-call override ('dryrun'/'on', e.g. from `/generate blog en dry`) wins;
+    otherwise it derives from settings.seo_autofill."""
+    if override in ("on", "dryrun"):
+        return override
+    if settings.seo_autofill_dryrun:
+        return "dryrun"
+    if settings.seo_autofill_enabled:
+        return "on"
+    return "off"
+
+
+def _seo_checklist_slim(slug: str) -> str:
+    """Slimmed checklist for the autofill path: meta/OG/excerpt/tags/links are
+    already on the draft, so only the human-only bits remain."""
+    return (
+        "⚠️ Antes de publicar: añade *feature image + alt* (campo Alt, NO Caption), luego:\n"
+        f"`seo-check {slug}`"
+    )
+
+
+def _bare_ghost_notice(ghost: "GhostPublisher", post: dict) -> str:
+    """Today's exact draft-published notice (editor link + full checklist).
+    Used for the flag-off path and as the autofill failure fallback."""
+    return (
+        f"\n\n---\n"
+        f"📤 *Borrador publicado en Ghost*\n"
+        f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}"
+        f"{_seo_checklist(post.get('slug', ''))}"
+    )
+
+
+def _seo_link_summary(inserted_pairs: list[dict], n_suggestions: int) -> str:
+    """Report ACTUALLY-INSERTED links only ("phrase → /slug/"), with a separate
+    count of well-formed suggestions that were skipped (no verbatim body match).
+    Never surfaces raw rejected suggestions."""
+    n = len(inserted_pairs)
+    skipped = max(0, n_suggestions - n)
+    head = f"{n} insertados" + (f", {skipped} omitidos" if skipped else "")
+    if not inserted_pairs:
+        return head
+    body = "; ".join(f"{p['phrase']} → /{p['slug']}/" for p in inserted_pairs)
+    return f"{head} — {body}"
+
+
+def _seo_live_message(ghost: "GhostPublisher", post: dict, seo: dict, inserted_pairs: list[dict]) -> str:
+    """Separate Telegram message for the live autofill path (SEO written to draft)."""
+    slug = post.get("slug", "")
+    warns = seo.get("seo_warnings") or []
+    warn_line = ("\n⚠️ revisar: " + "; ".join(warns)) if warns else ""
+    n_sug = len(seo.get("internal_links", []))
+    return (
+        f"📤 *Borrador en Ghost — SEO autorrelleno*\n"
+        f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}\n\n"
+        f"🔖 *SEO escrito en el draft:*\n"
+        f"• meta title ({len(seo['meta_title'])}): {seo['meta_title']}\n"
+        f"• meta desc ({len(seo['meta_description'])}): {seo['meta_description']}\n"
+        f"• excerpt: {len(seo['custom_excerpt'])} car.\n"
+        f"• tags: {', '.join(seo['tags'])}\n"
+        f"• internal links: {_seo_link_summary(inserted_pairs, n_sug)}"
+        f"{warn_line}\n\n"
+        f"🖼️ *Imagen sugerida:*\n"
+        f"`imagefinder --query \"{seo['image_query']}\" --context \"{seo['image_context']}\" --lang {ghost.lang}`\n\n"
+        f"{_seo_checklist_slim(slug)}"
+    )
+
+
+def _seo_dryrun_message(ghost: "GhostPublisher", post: dict, seo: dict, inserted_pairs: list[dict]) -> str:
+    """Separate Telegram message for DRY-RUN: bare draft created, SEO only proposed."""
+    slug = post.get("slug", "")
+    warns = seo.get("seo_warnings") or []
+    warn_line = ("\n⚠️ revisar: " + "; ".join(warns)) if warns else ""
+    n_sug = len(seo.get("internal_links", []))
+    return (
+        f"🧪 *DRY-RUN SEO* — draft creado SIN escribir SEO (solo revisión)\n"
+        f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}\n\n"
+        f"🔖 *SEO propuesto (NO escrito en Ghost):*\n"
+        f"• meta title ({len(seo['meta_title'])}): {seo['meta_title']}\n"
+        f"• meta desc ({len(seo['meta_description'])}): {seo['meta_description']}\n"
+        f"• excerpt ({len(seo['custom_excerpt'])} car.): {seo['custom_excerpt']}\n"
+        f"• tags: {', '.join(seo['tags'])}\n"
+        f"• internal links que se insertarían: {_seo_link_summary(inserted_pairs, n_sug)}"
+        f"{warn_line}\n\n"
+        f"🖼️ *Imagen sugerida:*\n"
+        f"`imagefinder --query \"{seo['image_query']}\" --context \"{seo['image_context']}\" --lang {ghost.lang}`\n\n"
+        f"{_seo_checklist_slim(slug)}"
+    )
+
+
+def _norm_title(t: str) -> str:
+    """Normaliza un título para compararlo con lo que Ghost almacenó: colapsa
+    whitespace y trunca a 255 chars (límite de posts.title en Ghost — un H1
+    largo del LLM puede volver truncado/normalizado, y la igualdad exacta
+    fallaría justo en los casos que el guard debe cubrir)."""
+    return " ".join((t or "").split())[:255]
+
+
+def _parse_ghost_ts(ts: str) -> float | None:
+    """created_at de la Admin API (ISO 8601, p.ej. 2026-07-04T20:26:00.000+00:00
+    o sufijo Z) → epoch. None si no parsea."""
+    try:
+        from datetime import datetime
+        return datetime.fromisoformat(ts.replace("Z", "+00:00")).timestamp()
+    except Exception:
+        return None
+
+
 class GhostPublisher:
     def __init__(self, lang: str = "es"):
+        self.lang = lang
         if lang == "en":
             self.url = (settings.ghost_url_en or "").rstrip("/")
             self.api_key = settings.ghost_api_key_en or ""
@@ -289,6 +417,16 @@ class GhostPublisher:
     def is_configured(self) -> bool:
         return bool(self.url and self.api_key)
 
+    def _build_html(self, markdown_content: str) -> str:
+        """Canonical markdown → Ghost body HTML conversion: strip the ResearchOwl
+        header, render markdown, drop the first <h1> (Ghost renders the title).
+        Shared so callers that need the linked HTML produce the SAME html that gets
+        posted (no drift between link-insertion and the mobiledoc body)."""
+        import markdown as _md
+        clean = _strip_researchowl_header(markdown_content)
+        html = _md.markdown(clean, extensions=["extra"])
+        return re.sub(r"<h1[^>]*>.*?</h1>", "", html, count=1, flags=re.DOTALL).lstrip()
+
     def _make_token(self) -> str:
         key_id, secret = self.api_key.split(":", 1)
         now = int(time.time())
@@ -300,19 +438,82 @@ class GhostPublisher:
         )
         return f"{signing}.{sig}"
 
+    def _admin_headers(self) -> dict:
+        """Headers canónicos de la Admin API (token JWT fresco por llamada).
+        Único sitio donde se construyen: cualquier cambio (Accept-Version,
+        encoding) se hace aquí y cubre todas las llamadas a Ghost."""
+        return {
+            "Authorization": f"Ghost {self._make_token()}",
+            "Accept-Version": "v5.0",
+            # Nunca heredar el default de aiohttp: con un backend brotli
+            # instalado anunciaría br y el decode roto de aiohttp 3.14 rompe
+            # la lectura de respuestas (incidente 2026-07-04, KNOWN-ISSUES.md).
+            "Accept-Encoding": SAFE_ACCEPT_ENCODING,
+        }
+
+    async def _admin_get(self, query: str, timeout: float = 15) -> dict | None:
+        """GET a {url}/ghost/api/admin/{query} con auth + headers canónicos.
+        None si non-200 (logueado); las excepciones de red/parseo suben tal
+        cual para que cada caller aplique su propia política best-effort."""
+        async with aiohttp.ClientSession(
+                timeout=aiohttp.ClientTimeout(total=timeout)) as sess:
+            async with sess.get(f"{self.url}/ghost/api/admin/{query}",
+                                headers=self._admin_headers()) as resp:
+                if resp.status != 200:
+                    body = await resp.text()
+                    logger.warning("Ghost admin GET non-200", query=query[:80],
+                                   status=resp.status, body=body[:200])
+                    return None
+                return await resp.json()
+
+    async def find_draft_by_title(self, title: str,
+                                  since: float | None = None) -> dict | None:
+        """Busca entre los drafts recientes uno con título exacto.
+
+        Mitad de recuperación del guard anti-duplicados de publish_draft: si el
+        POST fue aceptado por Ghost pero la lectura de la respuesta falló
+        (p.ej. decode roto), el draft existe aunque el caller viera una
+        excepción. Con `since` (epoch) solo casan drafts creados en este
+        intento — un draft viejo del mismo título de una run anterior NO cuenta
+        como "ya publicado". La comparación usa _norm_title (whitespace
+        colapsado + truncado a 255) para sobrevivir la normalización de Ghost.
+        Best-effort: None si no hay match o si falla."""
+        data = await self._admin_get(
+            "posts/?filter=status:draft&order=created_at%20desc&limit=15"
+            "&fields=id,title,slug,created_at"
+        )
+        if data is None:
+            return None
+        wanted = _norm_title(title)
+        for p in data.get("posts", []):
+            if _norm_title(p.get("title") or "") != wanted:
+                continue
+            if since is not None:
+                created = _parse_ghost_ts(p.get("created_at") or "")
+                # 60s de margen por posible desfase de reloj Ghost/bot.
+                if created is None or created < since - 60:
+                    continue
+            return p
+        return None
+
     async def publish_draft(self, title: str, markdown_content: str,
-                            tags: list[str] | None = None) -> dict:
-        import aiohttp as _aio
-        import markdown as _md
+                            tags: list[str] | None = None,
+                            seo: dict | None = None,
+                            body_html: str | None = None) -> dict:
+        """Create a Ghost DRAFT. status is ALWAYS "draft" — never published.
 
-        clean = _strip_researchowl_header(markdown_content)
-        html = _md.markdown(clean, extensions=["extra"])
-
-        # Ghost añade el título automáticamente — eliminar el primer <h1> para evitar duplicado
-        html = re.sub(r"<h1[^>]*>.*?</h1>", "", html, count=1, flags=re.DOTALL).lstrip()
+        Purely-additive SEO hooks (both default None → byte-identical to the
+        original behavior):
+          * body_html — pre-converted + internal-linked HTML from the caller; used
+            verbatim when given, else markdown_content is converted as before.
+          * seo — a generate_seo_fields dict; when given, its meta/OG/Twitter fields
+            are added to the post and its (allow-list) tags are used.
+        """
+        # Caller-supplied linked HTML wins; otherwise convert markdown as before.
+        html = body_html if body_html is not None else self._build_html(markdown_content)
 
         logger.info("Ghost publish_draft", html_length=len(html),
-                    html_preview=html[:200])
+                    html_preview=html[:200], seo=bool(seo))
 
         if not html.strip():
             raise ValueError("Ghost: HTML vacío tras conversión markdown — contenido no enviado")
@@ -328,28 +529,59 @@ class GhostPublisher:
             "sections": [[10, 0]],
         })
 
-        token = self._make_token()
-        body = {
-            "posts": [{
-                "title": title,
-                "mobiledoc": mobiledoc,
-                "status": "draft",
-                "tags": [{"name": t} for t in (tags or ["investigacion"])],
-            }]
+        # Language-aware default tag (fixes the hardcoded "investigacion" for EN).
+        tag_names = tags or [_DEFAULT_TAG.get(self.lang, "investigation")]
+        post_obj = {
+            "title": title,
+            "mobiledoc": mobiledoc,
+            "status": "draft",            # NEVER "published" — draft only, always.
+            "tags": [{"name": t} for t in tag_names],
         }
-        async with _aio.ClientSession() as sess:
+        if seo:
+            post_obj.update({
+                "meta_title": seo["meta_title"],
+                "meta_description": seo["meta_description"],
+                "custom_excerpt": seo["custom_excerpt"],
+                "og_title": seo["og_title"],
+                "og_description": seo["og_description"],
+                "twitter_title": seo["twitter_title"],
+                "twitter_description": seo["twitter_description"],
+            })
+        body = {"posts": [post_obj]}
+        attempt_start = time.time()
+        async with aiohttp.ClientSession() as sess:
             async with sess.post(
                 f"{self.url}/ghost/api/admin/posts/",
                 json=body,
-                headers={
-                    "Authorization": f"Ghost {token}",
-                    "Accept-Version": "v5.0",
-                },
+                headers=self._admin_headers(),
             ) as resp:
                 if resp.status not in (200, 201):
                     text = await resp.text()
                     raise ValueError(f"Ghost API {resp.status}: {text[:300]}")
-                return await resp.json()
+                # Guard anti-duplicados (incidente 2026-07-04): el 2xx confirma
+                # que Ghost YA creó el draft; si la lectura del body falla
+                # (p.ej. decode br roto), recuperamos el draft recién creado en
+                # vez de propagar — propagar hacía que el caller (o el usuario
+                # con /publish) reintentara y duplicara. El guard vive AQUÍ para
+                # cubrir a todos los callers, y solo se activa tras un POST
+                # aceptado: un fallo pre-POST jamás lo dispara.
+                try:
+                    return await resp.json()
+                except Exception as read_err:
+                    existing = None
+                    try:
+                        existing = await self.find_draft_by_title(
+                            title, since=attempt_start)
+                    except Exception as find_err:
+                        logger.warning("publish_draft: draft-exists check failed",
+                                       error=str(find_err))
+                    if existing:
+                        logger.warning(
+                            "publish_draft: POST aceptado pero lectura de la "
+                            "respuesta falló; draft recuperado sin re-publicar",
+                            post_id=existing["id"], error=str(read_err))
+                        return {"posts": [existing]}
+                    raise
 
 
 # ─── Output generation ────────────────────────────────────────────────────────
@@ -359,15 +591,91 @@ class OutputGenerator:
         self.db = db
         self.ollama = ollama
         self.processor = processor
+        # Set during a blog generation when the autofill/dry-run path runs: a short
+        # markdown SEO summary the bot sends as a SEPARATE Telegram message (so it is
+        # never buried inside the long .md document). None on the flag-off path.
+        self.last_publish_notice: str | None = None
+
+    async def _publish_blog_to_ghost(self, lang: str, full_output: str, topic: str,
+                                     session_id: int, seo_override: str | None) -> str:
+        """Publish a blog DRAFT to Ghost, gated by the SEO autofill mode.
+
+        Returns the ghost_notice to APPEND to the returned document (flag-off /
+        fallback bare-draft path only). For the autofill 'on'/'dryrun' paths it sets
+        self.last_publish_notice (a separate Telegram message) and returns "".
+
+        NEVER raises and NEVER blocks publishing: any autofill failure degrades to
+        today's exact bare-draft publish. status is always "draft".
+        """
+        ghost = GhostPublisher(lang=lang)
+        if not ghost.is_configured():
+            return ""
+        title = _extract_title(full_output) or topic
+        mode = _resolve_seo_mode(seo_override)
+
+        if mode in ("on", "dryrun"):
+            try:
+                from src.seo.autofill import (
+                    fetch_published_menu, generate_seo_fields, insert_internal_links,
+                )
+                article_md = _strip_researchowl_header(full_output)
+                menu = await fetch_published_menu(lang)
+                seo = await generate_seo_fields(
+                    article_md, menu, lang, title=title,
+                    db=self.db, session_id=session_id,
+                )
+                if seo is None:
+                    raise RuntimeError("generate_seo_fields returned None")
+                # Build the SAME html that will be posted, then link it deterministically.
+                html = ghost._build_html(article_md)
+                linked_html, inserted_pairs = insert_internal_links(
+                    html, seo["internal_links"], menu, lang)
+
+                if mode == "dryrun":
+                    # Bare draft (today's write path) — do NOT write seo fields; only
+                    # 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)
+                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)
+                logger.info("Auto-published blog to Ghost",
+                            mode=mode, post_id=post["id"], links=len(inserted_pairs))
+                return ""
+            except Exception as e:
+                logger.warning("SEO autofill failed; falling back to bare draft",
+                               error=str(e))
+                # Sin guard anti-duplicados aquí: vive DENTRO de publish_draft
+                # (solo se activa tras un POST aceptado). Un fallo pre-POST
+                # (Ollama caído, menú, links) cae aquí y DEBE publicar el
+                # contenido nuevo — un guard por título casaba con drafts
+                # viejos del mismo topic y lo descartaba en silencio.
+                # fall through to the bare-draft publish below
+
+        # OFF mode, or autofill failed → today's exact bare-draft publish.
+        try:
+            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)
+        except Exception as e:
+            logger.warning("Auto-publish to Ghost failed", error=str(e))
+            return ""
 
     async def generate(self, session_id: int, output_type: OutputType,
-                       progress_callback=None, lang: str = "es") -> str:
+                       progress_callback=None, lang: str = "es",
+                       seo_override: str | None = None) -> str:
         """Generate an output for a research session"""
+        self.last_publish_notice = None
         if output_type in (OutputType.REPORT_EXTENDED,
                            OutputType.BLOG_EXTENDED,
                            OutputType.PODCAST_EXTENDED):
             return await self.generate_extended(session_id, output_type, progress_callback,
-                                                lang=lang)
+                                                lang=lang, seo_override=seo_override)
 
         session = await self.db.get_session(session_id)
         if not session:
@@ -413,23 +721,11 @@ class OutputGenerator:
         # Save to DB
         await self.db.save_output(session_id, output_type, full_output)
 
-        # Auto-publish to Ghost for blog outputs
+        # Auto-publish to Ghost for blog outputs (autofill mode gated inside helper).
         ghost_notice = ""
         if output_type in (OutputType.BLOG, OutputType.BLOG_EXTENDED):
-            ghost = GhostPublisher(lang=lang)
-            if ghost.is_configured():
-                try:
-                    title = _extract_title(full_output) or topic
-                    result = await ghost.publish_draft(title, full_output)
-                    post = result["posts"][0]
-                    ghost_notice = (
-                        f"\n\n---\n"
-                        f"📤 *Borrador publicado en Ghost*\n"
-                        f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}"
-                    )
-                    logger.info("Auto-published blog to Ghost", post_id=post["id"])
-                except Exception as e:
-                    logger.warning("Auto-publish to Ghost failed", error=str(e))
+            ghost_notice = await self._publish_blog_to_ghost(
+                lang, full_output, topic, session_id, seo_override)
 
         logger.info("Output generated", type=output_type, length=len(full_output))
         return full_output + ghost_notice
@@ -442,10 +738,9 @@ class OutputGenerator:
 
     async def _generate_with_claude(self, prompt: str, system: str, output_type: OutputType,
                                      session_id: int | None = None) -> str:
-        import anthropic
         max_tokens = 4096 if output_type == OutputType.THREAD else 16000
         try:
-            client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+            client = get_anthropic_client()
             msg = await client.messages.create(
                 model=settings.claude_model,
                 max_tokens=max_tokens,
@@ -487,7 +782,8 @@ class OutputGenerator:
         return systems.get(output_type, "You are a helpful research assistant.")
 
     async def generate_extended(self, session_id: int, output_type: OutputType,
-                                progress_callback=None, lang: str = "es") -> str:
+                                progress_callback=None, lang: str = "es",
+                                seo_override: str | None = None) -> str:
         """
         Generación por secciones para outputs exhaustivos.
         1. Recupera muestra de contexto para el outline
@@ -588,23 +884,11 @@ class OutputGenerator:
 
         await self.db.save_output(session_id, output_type, full_output)
 
-        # Auto-publish to Ghost for extended blog outputs
+        # Auto-publish to Ghost for extended blog outputs (autofill mode gated inside).
         ghost_notice = ""
         if output_type == OutputType.BLOG_EXTENDED:
-            ghost = GhostPublisher(lang=lang)
-            if ghost.is_configured():
-                try:
-                    title = _extract_title(full_output) or topic
-                    result = await ghost.publish_draft(title, full_output)
-                    post = result["posts"][0]
-                    ghost_notice = (
-                        f"\n\n---\n"
-                        f"📤 *Borrador publicado en Ghost*\n"
-                        f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}"
-                    )
-                    logger.info("Auto-published extended blog to Ghost", post_id=post["id"])
-                except Exception as e:
-                    logger.warning("Auto-publish to Ghost failed (extended)", error=str(e))
+            ghost_notice = await self._publish_blog_to_ghost(
+                lang, full_output, topic, session_id, seo_override)
 
         logger.info("Extended output generated", type=output_type,
                     sections=len(sections), length=len(full_output))
@@ -613,9 +897,8 @@ class OutputGenerator:
     async def _generate_raw(self, prompt: str,
                              session_id: int | None = None) -> str:
         if settings.anthropic_api_key:
-            import anthropic
             try:
-                client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+                client = get_anthropic_client()
                 msg = await client.messages.create(
                     model=settings.claude_model,
                     max_tokens=2048,
@@ -819,8 +1102,7 @@ async def generate_diff_summary(
         )
 
     try:
-        import anthropic
-        client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+        client = get_anthropic_client()
         prompt = (
             f'Analiza el siguiente material de investigación sobre "{topic}" '
             f'y genera un resumen BREVE (máximo 300 palabras) de las novedades '
@@ -906,8 +1188,7 @@ async def generate_comparison(
     )
 
     try:
-        import anthropic
-        client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+        client = get_anthropic_client()
         msg = await client.messages.create(
             model=settings.claude_model,
             max_tokens=8192,

src/llm.py

@@ -0,0 +1,15 @@
+"""Cliente Anthropic compartido y cacheado.
+
+Antes cada llamada (scoring de cada chunk, generación, outline, diff…)
+instanciaba un AsyncAnthropic nuevo — cientos de veces por sesión, cada uno
+con su propio pool de conexiones. Se reutiliza una única instancia por proceso.
+"""
+from functools import lru_cache
+
+from src.config import settings
+
+
+@lru_cache(maxsize=1)
+def get_anthropic_client():
+    import anthropic
+    return anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)

src/news/monitor.py

@@ -0,0 +1,207 @@
+"""Monitor RSS de noticias UAP/OVNI.
+
+Capa independiente: NUNCA importa de src.bot (el bot importa de aquí). No toca el
+scheduler — F2. Best-effort por diseño: un feed caído se loguea y se salta; nunca
+tumba el run. El estado vive en la tabla news_seen vía los métodos de ResearchDB,
+que se inyectan (no se importan) para mantener el desacople.
+"""
+import asyncio
+import re
+import unicodedata
+from dataclasses import dataclass, field
+
+import feedparser
+import structlog
+
+log = structlog.get_logger()
+
+# Reddit devuelve 403 con el UA por defecto de feedparser/python; uno explícito
+# (y honesto) pasa. El resto de feeds lo aceptan sin problema.
+UA = "ExclusionZone-NewsBot/1.0 (+https://theexclusionzone.com)"
+
+FEEDS: list[dict] = [
+    {"name": "Gran Misterio",      "url": "https://granmisterio.org/feed/",                                                     "lang": "es"},
+    # Reemplazos ES tras retirar Espacio Misterio (XML roto). Verificados 2026-07-03:
+    # MysteryPlanet y Marcianitos publican a diario/casi a diario; El Ojo Crítico es
+    # mensual (poco volumen a propósito). Ufopolis/MundoEsoterico descartados (muerto/roto).
+    {"name": "MysteryPlanet",      "url": "https://mysteryplanet.com.ar/site/feed/",                                            "lang": "es"},
+    {"name": "Marcianitos Verdes", "url": "https://marcianitosverdes.haaan.com/feed/",                                          "lang": "es"},
+    {"name": "El Ojo Crítico",     "url": "https://elojocritico.info/feed/",                                                    "lang": "es"},
+    {"name": "Josep Guijarro (YT)", "url": "https://www.youtube.com/feeds/videos.xml?channel_id=UCIlGy26zi-BbX6zDYAMQ9Qg",      "lang": "es"},
+    {"name": "r/UFOs",             "url": "https://www.reddit.com/r/UFOs/new/.rss",                                             "lang": "en"},
+    {"name": "r/UFOB",             "url": "https://www.reddit.com/r/UFOB/new/.rss",                                             "lang": "en"},
+    {"name": "Liberation Times",   "url": "https://www.liberationtimes.com/?format=rss",                                        "lang": "en"},
+    {"name": "The Debrief",        "url": "https://thedebrief.org/feed/",                                                       "lang": "en"},
+]
+
+# Tokens cortos: word-boundary para evitar falsos positivos dentro de palabras
+# (p.ej. "aaro" en "aaron", "ufo" en "tufo"). Frases multi-palabra y términos
+# largos van por substring. Se decide dinámicamente: sin espacios y len <= 4.
+_SHORT_TOKEN_MAXLEN = 4
+
+# Límite de Telegram 4096; troceamos por debajo con margen.
+_TG_LIMIT = 4000
+
+# Reddit sin OAuth ratelimita por IP con ventana larga (>30s verificado): la 2ª
+# petición seguida a reddit.com devuelve 429 SIEMPRE, así que solo se pollea un
+# feed de reddit por run, rotando. Contador por proceso: tras un reinicio se
+# empieza por el primero, lo cual es inocuo.
+_reddit_turn = 0
+
+
+def _is_reddit(feed: dict) -> bool:
+    return "reddit.com" in feed["url"]
+
+
+@dataclass
+class NewsItem:
+    source: str
+    guid: str
+    title: str
+    link: str
+    published: str
+    summary: str
+    matched: list[str] = field(default_factory=list)
+
+
+def _normalize(text: str) -> str:
+    """minúsculas + NFKD sin marcas combinantes (quita tildes/diacríticos)."""
+    if not text:
+        return ""
+    decomposed = unicodedata.normalize("NFKD", text.lower())
+    return "".join(c for c in decomposed if not unicodedata.combining(c))
+
+
+def _is_short_token(kw: str) -> bool:
+    return " " not in kw and len(kw) <= _SHORT_TOKEN_MAXLEN
+
+
+def _match(title: str, summary: str, keywords: list[str]) -> list[str]:
+    """Devuelve las keywords (en su forma original) que matchean en title+summary,
+    sobre texto normalizado. Word-boundary para tokens cortos, substring para el
+    resto. Sin duplicados, preservando el orden de `keywords`."""
+    text = _normalize(f"{title} {summary}")
+    if not text:
+        return []
+    matched: list[str] = []
+    for kw in keywords:
+        nkw = _normalize(kw)
+        if not nkw:
+            continue
+        if _is_short_token(nkw):
+            if re.search(rf"\b{re.escape(nkw)}\b", text):
+                matched.append(kw)
+        else:
+            if nkw in text:
+                matched.append(kw)
+    return matched
+
+
+async def poll_feeds(db, settings) -> list[NewsItem]:
+    """Recorre los feeds, registra items matcheados en news_seen y devuelve los
+    NUEVOS (no vistos) de fuentes ya sembradas. Cold-start por fuente: el primer
+    poll de cada feed siembra en silencio (notified=1) y no devuelve nada.
+
+    Best-effort: cada feed va en su propio try/except — uno caído nunca tumba el
+    run. feedparser.parse es bloqueante (I/O de red), así que se ejecuta en un
+    hilo para no congelar el event loop del bot durante el poll.
+
+    Reddit: solo se pollea UN feed de reddit.com por run, rotando entre ellos
+    (ver _reddit_turn) — dos peticiones seguidas garantizan un 429 en la 2ª.
+    """
+    global _reddit_turn
+    keywords = [k.strip() for k in settings.NEWS_KEYWORDS.split(",") if k.strip()]
+    new_items: list[NewsItem] = []
+
+    reddit_feeds = [f for f in FEEDS if _is_reddit(f)]
+    skip: set[str] = set()
+    if len(reddit_feeds) > 1:
+        chosen = reddit_feeds[_reddit_turn % len(reddit_feeds)]
+        _reddit_turn += 1
+        skip = {f["name"] for f in reddit_feeds if f["name"] != chosen["name"]}
+
+    for feed in FEEDS:
+        if feed["name"] in skip:
+            continue
+        try:
+            parsed = await asyncio.to_thread(feedparser.parse, feed["url"], agent=UA)
+            status = getattr(parsed, "status", None)
+            # Un 429/403 llega con bozo=False y 0 entries: sin este guard se
+            # tragaba en silencio (así estuvo r/UFOB sin sembrar desde F1).
+            if status and status >= 400:
+                log.warning("feed %s HTTP %s; salto", feed["name"], status)
+                continue
+            if parsed.bozo and not parsed.entries:
+                log.warning("feed bozo sin entries: %s (%s)",
+                            feed["name"], getattr(parsed, "bozo_exception", ""))
+                continue
+
+            seeded = await db.source_seeded(feed["name"])  # cold-start por fuente
+            for e in parsed.entries:
+                guid = getattr(e, "id", None) or getattr(e, "link", "")
+                if not guid:
+                    continue
+                matched = _match(e.get("title", ""), e.get("summary", ""), keywords)
+                if not matched:
+                    continue
+                notified = 1 if not seeded else 0  # 1er run: sembrar sin notificar
+                is_new = await db.record_news_item(
+                    feed["name"], guid, e.get("title", ""), e.get("link", ""),
+                    e.get("published", ""), ",".join(matched), notified,
+                )
+                if is_new and seeded:
+                    new_items.append(NewsItem(
+                        feed["name"], guid, e.get("title", ""), e.get("link", ""),
+                        e.get("published", ""), e.get("summary", ""), matched,
+                    ))
+        except Exception as ex:  # aislado: un feed no tumba el run
+            log.exception("fallo feed %s: %s", feed["name"], ex)
+
+    return new_items
+
+
+def item_from_row(row: dict) -> NewsItem:
+    """Adapta una fila de get_recent_news (dict de news_seen) a NewsItem para que
+    format_digest sea agnóstico de la fuente (poll vs DB)."""
+    matched = row.get("matched_keywords") or ""
+    return NewsItem(
+        source=row.get("source", ""),
+        guid=row.get("guid", ""),
+        title=row.get("title", "") or "",
+        link=row.get("link", "") or "",
+        published=row.get("published_at", "") or "",
+        summary="",
+        matched=[m for m in matched.split(",") if m],
+    )
+
+
+def format_digest(items: list[NewsItem], header: str = "🛸 Novedades UAP/OVNI") -> list[str]:
+    """Agrupa por fuente y formatea cada item como '• <title>\\n  <link>'. Trocea
+    en mensajes < _TG_LIMIT chars (devuelve varios si hace falta). Texto plano
+    (no Markdown): el handler envía sin parse_mode. Lista vacía -> []."""
+    if not items:
+        return []
+
+    # Agrupar por fuente preservando el orden de primera aparición.
+    groups: dict[str, list[NewsItem]] = {}
+    for it in items:
+        groups.setdefault(it.source, []).append(it)
+
+    blocks: list[str] = []  # un bloque por fuente, indivisible salvo que él solo exceda
+    for source, its in groups.items():
+        lines = [f"── {source} ──"]
+        for it in its:
+            title = (it.title or "(sin título)").strip()
+            lines.append(f"• {title}\n  {it.link}".rstrip())
+        blocks.append("\n".join(lines))
+
+    chunks: list[str] = []
+    cur = header
+    for block in blocks:
+        if len(cur) + len(block) + 2 > _TG_LIMIT and cur.strip() != header:
+            chunks.append(cur)
+            cur = f"{header} (cont.)"
+        cur += "\n\n" + block
+    if cur.strip():
+        chunks.append(cur)
+    return chunks

src/processor/processor.py

@@ -23,6 +23,7 @@ class OllamaClient:
     def __init__(self):
         self.base_url = settings.ollama_url.rstrip("/")
         self.model = settings.ollama_model
+        self.embed_model = settings.ollama_embed_model
 
     async def generate(self, prompt: str, system: str = None,
                        timeout: int = 120, temperature: float = 0.7) -> str:
@@ -47,7 +48,7 @@ class OllamaClient:
 
     async def embed(self, text: str) -> Optional[list[float]]:
         """Get embedding vector for a text"""
-        payload = {"model": self.model, "prompt": text}
+        payload = {"model": self.embed_model, "prompt": text}
         try:
             async with httpx.AsyncClient(timeout=60) as client:
                 resp = await client.post(f"{self.base_url}/api/embeddings", json=payload)
@@ -69,9 +70,28 @@ class OllamaClient:
 def simple_chunk(text: str, chunk_size: int = 800, overlap: int = 100) -> list[str]:
     """
     Split text into overlapping chunks by approximate word count.
-    Respects paragraph boundaries when possible.
+    Respects paragraph/line boundaries when possible.
+
+    Acepta párrafos separados por uno o más saltos de línea (Wikipedia y
+    trafilatura usan '\n' simple, lo que antes dejaba el documento entero como
+    un único 'párrafo' → un solo chunk gigante). Además subdivide por palabras
+    cualquier párrafo que por sí solo supere chunk_size.
     """
-    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
+    raw_paragraphs = [p.strip() for p in re.split(r"\n+", text) if p.strip()]
+
+    # Subdivide párrafos sobredimensionados en piezas de (chunk_size - overlap)
+    # palabras; así, al reinyectar 'overlap' palabras de solapamiento, ningún
+    # chunk resultante supera chunk_size.
+    piece_size = max(1, chunk_size - max(0, overlap))
+    paragraphs: list[str] = []
+    for para in raw_paragraphs:
+        words = para.split()
+        if len(words) <= chunk_size:
+            paragraphs.append(para)
+        else:
+            for i in range(0, len(words), piece_size):
+                paragraphs.append(" ".join(words[i:i + piece_size]))
+
     chunks = []
     current = []
     current_words = 0
@@ -80,10 +100,12 @@ def simple_chunk(text: str, chunk_size: int = 800, overlap: int = 100) -> list[s
         para_words = len(para.split())
         if current_words + para_words > chunk_size and current:
             chunks.append("\n\n".join(current))
-            # overlap: keep last paragraph
-            if overlap > 0 and current:
-                current = [current[-1]]
-                current_words = len(current[0].split())
+            # overlap: arrastra un tail de 'overlap' palabras (no el párrafo
+            # completo — eso duplicaba el tamaño cuando los párrafos eran grandes)
+            if overlap > 0:
+                tail = "\n\n".join(current).split()[-overlap:]
+                current = [" ".join(tail)]
+                current_words = len(tail)
             else:
                 current = []
                 current_words = 0
@@ -124,7 +146,6 @@ class ContentProcessor:
     async def process_session(self, session_id: int, topic: str,
                                progress_callback=None) -> dict:
         """Process all scraped sources for a session"""
-        from src.db.database import ResearchDB
         sources = await self.db.get_all_sources(session_id)
         scraped = [s for s in sources if s["status"] == "scraped"]
 
@@ -132,7 +153,6 @@ class ContentProcessor:
         scraped = await self._dedup_sources(session_id, scraped)
         logger.info("After dedup", unique=len(scraped))
         total_chunks = 0
-        total_words = 0
 
         semaphore = asyncio.Semaphore(3)  # process 3 sources at once
 
@@ -223,33 +243,45 @@ class ContentProcessor:
             return 0
 
         chunks = simple_chunk(content, settings.chunk_size, settings.chunk_overlap)
+        # Solo se puntúan/almacenan chunks con suficiente contenido
+        candidates = [(i, ch) for i, ch in enumerate(chunks)
+                      if len(ch.split()) >= 30]
         logger.info("Processing source", source_id=source_id,
                     content_len=len(content), num_chunks=len(chunks),
+                    candidates=len(candidates),
                     quality_threshold=settings.quality_threshold)
+        if not candidates:
+            return 0
+
+        # Scoring en lote: 1 (o pocas) llamadas a Claude por fuente en vez de
+        # una por chunk — antes una fuente de 19 chunks = 19 llamadas.
+        qualities = await self._score_quality_batch(
+            [ch for _, ch in candidates], topic, session_id
+        )
+
         stored = 0
         filtered_quality = 0
 
-        for i, chunk in enumerate(chunks):
-            words = len(chunk.split())
-            if words < 30:
-                continue
-
-            quality = await self._score_quality(chunk, topic, session_id)
+        for (i, chunk), quality in zip(candidates, qualities):
             if quality < settings.quality_threshold:
                 filtered_quality += 1
                 logger.debug("Chunk filtered by quality", source_id=source_id,
                              chunk_index=i, quality=round(quality, 2),
-                             threshold=settings.quality_threshold, words=words)
+                             threshold=settings.quality_threshold,
+                             words=len(chunk.split()))
                 continue
 
-            embedding = await self.ollama.embed(chunk[:1000])
+            # Embeber el chunk completo (ya acotado a ~chunk_size palabras).
+            # Antes truncaba a 1000 chars → el vector solo representaba el
+            # principio de cada chunk, degradando el ranking del RAG.
+            embedding = await self.ollama.embed(chunk)
 
             await self.db.add_chunk(
                 session_id=session_id,
                 source_id=source_id,
                 content=chunk,
                 chunk_index=i,
-                token_count=words,
+                token_count=len(chunk.split()),
                 quality_score=quality,
                 embedding=embedding
             )
@@ -274,9 +306,84 @@ class ContentProcessor:
             return await self._score_with_claude(chunk, topic, session_id)
         return await self._score_with_ollama(chunk, topic)
 
+    # ─── Batch scoring ──────────────────────────────────────────────────────
+
+    _BATCH_SCORE_SIZE = 25
+
+    async def _score_quality_batch(self, chunk_texts: list[str], topic: str,
+                                    session_id: int | None = None) -> list[float]:
+        """Puntúa varios chunks a la vez. Devuelve scores 0-1 en el mismo orden."""
+        if not chunk_texts:
+            return []
+        if not settings.anthropic_api_key:
+            # Ollama es local; no merece la pena batchear, se mantiene por-chunk
+            return [await self._score_with_ollama(c, topic) for c in chunk_texts]
+
+        results: list[float] = []
+        for i in range(0, len(chunk_texts), self._BATCH_SCORE_SIZE):
+            sub = chunk_texts[i:i + self._BATCH_SCORE_SIZE]
+            scores = await self._score_with_claude_batch(sub, topic, session_id)
+            if scores is None:  # fallo del batch → fallback por-chunk con Ollama
+                scores = [await self._score_with_ollama(c, topic) for c in sub]
+            results.extend(scores)
+        return results
+
+    @staticmethod
+    def _parse_batch_scores(text: str, n: int) -> list[float]:
+        """Extrae n scores normalizados (0-1) de la respuesta del modelo.
+
+        Toma el último número de cada línea (robusto ante numeración tipo
+        '1. 8'); rellena con 0.6 neutro si faltan líneas."""
+        scores: list[float] = []
+        for line in text.splitlines():
+            nums = re.findall(r'\d+(?:\.\d+)?', line)
+            if nums:
+                scores.append(min(1.0, float(nums[-1]) / 10.0))
+        if len(scores) >= n:
+            return scores[:n]
+        return scores + [0.6] * (n - len(scores))
+
+    async def _score_with_claude_batch(self, chunks: list[str], topic: str,
+                                        session_id: int | None = None):
+        """Puntúa hasta _BATCH_SCORE_SIZE chunks en una sola llamada.
+        Devuelve lista de scores 0-1, o None si la llamada falla."""
+        from src.llm import get_anthropic_client
+        listing = "\n\n".join(f"[{i + 1}]\n{c[:400]}" for i, c in enumerate(chunks))
+        prompt = (
+            f'Rate each of the following {len(chunks)} texts 0-10 for relevance '
+            f'to the topic "{topic}". Be generous — tangentially related = 4+, '
+            f'only below 3 if completely unrelated.\n'
+            f'Reply with EXACTLY {len(chunks)} lines, one integer 0-10 per line, '
+            f'in the same order as the texts. Just the number (e.g. 7), '
+            f'no labels, no other text.\n\n{listing}'
+        )
+        try:
+            client = get_anthropic_client()
+            msg = await client.messages.create(
+                model=settings.claude_model,
+                max_tokens=8 * len(chunks) + 50,
+                messages=[{"role": "user", "content": prompt}]
+            )
+            if session_id is not None:
+                try:
+                    await self.db.log_api_call(
+                        session_id, "scoring", settings.claude_model,
+                        msg.usage.input_tokens, msg.usage.output_tokens
+                    )
+                except Exception as log_err:
+                    logger.warning("Failed to log API usage", error=str(log_err))
+            scores = self._parse_batch_scores(msg.content[0].text.strip(), len(chunks))
+            logger.debug("Claude batch scored", n=len(chunks),
+                         avg=round(sum(scores) / len(scores), 2))
+            return scores
+        except Exception as e:
+            logger.warning("Claude batch scoring failed, will fallback",
+                           error=str(e), n=len(chunks))
+            return None
+
     async def _score_with_claude(self, chunk: str, topic: str,
                                   session_id: int | None = None) -> float:
-        import anthropic
+        from src.llm import get_anthropic_client
         prompt = (
             f'Rate 0-10 how relevant this text is to the topic "{topic}". '
             f'Be generous — if the text is tangentially related, score 4+. '
@@ -284,7 +391,7 @@ class ContentProcessor:
             f'Reply with only a number.\n\nText:\n{chunk[:500]}'
         )
         try:
-            client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+            client = get_anthropic_client()
             msg = await client.messages.create(
                 model=settings.claude_model,
                 max_tokens=10,

src/scraper/exhaustive.py

@@ -7,7 +7,7 @@ import random
 import re
 import time
 from typing import Optional
-from urllib.parse import urljoin, urlparse, quote_plus
+from urllib.parse import urljoin, urlparse, parse_qs, quote, quote_plus
 
 import aiohttp
 import feedparser
@@ -18,7 +18,7 @@ from duckduckgo_search import DDGS
 from youtube_transcript_api import YouTubeTranscriptApi, NoTranscriptFound
 from tenacity import retry, stop_after_attempt, wait_exponential
 
-from src.config import settings
+from src.config import settings, SAFE_ACCEPT_ENCODING
 from src.db.database import ResearchDB
 
 logger = structlog.get_logger()
@@ -27,7 +27,13 @@ HEADERS = {
     "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
     "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
     "Accept-Language": "en-US,en;q=0.9,es;q=0.8",
-    "Accept-Encoding": "gzip, deflate, br",
+    # Sin "br": el descompresor incremental de aiohttp 3.14 falla con streams
+    # brotli válidos según el troceo de chunks (disclosure.org, todaywhy.com;
+    # los mismos bytes descomprimen bien offline). Con gzip el camino zlib es
+    # sólido. OJO: ya NO hay ningún backend brotli instalado (397546a quitó
+    # brotlicffi) — si un servidor responde br sin que se haya anunciado,
+    # aiohttp falla sin recuperación posible y esa fuente se pierde.
+    "Accept-Encoding": SAFE_ACCEPT_ENCODING,
     "DNT": "1",
 }
 
@@ -36,7 +42,7 @@ REDDIT_HEADERS = {
     "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
     "Accept": "application/json, text/javascript, */*; q=0.01",
     "Accept-Language": "en-US,en;q=0.9",
-    "Accept-Encoding": "gzip, deflate, br",
+    "Accept-Encoding": SAFE_ACCEPT_ENCODING,  # sin "br", ver HEADERS
     "Referer": "https://www.reddit.com/",
     "X-Requested-With": "XMLHttpRequest",
 }
@@ -53,6 +59,10 @@ YOUTUBE_RE = re.compile(r"(?:youtube\.com/watch\?v=|youtu\.be/)([a-zA-Z0-9_-]{11
 PDF_RE = re.compile(r"\.pdf(\?|$)", re.IGNORECASE)
 REDDIT_RE = re.compile(r"reddit\.com/(r/\w+/comments/\w+)")
 WIKIPEDIA_RE = re.compile(r"wikipedia\.org/wiki/(.+)")
+# Segmentos de path típicos de feeds, con límite de palabra: el substring puro
+# clasificaba /feedback o /atomic como rss.
+RSS_RE = re.compile(r"/(rss|feeds?|atom)(/|\.xml|\.rss|$)|\.rss$|format=rss",
+                    re.IGNORECASE)
 
 
 async def fetch_with_retry(fetch_fn, source_name: str, max_retries: int = 3):
@@ -82,22 +92,48 @@ def detect_source_type(url: str) -> str:
         return "wikipedia"
     if "arxiv.org" in url:
         return "arxiv"
-    if any(d in url for d in ["rss", "feed", "atom"]):
+    if RSS_RE.search(url):
         return "rss"
     return "web"
 
 
 def is_blacklisted(url: str) -> bool:
     try:
-        domain = urlparse(url).netloc.lower().replace("www.", "")
-        return any(bl in domain for bl in BLACKLIST_DOMAINS)
+        domain = urlparse(url).netloc.lower().split(":")[0]
+        if domain.startswith("www."):
+            domain = domain[4:]
+        # Exact domain or subdomain match — NOT substring (evitaba bloquear
+        # netflix.com / phoenix.com por contener "x.com", etc.)
+        return any(domain == bl or domain.endswith("." + bl)
+                   for bl in BLACKLIST_DOMAINS)
     except Exception:
         return True
 
 
+def _unwrap_news_link(link: str) -> str:
+    """Bing News envuelve los links de entry en apiclick.aspx; la URL real del
+    publisher viene en texto plano en el parámetro ?url=. Cualquier otro link
+    pasa sin tocar. Si es un apiclick pero no trae URL usable, devuelve "" —
+    el caller lo descarta por startswith("http") — en vez de dejar pasar el
+    redirect de bing.com, que se rasparía como página de consent/basura."""
+    try:
+        parsed = urlparse(link)
+        if (parsed.netloc.lower().endswith("bing.com")
+                and parsed.path == "/news/apiclick.aspx"):
+            real = parse_qs(parsed.query).get("url", [""])[0].strip()
+            if real.startswith("//"):  # protocol-relative
+                real = "https:" + real
+            return real if real.startswith("http") else ""
+    except Exception:
+        return ""
+    return link
+
+
 def normalize_url(url: str) -> str:
+    # Strip only the fragment. NO borrar el query string: rompía URLs de
+    # YouTube (watch?v=...) y artículos que enrutan por query (?id=, ?p=).
     parsed = urlparse(url)
-    clean = parsed._replace(fragment="", query="")
+    clean = parsed._replace(fragment="")
     return clean.geturl().rstrip("/")
 
 
@@ -155,13 +191,19 @@ class ExhaustiveScraper:
 
     async def seed(self):
         """Initial broad search across multiple sources"""
-        logger.info("Seeding research", topic=self.topic)
+        logger.info("Seeding research", topic=self.topic,
+                    youtube=settings.enable_youtube, reddit=settings.enable_reddit)
         tasks = [
             self._seed_search(),
             self._seed_wikipedia(),
-            self._seed_reddit(),
-            self._seed_youtube(),
+            self._seed_archive(),
         ]
+        if settings.enable_reddit:
+            tasks.append(self._seed_reddit())
+        if settings.enable_youtube:
+            tasks.append(self._seed_youtube())
+        if settings.enable_news_seed:
+            tasks.append(self._seed_news())
         await asyncio.gather(*tasks, return_exceptions=True)
 
     async def _generate_ddg_queries(self) -> list[str]:
@@ -180,9 +222,9 @@ class ExhaustiveScraper:
             return fallback
 
         try:
-            import anthropic
+            from src.llm import get_anthropic_client
             logger.info("Generating DDG queries with Claude", topic=self.topic)
-            client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+            client = get_anthropic_client()
             prompt = (
                 f'Generate exactly 8 DuckDuckGo search queries to research: "{self.topic}"\n\n'
                 f'Rules:\n'
@@ -215,17 +257,21 @@ class ExhaustiveScraper:
     async def _search_searxng(self, query: str) -> list[dict]:
         """Busca en SearXNG y retorna lista de {href, title}. Retorna [] si no disponible."""
         import aiohttp
-        searxng_url = "http://searxng-svc.researchowl.svc.cluster.local:8080/search"
+        searxng_url = settings.searxng_url
         params = {
             "q": query,
             "format": "json",
-            "engines": "duckduckgo,google,bing,brave",
+            "engines": settings.searxng_engines,
             "language": "all",
         }
         headers = {
             "Accept": "application/json",
             "X-Forwarded-For": "127.0.0.1",
             "User-Agent": "ResearchOwl/1.0",
+            # Sin esto la sesión hereda el default de aiohttp, que anunciaría
+            # br si algún día se reinstala un backend brotli — y el decode roto
+            # de 3.14 tumbaría TODAS las búsquedas SearXNG (el camino primario).
+            "Accept-Encoding": SAFE_ACCEPT_ENCODING,
         }
         try:
             async with aiohttp.ClientSession() as session:
@@ -251,6 +297,21 @@ class ExhaustiveScraper:
             logger.warning("SearXNG failed", query=query, error=str(e))
             return []
 
+    # DDGS es síncrono (requests bloqueante por dentro): llamarlo directamente
+    # desde código async congela el event loop entero — bot de Telegram incluido —
+    # mientras dura la búsqueda. Estos helpers deben ejecutarse SIEMPRE vía
+    # run_in_executor.
+
+    @staticmethod
+    def _ddg_text_sync(query: str) -> list[dict]:
+        with DDGS() as ddgs:
+            return list(ddgs.text(query, max_results=settings.max_pages_per_search))
+
+    @staticmethod
+    def _ddg_videos_sync(query: str) -> list[dict]:
+        with DDGS() as ddgs:
+            return list(ddgs.videos(query, max_results=10))
+
     async def _seed_search(self):
         """SearXNG primary + DDG fallback per query"""
         queries = await self._generate_ddg_queries()
@@ -261,12 +322,10 @@ class ExhaustiveScraper:
             if not results:
                 logger.info("SearXNG vacío, usando DDG", query=query)
                 try:
-                    with DDGS() as ddgs:
-                        ddg_results = list(ddgs.text(
-                            query,
-                            max_results=settings.max_pages_per_search
-                        ))
-                    results = ddg_results
+                    loop = asyncio.get_running_loop()
+                    results = await loop.run_in_executor(
+                        None, self._ddg_text_sync, query
+                    )
                     logger.info("DDG fallback ok", query=query, results=len(results))
                 except Exception as e:
                     logger.warning("DDG fallback failed", query=query, error=str(e))
@@ -284,30 +343,92 @@ class ExhaustiveScraper:
             await asyncio.sleep(random.uniform(1, 3))
 
     async def _seed_wikipedia(self):
-        """Search Wikipedia API for correct article URLs.
-        Tries English first, falls back to Spanish if no results found."""
+        """Siembra Wikipedia en AMBOS idiomas siempre, con búsqueda full-text
+        (list=search). El opensearch anterior era prefix-de-título y devolvía 0
+        para topics verbosos ("incidente ovni de Manises 1979") aunque el
+        artículo exista; y el break tras en dejaba es solo como fallback.
+        Cada idioma va aislado: un fallo no afecta al otro."""
         http = await self._get_http()
-        added = 0
 
         for lang in ("en", "es"):
             try:
                 api_url = (
-                    f"https://{lang}.wikipedia.org/w/api.php?action=opensearch"
-                    f"&search={quote_plus(self.topic)}&limit=10&format=json"
+                    f"https://{lang}.wikipedia.org/w/api.php?action=query"
+                    f"&list=search&srsearch={quote_plus(self.topic)}"
+                    f"&srlimit=8&format=json"
                 )
                 async with http.get(api_url) as resp:
                     data = await resp.json()
-                    urls = data[3] if len(data) > 3 else []
-                    for url in urls:
-                        if url:
-                            await self.db.add_source(self.session_id, url, "wikipedia", depth=0)
-                            added += 1
-                logger.info("Wikipedia seed", lang=lang, found=len(urls))
-                if added > 0:
-                    break  # English results found — no need to try Spanish
+                hits = data.get("query", {}).get("search", [])
+                added = 0
+                for h in hits:
+                    title = h.get("title")
+                    if not title:
+                        continue
+                    url = (f"https://{lang}.wikipedia.org/wiki/"
+                           f"{quote(title.replace(' ', '_'))}")
+                    await self.db.add_source(self.session_id, url, "wikipedia",
+                                             depth=0, title=title)
+                    added += 1
+                logger.info("Wikipedia seed", lang=lang, found=added)
             except Exception as e:
                 logger.warning("Wikipedia API seed failed", lang=lang, error=str(e))
 
+    async def _seed_archive(self):
+        """Siembra Internet Archive (advancedsearch, mediatype:texts): documentos
+        FOIA, informes y libros históricos. Se siembra la página /details/<id>;
+        sus PDFs los descubre la recursión (la propia página los enlaza) y los
+        extrae _extract_pdf. Sin fallback de query: el AND estricto de archive.org
+        con el topic completo prima precisión — relajarlo mete ruido (verificado:
+        'Manises' a secas devuelve loza valenciana). Best-effort: nunca rompe el seed."""
+        try:
+            http = await self._get_http()
+            q = quote_plus(f"({self.topic}) AND mediatype:(texts)")
+            url = (
+                f"https://archive.org/advancedsearch.php?q={q}"
+                "&fl[]=identifier&fl[]=title&rows=12&page=1&output=json"
+            )
+            async with http.get(url) as resp:
+                if resp.status != 200:
+                    logger.warning("Archive.org seed non-200", status=resp.status)
+                    return
+                data = await resp.json(content_type=None)
+            docs = data.get("response", {}).get("docs", [])
+            added = 0
+            for d in docs:
+                ident = d.get("identifier")
+                if not ident:
+                    continue
+                item_url = f"https://archive.org/details/{ident}"
+                title = d.get("title")
+                await self.db.add_source(
+                    self.session_id, item_url, "web", depth=0,
+                    title=title if isinstance(title, str) else None
+                )
+                added += 1
+            logger.info("Archive.org seed", found=len(docs), added=added)
+        except Exception as e:
+            logger.warning("Archive.org seed failed", error=str(e))
+
+    async def _seed_news(self):
+        """Siembra Bing News RSS para cobertura de actualidad. Solo registra
+        el feed como fuente tipo rss: _extract_rss (dispatch del pipeline) lo
+        parsea y siembra sus entries — sin duplicar lógica de parseo aquí.
+        Bing y no Google News: los links de Google (news.google.com/rss/articles/
+        CBMi…) van a un muro de consent + redirect cifrado solo resoluble vía su
+        API interna batchexecute (verificado 2026-07-04); los de Bing llevan la
+        URL real del publisher en ?url= (ver _unwrap_news_link).
+        Best-effort: nunca rompe el seed."""
+        try:
+            url = f"https://www.bing.com/news/search?q={quote_plus(self.topic)}&format=rss"
+            await self.db.add_source(
+                self.session_id, url, "rss", depth=0,
+                title=f"Bing News: {self.topic}"
+            )
+            logger.info("News seed added", topic=self.topic)
+        except Exception as e:
+            logger.warning("News seed failed", error=str(e))
+
     async def _seed_reddit(self):
         """Search Reddit — sequential to avoid rate limiting"""
         try:
@@ -337,18 +458,18 @@ class ExhaustiveScraper:
     async def _seed_youtube(self):
         """Search YouTube via DDG for video transcripts"""
         try:
-            with DDGS() as ddgs:
-                results = list(ddgs.videos(
-                    f"{self.topic} documentary explanation",
-                    max_results=10
-                ))
-                for r in results:
-                    url = r.get("content", "")
-                    if "youtube.com" in url or "youtu.be" in url:
-                        await self.db.add_source(
-                            self.session_id, url, "youtube", depth=0,
-                            title=r.get("title")
-                        )
+            loop = asyncio.get_running_loop()
+            results = await loop.run_in_executor(
+                None, self._ddg_videos_sync,
+                f"{self.topic} documentary explanation"
+            )
+            for r in results:
+                url = r.get("content", "")
+                if "youtube.com" in url or "youtu.be" in url:
+                    await self.db.add_source(
+                        self.session_id, url, "youtube", depth=0,
+                        title=r.get("title")
+                    )
         except Exception as e:
             logger.warning("YouTube seed failed", error=str(e))
 
@@ -413,6 +534,14 @@ class ExhaustiveScraper:
             url = source["url"]
             source_id = source["id"]
 
+            # Saltar fuentes desactivadas (también las descubiertas dentro de
+            # páginas web, no solo las del seed) sin gastar una petición de red.
+            if ((source_type == "youtube" and not settings.enable_youtube) or
+                    (source_type == "reddit" and not settings.enable_reddit)):
+                await self.db.update_source(source_id, status="skipped",
+                                            error=f"{source_type} disabled")
+                return 0
+
             try:
                 try:
                     cached = await self.db.get_cached_content(url)
@@ -454,6 +583,9 @@ class ExhaustiveScraper:
                         lambda: self._extract_reddit(url), url
                     )
                     await asyncio.sleep(settings.request_delay)
+                elif source_type == "rss":
+                    # El feed no tiene contenido propio: es puro descubrimiento.
+                    return await self._extract_rss(source_id, url, source["depth"])
                 elif source_type == "pdf":
                     content, title = await fetch_with_retry(
                         lambda: self._extract_pdf(url), url
@@ -533,7 +665,8 @@ class ExhaustiveScraper:
 
         # Extract title and new URLs with BS4
         soup = BeautifulSoup(html, "lxml")
-        title = soup.title.string.strip() if soup.title else url
+        # .string es None si el <title> tiene tags anidados; get_text es robusto
+        title = soup.title.get_text(strip=True) if soup.title else url
 
         new_urls = []
         if depth < settings.max_depth:
@@ -593,7 +726,7 @@ class ExhaustiveScraper:
             return None, None
 
         video_id = match.group(1)
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
 
         def _fetch():
             return YouTubeTranscriptApi.get_transcript(
@@ -650,6 +783,42 @@ class ExhaustiveScraper:
             logger.warning("Reddit extraction failed", url=url, error=str(e))
             return None, None
 
+    async def _extract_rss(self, source_id: int, url: str, depth: int) -> int:
+        """RSS/Atom: siembra las entries del feed como fuentes nuevas (filtradas
+        por relevancia de título/URL) y marca el feed como skipped — no tiene
+        contenido propio que trocear. feedparser.parse es bloqueante: va en hilo,
+        igual que en el monitor de noticias. Devuelve nº de fuentes añadidas."""
+        parsed = await asyncio.to_thread(feedparser.parse, url,
+                                         agent=HEADERS["User-Agent"])
+        entries = parsed.entries or []
+        added = 0
+        if depth < settings.max_depth:
+            # Se escanean TODAS las entries y el cap aplica a las añadidas:
+            # capar antes de filtrar perdía las relevantes que no caen entre
+            # las 20 más recientes (verificado con thedebrief: 0/20 vs 3/100).
+            for e in entries:
+                if added >= 20:  # cap, como los 30 links/página de web
+                    break
+                link = normalize_url(_unwrap_news_link(getattr(e, "link", "") or ""))
+                title = getattr(e, "title", "") or ""
+                if (not link.startswith("http") or is_blacklisted(link)
+                        or not self._url_is_relevant(link, title)):
+                    continue
+                if await self.db.source_exists(self.session_id, link):
+                    continue
+                await self.db.add_source(
+                    self.session_id, link, detect_source_type(link),
+                    depth=depth + 1, title=title or None
+                )
+                added += 1
+        await self.db.update_source(
+            source_id, status="skipped",
+            error=f"rss feed: {added} entries sembradas"
+        )
+        logger.info("RSS feed seeded", url=url[:60],
+                    entries=len(entries), added=added)
+        return added
+
     async def _extract_pdf(self, url: str) -> tuple[Optional[str], Optional[str]]:
         """Download and extract PDF text"""
         import pdfplumber

src/seo/.rules.sha256

@@ -0,0 +1 @@
+b8faa93b1f7727d3870e18f69b68283d9a97ed9da819ef0cfa79a60cc2c4ab70

src/seo/__init__.py

@@ -0,0 +1,2 @@
+"""Vendored SEO rule engine (see rules.py). Re-export check_post/score for the bot."""
+from .rules import check_post, score, internal_links, Violation  # noqa: F401

src/seo/_vendor_header.py

@@ -0,0 +1,17 @@
+# -----------------------------------------------------------------------------
+# VENDORED COPY — DO NOT EDIT THIS FILE BY HAND.
+#
+# Canonical source of truth:
+#   git.chemavx.xyz/chemavx/chemavx-seo-tools  ->  seo_rules.py
+#   (local working copy: ~/seo-tools/seo_rules.py)
+#
+# The bot reuses the shared SEO rule engine inside the container, where
+# seo-tools is not installed. Everything below the BEGIN marker is a
+# byte-for-byte copy of the canonical file.
+#
+# To update:   edit the canonical, then run  `make sync-seo`  (re-copies here).
+# Drift guard: CI step "Verify vendored SEO engine" clones the canonical and
+#              diffs it against the content below the marker; the build FAILS on
+#              any divergence. Locally, `make check-seo-sync` does the same.
+# -----------------------------------------------------------------------------
+# ===== BEGIN VENDORED seo_rules.py (exact copy of canonical; do not edit below) =====

src/seo/autofill.py

@@ -0,0 +1,652 @@
+"""SEO autofill — best-effort meta/OG/Twitter/tags/internal-links for a draft.
+
+Step 2 (this file): pure, testable units. NOTHING here is wired into the live
+publish path yet (that is Step 3, behind `settings.seo_autofill`). Every public
+function is isolation-safe: on ANY failure it degrades (menu → [], fields → None,
+link insertion → unchanged html), so it can never block or break article
+publishing, and it never changes a post's `status` away from `draft`.
+
+Three units:
+  * fetch_published_menu(lang)  — live "menu" of existing posts to link to.
+  * generate_seo_fields(...)    — one Haiku JSON call → validated SEO dict.
+  * insert_internal_links(...)  — deterministic, LLM never touches HTML.
+
+The SEO rules engine (length limits, link counting) is the SAME vendored module
+the auditor (Tool A) and validator (Tool C) use: src/seo/rules.py.
+"""
+from __future__ import annotations
+
+import json
+import re
+
+import structlog
+
+from src.config import settings
+from src.llm import get_anthropic_client
+from src.seo import rules as R
+
+logger = structlog.get_logger(__name__)
+
+# Canonical site host per language. The wrapped <a> hrefs use the www host (the
+# site's canonical form); rules.internal_links still counts them because its
+# SITE_HOST ("theexclusionzone.com") is a substring of "www.theexclusionzone.com".
+SITE_BY_LANG = {
+    "en": "www.theexclusionzone.com",
+    "es": "www.zonadeexclusion.com",
+}
+
+# Tag allow-list — the model may ONLY pick from these; invented tags are dropped
+# deterministically after the call (never trust the LLM to self-limit). EN is the
+# live vocabulary Jose confirmed; ES is the live core taxonomy on zonadeexclusion
+# (fetched 2026-07-03), mirroring EN. Deliberately excluded from ES: the one-post
+# long-tail tags the model invented before constraining, and "investigacion-2"
+# (a Ghost case-collision duplicate of "investigacion").
+ALLOWED_TAGS = {
+    "en": [
+        "uap", "declassified", "military-cases", "classic-cases",
+        "investigation", "spain-cases", "latin-america",
+    ],
+    "es": [
+        "uap", "desclasificados", "casos-militares", "casos-clasicos",
+        "investigacion", "casos-espana",
+    ],
+}
+
+# Language-aware default tag when nothing from the allow-list fits / the model
+# returns none. EN canonical is "investigation" (NOT the legacy "investigacion").
+DEFAULT_TAG = {"en": "investigation", "es": "investigacion"}
+
+MAX_INTERNAL_LINKS = 3
+
+# Blocking violations at generation time = only the length/missing rules on the
+# fields WE generate. feature_image*, internal_links.too_few, social *.empty and
+# the INFO rules are expected / non-blocking at draft time.
+_BLOCKING_PREFIXES = ("meta_title.", "meta_description.", "custom_excerpt.")
+
+
+# ─── 1. Published menu ──────────────────────────────────────────────────────
+
+async def fetch_published_menu(lang: str) -> list[dict]:
+    """Live list of published posts on the same site/lang: [{slug, title}, ...].
+
+    Reuses GhostPublisher's per-lang URL + JWT minting. On ANY failure
+    (unconfigured, timeout, non-200, bad JSON) → return [] and log; never 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():
+            logger.warning("seo.menu: Ghost not configured for lang", lang=lang)
+            return []
+
+        # _admin_get lleva los headers canónicos (auth, Accept-Version y el
+        # Accept-Encoding sin br — ver KNOWN-ISSUES.md) y loguea los non-200.
+        data = await pub._admin_get(
+            "posts/?filter=status:published&fields=id,slug,title&limit=all",
+            timeout=30,
+        )
+        if data is None:
+            return []
+
+        menu = [
+            {"slug": p["slug"], "title": p.get("title", "")}
+            for p in data.get("posts", [])
+            if p.get("slug")
+        ]
+        logger.info("seo.menu: fetched", lang=lang, count=len(menu))
+        return menu
+    except Exception as e:  # noqa: BLE001 — isolation guarantee
+        logger.warning("seo.menu: fetch failed, using empty menu",
+                       lang=lang, error=str(e))
+        return []
+
+
+# ─── 2. SEO field generation (one Haiku JSON call) ──────────────────────────
+
+def _system_prompt(lang: str) -> str:
+    allow = ", ".join(ALLOWED_TAGS.get(lang, []))
+    out_lang = "SPANISH" if lang == "es" else "ENGLISH"
+    # The anti-legacy "never use investigacion" rule is EN-only: on the ES site
+    # "investigacion" IS the canonical tag (and the allow-list already excludes
+    # the "investigacion-2" duplicate).
+    legacy_clause = ", never use \"investigacion\"" if lang == "en" else ""
+    tag_clause = (
+        f"TAGS: choose 2-4 tags ONLY from this exact list (never invent a tag, "
+        f"never translate it{legacy_clause}): {allow}.\n"
+        if allow else
+        "TAGS: 2-4 lowercase-hyphenated topical tags appropriate to the article.\n"
+    )
+    return (
+        "You are an SEO editor for an investigative blog about UAP/UFO history.\n"
+        "You are given a FINISHED article and a MENU of existing published posts on the site.\n"
+        "Return ONLY a single JSON object — no prose, no markdown fences — with these fields.\n\n"
+        "HARD LIMITS (count characters; never exceed — and aim BELOW the cap for safety):\n"
+        f"- meta_title:        <= {R.META_TITLE_MAX} characters (aim ~50). Compelling, specific, "
+        "front-load the key entity.\n"
+        f"- meta_description:  <= {R.META_DESC_MAX} characters (aim ~125 — one SHORT sentence). "
+        "Earn the click; describe what the article answers, not clickbait.\n"
+        f"- custom_excerpt:    <= {R.CUSTOM_EXCERPT_MAX} characters (aim ~260). 1-2 sentences "
+        "summarizing the article's substance.\n\n"
+        + tag_clause +
+        "\nINTERNAL LINKS — quality over count, mirror the site's manual policy:\n"
+        "- The \"phrase\" MUST be a SHORT entity name (typically 1-4 words: a person, program, "
+        "office, or named incident) copied VERBATIM from the ARTICLE body. NEVER use a menu "
+        "title or slug as the phrase — the phrase has to literally appear in the article text.\n"
+        "- Link that phrase to a MENU post ONLY if that post's PRIMARY SUBJECT — what it is "
+        "actually ABOUT (judge from its slug + title) — IS that same entity. Not merely a post "
+        "that mentions it. If unsure the target is really ABOUT the entity, DO NOT link it.\n"
+        "  GOOD: phrase \"AARO\" → aaro-... post (short phrase from the body; that post is ABOUT AARO).\n"
+        "  GOOD: phrase \"GIMBAL\" → gimbal-gofast-... post (it is ABOUT the GIMBAL video).\n"
+        "  BAD:  phrase \"AATIP\" → gimbal-gofast-... post (that post is NOT about AATIP). Skip it.\n"
+        "  BAD:  using the full menu title \"AARO's UFO Investigation: What Eight Decades...\" as the "
+        "phrase (that string is not in the article body — it will fail to match). Use \"AARO\".\n"
+        "- 0 to 3 links. NEVER force a link to hit a number. ONE strong, on-subject link beats two "
+        "loose ones. Skip if no target is genuinely about the phrase.\n"
+        "- Use only slugs from the MENU. Never link the article to itself.\n"
+        "- Give the phrase EXACTLY as it appears in the article (case-sensitive), so it can be matched.\n\n"
+        "IMAGE: suggest a concrete image search query (subjects/objects a stock site would have — "
+        "NOT the headline) and a one-sentence context describing what the article is about.\n\n"
+        f"Write meta_title, meta_description, custom_excerpt, image_query and image_context in {out_lang}.\n\n"
+        "JSON schema (all fields required):\n"
+        '{"meta_title": "...", "meta_description": "...", "custom_excerpt": "...", '
+        '"tags": ["..."], "internal_links": [{"phrase": "...", "slug": "..."}], '
+        '"image_query": "...", "image_context": "..."}'
+    )
+
+
+def _user_message(article_text: str, link_menu: list[dict]) -> str:
+    menu_lines = "\n".join(
+        f"- {m['slug']} — {m.get('title','')}" for m in link_menu
+    ) or "(no existing posts)"
+    return (
+        "MENU of existing published posts (slug — title):\n"
+        f"{menu_lines}\n\n"
+        "ARTICLE:\n"
+        f"{article_text}"
+    )
+
+
+def _parse_json_object(text: str) -> dict:
+    """Strip optional ``` fences and parse the first JSON object. Raises on failure."""
+    t = text.strip()
+    t = re.sub(r"^```(?:json)?\s*", "", t)
+    t = re.sub(r"\s*```$", "", t).strip()
+    # If extra prose snuck in, grab the outermost {...}.
+    if not t.startswith("{"):
+        m = re.search(r"\{.*\}", t, re.DOTALL)
+        if m:
+            t = m.group(0)
+    obj = json.loads(t)
+    if not isinstance(obj, dict):
+        raise ValueError("model did not return a JSON object")
+    return obj
+
+
+_REQUIRED_STR = ("meta_title", "meta_description", "custom_excerpt",
+                 "image_query", "image_context")
+
+
+def _coerce(obj: dict, lang: str) -> dict:
+    """Validate shape + constrain tags to the allow-list. Raises on missing required strings."""
+    for k in _REQUIRED_STR:
+        if not isinstance(obj.get(k), str) or not obj[k].strip():
+            raise ValueError(f"missing/empty required field: {k}")
+
+    raw_tags = obj.get("tags") or []
+    if not isinstance(raw_tags, list):
+        raw_tags = []
+    allow = ALLOWED_TAGS.get(lang)
+    if allow:
+        seen, tags = set(), []
+        allow_set = set(allow)
+        for t in raw_tags:
+            if isinstance(t, str):
+                s = t.strip().lower()
+                if s in allow_set and s not in seen:
+                    seen.add(s)
+                    tags.append(s)
+        if not tags:
+            tags = [DEFAULT_TAG.get(lang, "investigation")]
+    else:
+        tags = [t.strip().lower() for t in raw_tags
+                if isinstance(t, str) and t.strip()] or [DEFAULT_TAG.get(lang, "investigacion")]
+
+    links = []
+    for item in (obj.get("internal_links") or []):
+        if isinstance(item, dict) and item.get("phrase") and item.get("slug"):
+            links.append({"phrase": str(item["phrase"]), "slug": str(item["slug"])})
+
+    return {
+        "meta_title": obj["meta_title"].strip(),
+        "meta_description": obj["meta_description"].strip(),
+        "custom_excerpt": obj["custom_excerpt"].strip(),
+        "tags": tags,
+        "internal_links": links,
+        "image_query": obj["image_query"].strip(),
+        "image_context": obj["image_context"].strip(),
+    }
+
+
+def _markdown_to_html(article_text: str) -> str:
+    """Same conversion publish_draft uses, so validation sees the real body."""
+    import markdown as _md
+    html = _md.markdown(article_text, extensions=["extra"])
+    return re.sub(r"<h1[^>]*>.*?</h1>", "", html, count=1, flags=re.DOTALL).lstrip()
+
+
+def _synthetic_post(fields: dict, html: str, title: str, slug: str) -> dict:
+    mt, md = fields["meta_title"], fields["meta_description"]
+    return {
+        "html": html,
+        "title": title,
+        "slug": slug or "",
+        "meta_title": mt,
+        "meta_description": md,
+        "custom_excerpt": fields["custom_excerpt"],
+        "og_title": mt, "og_description": md,
+        "twitter_title": mt, "twitter_description": md,
+        "feature_image": "",       # human adds later
+        "feature_image_alt": "",   # human adds later
+    }
+
+
+def _blocking(violations) -> list:
+    return [v for v in violations if v.rule.startswith(_BLOCKING_PREFIXES)]
+
+
+# Length-limited fields we generate. The retry aims at (limit - margin), well
+# UNDER the hard limit: Haiku cannot count to an exact char count and reliably
+# overshoots its target by 20-50 chars, so the margin must absorb that overshoot.
+# Margins are per-field (bigger for the long free-text fields that overshoot most).
+_LEN_LIMITS = {
+    "meta_title": R.META_TITLE_MAX,
+    "meta_description": R.META_DESC_MAX,
+    "custom_excerpt": R.CUSTOM_EXCERPT_MAX,
+}
+_LEN_MARGIN = {
+    "meta_title": 10,        # target 50
+    "meta_description": 35,   # target 110
+    "custom_excerpt": 45,     # target 255
+}
+
+
+# Minimum useful length per field. If a CLEAN boundary-trim would drop a field
+# below this, we DON'T trim it — better a slightly-over field a human nudges than
+# a butchered stub. Falls back to path-1 (keep + flag) for that field only.
+_MIN_USEFUL = {
+    "meta_title": 25,
+    "meta_description": 80,
+    "custom_excerpt": 120,
+}
+
+# Connective / function words (EN + ES) we must not leave dangling at the end of
+# a word-boundary trim — they all "expect more" after them, so ending on one reads
+# as cut-off. Compared lowercased, after stripping trailing punctuation. Includes
+# correlatives (neither/nor), interrogatives (why/what), and contracted auxiliaries
+# (won't/can't) that surfaced as bad trims in testing.
+_DANGLING_WORDS = {
+    # English — articles / prepositions / basic connectives
+    "and", "or", "but", "the", "a", "an", "of", "to", "in", "on", "for", "with",
+    "at", "by", "from", "as", "that", "which", "is", "was", "were", "this",
+    "its", "their", "his", "her", "into", "over", "after", "about", "between",
+    "during", "against", "among", "without", "within", "upon", "toward", "towards",
+    "via", "per", "amid", "despite", "near", "off", "out", "up", "down",
+    # English — correlatives / subordinators / interrogatives / negation
+    "nor", "neither", "either", "both", "whether", "while", "since", "though",
+    "although", "unless", "until", "because", "if", "than", "then", "yet", "so",
+    "not", "no", "why", "how", "when", "where", "what", "who", "whom", "whose",
+    # English — contracted auxiliaries (read as mid-clause)
+    "won't", "can't", "cannot", "don't", "doesn't", "didn't", "isn't", "aren't",
+    "wasn't", "weren't", "hasn't", "haven't", "still", "just", "ever",
+    # Spanish
+    "y", "o", "u", "pero", "el", "la", "los", "las", "un", "una", "de", "del",
+    "en", "con", "por", "para", "que", "su", "sus", "al", "como", "se", "lo",
+    "ni", "sino", "porque", "aunque", "mientras", "cuando", "donde", "segun",
+    "según", "sin", "entre", "sobre", "tras", "ante", "hacia", "hasta", "desde",
+    "ya", "muy", "mas", "más", "menos", "tan", "no",
+}
+
+# Characters safe to strip from the end of a trimmed phrase (whitespace, commas,
+# semicolons, colons, dashes/em-dashes). Sentence-final . ! ? are intentionally
+# NOT here — if a trim happens to land on one we keep it.
+_TRAIL_PUNCT = " \t,;:—–- "
+
+
+def _drop_sentences_to_fit(text: str, limit: int) -> str:
+    """Drop WHOLE trailing sentences until the text fits within `limit`.
+
+    Splits on sentence terminators (. ! ?) keeping the delimiter, then returns the
+    longest run of complete leading sentences that is <= limit. Never a mid-
+    sentence cut. Returns "" if even the first sentence is over limit (caller's
+    min-useful guardrail then keeps the original)."""
+    parts = [p for p in re.findall(r"[^.!?]*[.!?]+|\S[^.!?]*$", text.strip()) if p.strip()]
+    if not parts:
+        return ""
+    out = ""
+    for p in parts:
+        candidate = out + p
+        if len(candidate.strip()) <= limit:
+            out = candidate
+        else:
+            break
+    return out.strip()
+
+
+def _trim_to_word_boundary(text: str, limit: int) -> str:
+    """Trim to the last WORD boundary that fits, then strip trailing punctuation
+    and any dangling connective word. No mid-word cut, no ellipsis. The result
+    reads as a complete-ish phrase. Returns "" if nothing survives the cleanup."""
+    text = text.strip()
+    if len(text) <= limit:
+        return text
+    out = ""
+    for w in text.split():
+        candidate = w if not out else f"{out} {w}"
+        if len(candidate) <= limit:
+            out = candidate
+        else:
+            break
+    # Clean the tail: alternately strip trailing punctuation and dangling words
+    # until the phrase ends on a content word.
+    while out:
+        stripped = out.rstrip(_TRAIL_PUNCT)
+        if stripped != out:
+            out = stripped
+            continue
+        # Capture the trailing word INCLUDING internal apostrophes, so contractions
+        # ("won't", "doesn't") match the dangling list instead of just their tail.
+        m = re.search(r"([^\W\d_]+(?:['’][^\W\d_]+)*)$", out, re.UNICODE)
+        if m and m.group(1).lower() in _DANGLING_WORDS:
+            out = out[:m.start()].rstrip(_TRAIL_PUNCT)
+            continue
+        break
+    return out.strip()
+
+
+def _shorten_over_limit(fields: dict) -> tuple[dict, list[dict]]:
+    """FINAL, deterministic, boundary-aware shortener — runs only after the LLM +
+    retry have tried, only on fields STILL over their hard limit. This is NOT the
+    forbidden ugly truncation: custom_excerpt drops whole trailing sentences; the
+    single-line metas trim to a word boundary and clean dangling punctuation.
+
+    Quality guardrail: if a clean shorten would fall below the field's minimum
+    useful length, we keep the LLM's over-limit text and flag it (path-1 fallback
+    for that field only). Mutates `fields` in place; also returns it.
+
+    Returns (fields, log) where log entries are
+    {field, before, after, applied, reason} for reporting/auditing."""
+    log: list[dict] = []
+    for field, limit in _LEN_LIMITS.items():
+        before = fields.get(field, "")
+        if len(before) <= limit:
+            continue
+        if field == "custom_excerpt":
+            after = _drop_sentences_to_fit(before, limit)
+        else:
+            # Single-line metas: prefer a clean WHOLE-sentence prefix (many metas
+            # are 2 sentences — keeping just the first reads as a complete thought).
+            # Only fall back to a word-boundary trim if that prefix is too short.
+            after = _drop_sentences_to_fit(before, limit)
+            if not after or len(after) < _MIN_USEFUL[field]:
+                after = _trim_to_word_boundary(before, limit)
+
+        if after and len(after) <= limit and len(after) >= _MIN_USEFUL[field]:
+            fields[field] = after
+            log.append({"field": field, "before": before, "after": after,
+                        "applied": True, "reason": "shortened"})
+            logger.info("seo.shorten: field shortened",
+                        field=field, before_len=len(before), after_len=len(after))
+        else:
+            reason = "would-be-too-short" if after else "no-clean-boundary"
+            log.append({"field": field, "before": before, "after": before,
+                        "applied": False, "reason": reason})
+            logger.info("seo.shorten: kept over-limit (clean trim unsafe)",
+                        field=field, before_len=len(before),
+                        trimmed_len=len(after), reason=reason)
+    return fields, log
+
+
+def _retry_instruction(fields: dict) -> str:
+    """Forceful, field-specific shrink instruction listing actual length, hard
+    limit, the exact overage, and a sub-limit target. Empty string if nothing over."""
+    lines = []
+    for field, limit in _LEN_LIMITS.items():
+        cur = len(fields.get(field, ""))
+        if cur > limit:
+            target = limit - _LEN_MARGIN[field]
+            cut = cur - target
+            lines.append(
+                f"- {field} is currently {cur} characters. The HARD limit is {limit}. "
+                f"Rewrite it to AT MOST {target} characters (aim for {target}, NOT {limit}; "
+                f"shorter is better than longer) — cut at least {cut} characters by removing a "
+                f"clause, adjective, or example. Keep the same meaning and language."
+            )
+    if not lines:
+        return ""
+    return (
+        "\n\nSTOP. YOUR PREVIOUS OUTPUT EXCEEDED A HARD CHARACTER LIMIT. "
+        "Fix ONLY the field(s) below; leave every other field exactly as you had it:\n"
+        + "\n".join(lines)
+        + "\nReturn the FULL JSON object again with ALL fields present, only these shortened. "
+        "When in doubt, cut MORE — a shorter field is always acceptable, an over-limit one is not."
+    )
+
+
+async def generate_seo_fields(
+    article_text: str,
+    link_menu: list[dict],
+    lang: str,
+    *,
+    title: str = "",
+    slug: str = "",
+    db=None,
+    session_id: int | None = None,
+) -> dict | None:
+    """Second, dedicated Haiku call → validated SEO dict, or None on hard failure.
+
+    Returns: meta_title, meta_description, custom_excerpt, tags[],
+    internal_links[{phrase,slug}], image_query, image_context, og_/twitter_*
+    (derived deterministically), and seo_warnings[] for any non-fatal issues.
+    On ANY exception → None (caller falls back to a bare draft).
+    """
+    try:
+        client = get_anthropic_client()
+        system = _system_prompt(lang)
+        user = _user_message(article_text, link_menu)
+
+        async def _raw(messages: list, max_tokens: int = 1024,
+                       temperature: float | None = None) -> str:
+            kwargs = {
+                "model": settings.claude_model,
+                "max_tokens": max_tokens,
+                "system": system,
+                "messages": messages,
+            }
+            if temperature is not None:
+                kwargs["temperature"] = temperature
+            msg = await client.messages.create(**kwargs)
+            if db is not None and session_id is not None:
+                try:
+                    await db.log_api_call(
+                        session_id, "seo_fields", settings.claude_model,
+                        msg.usage.input_tokens, msg.usage.output_tokens,
+                    )
+                except Exception as log_err:  # noqa: BLE001
+                    logger.warning("seo.fields: usage log failed", error=str(log_err))
+            return msg.content[0].text
+
+        base_msgs = [{"role": "user", "content": user}]
+        text1 = await _raw(base_msgs)
+        fields = _coerce(_parse_json_object(text1), lang)
+        fields["internal_links"] = _sanitize_links(fields["internal_links"], link_menu)
+
+        # Validate against the shared engine using the real body (md→html + links).
+        body_html = _markdown_to_html(article_text)
+        linked_html, _ = insert_internal_links(body_html, fields["internal_links"], link_menu, lang)
+        violations = R.check_post(_synthetic_post(fields, linked_html, title, slug))
+        blocking = _blocking(violations)
+
+        if blocking:
+            detail = "; ".join(
+                f"{v.rule}: {v.message}" for v in blocking
+            )
+            instr = _retry_instruction(fields)
+            logger.info("seo.fields: blocking violation, one stricter retry", detail=detail)
+            try:
+                # Real edit turn: hand the model its OWN previous JSON to shorten,
+                # rather than re-rolling from scratch (which kept overshooting). Lower
+                # max_tokens to physically discourage rambling.
+                retry_msgs = base_msgs + [
+                    {"role": "assistant", "content": text1},
+                    {"role": "user", "content": instr},
+                ]
+                # temperature=0 so the shorten instruction binds deterministically.
+                retry = _coerce(_parse_json_object(
+                    await _raw(retry_msgs, max_tokens=768, temperature=0.0)), lang)
+                retry["internal_links"] = _sanitize_links(retry["internal_links"], link_menu)
+                rlinked, _ = insert_internal_links(
+                    _markdown_to_html(article_text), retry["internal_links"], link_menu, lang)
+                rviol = R.check_post(_synthetic_post(retry, rlinked, title, slug))
+                if not _blocking(rviol):
+                    fields, violations, blocking = retry, rviol, []
+                else:
+                    # Keep the retry's text but record it stayed over (never truncate).
+                    fields, violations, blocking = retry, rviol, _blocking(rviol)
+            except Exception as e:  # noqa: BLE001
+                logger.warning("seo.fields: retry failed, keeping first output", error=str(e))
+
+        # Final boundary-aware shortener — only for fields the LLM + retry left
+        # over limit. Clean (sentence-drop / word-boundary), never mid-word, and
+        # skipped (kept + flagged) if it would butcher the field below min-useful.
+        shorten_log: list[dict] = []
+        if blocking:
+            fields, shorten_log = _shorten_over_limit(fields)
+            slinked, _ = insert_internal_links(
+                _markdown_to_html(article_text), fields["internal_links"], link_menu, lang)
+            violations = R.check_post(_synthetic_post(fields, slinked, title, slug))
+            blocking = _blocking(violations)
+
+        mt, md = fields["meta_title"], fields["meta_description"]
+        fields["og_title"] = fields["twitter_title"] = mt
+        fields["og_description"] = fields["twitter_description"] = md
+        fields["seo_warnings"] = [
+            f"{v.rule}: {v.message}" for v in blocking
+        ]
+        if shorten_log:
+            fields["seo_shortened"] = [
+                {"field": e["field"], "applied": e["applied"], "reason": e["reason"]}
+                for e in shorten_log
+            ]
+        return fields
+    except Exception as e:  # noqa: BLE001 — isolation guarantee
+        logger.warning("seo.fields: generation failed, falling back to bare draft",
+                       error=str(e))
+        return None
+
+
+# ─── 3. Deterministic internal-link insertion ──────────────────────────────
+
+def _sanitize_links(suggestions: list[dict], menu: list[dict]) -> list[dict]:
+    """Defense-in-depth: drop malformed link suggestions before insertion/reporting.
+
+    The "phrase" must be SHORT verbatim ARTICLE text — never a slug or a menu
+    title. A suggestion is dropped when its phrase equals its own slug, any menu
+    slug, or any menu title (the slug-as-phrase / title-as-phrase traps). Even if
+    the model ignores the prompt rule, the output stays clean. Drops are logged at
+    debug and never raise; survivors still go through the verbatim-in-body guard.
+    """
+    menu_slugs = {(m.get("slug") or "").strip().casefold() for m in menu}
+    menu_titles = {(m.get("title") or "").strip().casefold() for m in menu}
+    menu_slugs.discard("")
+    menu_titles.discard("")
+    clean: list[dict] = []
+    for s in suggestions:
+        phrase = (s.get("phrase") or "").strip()
+        slug = (s.get("slug") or "").strip()
+        if not phrase or not slug:
+            continue
+        pf = phrase.casefold()
+        if pf == slug.casefold() or pf in menu_slugs or pf in menu_titles:
+            logger.debug("seo.links: dropped malformed suggestion (slug/title as phrase)",
+                         phrase=phrase, slug=slug)
+            continue
+        clean.append({"phrase": phrase, "slug": slug})
+    return clean
+
+
+def insert_internal_links(
+    html: str,
+    suggestions: list[dict],
+    menu: list[dict],
+    lang: str = "en",
+) -> tuple[str, int]:
+    """Wrap the FIRST verbatim, word-boundary occurrence of each suggested phrase
+    in an <a> to its menu slug. Deterministic — the LLM never edits HTML.
+
+    Rules: slug must be in the menu; phrase must appear verbatim in a TEXT node
+    (never inside a tag, never inside an existing <a>); word-boundary match (no
+    partial words); each phrase/slug used at most once; cap at MAX_INTERNAL_LINKS.
+    A phrase that is missing or already linked is silently skipped (never forced,
+    never an error). Returns (modified_html, inserted_pairs) where inserted_pairs
+    is the list of {phrase, slug} dicts actually wrapped (len == links inserted).
+    """
+    if not html or not suggestions:
+        return html, []
+
+    site = SITE_BY_LANG.get(lang, SITE_BY_LANG["en"])
+    menu_slugs = {m["slug"] for m in menu if m.get("slug")}
+
+    # Split into tag tokens and text tokens; we only ever edit text tokens, and
+    # only when not inside an <a>…</a> (anchor depth 0). This guarantees no nested
+    # links and no edits inside tag attributes.
+    tokens = re.split(r"(<[^>]+>)", html)
+    inserted = 0
+    inserted_pairs: list[dict] = []
+    used_phrases: set[str] = set()
+    used_slugs: set[str] = set()
+
+    for sug in suggestions:
+        if inserted >= MAX_INTERNAL_LINKS:
+            break
+        phrase = (sug.get("phrase") or "").strip()
+        slug = (sug.get("slug") or "").strip()
+        if not phrase or not slug:
+            continue
+        if slug not in menu_slugs:
+            continue
+        if phrase in used_phrases or slug in used_slugs:
+            continue
+
+        pat = re.compile(r"(?<!\w)" + re.escape(phrase) + r"(?!\w)")
+        anchor_depth = 0
+        done = False
+        for i, tok in enumerate(tokens):
+            if tok.startswith("<") and tok.endswith(">"):
+                low = tok.lower()
+                if low.startswith("<a") and not low.startswith("</a"):
+                    anchor_depth += 1
+                elif low.startswith("</a"):
+                    anchor_depth = max(0, anchor_depth - 1)
+                continue
+            if anchor_depth > 0 or not tok:
+                continue
+            m = pat.search(tok)
+            if not m:
+                continue
+            replacement = (
+                f'<a href="https://{site}/{slug}/">{m.group(0)}</a>'
+            )
+            tokens[i] = tok[:m.start()] + replacement + tok[m.end():]
+            inserted += 1
+            inserted_pairs.append({"phrase": phrase, "slug": slug})
+            used_phrases.add(phrase)
+            used_slugs.add(slug)
+            done = True
+            break
+        if not done:
+            logger.debug("seo.links: phrase not found / already linked, skipped",
+                         phrase=phrase, slug=slug)
+
+    return "".join(tokens), inserted_pairs

src/seo/rules.py

@@ -0,0 +1,220 @@
+# -----------------------------------------------------------------------------
+# VENDORED COPY — DO NOT EDIT THIS FILE BY HAND.
+#
+# Canonical source of truth:
+#   git.chemavx.xyz/chemavx/chemavx-seo-tools  ->  seo_rules.py
+#   (local working copy: ~/seo-tools/seo_rules.py)
+#
+# The bot reuses the shared SEO rule engine inside the container, where
+# seo-tools is not installed. Everything below the BEGIN marker is a
+# byte-for-byte copy of the canonical file.
+#
+# To update:   edit the canonical, then run  `make sync-seo`  (re-copies here).
+# Drift guard: CI step "Verify vendored SEO engine" clones the canonical and
+#              diffs it against the content below the marker; the build FAILS on
+#              any divergence. Locally, `make check-seo-sync` does the same.
+# -----------------------------------------------------------------------------
+# ===== BEGIN VENDORED seo_rules.py (exact copy of canonical; do not edit below) =====
+"""
+Reusable SEO rule engine for The Exclusion Zone (EN) — theexclusionzone.com (Ghost).
+
+Pure functions, NO I/O. Feed it a Ghost Admin API post dict (with `html` format
+included) and it returns a list of Violation(rule, severity, message, fix).
+
+Shared by:
+  - seo_audit.py        — Tool A, site-wide auditor (this engine, run over all posts)
+  - (future) seo_validate.py — Tool C, pre-publish validator (same engine, one draft)
+
+Design note: every check is an independent function registered in RULES. To add a
+rule, write a function (post) -> list[Violation] and append it to RULES. The auditor
+and the validator both just call check_post(); they never re-implement a check.
+"""
+import re
+from collections import namedtuple
+
+SITE_HOST = "theexclusionzone.com"
+
+# ---- thresholds (single source of truth, reused by validator) -------------
+META_TITLE_MAX     = 60
+META_DESC_MAX      = 145
+CUSTOM_EXCERPT_MAX = 300
+MIN_INTERNAL_LINKS = 2
+
+# ---- severity weights (used to rank "worst first") ------------------------
+HIGH, MED, LOW, INFO = 3, 2, 1, 0
+SEV_NAME = {HIGH: "HIGH", MED: "MED", LOW: "LOW", INFO: "INFO"}
+
+# The Edition-main theme injects BlogPosting JSON-LD globally in default.hbs
+# ({{#is "post"}} ... <script type="application/ld+json"> @type BlogPosting),
+# plus Ghost's own {{ghost_head}}. So JSON-LD is handled site-wide and is NOT a
+# per-post rule. Flip to False only if that theme block is ever removed.
+THEME_HANDLES_JSONLD = True
+
+Violation = namedtuple("Violation", "rule severity message fix")
+
+
+def _s(v):
+    return v if isinstance(v, str) else ""
+
+
+def _empty(v):
+    return not (isinstance(v, str) and v.strip())
+
+
+# First path segments that are NOT article posts on a Ghost site (tags, authors,
+# pagination, static content, etc.). Keeps root-relative link counting from
+# treating /tag/foo or /content/images/... as an internal article link.
+NON_POST_PREFIXES = {
+    "tag", "tags", "author", "page", "p", "content", "assets",
+    "rss", "ghost", "members", "404", "sitemap",
+}
+
+
+def _internal_slug(href, self_slug):
+    """Return the article slug an href points to if it's an internal POST link,
+    else None. Accepts both absolute (contains SITE_HOST) and root-relative
+    ("/slug/") forms; rejects protocol-relative ("//host"), anchors, mailto,
+    external links, static assets, and known non-post sections."""
+    if SITE_HOST in href:
+        path = re.sub(r"^https?://[^/]+/", "", href)
+    elif href.startswith("/") and not href.startswith("//"):
+        path = href[1:]
+    else:
+        return None
+    slug = path.strip("/").split("/")[0]
+    if not slug or slug == self_slug:
+        return None
+    if slug in NON_POST_PREFIXES or "." in slug:  # section page or static asset
+        return None
+    return slug
+
+
+def internal_links(post):
+    """Distinct internal article slugs linked from the body, excluding self-links.
+
+    Counts BOTH absolute internal links (href containing SITE_HOST) and
+    root-relative links ("/slug/"), so a Ghost-relative link isn't miscounted as
+    "too few". Tool A (auditor) and Tool C (validator) share this, staying in sync.
+    """
+    html = _s(post.get("html"))
+    self_slug = post.get("slug")
+    out = set()
+    for m in re.finditer(r'href="([^"#]+)"', html):
+        slug = _internal_slug(m.group(1), self_slug)
+        if slug:
+            out.add(slug)
+    return out
+
+
+# --- individual rules -------------------------------------------------------
+
+def r_meta_title(p):
+    mt = _s(p.get("meta_title"))
+    title = _s(p.get("title"))
+    if _empty(mt):
+        sev = MED if len(title) > META_TITLE_MAX else LOW
+        return [Violation("meta_title.missing", sev,
+                          f"meta_title missing → falls back to title ({len(title)} chars"
+                          + (f", which is >{META_TITLE_MAX}!" if len(title) > META_TITLE_MAX else "") + ")",
+                          "set meta_title")]
+    if len(mt) > META_TITLE_MAX:
+        return [Violation("meta_title.too_long", MED,
+                          f"meta_title {len(mt)} chars > {META_TITLE_MAX}", "shorten meta_title")]
+    return []
+
+
+def r_meta_description(p):
+    md = _s(p.get("meta_description"))
+    if _empty(md):
+        return [Violation("meta_description.missing", HIGH,
+                          "meta_description MISSING (no SERP snippet control)", "write meta_description")]
+    if len(md) > META_DESC_MAX:
+        return [Violation("meta_description.too_long", MED,
+                          f"meta_description {len(md)} chars > {META_DESC_MAX} (will be truncated in SERP)",
+                          "shorten meta_description")]
+    return []
+
+
+def r_custom_excerpt(p):
+    ce = _s(p.get("custom_excerpt"))
+    if ce and len(ce) > CUSTOM_EXCERPT_MAX:
+        return [Violation("custom_excerpt.too_long", LOW,
+                          f"custom_excerpt {len(ce)} chars > {CUSTOM_EXCERPT_MAX}", "shorten custom_excerpt")]
+    return []
+
+
+def r_social_fields(p):
+    """OG/Twitter empties — the safest auto-fixable category (mirror meta_*)."""
+    out = []
+    fallbacks = {
+        "og_title":          "meta_title",
+        "og_description":    "meta_description",
+        "twitter_title":     "meta_title",
+        "twitter_description": "meta_description",
+    }
+    for field, src in fallbacks.items():
+        if _empty(p.get(field)):
+            out.append(Violation(f"{field}.empty", LOW,
+                                 f"{field} empty", f"mirror from {src}"))
+    return out
+
+
+def r_feature_image(p):
+    out = []
+    if _empty(p.get("feature_image")):
+        out.append(Violation("feature_image.missing", MED,
+                             "feature_image missing (no social/share card image)", "add feature image"))
+    else:
+        if _empty(p.get("feature_image_alt")):
+            out.append(Violation("feature_image_alt.missing", LOW,
+                                 "feature_image_alt missing (a11y + image SEO)", "add feature image alt text"))
+    return out
+
+
+def r_internal_links(p):
+    n = len(internal_links(p))
+    if n < MIN_INTERNAL_LINKS:
+        return [Violation("internal_links.too_few", MED,
+                          f"{n} internal link(s) < {MIN_INTERNAL_LINKS} (weak interlinking)",
+                          "add internal links to related articles")]
+    return []
+
+
+def r_title_equals_meta(p):
+    mt = _s(p.get("meta_title"))
+    title = _s(p.get("title"))
+    if mt and mt == title:
+        return [Violation("title_eq_meta_title", INFO,
+                          "meta_title identical to title (often fine; review)", "")]
+    return []
+
+
+def r_jsonld(p):
+    if THEME_HANDLES_JSONLD:
+        return []
+    return [Violation("jsonld.missing", MED, "no BlogPosting JSON-LD", "add JSON-LD")]
+
+
+RULES = [
+    r_meta_title,
+    r_meta_description,
+    r_custom_excerpt,
+    r_social_fields,
+    r_feature_image,
+    r_internal_links,
+    r_title_equals_meta,
+    r_jsonld,
+]
+
+
+def check_post(post):
+    """Run every rule against one post dict → list[Violation]."""
+    out = []
+    for rule in RULES:
+        out.extend(rule(post))
+    return out
+
+
+def score(violations):
+    """Total severity weight (for ranking posts worst-first); INFO counts 0."""
+    return sum(v.severity for v in violations)

tests/test_scraper.py

@@ -11,6 +11,16 @@ def test_detect_source_type():
     assert detect_source_type("https://example.com/article") == "web"
 
 
+def test_detect_source_type_rss():
+    assert detect_source_type("https://thedebrief.org/feed/") == "rss"
+    assert detect_source_type("https://example.com/rss") == "rss"
+    assert detect_source_type("https://example.com/feeds/all.atom.xml") == "rss"
+    assert detect_source_type("https://www.liberationtimes.com/?format=rss") == "rss"
+    # Falsos positivos del substring viejo: no son feeds.
+    assert detect_source_type("https://example.com/feedback") == "web"
+    assert detect_source_type("https://example.com/atomic-theory") == "web"
+
+
 def test_is_blacklisted():
     assert is_blacklisted("https://facebook.com/something") == True
     assert is_blacklisted("https://en.wikipedia.org/wiki/Test") == False
@@ -26,3 +36,22 @@ def test_simple_chunk():
     chunks = simple_chunk(text, chunk_size=100, overlap=20)
     assert len(chunks) > 1
     assert all(isinstance(c, str) for c in chunks)
+
+
+def test_unwrap_news_link():
+    from src.scraper.exhaustive import _unwrap_news_link
+    wrapped = ("http://www.bing.com/news/apiclick.aspx?ref=FexRss&aid=&tid=x"
+               "&url=https://example.com/article&c=1&mkt=es-es")
+    assert _unwrap_news_link(wrapped) == "https://example.com/article"
+    # Links normales pasan intactos
+    assert _unwrap_news_link("https://thedebrief.org/foo") == "https://thedebrief.org/foo"
+    # url= protocol-relative se normaliza a https
+    assert _unwrap_news_link(
+        "http://www.bing.com/news/apiclick.aspx?url=//example.com/article"
+    ) == "https://example.com/article"
+    # apiclick sin url= usable se descarta ("" → el caller lo salta), nunca se
+    # deja pasar el redirect de bing.com como fuente
+    assert _unwrap_news_link("http://www.bing.com/news/apiclick.aspx?ref=x") == ""
+    assert _unwrap_news_link(
+        "http://www.bing.com/news/apiclick.aspx?url=javascript:alert(1)"
+    ) == ""

tests/test_seo_autofill.py

@@ -0,0 +1,39 @@
+from src.seo.autofill import ALLOWED_TAGS, DEFAULT_TAG, _coerce, _system_prompt
+
+BASE = {
+    "meta_title": "t",
+    "meta_description": "d",
+    "custom_excerpt": "e",
+    "image_query": "q",
+    "image_context": "c",
+}
+
+
+def test_coerce_es_drops_invented_tags():
+    obj = dict(BASE, tags=["uap", "humanoides", "Desclasificados", "investigacion-2"])
+    out = _coerce(obj, "es")
+    assert out["tags"] == ["uap", "desclasificados"]
+
+
+def test_coerce_es_falls_back_to_default():
+    obj = dict(BASE, tags=["pentagono", "encuentros-cercanos"])
+    out = _coerce(obj, "es")
+    assert out["tags"] == [DEFAULT_TAG["es"]]
+
+
+def test_coerce_en_still_constrained():
+    obj = dict(BASE, tags=["uap", "investigacion"])
+    out = _coerce(obj, "en")
+    assert out["tags"] == ["uap"]
+
+
+def test_system_prompt_lists_allowed_tags_per_lang():
+    es = _system_prompt("es")
+    en = _system_prompt("en")
+    for tag in ALLOWED_TAGS["es"]:
+        assert tag in es
+    # La regla anti-legacy 'never use "investigacion"' es solo para EN: en ES
+    # "investigacion" es el tag canónico del allow-list.
+    assert 'never use "investigacion"' in en
+    assert 'never use "investigacion"' not in es
+    assert "ONLY from this exact list" in es