Tighten Asteria candidate board quality

Author: sametbasbug

news_pipeline/news_pipeline/cli/commands/heartbeat_prepare_one.py

@@ -20,9 +20,12 @@
 from news_pipeline.queue.service import QueueService
 from news_pipeline.storage.json_store import JsonStore
 
-MAX_PER_SOURCE = 3
+MAX_PER_SOURCE = 2
+DEFAULT_MAX_SOURCE_AGE_HOURS = 30
 DEFAULT_PREWARM_BOARD_PATH = "news_pipeline/data/heartbeat/prepared-board.json"
 MIN_HEALTHY_BOARD_ELIGIBLE = 6
+MIN_THIN_SOURCE_TEXT_CHARS = 500
+MIN_BOARD_FILL_SCORE = 0.50
 HOT_CATEGORY_RECENT_WINDOW = 3
 HOT_CATEGORY_REPEAT_THRESHOLD = 2
 HOT_CATEGORY_POLICY = "skip_target_fill_only"
@@ -94,6 +97,10 @@
     "red card controversy",
     "football chiefs",
     "training grounds",
+    "gifted by erdogan",
+    "pistols with ammunition",
+    "revolver gifted",
+    "parting gift",
 }
 SPONSORED_URL_MARKERS = {
     "/sponsored-content/",
@@ -152,6 +159,8 @@
     "football chiefs",
     "red card",
     "infantino",
+    "revolver gifted",
+    "parting gift",
 }
 
 
@@ -237,6 +246,10 @@ def _source_host(item: Any) -> str:
     return host[4:] if host.startswith("www.") else host
 
 
+def _normalized_article(root: Path, item: Any) -> NormalizedArticle | None:
+    return JsonStore(root / "news_pipeline/data/normalized", NormalizedArticle).load(item.normalized_id)
+
+
 def _is_sponsored_source(item: Any) -> bool:
     url = _source_url(item).lower()
     return any(marker in url for marker in SPONSORED_URL_MARKERS)
@@ -246,6 +259,20 @@ def _has_known_unreadable_primary_source(item: Any) -> bool:
     return _source_host(item) in UNREADABLE_PRIMARY_HOSTS
 
 
+def _source_text_quality_rejection(root: Path, item: Any) -> str | None:
+    article = _normalized_article(root, item)
+    if article is None:
+        return "missing normalized article"
+    source = _normalized(_source_name(item))
+    text = _normalized(" ".join(part for part in [article.summary or "", article.content_snippet or ""] if part))
+    text_len = len(text)
+    if source == "france 24 world" and re.search(r"\bfrance 24'?s\b.*\b(reports|tells us|explains)\b", text):
+        return "thin France 24 video/report fallback"
+    if source == "politico europe" and text_len < MIN_THIN_SOURCE_TEXT_CHARS:
+        return f"thin Politico snippet ({text_len} chars < {MIN_THIN_SOURCE_TEXT_CHARS})"
+    return None
+
+
 def _frontmatter_value(text: str, key: str) -> str:
     match = re.search(rf"(?m)^{re.escape(key)}:\s*[\"']?(.*?)[\"']?\s*$", text)
     return match.group(1).strip() if match else ""
@@ -352,14 +379,12 @@ def _live_source_urls(root: Path) -> set[str]:
 
 
 def _headline_text(root: Path, item: Any) -> str:
-    normalized_store = JsonStore(root / "news_pipeline/data/normalized", NormalizedArticle)
-    article = normalized_store.load(item.normalized_id)
+    article = _normalized_article(root, item)
     return article.title if article else item.draft_title
 
 
 def _duplicate_probe_text(root: Path, item: Any) -> tuple[str, str]:
-    normalized_store = JsonStore(root / "news_pipeline/data/normalized", NormalizedArticle)
-    article = normalized_store.load(item.normalized_id)
+    article = _normalized_article(root, item)
     title_parts = [item.draft_title or ""]
     description_parts = [item.draft_description or ""]
     if article:
@@ -370,8 +395,7 @@ def _duplicate_probe_text(root: Path, item: Any) -> tuple[str, str]:
 
 
 def _item_company_hits(root: Path, item: Any) -> set[str]:
-    normalized_store = JsonStore(root / "news_pipeline/data/normalized", NormalizedArticle)
-    article = normalized_store.load(item.normalized_id)
+    article = _normalized_article(root, item)
     article_text = ""
     if article:
         article_text = article.title or ""
@@ -386,8 +410,7 @@ def _item_company_hits(root: Path, item: Any) -> set[str]:
 
 
 def _item_topic_family_hits(root: Path, item: Any) -> set[str]:
-    normalized_store = JsonStore(root / "news_pipeline/data/normalized", NormalizedArticle)
-    article = normalized_store.load(item.normalized_id)
+    article = _normalized_article(root, item)
     text = " ".join(
         [
             item.draft_title or "",
@@ -412,8 +435,7 @@ def _headline_has_term(headline: str, term: str) -> bool:
 
 
 def _public_value_text(root: Path, item: Any) -> str:
-    normalized_store = JsonStore(root / "news_pipeline/data/normalized", NormalizedArticle)
-    article = normalized_store.load(item.normalized_id)
+    article = _normalized_article(root, item)
     parts = [
         item.draft_title or "",
         item.draft_description or "",
@@ -548,6 +570,9 @@ def _passes_basic_board_filter(root: Path, item: Any, max_source_age_hours: int,
         return False, "sponsored or paid content"
     if _has_known_unreadable_primary_source(item):
         return False, "known unreadable primary source"
+    source_quality_reason = _source_text_quality_rejection(root, item)
+    if source_quality_reason:
+        return False, source_quality_reason
     fresh, stale_reason = _source_is_fresh(root, item, max_source_age_hours)
     if not fresh:
         return False, stale_reason
@@ -614,7 +639,9 @@ def _select_headline_board(root: Path, items: list[Any], limit: int, max_source_
     def add(item: Any) -> bool:
         if item.queue_id in selected_ids:
             return False
-        _, board_reasons = score_map.get(item.queue_id, (round(float(item.editorial_priority), 3), []))
+        board_score, board_reasons = score_map.get(item.queue_id, (round(float(item.editorial_priority), 3), []))
+        if board_score < MIN_BOARD_FILL_SCORE:
+            return False
         if _has_board_hard_veto(board_reasons):
             return False
         headline = _normalized(_headline_text(root, item))
@@ -688,13 +715,13 @@ def add(item: Any) -> bool:
         "scienceSpaceBoardLimit": SCIENCE_SPACE_BOARD_LIMIT if science_space_pressure else None,
         "maxPerSource": MAX_PER_SOURCE,
         "minCategoryTargetScore": MIN_CATEGORY_TARGET_SCORE,
+        "minBoardFillScore": MIN_BOARD_FILL_SCORE,
     }
     return selected, skipped, {"scores": score_map, "diagnostics": diagnostics}
 
 
 def _article_payload(root: Path, item: Any) -> dict[str, Any]:
-    normalized_store = JsonStore(root / "news_pipeline/data/normalized", NormalizedArticle)
-    article = normalized_store.load(item.normalized_id)
+    article = _normalized_article(root, item)
     source_age_hours = None
     if article and article.published_at:
         source_age_hours = round((datetime.now(UTC) - article.published_at.astimezone(UTC)).total_seconds() / 3600, 2)
@@ -791,14 +818,15 @@ def _full_collect_retry_reason(
     if not stats:
         return None
     eligible_count = int(board_meta["diagnostics"].get("eligibleCount") or 0)
+    pack_count = len(packs)
     if stats.get("sources_collected", 0) == 0:
-        if packs and eligible_count >= MIN_HEALTHY_BOARD_ELIGIBLE:
+        if pack_count >= MIN_HEALTHY_BOARD_ELIGIBLE and eligible_count >= MIN_HEALTHY_BOARD_ELIGIBLE:
             return None
         return "cadence produced zero collected sources"
     if not packs and stats.get("skipped_cadence", 0) > 0:
         return "no board candidates after cadence-limited collect"
-    if eligible_count < MIN_HEALTHY_BOARD_ELIGIBLE and stats.get("skipped_cadence", 0) > 0:
-        return f"thin board ({eligible_count} eligible < {MIN_HEALTHY_BOARD_ELIGIBLE}) after cadence-limited collect"
+    if (pack_count < MIN_HEALTHY_BOARD_ELIGIBLE or eligible_count < MIN_HEALTHY_BOARD_ELIGIBLE) and stats.get("skipped_cadence", 0) > 0:
+        return f"thin board ({pack_count} selected, {eligible_count} eligible; target {MIN_HEALTHY_BOARD_ELIGIBLE}) after cadence-limited collect"
     return None
 
 
@@ -933,7 +961,7 @@ def prepare_one_command(
     cleanup: bool = typer.Option(True, "--cleanup/--no-cleanup", help="Run queue cleanup before preparing the editorial pack."),
     full_collect: bool = typer.Option(False, "--full-collect", help="Bypass source cadence during collect."),
     min_score: float = typer.Option(0.68, "--min-score", help="Strict publish threshold used for diagnostics."),
-    max_source_age_hours: int = typer.Option(24, "--max-source-age-hours", help="Maximum source age for headline board freshness and auto publish diagnostics."),
+    max_source_age_hours: int = typer.Option(DEFAULT_MAX_SOURCE_AGE_HOURS, "--max-source-age-hours", help="Maximum source age for headline board freshness and auto publish diagnostics."),
     limit: int = typer.Option(20, "--limit", help="How many headline candidates to return."),
     json_output: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),
 ) -> None:
@@ -983,7 +1011,7 @@ def board_prewarm_command(
     cleanup: bool = typer.Option(True, "--cleanup/--no-cleanup", help="Run queue cleanup before preparing the prewarmed board."),
     full_collect: bool = typer.Option(False, "--full-collect", help="Bypass source cadence during collect."),
     min_score: float = typer.Option(0.68, "--min-score", help="Strict publish threshold used for diagnostics."),
-    max_source_age_hours: int = typer.Option(24, "--max-source-age-hours", help="Maximum source age for headline board freshness and auto publish diagnostics."),
+    max_source_age_hours: int = typer.Option(DEFAULT_MAX_SOURCE_AGE_HOURS, "--max-source-age-hours", help="Maximum source age for headline board freshness and auto publish diagnostics."),
     limit: int = typer.Option(6, "--limit", help="How many headline candidates to return."),
     output: Path = typer.Option(Path(DEFAULT_PREWARM_BOARD_PATH), "--output", "-o", help="Prewarmed board artifact path."),
     max_age_minutes: int = typer.Option(15, "--max-age-minutes", min=1, help="Freshness window consumers should enforce."),

news_pipeline/news_pipeline/cli/commands/heartbeat_publish_one.py

@@ -24,7 +24,7 @@
 
 DEFAULT_MIN_SCORE = 0.68
 ASTERIA_POLISHED_MIN_SCORE = 0.68
-DEFAULT_MAX_SOURCE_AGE_HOURS = 24
+DEFAULT_MAX_SOURCE_AGE_HOURS = 30
 DEFAULT_MIN_INTERVAL_SECONDS = 3000
 STATE_PATH = Path("news_pipeline/data/state/heartbeat-publish-one.json")
 MAX_STEP_LOG_CHARS = 900

news_pipeline/tests/test_safety_gates.py

@@ -712,6 +712,65 @@ def test_prepare_board_excludes_sponsored_content_url(tmp_path: Path) -> None:
     assert skipped[0]["reason"] == "sponsored or paid content"
 
 
+def test_prepare_board_excludes_thin_france24_video_report_fallback(tmp_path: Path) -> None:
+    article = _article(
+        "france24-video-report",
+        url="https://www.france24.com/en/trump-says-will-cut-off-all-trade-with-spain-over-nato-contribution-row",
+    )
+    article.source_id = "france24-world"
+    article.source_name = "France 24 World"
+    article.title = "Trump says will cut off all trade with Spain over NATO contribution row"
+    article.summary = "US President Donald Trump threatened to halt trade with Spain over NATO spending. FRANCE 24's Monte Francis reports."
+    article.content_snippet = article.summary
+    item = _item("france24-video-report", normalized_id=article.id, url=str(article.canonical_url), priority=0.74)
+    item.draft_sources = [DraftSource(name="France 24 World", url=str(article.canonical_url))]
+    _save_runtime(tmp_path, item, article)
+
+    selected, skipped, _ = _select_headline_board(tmp_path, [item], limit=10, max_source_age_hours=72)
+
+    assert selected == []
+    assert skipped[0]["reason"] == "thin France 24 video/report fallback"
+
+
+def test_prepare_board_excludes_thin_politico_snippet(tmp_path: Path) -> None:
+    article = _article(
+        "politico-thin-snippet",
+        url="https://www.politico.eu/article/nato-ankara-summit-wasted-opportunity/",
+    )
+    article.source_id = "politico-eu"
+    article.source_name = "Politico Europe"
+    article.title = "NATO's Ankara summit was a wasted opportunity"
+    article.summary = "Europe could have used the meeting to show the U.S. how it will take the lead."
+    article.content_snippet = "The U.S. may have the strongest military in the world."
+    item = _item("politico-thin-snippet", normalized_id=article.id, url=str(article.canonical_url), priority=0.72)
+    item.draft_sources = [DraftSource(name="Politico Europe", url=str(article.canonical_url))]
+    _save_runtime(tmp_path, item, article)
+
+    selected, skipped, _ = _select_headline_board(tmp_path, [item], limit=10, max_source_age_hours=72)
+
+    assert selected == []
+    assert skipped[0]["reason"] == "thin Politico snippet (133 chars < 500)"
+
+
+def test_prepare_board_keeps_readable_cnbc_even_with_short_normalized_snippet(tmp_path: Path) -> None:
+    article = _article("cnbc-readable-short", url="https://www.cnbc.com/2026/07/09/meta-ai-coding-market.html")
+    article.source_id = "cnbc-tech"
+    article.source_name = "CNBC Technology"
+    article.title = "Meta jumps into AI coding market in effort to chase Anthropic and OpenAI"
+    article.summary = "Meta is upgrading its artificial intelligence coding model."
+    article.content_snippet = "Meta is rolling out a major update as it attempts to compete with OpenAI and Anthropic in critical areas of the AI coding market."
+    item = _item("cnbc-readable-short", normalized_id=article.id, url=str(article.canonical_url), priority=0.706)
+    item.draft_title = article.title
+    item.draft_description = article.content_snippet
+    item.draft_category = "Teknoloji"
+    item.draft_sources = [DraftSource(name="CNBC Technology", url=str(article.canonical_url))]
+    _save_runtime(tmp_path, item, article)
+
+    selected, _, _ = _select_headline_board(tmp_path, [item], limit=10, max_source_age_hours=72)
+
+    assert [item.queue_id for item in selected] == ["cnbc-readable-short"]
+
+
 def test_prepare_board_penalizes_low_public_value_oddity_below_board_floor(tmp_path: Path) -> None:
     article = _article("snake-flood", url="https://example.org/world/snakes-flood")
     article.source_id = "guardian-world"
@@ -733,6 +792,23 @@ def test_prepare_board_penalizes_low_public_value_oddity_below_board_floor(tmp_p
     assert "public_value_penalty:oddity_or_entertainment" in reasons
 
 
+def test_prepare_board_excludes_low_public_value_ceremonial_gift_story(tmp_path: Path) -> None:
+    article = _article("nato-gift", url="https://example.org/demo/nato-gift")
+    article.title = "Newly released footage shows revolver gifted by Erdogan to NATO leaders"
+    article.summary = "Images show a revolver and bullets given as a parting gift after the NATO summit."
+    item = _item("nato-gift", normalized_id=article.id, url=str(article.canonical_url), priority=0.712)
+    item.draft_title = article.title
+    item.draft_description = article.summary
+    item.draft_category = "Siyaset"
+    item.draft_sources = [DraftSource(name="Euronews World", url=str(article.canonical_url))]
+    _save_runtime(tmp_path, item, article)
+
+    selected, skipped, _ = _select_headline_board(tmp_path, [item], limit=10, max_source_age_hours=72)
+
+    assert selected == []
+    assert skipped[0]["reason"] == "blocked low-signal headline"
+
+
 def test_publish_one_retries_next_candidate_after_duplicate_publish_gate(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.chdir(tmp_path)
     first_article = _article("dup-first", url="https://example.org/demo/dup-first")
@@ -1353,6 +1429,25 @@ def test_prepare_one_full_collect_retry_still_runs_when_cadence_board_is_thin()
     assert reason == "cadence produced zero collected sources"
 
 
+def test_prepare_one_full_collect_retry_when_selected_board_is_thin_even_if_eligible_exists() -> None:
+    reason = heartbeat_prepare_one._full_collect_retry_reason(
+        collect=True,
+        full_collect=False,
+        steps=[
+            {
+                "name": "collect",
+                "ok": True,
+                "stdout": "sources_collected=4 items=4 skipped_cadence=14 failed=0",
+                "stderr": "",
+            }
+        ],
+        packs=[{"queueId": f"candidate-{index}"} for index in range(5)],
+        board_meta={"diagnostics": {"eligibleCount": 14}},
+    )
+
+    assert reason == "thin board (5 selected, 14 eligible; target 6) after cadence-limited collect"
+
+
 
 def test_headline_board_penalizes_recent_topic_family_saturation(tmp_path: Path) -> None:
     content = tmp_path / "src/content/equinoxHaber"
@@ -1566,6 +1661,45 @@ def test_headline_board_excludes_low_score_category_fill_even_behind_stronger_gl
     assert meta["diagnostics"]["minCategoryTargetScore"] == 0.68
 
 
+def test_headline_board_does_not_fill_with_very_low_board_score_repeat(tmp_path: Path) -> None:
+    content = tmp_path / "src/content/equinoxHaber"
+    content.mkdir(parents=True)
+    for index in range(2):
+        (content / f"recent-openai-{index}.md").write_text(
+            f"""---
+title: "OpenAI, yeni ChatGPT sürümünü duyurdu {index}"
+description: "OpenAI ve ChatGPT gündemi işlendi."
+pubDate: '2026-06-15T1{index}:00:00+03:00'
+category: "Teknoloji"
+tags: ["OpenAI", "ChatGPT"]
+sources:
+  - name: "Tech Source"
+    url: "https://example.org/recent-openai-{index}"
+---
+Gövde.
+""",
+            encoding="utf-8",
+        )
+    article = _article("openai-repeat", url="https://example.org/demo/openai-repeat")
+    article.title = "OpenAI rolls out GPT-5.6 after government greenlight"
+    article.summary = "OpenAI and ChatGPT remain central to the product announcement."
+    article.tags = ["OpenAI", "ChatGPT"]
+    item = _item("openai-repeat", normalized_id=article.id, url=str(article.canonical_url), priority=0.68)
+    item.draft_title = article.title
+    item.draft_description = article.summary
+    item.draft_category = "Teknoloji"
+    item.draft_tags = ["OpenAI", "ChatGPT"]
+    item.draft_sources = [DraftSource(name="The Verge", url=str(article.canonical_url))]
+    _save_runtime(tmp_path, item, article)
+
+    selected, _, meta = _select_headline_board(tmp_path, [item], limit=10, max_source_age_hours=72)
+
+    assert selected == []
+    score, _ = meta["scores"][item.queue_id]
+    assert score < heartbeat_prepare_one.MIN_BOARD_FILL_SCORE
+    assert meta["diagnostics"]["minBoardFillScore"] == 0.50
+
+
 def test_politics_is_not_treated_as_hot_category(tmp_path: Path) -> None:
     content = tmp_path / "src/content/equinoxHaber"
     content.mkdir(parents=True)