Expand multi-signal coverage to RSS and Reddit plugins

Author: webstackdev

core/tasks.py

@@ -41,6 +41,7 @@
 __all__ = [
     "recompute_source_quality",
     "recompute_authority_scores",
+    "run_all_source_quality_recomputations",
     "run_all_authority_recomputations",
     "run_relevance_scoring_skill",
     "run_summarization_skill",
@@ -94,6 +95,19 @@ def run_all_authority_recomputations():
     return len(project_ids)
 
 
+@shared_task(name="core.tasks.run_all_source_quality_recomputations")
+def run_all_source_quality_recomputations() -> int:
+    """Queue source-quality recomputation for every project."""
+
+    project_ids = list(Project.objects.values_list("id", flat=True))
+    for project_id in project_ids:
+        if settings.CELERY_TASK_ALWAYS_EAGER:
+            recompute_source_quality(project_id)
+        else:
+            _enqueue_task(recompute_source_quality, project_id)
+    return len(project_ids)
+
+
 @shared_task(name="core.tasks.process_content")
 def process_content(content_id: int):
     """Run the main AI pipeline for a stored content item."""
@@ -616,6 +630,8 @@ def _engagement_total_for_content(
             + _metadata_number(metadata.get("reply_count"))
             + _metadata_number(metadata.get("repost_count"))
         )
+    if content.source_plugin == "reddit":
+        return _metadata_number(metadata.get("score"))
     if content.source_plugin == "mastodon":
         return (
             _metadata_number(metadata.get("favorite_count"))

core/tests/test_entrypoints.py

@@ -42,3 +42,16 @@ def test_wsgi_module_sets_default_settings_and_builds_application(mocker):
 
 def test_celery_app_redirects_worker_stdout_at_info_level():
     assert app.conf.worker_redirect_stdouts_level == "INFO"
+
+
+def test_celery_app_schedules_source_quality_before_authority_recompute():
+    beat_schedule = app.conf.beat_schedule
+
+    assert (
+        beat_schedule["run-all-source-quality-recomputations-nightly"]["task"]
+        == "core.tasks.run_all_source_quality_recomputations"
+    )
+    assert (
+        beat_schedule["run-all-authority-recomputations-nightly"]["task"]
+        == "core.tasks.run_all_authority_recomputations"
+    )

core/tests/test_tasks.py

@@ -9,6 +9,7 @@
     queue_content_skill,
     recompute_authority_scores,
     run_all_authority_recomputations,
+    run_all_source_quality_recomputations,
     run_relevance_scoring_skill,
     run_summarization_skill,
 )
@@ -86,6 +87,43 @@ def test_run_all_authority_recomputations_executes_inline_when_eager(
     delay_mock.assert_not_called()
 
 
+def test_run_all_source_quality_recomputations_enqueues_all_projects(
+    source_plugin_context, mocker
+):
+    delay_mock = mocker.patch("core.tasks.recompute_source_quality.delay")
+    other_project = Project.objects.create(
+        name="Source Quality Project",
+        topic_description="Signals",
+    )
+
+    enqueued_count = run_all_source_quality_recomputations()
+
+    assert enqueued_count == 2
+    delay_mock.assert_any_call(source_plugin_context.project.id)
+    delay_mock.assert_any_call(_require_pk(other_project))
+    assert delay_mock.call_count == 2
+
+
+def test_run_all_source_quality_recomputations_executes_inline_when_eager(
+    source_plugin_context, settings, mocker
+):
+    settings.CELERY_TASK_ALWAYS_EAGER = True
+    recompute_mock = mocker.patch("core.tasks.recompute_source_quality")
+    delay_mock = mocker.patch("core.tasks.recompute_source_quality.delay")
+    other_project = Project.objects.create(
+        name="Inline Source Quality Project",
+        topic_description="Signals",
+    )
+
+    enqueued_count = run_all_source_quality_recomputations()
+
+    assert enqueued_count == 2
+    recompute_mock.assert_any_call(source_plugin_context.project.id)
+    recompute_mock.assert_any_call(_require_pk(other_project))
+    assert recompute_mock.call_count == 2
+    delay_mock.assert_not_called()
+
+
 def test_recompute_authority_scores_updates_entities_and_creates_snapshots(
     source_plugin_context, mocker
 ):
@@ -198,6 +236,62 @@ def test_recompute_authority_scores_updates_entities_and_creates_snapshots(
     assert primary_snapshot.weights_at_compute["engagement"] == pytest.approx(0.15)
 
 
+def test_recompute_authority_scores_uses_bluesky_and_reddit_engagement_metadata(
+    source_plugin_context, mocker
+):
+    mocker.patch("content.signals.queue_topic_centroid_recompute")
+    project = source_plugin_context.project
+    ProjectConfig.objects.create(project=project)
+    entity = source_plugin_context.entity
+    bluesky_content = Content.objects.create(
+        project=project,
+        entity=entity,
+        url="https://bsky.app/profile/example/post/abc",
+        canonical_url="https://bsky.app/profile/example/post/abc",
+        title="Bluesky authority content",
+        author="example.bsky.social",
+        source_plugin=SourcePluginName.BLUESKY,
+        published_date="2026-04-28T12:00:00Z",
+        content_text="Bluesky authority content.",
+        source_metadata={"like_count": 4, "reply_count": 1, "repost_count": 2},
+    )
+    reddit_content = Content.objects.create(
+        project=project,
+        entity=entity,
+        url="https://www.reddit.com/r/python/comments/abc123/test",
+        canonical_url="https://www.reddit.com/r/python/comments/abc123/test",
+        title="Reddit authority content",
+        author="redditor",
+        source_plugin=SourcePluginName.REDDIT,
+        published_date="2026-04-28T13:00:00Z",
+        content_text="Reddit authority content.",
+        source_metadata={"score": 10, "comment_count": 3},
+    )
+    EntityMention.objects.create(
+        project=project,
+        content=bluesky_content,
+        entity=entity,
+        role=EntityMentionRole.AUTHOR,
+        sentiment="positive",
+        span="Example",
+        confidence=0.9,
+    )
+    EntityMention.objects.create(
+        project=project,
+        content=reddit_content,
+        entity=entity,
+        role=EntityMentionRole.AUTHOR,
+        sentiment="neutral",
+        span="Example",
+        confidence=0.9,
+    )
+
+    recompute_authority_scores(project.id)
+
+    snapshot = EntityAuthoritySnapshot.objects.get(entity=entity)
+    assert snapshot.engagement_component == pytest.approx(1.0)
+
+
 def test_queue_content_skill_enqueues_relevance_task(source_plugin_context, mocker):
     content = Content.objects.create(
         project=source_plugin_context.project,

frontend/tsconfig.tsbuildinfo

@@ -152,6 +152,7 @@ def _build_content_item(self, post, published_date: datetime) -> ContentItem:
                 "author_did": self._nested_value(post, "author", "did") or "",
                 "author_handle": author_handle,
                 "embedded_url": external_url or "",
+                "like_count": getattr(post, "like_count", 0) or 0,
                 "post_uri": getattr(post, "uri", ""),
                 "reply_count": getattr(post, "reply_count", 0) or 0,
                 "repost_count": getattr(post, "repost_count", 0) or 0,

ingestion/plugins/bluesky.py

@@ -52,6 +52,14 @@ def fetch_new_content(self, since: datetime | None) -> list[ContentItem]:
                     published_date=published_date,
                     content_text=(submission.selftext or submission.title).strip(),
                     source_plugin=SourcePluginName.REDDIT,
+                    source_metadata={
+                        "comment_count": getattr(submission, "num_comments", 0) or 0,
+                        "permalink": getattr(submission, "permalink", "") or "",
+                        "score": getattr(submission, "score", 0) or 0,
+                        "submission_id": submission.id,
+                        "subreddit": self.source_config.config["subreddit"],
+                        "upvote_ratio": getattr(submission, "upvote_ratio", 0.0) or 0.0,
+                    },
                 )
             )
         return items

ingestion/plugins/reddit.py

@@ -90,6 +90,7 @@ def test_bluesky_fetch_new_content_prefers_embedded_links_and_filters_replies(
         indexed_at=old_iso,
         author=SimpleNamespace(did="did:plc:alice", handle="Alice.BSky.social"),
         record=SimpleNamespace(text="Old post", created_at=old_iso),
+        like_count=0,
         reply_count=0,
         repost_count=0,
         embed=None,
@@ -99,6 +100,7 @@ def test_bluesky_fetch_new_content_prefers_embedded_links_and_filters_replies(
         indexed_at=iso_now,
         author=SimpleNamespace(did="did:plc:alice", handle="Alice.BSky.social"),
         record=SimpleNamespace(text="A reply", created_at=iso_now),
+        like_count=1,
         reply_count=1,
         repost_count=0,
         embed=None,
@@ -114,6 +116,7 @@ def test_bluesky_fetch_new_content_prefers_embedded_links_and_filters_replies(
                 title="  Linked article  ",
             )
         ),
+        like_count=5,
         reply_count=2,
         repost_count=3,
     )
@@ -122,6 +125,7 @@ def test_bluesky_fetch_new_content_prefers_embedded_links_and_filters_replies(
         indexed_at=iso_now,
         author=SimpleNamespace(did="did:plc:alice", handle="Alice.BSky.social"),
         record=SimpleNamespace(text="Duplicate", created_at=iso_now),
+        like_count=0,
         reply_count=0,
         repost_count=0,
         embed=None,
@@ -151,6 +155,7 @@ def test_bluesky_fetch_new_content_prefers_embedded_links_and_filters_replies(
         "author_did": "did:plc:alice",
         "author_handle": "alice.bsky.social",
         "embedded_url": "https://example.com/article",
+        "like_count": 5,
         "post_uri": "at://did:plc:alice/app.bsky.feed.post/fresh",
         "reply_count": 2,
         "repost_count": 3,

ingestion/tests/test_bluesky.py

@@ -63,6 +63,9 @@ def test_reddit_fetch_new_content_deduplicates_and_filters_by_since(
         title="  Duplicate title  ",
         selftext="",
         author="redditor",
+        num_comments=7,
+        score=12,
+        upvote_ratio=0.91,
         created_utc=now.timestamp(),
     )
     duplicate_submission = SimpleNamespace(
@@ -72,6 +75,9 @@ def test_reddit_fetch_new_content_deduplicates_and_filters_by_since(
         title="Duplicate title",
         selftext="Ignored duplicate",
         author="redditor",
+        num_comments=7,
+        score=12,
+        upvote_ratio=0.91,
         created_utc=now.timestamp(),
     )
     subreddit = SimpleNamespace(
@@ -89,6 +95,14 @@ def test_reddit_fetch_new_content_deduplicates_and_filters_by_since(
     assert items[0].author == "redditor"
     assert items[0].content_text == "Duplicate title"
     assert items[0].source_plugin == SourcePluginName.REDDIT
+    assert items[0].source_metadata == {
+        "comment_count": 7,
+        "permalink": "/r/python/comments/dup-1/test",
+        "score": 12,
+        "submission_id": duplicate_id,
+        "subreddit": "python",
+        "upvote_ratio": 0.91,
+    }
 
 
 def test_reddit_health_check_returns_true(reddit_context, mocker):

ingestion/tests/test_reddit.py

@@ -189,6 +189,9 @@ def test_run_ingestion_creates_content_from_reddit_posts(source_plugin_context,
         title="Reddit Post",
         selftext="Post body",
         author="redditor",
+        num_comments=6,
+        score=14,
+        upvote_ratio=0.88,
         created_utc=datetime(2026, 4, 20, 12, 0, tzinfo=timezone.utc).timestamp(),
     )
     subreddit = SimpleNamespace(
@@ -205,6 +208,8 @@ def test_run_ingestion_creates_content_from_reddit_posts(source_plugin_context,
     process_content_delay_mock.assert_called_once_with(_require_pk(content))
     assert content.source_plugin == SourcePluginName.REDDIT
     assert content.entity is None
+    assert content.source_metadata["score"] == 14
+    assert content.source_metadata["comment_count"] == 6
 
 
 def test_ingest_source_config_deduplicates_bluesky_posts_by_post_uri(

ingestion/tests/test_tasks.py

@@ -23,6 +23,10 @@
         "task": "core.tasks.refresh_linkedin_tokens",
         "schedule": 60 * 60,
     },
+    "run-all-source-quality-recomputations-nightly": {
+        "task": "core.tasks.run_all_source_quality_recomputations",
+        "schedule": crontab(hour=1, minute=45),
+    },
     "run-all-authority-recomputations-nightly": {
         "task": "core.tasks.run_all_authority_recomputations",
         "schedule": crontab(hour=2, minute=0),