fix: preserve short release titles and cover source generation contracts

omni/feeds.py

@@ -30,7 +30,7 @@ def _item(src, kind, url, title, text="", author="", published="", extra=None):
     ex["tags"] = list(src.tags)
     t = _txt(title)
     body = _txt(text)
-    if sum(ch.isalpha() for ch in t) < 12 and body:
+    if body and (not t or getattr(src, "adapter", "").startswith("tg_")) and sum(ch.isalpha() for ch in t) < 12:
         cand = [ln for ln in body.splitlines() if sum(ch.isalpha() for ch in ln) >= 12]
         if cand:
             t = max(cand, key=len)[:140]

tests/test_ingest_quality.py

@@ -101,3 +101,18 @@ def test_fetch_bad_record_does_not_discard_valid_siblings(monkeypatch):
     bad.text = None
     monkeypatch.setitem(feeds.ADAPTERS, "test_adapter", lambda source, n: [bad, good])
     assert feeds.fetch(src) == [good]
+
+
+@pytest.mark.parametrize("title", ["GPT-5 launched", "Release 2.0", "v2.1.0"])
+def test_feed_keeps_short_release_titles(title):
+    src = S("release_test", "models", "atom", "https://example.invalid/feed")
+    row = feeds._item(src, "entry", "https://example.invalid/release", title,
+                      "Long description that must not replace the release title.")
+    assert row.title == title
+
+
+def test_telegram_still_uses_body_for_empty_heading():
+    src = S("tg_test", "security", "tg", "https://t.me/s/example", "tg_web")
+    body = "An advisory with a meaningful first line."
+    row = feeds._item(src, "post", "https://t.me/example/1", "...", body)
+    assert row.title == body

tests/test_panel_api.py

@@ -296,6 +296,63 @@ def test_fresh_generator_preserves_tag_tuples(tmp_path, tags):
     assert isinstance(generated[0].tags, tuple)
 
 
+def test_fresh_generator_dedupes_registry_and_batch(tmp_path, monkeypatch):
+    import runpy
+    from omni import sources
+
+    calls = []
+    existing = sources.S("Existing_Feed", "security", "rss", "https://example.invalid/known")
+
+    def registry(include_fresh=True):
+        calls.append(include_fresh)
+        return [existing]
+
+    monkeypatch.setattr(sources, "all_sources", registry)
+    probe = runpy.run_path(str(ROOT / "tools" / "probe_fresh.py"))
+    rows = [
+        {"name": "existing-feed", "url": "https://example.invalid/other"},
+        {"name": "url_collision", "url": "http://www.example.invalid/known/"},
+        {"name": "novel", "url": "https://example.invalid/new"},
+        {"name": "novel_mirror", "url": "http://www.example.invalid/new/"},
+        {"name": "NOVEL", "url": "https://example.invalid/newer"},
+    ]
+    path = tmp_path / "sources.py"
+    assert probe["emit"](rows, [], path) == 1
+    assert calls == [False]
+    assert [s.name for s in runpy.run_path(str(path))["fresh"](sources.S)] == ["novel"]
+
+
+def test_fresh_generator_preserves_dynamic_window_and_python_literals(tmp_path):
+    import runpy
+    from urllib.parse import parse_qs, urlsplit
+    from omni import sources
+
+    probe = runpy.run_path(str(ROOT / "tools" / "probe_fresh.py"))
+    row = {"name": "nvd_recent", "category": "security", "kind": "json", "adapter": "nvd",
+           "url": "https://example.invalid/old", "dynamic": "NVD_LAST_WEEK",
+           "extra": {"enabled": True, "optional": None}}
+    candidate, marker = probe["candidate_source"](row)
+    assert marker == candidate.extra["_dynamic"] == "NVD_LAST_WEEK"
+    assert "lastModStartDate=" in candidate.url
+    path = tmp_path / "sources.py"
+    assert probe["emit"]([dict(row, extra=candidate.extra)], [], path, dedupe=False) == 1
+    generated = runpy.run_path(str(path))["fresh"](sources.S)[0]
+    query = parse_qs(urlsplit(generated.url).query)
+    assert query["lastModStartDate"] and query["lastModEndDate"]
+    assert generated.extra == {"enabled": True, "optional": None}
+
+
+def test_fresh_generator_rejects_unknown_dynamic_marker(tmp_path):
+    import runpy
+
+    probe = runpy.run_path(str(ROOT / "tools" / "probe_fresh.py"))
+    row = {"name": "future", "url": "https://example.invalid/feed", "dynamic": "UNKNOWN"}
+    with pytest.raises(ValueError, match="dynamic"):
+        probe["candidate_source"](row)
+    with pytest.raises(ValueError, match="dynamic"):
+        probe["emit"]([row], [], tmp_path / "bad.py", dedupe=False)
+
+
 def test_timeline_local_day_and_rolling_24_hours(monkeypatch):
     import datetime
     from omni import panel_api, store